Orchestrator¶
The OrchestratorAgent lets you build a router that automatically delegates tasks to specialist agents. Each specialist is registered as a tool — the orchestrator's LLM reads their descriptions and calls whichever fits the task.
How it works¶
User prompt
│
▼
Orchestrator LLM
│ reads agent descriptions, picks the right one
▼
agent.run(task) ←── specialist executes with its own model + tools
│
▼
result fed back to Orchestrator LLM
│
▼
Final answer to user
If a task spans multiple specialists, the orchestrator calls them in sequence automatically — no extra configuration needed.
Installation path¶
Basic example¶
from exagent import Agent, tool
from exagent.multi_agent.orchestrator import OrchestratorAgent
# --- Specialist 1: Calculator ---
@tool
def calculate(expression: str) -> str:
"""Evaluate an arithmetic expression."""
allowed = set("0123456789+-*/(). ")
if not all(c in allowed for c in expression):
return "Error: disallowed characters."
return str(eval(expression, {"__builtins__": {}}))
class CalculatorAgent(Agent):
def __init__(self):
self.system_description = "You are a calculator assistant."
self.set_model("openai", "gpt-4.1-mini")
self.add_tool(calculate)
super().__init__()
# --- Specialist 2: Text Processor ---
@tool
def word_count(text: str) -> str:
"""Count the number of words in a text."""
return f"{len(text.split())} words"
class TextAgent(Agent):
def __init__(self):
self.system_description = "You are a text processing assistant."
self.set_model("openai", "gpt-4.1-mini")
self.add_tool(word_count)
super().__init__()
# --- Orchestrator ---
class MyOrchestrator(OrchestratorAgent):
def __init__(self):
self.set_model("openai", "gpt-4.1-mini")
self.add_agent(
CalculatorAgent(),
name="calculator",
description="Handles arithmetic and numeric calculations.",
)
self.add_agent(
TextAgent(),
name="text_processor",
description="Handles word counts, character counts, and text operations.",
)
super().__init__()
orch = MyOrchestrator()
print(orch.run("What is 128 * 64?"))
# → routed to calculator
print(orch.run("How many words are in 'the quick brown fox'?"))
# → routed to text_processor
add_agent()¶
orchestrator.add_agent(
agent, # Agent — fully configured with model + tools
name, # str — tool name the LLM uses to call this agent
description=None, # str — when to use this agent (falls back to agent.system_description)
)
Write clear descriptions
The orchestrator's LLM reads description to choose between agents. Be specific about what each agent handles — and what it does not handle.
Too vague: "Handles data tasks"
Better: "Handles arithmetic, percentages, and numeric expression evaluation. Does not handle text formatting."
Multi-step routing¶
When a task requires two specialists, the orchestrator calls them in sequence within a single run() or stream() call:
Orchestrator → calls calculator("840 * 0.15") → "126.0"
→ calls text_processor("126.0") → "4 characters"
→ replies: "15% of 840 is 126.0, which has 4 characters."
Parallel vs serial agent calls¶
By default the orchestrator runs in serial mode — the model calls one agent at a time and waits for the result before deciding the next step. This is the safe default because orchestrator tasks almost always have data dependencies (agent B needs agent A's output).
# Default — serial, no configuration needed
class MyOrchestrator(OrchestratorAgent):
def __init__(self):
self.set_model("openai", "gpt-4.1-mini")
...
super().__init__()
Without serial mode, the model may dispatch multiple agents in one shot and guess the inputs for later steps — leading to wasted calls and self-correction turns:
# What can go wrong in parallel mode for a dependent task:
User: "Calculate 455 * 233 and count the characters in the result."
→ calculator('455 * 233') # correct
→ text_processor('count 455*233') # wrong — guessed the input before calculator returned
→ text_processor('106015') # self-correction turn needed
In serial mode the same task resolves cleanly in two turns:
Enabling parallel calls¶
Set parallel = True as a class attribute to allow the model to call multiple agents in a single turn. This is faster for tasks where the agents are genuinely independent.
class MyOrchestrator(OrchestratorAgent):
parallel = True # opt in to parallel calls
def __init__(self):
self.set_model("openai", "gpt-4.1-mini")
self.add_agent(WeatherAgent(), name="weather", description="...")
self.add_agent(NewsAgent(), name="news", description="...")
super().__init__()
# Good use of parallel: both calls are independent
orch.run("What is the weather in Tokyo and what are today's top headlines?")
# → weather('Tokyo') and news() called at the same time ✓
OpenAI only
parallel_tool_calls is an OpenAI Responses API parameter. When using Anthropic, the parallel attribute has no effect — serial vs parallel behaviour is controlled by the model itself.
Each specialist keeps its own model¶
The orchestrator does not control which model the specialists use. Each agent runs with its own already-configured model. This means you can mix providers:
class MyOrchestrator(OrchestratorAgent):
def __init__(self):
self.set_model("openai", "gpt-4.1-mini") # orchestrator model
self.add_agent(
ResearchAgent(), # uses claude-opus-4-6 internally
name="researcher",
description="...",
)
self.add_agent(
SummaryAgent(), # uses gpt-4.1-mini internally
name="summariser",
description="...",
)
super().__init__()
Streaming¶
stream() works exactly like on a regular agent. tool_call and tool_result events show which specialist was chosen and what it returned:
for event in orch.stream("What is 128 * 64?"):
if event["type"] == "text_delta":
print(event["text"], end="", flush=True)
elif event["type"] == "tool_call":
print(f"\n[delegate] → {event['tool_call'].name}")
elif event["type"] == "tool_result":
print(f"[result] {event['name']} → {event['content']}")
elif event["type"] == "done":
print()
Interactive shell¶
Pass any orchestrator to shell() and chat with it interactively:
The shell automatically detects orchestrators and uses [delegate] / [result] labels instead of [tool_call] / [tool_result], so the routing is clearly visible.
Communication is one-way¶
Sub-agents delegate to specialists only — specialists cannot call back to the orchestrator. If a task needs two agents, the orchestrator calls them in sequence. There is no peer-to-peer communication between specialists.