Ai Coordinating Agents

Build Multi-Agent Systems

Guide the user through building multiple AI agents that collaborate — a supervisor delegates tasks, specialists handle their domains, and results flow back. Uses DSPy for each agent's reasoning and LangGraph for orchestration, handoff, and parallel execution.

Step 1: Identify the agents

Ask the user: 1. What's the overall task? (research a topic, handle support, create content, analyze data?) 2. What specialist roles do you need? (researcher, writer, reviewer, analyst, etc.) 3. How do agents hand off work? (supervisor routes, chain passes forward, parallel fan-out?) 4. Do any agents need tools? (search, database, APIs, code execution?)

Common multi-agent patterns

Step 2: Build each agent as a DSPy module

Each agent gets its own signature, reasoning strategy, and (optionally) tools.

import dspy

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

Simple agent — just a DSPy module

import dspy

class ResearchSummary(dspy.Signature):
    """Research the topic and provide a detailed summary with key findings."""
    topic: str = dspy.InputField()
    sources: list[str] = dspy.InputField(desc="Search results or documents to analyze")
    summary: str = dspy.OutputField(desc="Detailed research summary")
    key_findings: list[str] = dspy.OutputField(desc="Top 3-5 key findings")

class ResearchAgent(dspy.Module):
    def __init__(self, retriever):
        self.retriever = retriever
        self.analyze = dspy.ChainOfThought(ResearchSummary)

    def forward(self, topic):
        sources = self.retriever(topic).passages
        return self.analyze(topic=topic, sources=sources)

Agent with tools — use ReAct

def search_web(query: str) -> str:
    """Search the web for current information."""
    # your search implementation
    return results

def query_database(sql: str) -> str:
    """Query the analytics database."""
    # your database implementation
    return results

class DataAnalyst(dspy.Module):
    def __init__(self):
        self.agent = dspy.ReAct(
            "question, context -> analysis, recommendation",
            tools=[search_web, query_database],
            max_iters=5,
        )

    def forward(self, question, context=""):
        return self.agent(question=question, context=context)

Agent with LangChain tools

Convert pre-built LangChain tools for use in DSPy agents:

from langchain_community.tools import DuckDuckGoSearchRun

search_tool = dspy.Tool.from_langchain(DuckDuckGoSearchRun())

class WebResearcher(dspy.Module):
    def __init__(self):
        self.agent = dspy.ReAct(
            "question -> findings",
            tools=[search_tool],
            max_iters=5,
        )

    def forward(self, question):
        return self.agent(question=question)

Step 3: Add a supervisor (LangGraph)

The supervisor decides which agent to call next based on the current state.

Define the shared state

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator

class TeamState(TypedDict):
    task: str                                     # the overall task
    messages: Annotated[list[dict], operator.add]  # communication log
    current_agent: str                             # who's working now
    results: dict                                  # collected results from agents
    status: str                                    # "in_progress", "done", "needs_review"

Build the supervisor

class RouteTask(dspy.Signature):
    """Decide which specialist agent should handle the next step."""
    task: str = dspy.InputField(desc="The overall task")
    completed_work: str = dspy.InputField(desc="Work completed so far")
    available_agents: list[str] = dspy.InputField()
    next_agent: str = dspy.OutputField(desc="Which agent to call next")
    sub_task: str = dspy.OutputField(desc="Specific instruction for that agent")
    is_complete: bool = dspy.OutputField(desc="Whether the overall task is done")

supervisor_module = dspy.ChainOfThought(RouteTask)

def supervisor(state: TeamState) -> dict:
    completed = "\n".join(
        f"{k}: {v}" for k, v in state["results"].items()
    )
    result = supervisor_module(
        task=state["task"],
        completed_work=completed or "Nothing yet",
        available_agents=["researcher", "writer", "reviewer"],
    )

    if result.is_complete:
        return {"status": "done", "current_agent": "none"}

    return {
        "current_agent": result.next_agent,
        "messages": [{"role": "supervisor", "content": f"@{result.next_agent}: {result.sub_task}"}],
    }

