Multi-Agent Supervisor

The Supervisor orchestrates multiple agent graphs by routing work between them based on a routing function.

Basic Usage

from flowgentra_ai import Supervisor, SharedState

# 1. Build agent graphs (each is a compiled StateGraph)
# ... (see Graph Workflows guide)

# 2. Define a routing function
def router(state):
    """Returns the name of the next agent, or 'FINISH' to stop."""
    task = state.get_string("task") or ""
    if "research" in task:
        return "researcher"
    if "write" in task:
        return "writer"
    return "FINISH"

# 3. Create the supervisor
sup = Supervisor(router)
sup.add_agent("researcher", research_graph)
sup.add_agent("writer", writer_graph)

# 4. Run
result = sup.run(SharedState({"task": "research AI trends"}))
print(result.to_dict())

Configuration

# Set maximum routing rounds (default: 10)
sup.max_rounds(5)

# Change the finish marker (default: "FINISH")
sup.finish_marker("DONE")

# List registered agents
print(sup.agent_names())  # ["researcher", "writer"]

How It Works

1. The supervisor calls the router function with the current state
2. The router returns the name of an agent to dispatch to
3. The selected agent graph runs via invoke() with the current state
4. The updated state is passed back to the router
5. This repeats until the router returns the finish marker or max rounds is hit

Example: Research + Write Pipeline

from flowgentra_ai import (
    StateGraphBuilder, SharedState, END,
    Supervisor, LLMConfig, LLMClient, Message,
)

client = LLMClient.from_config(LLMConfig("openai", "gpt-4", api_key="sk-..."))

# Research agent
def research(state):
    topic = state["topic"]
    resp = client.chat([Message.user(f"Research key facts about: {topic}")])
    state["research"] = resp.content
    return state

research_builder = StateGraphBuilder()
research_builder.add_node("research", research)
research_builder.set_entry_point("research")
research_builder.add_edge("research", END)
research_graph = research_builder.compile()

# Writer agent
def write(state):
    research = state["research"]
    resp = client.chat([Message.user(f"Write a summary based on: {research}")])
    state["article"] = resp.content
    return state

writer_builder = StateGraphBuilder()
writer_builder.add_node("write", write)
writer_builder.set_entry_point("write")
writer_builder.add_edge("write", END)
writer_graph = writer_builder.compile()

# Supervisor
def router(state):
    if not state.get_string("research"):
        return "researcher"
    if not state.get_string("article"):
        return "writer"
    return "FINISH"

sup = Supervisor(router)
sup.add_agent("researcher", research_graph)
sup.add_agent("writer", writer_graph)
sup.max_rounds(5)

result = sup.run(SharedState({"topic": "Rust programming"}))
print(result["article"])