Coordinator Agent
The coordinator agent is the brain of the multi-agent system. It plans the work, assigns tasks to the right agents, monitors progress, and synthesises the final result.
8 min•By Priygop Team•Updated 2026
Coordinator Responsibilities
- Decompose the goal: break it into subtasks that map to specific agent capabilities
- Assign tasks: send each subtask to the most appropriate specialist agent
- Monitor progress: track which agents have completed their tasks
- Handle failures: if an agent fails, retry or reassign the task
- Synthesise results: combine outputs from multiple agents into a coherent final result
- Manage human escalation: if the goal cannot be completed, escalate with a clear status report
Coordinator Implementation
Coordinator Implementation
# Coordinator agent that manages a team of specialist agents
class CoordinatorAgent:
def __init__(self, agents: dict):
"""
agents: dict of agent_name -> agent callable
Each agent is a function(task: str, context: dict) -> dict
"""
self.agents = agents
def decompose_goal(self, goal: str) -> list:
"""
In a real system, a language model decomposes the goal.
Returns a list of task assignments.
"""
return [
{"task_id": "T1", "agent": "researcher", "instruction": f"Research: {goal}",
"context": {}, "depends_on": []},
{"task_id": "T2", "agent": "analyser", "instruction": "Analyse research",
"context": {"source": "T1"}, "depends_on": ["T1"]},
{"task_id": "T3", "agent": "writer", "instruction": "Write summary report",
"context": {"source": "T2"}, "depends_on": ["T2"]},
{"task_id": "T4", "agent": "reviewer", "instruction": "Review report quality",
"context": {"source": "T3"}, "depends_on": ["T3"]},
]
def run(self, goal: str) -> dict:
task_list = self.decompose_goal(goal)
results = {}
print(f"Coordinator: {len(task_list)} tasks for goal: '{goal}'")
for task in task_list:
# Wait for dependencies
for dep in task["depends_on"]:
if dep not in results:
return {"status": "error", "error": f"Missing dependency: {dep}"}
# Enrich context with dependency results
context = task["context"].copy()
for dep in task["depends_on"]:
context[f"dep_{dep}"] = results[dep]
agent_name = task["agent"]
if agent_name not in self.agents:
return {"status": "error", "error": f"Agent '{agent_name}' not found"}
print(f" Assigning {task['task_id']} to {agent_name}")
result = self.agents[agent_name](task["instruction"], context)
results[task["task_id"]] = result
return {"status": "complete", "final_output": results.get("T4"), "steps": len(task_list)}
# Mock specialist agents
agents = {
"researcher": lambda task, ctx: {"summary": "Researched: " + task, "quality": 9},
"analyser": lambda task, ctx: {"analysis": "Analysed data", "quality": 8},
"writer": lambda task, ctx: {"report": "Written report", "quality": 9},
"reviewer": lambda task, ctx: {"approved": True, "score": 9, "notes": "Excellent"},
}
coordinator = CoordinatorAgent(agents)
result = coordinator.run("Top AI frameworks for 2024")
print(f"\nFinal: {result['status']}, Steps: {result['steps']}")Key Takeaways
- The coordinator agent is the brain of the multi-agent system.
- Decompose the goal: break it into subtasks that map to specific agent capabilities
- Assign tasks: send each subtask to the most appropriate specialist agent
- Monitor progress: track which agents have completed their tasks