Wire up the agents as graph nodes

researcher = ResearchAgent(retriever=my_retriever)
writer_module = dspy.ChainOfThought(WriteContent)
reviewer_module = dspy.ChainOfThought(ReviewContent)

def researcher_node(state: TeamState) -> dict:
    task_msg = state["messages"][-1]["content"]
    result = researcher(topic=task_msg)
    return {
        "results": {**state["results"], "research": result.summary},
        "messages": [{"role": "researcher", "content": result.summary}],
    }

def writer_node(state: TeamState) -> dict:
    result = writer_module(
        task=state["task"],
        research=state["results"].get("research", ""),
    )
    return {
        "results": {**state["results"], "draft": result.output},
        "messages": [{"role": "writer", "content": result.output}],
    }

def reviewer_node(state: TeamState) -> dict:
    result = reviewer_module(
        draft=state["results"].get("draft", ""),
        task=state["task"],
    )
    return {
        "results": {**state["results"], "review": result.feedback},
        "messages": [{"role": "reviewer", "content": result.feedback}],
    }

Build the graph

graph = StateGraph(TeamState)

# Add nodes
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher_node)
graph.add_node("writer", writer_node)
graph.add_node("reviewer", reviewer_node)

# Supervisor decides who goes next
graph.add_edge(START, "supervisor")

def route_to_agent(state: TeamState) -> str:
    if state["status"] == "done":
        return "done"
    return state["current_agent"]

graph.add_conditional_edges(
    "supervisor",
    route_to_agent,
    {
        "researcher": "researcher",
        "writer": "writer",
        "reviewer": "reviewer",
        "done": END,
    },
)

# All agents report back to supervisor
graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", "supervisor")
graph.add_edge("reviewer", "supervisor")

app = graph.compile()

Run it

result = app.invoke({
    "task": "Write a blog post about the benefits of remote work",
    "messages": [],
    "current_agent": "",
    "results": {},
    "status": "in_progress",
})
# Supervisor routes: researcher → writer → reviewer → done
print(result["results"]["draft"])

Step 4: Agent handoff pattern

When one agent passes work directly to another (no supervisor).

Shared context via state

class HandoffState(TypedDict):
    task: str
    context: Annotated[list[str], operator.add]  # accumulated context
    output: str

def agent_a(state: HandoffState) -> dict:
    result = module_a(task=state["task"])
    return {"context": [f"Agent A found: {result.output}"]}

def agent_b(state: HandoffState) -> dict:
    full_context = "\n".join(state["context"])
    result = module_b(task=state["task"], context=full_context)
    return {"context": [f"Agent B added: {result.output}"]}

def agent_c(state: HandoffState) -> dict:
    full_context = "\n".join(state["context"])
    result = module_c(task=state["task"], context=full_context)
    return {"output": result.output}

graph = StateGraph(HandoffState)
graph.add_node("a", agent_a)
graph.add_node("b", agent_b)
graph.add_node("c", agent_c)
graph.add_edge(START, "a")
graph.add_edge("a", "b")
graph.add_edge("b", "c")
graph.add_edge("c", END)

Conditional handoff

Route to different specialists based on intermediate results:

def route_after_classify(state) -> str:
    if state["category"] == "billing":
        return "billing_specialist"
    elif state["category"] == "technical":
        return "tech_specialist"
    return "general_agent"

graph.add_conditional_edges("classifier", route_after_classify, {
    "billing_specialist": "billing",
    "tech_specialist": "tech",
    "general_agent": "general",
})

Step 5: Parallel agents

Fan out to multiple agents simultaneously and merge results.

from langgraph.constants import Send

class ParallelState(TypedDict):
    task: str
    subtasks: list[str]
    results: Annotated[list[dict], operator.add]
    final_output: str

def split_task(state: ParallelState) -> list:
    """Fan out subtasks to worker agents."""
    return [Send("worker", {"task": state["task"], "subtask": st}) for st in state["subtasks"]]

def worker(state: dict) -> dict:
    """Each worker handles one subtask."""
    worker_module = dspy.ChainOfThought("task, subtask -> result")
    result = worker_module(task=state["task"], subtask=state["subtask"])
    return {"results": [{"subtask": state["subtask"], "result": result.result}]}

def merge_results(state: ParallelState) -> dict:
    """Combine all worker results into a final output."""
    merger = dspy.ChainOfThought("task, partial_results -> final_output")
    partial = "\n".join(f"- {r['subtask']}: {r['result']}" for r in state["results"])
    result = merger(task=state["task"], partial_results=partial)
    return {"final_output": result.final_output}

graph = StateGraph(ParallelState)
graph.add_node("worker", worker)
graph.add_node("merge", merge_results)
graph.add_conditional_edges(START, split_task)
graph.add_edge("worker", "merge")
graph.add_edge("merge", END)

Step 6: Human-in-the-loop

Pause before agents take critical actions.

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

# Interrupt before any agent that takes external actions
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute_action", "send_email", "update_database"],
)

config = {"configurable": {"thread_id": "task-001"}}

# Run until interrupt
result = app.invoke(input_state, config)
# -> Pauses before "execute_action" node

# Human reviews the proposed action in result state
print(result["proposed_action"])

# If approved, resume from checkpoint
result = app.invoke(None, config)

Step 7: Optimize the team

Per-agent metrics

Optimize each agent's prompts independently first:

def researcher_metric(example, prediction, trace=None):
    """Are the research findings relevant and complete?"""
    judge = dspy.Predict(JudgeResearch)
    return judge(topic=example.topic, findings=prediction.summary).is_good

optimizer = dspy.MIPROv2(metric=researcher_metric, auto="light")
optimized_researcher = optimizer.compile(researcher, trainset=research_trainset)

End-to-end team metric

Then optimize all agents together with a team-level metric:

def team_metric(example, prediction, trace=None):
    """Is the final output high quality?"""
    judge = dspy.Predict(JudgeOutput)
    return judge(
        task=example.task,
        expected=example.output,
        actual=prediction.final_output,
    ).is_good

# Create a module that wraps the full team
class TeamModule(dspy.Module):
    def __init__(self):
        self.supervisor = supervisor_module
        self.researcher = optimized_researcher
        self.writer = writer_module
        self.reviewer = reviewer_module

    def forward(self, task):
        # Run the LangGraph app
        result = app.invoke({"task": task, "messages": [], "current_agent": "", "results": {}, "status": "in_progress"})
        return dspy.Prediction(final_output=result["results"].get("draft", ""))

optimizer = dspy.MIPROv2(metric=team_metric, auto="medium")
optimized_team = optimizer.compile(TeamModule(), trainset=team_trainset)

When NOT to use multi-agent

Multi-agent adds orchestration complexity. Consider simpler alternatives first:

• One agent can do the job — if your task needs tools but not multiple specialists, use a single dspy.ReAct agent (see /ai-taking-actions). A single agent with 5 tools is simpler than 3 agents with 2 tools each.
• Fixed pipeline with no routing — if agents always run in the same order (write → edit → review) with no conditional branching, a plain DSPy pipeline module is simpler than LangGraph (see /ai-building-pipelines).
• You are over-specializing — if each "agent" is just a single dspy.Predict call with no tools or state, you do not need agents. Use a multi-step DSPy module instead.

Use multi-agent when you genuinely need dynamic routing (supervisor decides who goes next), parallel execution (fan-out to multiple specialists), or human-in-the-loop checkpoints between steps.

Gotchas

• Claude puts orchestration logic inside DSPy modules. Routing decisions, agent selection, and state transitions belong in LangGraph (conditional edges, route_to_agent). DSPy modules should only handle the reasoning each agent does — classify, research, write, review. If forward() contains if agent == "writer" branching, move that logic to LangGraph edges.
• Claude creates one giant shared state with every field. Each agent only needs a few fields from the state. A bloated TypedDict with 15+ fields makes the graph hard to debug and wastes context. Keep the shared state minimal — task, messages, results, status — and let agents pass specifics through the results dict.
• Claude forgets to cap supervisor iterations. Without a limit, the supervisor can loop forever — routing researcher → writer → reviewer → researcher indefinitely. Add a max_steps counter to the state and a check in the supervisor that forces is_complete = True after N iterations (typically 5-10).
• Claude optimizes the full team before individual agents. Multi-agent optimization is expensive and hard to debug. Always optimize each agent independently first (with per-agent metrics), then freeze the good ones and optimize the team end-to-end. This bottom-up approach is faster and produces better results.
• Claude uses `dspy.Parallel` when it should use LangGraph `Send()`. dspy.Parallel is for independent LM calls within a single module. For parallel agents with different roles, tools, and state, use LangGraph's Send() pattern — it gives you proper state management, error handling, and the ability to interrupt individual agents.

Cross-references

Install any skill: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill <name>

• Single agent with tools — start here instead of multi-agent if one agent suffices -- see /ai-taking-actions
• Stateless pipelines — when agents always run in the same order without routing -- see /ai-building-pipelines
• Conversational agents — if agents need to hold multi-turn conversations -- see /ai-building-chatbots
• Measure and improve agents — evaluate and optimize your multi-agent system -- see /ai-improving-accuracy
• ReAct agents — the DSPy module powering tool-using agents -- see /dspy-react
• Install `/ai-do` if you do not have it — it routes any AI problem to the right skill and is the fastest way to work: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill ai-do

Additional resources

• For worked examples (research team, support escalation), see examples.md
• LangGraph documentation
• LangGraph GitHub

Multi-Agent Examples

Example 1: Research team

A supervisor coordinates three specialists: a web researcher, an analyst, and a writer. The supervisor delegates tasks, collects results, and decides when the work is done.

Agent modules

import dspy
from langchain_community.tools import DuckDuckGoSearchRun

# Web researcher — uses search tools
search_tool = dspy.Tool.from_langchain(DuckDuckGoSearchRun())

class WebResearcher(dspy.Module):
    def __init__(self):
        self.agent = dspy.ReAct(
            "topic, focus_area -> findings",
            tools=[search_tool],
            max_iters=5,
        )

    def forward(self, topic, focus_area=""):
        return self.agent(topic=topic, focus_area=focus_area or topic)


# Analyst — synthesizes research into insights
class AnalyzeFindings(dspy.Signature):
    """Analyze research findings and extract key insights, trends, and implications."""
    topic: str = dspy.InputField()
    raw_findings: str = dspy.InputField(desc="Research findings from various sources")
    insights: list[str] = dspy.OutputField(desc="Key insights and takeaways")
    trends: list[str] = dspy.OutputField(desc="Notable trends")
    recommendation: str = dspy.OutputField(desc="Overall recommendation based on analysis")

class Analyst(dspy.Module):
    def __init__(self):
        self.analyze = dspy.ChainOfThought(AnalyzeFindings)

    def forward(self, topic, raw_findings):
        return self.analyze(topic=topic, raw_findings=raw_findings)


# Writer — produces the final report
class WriteReport(dspy.Signature):
    """Write a clear, well-structured report based on research and analysis."""
    topic: str = dspy.InputField()
    insights: str = dspy.InputField(desc="Analysis insights and recommendations")
    report: str = dspy.OutputField(desc="Well-structured report with sections and key takeaways")

class Writer(dspy.Module):
    def __init__(self):
        self.write = dspy.ChainOfThought(WriteReport)

    def forward(self, topic, insights):
        return self.write(topic=topic, insights=insights)


def writer_reward(args, pred):
    """Soft reward encouraging substantive reports."""
    score = 1.0
    if len(pred.report.split()) <= 200:
        score -= 0.2  # soft: report should be at least 200 words
    return score

writer = dspy.Refine(module=Writer(), N=3, reward_fn=writer_reward, threshold=0.8)

Supervisor and graph

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator

class ResearchState(TypedDict):
    topic: str
    messages: Annotated[list[dict], operator.add]
    results: dict
    current_agent: str
    status: str

class PlanNextStep(dspy.Signature):
    """Decide the next step in the research process."""
    topic: str = dspy.InputField()
    completed_steps: str = dspy.InputField()
    next_agent: str = dspy.OutputField(desc="One of: researcher, analyst, writer, done")
    instruction: str = dspy.OutputField(desc="What to tell the agent")

planner = dspy.ChainOfThought(PlanNextStep)
researcher = WebResearcher()
analyst = Analyst()
# writer is defined above as dspy.Refine(module=Writer(), ...)

def supervisor_node(state: ResearchState) -> dict:
    completed = "\n".join(f"- {k}: done" for k in state["results"])
    result = planner(topic=state["topic"], completed_steps=completed or "Nothing yet")

    if result.next_agent == "done":
        return {"status": "done"}

    return {
        "current_agent": result.next_agent,
        "messages": [{"role": "supervisor", "content": result.instruction}],
    }

def researcher_node(state: ResearchState) -> dict:
    result = researcher(topic=state["topic"])
    return {
        "results": {**state["results"], "research": result.findings},
        "messages": [{"role": "researcher", "content": result.findings}],
    }

def analyst_node(state: ResearchState) -> dict:
    result = analyst(
        topic=state["topic"],
        raw_findings=state["results"].get("research", ""),
    )
    insights = f"Insights: {result.insights}\nTrends: {result.trends}\nRecommendation: {result.recommendation}"
    return {
        "results": {**state["results"], "analysis": insights},
        "messages": [{"role": "analyst", "content": insights}],
    }

def writer_node(state: ResearchState) -> dict:
    result = writer(
        topic=state["topic"],
        insights=state["results"].get("analysis", ""),
    )
    return {
        "results": {**state["results"], "report": result.report},
        "messages": [{"role": "writer", "content": result.report}],
    }

# Build graph
graph = StateGraph(ResearchState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("researcher", researcher_node)
graph.add_node("analyst", analyst_node)
graph.add_node("writer", writer_node)

graph.add_edge(START, "supervisor")

def route(state: ResearchState) -> str:
    if state["status"] == "done":
        return "done"
    return state["current_agent"]

graph.add_conditional_edges("supervisor", route, {
    "researcher": "researcher",
    "analyst": "analyst",
    "writer": "writer",
    "done": END,
})

graph.add_edge("researcher", "supervisor")
graph.add_edge("analyst", "supervisor")
graph.add_edge("writer", "supervisor")

app = graph.compile()

Usage

result = app.invoke({
    "topic": "Impact of AI on software development productivity in 2025",
    "messages": [],
    "results": {},
    "current_agent": "",
    "status": "in_progress",
})

print(result["results"]["report"])
# Full research report with findings, analysis, and recommendations

---

Example 2: Support escalation (L1 → L2 specialists)

An L1 classifier routes tickets to specialized L2 agents. Each L2 agent has domain-specific tools and knowledge.

L1 classifier

from typing import Literal

class TriageTicket(dspy.Signature):
    """Classify the support ticket and decide which specialist team should handle it."""
    ticket: str = dspy.InputField()
    customer_tier: str = dspy.InputField(desc="free, pro, or enterprise")
    category: Literal["billing", "technical", "account", "security"] = dspy.OutputField()
    priority: Literal["low", "medium", "high", "critical"] = dspy.OutputField()

class L1Classifier(dspy.Module):
    def __init__(self):
        self.triage = dspy.ChainOfThought(TriageTicket)

    def forward(self, ticket, customer_tier):
        return self.triage(ticket=ticket, customer_tier=customer_tier)


def l1_classifier_reward(args, pred):
    """Hard constraint: enterprise critical issues must not be auto-classified — force escalation."""
    customer_tier = args.get("customer_tier", "")
    if customer_tier == "enterprise" and pred.priority == "critical":
        return 0.0  # hard: enterprise critical issues go directly to human support
    return 1.0

classifier = dspy.Refine(module=L1Classifier(), N=3, reward_fn=l1_classifier_reward, threshold=1.0)

L2 specialist agents

class BillingResponse(dspy.Signature):
    """Handle a billing support issue. Be specific about amounts and dates."""
    ticket: str = dspy.InputField()
    account_info: str = dspy.InputField(desc="Customer's billing history")
    response: str = dspy.OutputField()
    action_needed: str = dspy.OutputField(desc="refund, credit, none, or escalate_to_human")

class TechnicalResponse(dspy.Signature):
    """Handle a technical support issue. Include specific troubleshooting steps."""
    ticket: str = dspy.InputField()
    docs: list[str] = dspy.InputField(desc="Relevant technical documentation")
    response: str = dspy.OutputField()
    resolved: bool = dspy.OutputField()

class BillingAgent(dspy.Module):
    def __init__(self):
        self.respond = dspy.ChainOfThought(BillingResponse)

    def forward(self, ticket, account_info):
        return self.respond(ticket=ticket, account_info=account_info)

class TechAgent(dspy.Module):
    def __init__(self, retriever):
        self.retriever = retriever
        self.respond = dspy.ChainOfThought(TechnicalResponse)

    def forward(self, ticket):
        docs = self.retriever(ticket).passages
        return self.respond(ticket=ticket, docs=docs)

LangGraph escalation flow

class SupportState(TypedDict):
    ticket: str
    customer_tier: str
    category: str
    priority: str
    response: str
    action_needed: str
    escalated_to_human: bool

# classifier is defined above as dspy.Refine(module=L1Classifier(), ...)
billing_agent = BillingAgent()
tech_agent = TechAgent(retriever=tech_docs_retriever)

def classify_node(state: SupportState) -> dict:
    result = classifier(ticket=state["ticket"], customer_tier=state["customer_tier"])
    return {"category": result.category, "priority": result.priority}

def billing_node(state: SupportState) -> dict:
    account_info = lookup_billing(state["ticket"])  # your billing lookup
    result = billing_agent(ticket=state["ticket"], account_info=account_info)
    return {"response": result.response, "action_needed": result.action_needed}

def tech_node(state: SupportState) -> dict:
    result = tech_agent(ticket=state["ticket"])
    return {"response": result.response}

def human_escalation(state: SupportState) -> dict:
    return {"escalated_to_human": True, "response": "Escalated to human support team."}

def route_to_specialist(state: SupportState) -> str:
    if state["priority"] == "critical":
        return "human"
    return state["category"]

graph = StateGraph(SupportState)
graph.add_node("classify", classify_node)
graph.add_node("billing", billing_node)
graph.add_node("technical", tech_node)
graph.add_node("human", human_escalation)

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_to_specialist, {
    "billing": "billing",
    "technical": "technical",
    "account": "human",
    "security": "human",
    "human": "human",
})

# Billing agent can escalate to human for refunds
def billing_followup(state: SupportState) -> str:
    if state.get("action_needed") == "escalate_to_human":
        return "human"
    return "done"

graph.add_conditional_edges("billing", billing_followup, {"human": "human", "done": END})
graph.add_edge("technical", END)
graph.add_edge("human", END)

app = graph.compile()

Usage

# Technical issue — routed to tech agent
result = app.invoke({
    "ticket": "API returns 500 error when uploading files larger than 10MB",
    "customer_tier": "pro",
    "category": "", "priority": "", "response": "",
    "action_needed": "", "escalated_to_human": False,
})
print(result["response"])
# Specific troubleshooting steps from tech docs

# Critical billing issue — escalated to human
result = app.invoke({
    "ticket": "We were charged $50,000 instead of $500, need immediate resolution",
    "customer_tier": "enterprise",
    "category": "", "priority": "", "response": "",
    "action_needed": "", "escalated_to_human": False,
})
print(result["escalated_to_human"])  # True