Multi-Agent Systems¶

Multi-agent systems decompose complex tasks across specialized agents — a researcher, a coder, a writer, a reviewer. Each agent does what it's best at, and a coordinator orchestrates the handoffs. This covers architectures, coordination patterns, and communication strategies.

Table of Contents¶

Installation / Setup
Core Syntax
Common Operations
Practical Examples
Common Patterns
Best Practices
Common Mistakes
Troubleshooting
Related Cheat Sheets
Official Documentation

Installation / Setup¶

pip install langgraph langchain-openai

Core Syntax¶

Multi-agent architectures¶

Architecture	How it works	Best for
Supervisor	Central coordinator assigns tasks	Clear task delegation
Workers	Specialized agents handle sub-tasks	Parallel sub-tasks
Handoffs	Agents pass work to each other	Sequential workflows
Debate	Agents argue different perspectives	Balanced analysis
Hierarchical	Managers manage workers	Large-scale systems

Supervisor pattern¶

graph TD
    A[User] --> B[Supervisor]
    B --> C[Researcher]
    B --> D[Coder]
    B --> E[Writer]
    C --> B
    D --> B
    E --> B
    B --> F[Final Output]

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

# specialized agents
researcher = create_react_agent(llm, [search_tool], state_modifier="You research topics.")
coder = create_react_agent(llm, [run_python], state_modifier="You write Python code.")
writer = create_react_agent(llm, [], state_modifier="You write clear reports.")

# supervisor decides who to call
supervisor_prompt = """You are a supervisor. Route to the right worker:
- researcher: for information lookup
- coder: for code writing
- writer: for report writing

Current task: {task}"""

def supervisor(task):
    response = llm.invoke(supervisor_prompt.format(task=task))
    route = response.content.strip().lower()
    if "research" in route:
        return researcher.invoke({"messages": [("human", task)]})
    elif "code" in route:
        return coder.invoke({"messages": [("human", task)]})
    else:
        return writer.invoke({"messages": [("human", task)]})

LangGraph multi-agent¶

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

class State(TypedDict):
    messages: Annotated[list, add]
    next: str

def supervisor_node(state: State):
    response = llm.invoke([
        {"role": "system", "content": "Route to researcher, coder, or FINISH."},
        *state["messages"],
    ])
    return {"next": response.content.strip()}

def researcher_node(state: State):
    result = researcher.invoke(state)
    return {"messages": [result["messages"][-1]]}

def coder_node(state: State):
    result = coder.invoke(state)
    return {"messages": [result["messages"][-1]]}

graph = StateGraph(State)
graph.add_node("supervisor", supervisor_node)
graph.add_node("researcher", researcher_node)
graph.add_node("coder", coder_node)

graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", lambda s: s["next"], {
    "researcher": "researcher",
    "coder": "coder",
    "FINISH": END,
})
graph.add_edge("researcher", "supervisor")
graph.add_edge("coder", "supervisor")

Handoff pattern¶

def handoff(from_agent, to_agent, task):
    """Transfer work from one agent to another."""
    result = from_agent.invoke(task)
    # extract relevant context
    context = result["messages"][-1].content
    # hand off
    return to_agent.invoke({"messages": [("human", f"Continue from previous agent:\n{context}")]})

Common Operations¶

Agent communication via shared state¶

# agents communicate through shared state
class SharedState:
    def __init__(self):
        self.data = {}

    def set(self, key, value):
        self.data[key] = value

    def get(self, key, default=None):
        return self.data.get(key, default)

# researcher sets data
state.set("findings", "Key insight: ...")

# coder uses data
findings = state.get("findings")
code = coder.invoke(f"Write code based on: {findings}")

Parallel agent execution¶

from concurrent.futures import ThreadPoolExecutor

def parallel_agents(task, agents):
    """Run multiple agents in parallel on different sub-tasks."""
    sub_tasks = decompose(task)
    with ThreadPoolExecutor(max_workers=len(agents)) as pool:
        results = list(pool.map(
            lambda a: a.invoke({"messages": [("human", a.task)]}),
            agents
        ))
    return results

Debate pattern¶

def debate(topic, num_rounds=3):
    agent_a = create_react_agent(llm, [], state_modifier="You argue FOR the topic.")
    agent_b = create_react_agent(llm, [], state_modifier="You argue AGAINST the topic.")

    history = []
    for round in range(num_rounds):
        a_response = agent_a.invoke({"messages": [
            ("human", f"Argue for: {topic}\n\nPrevious: {history}")]})
        history.append(("for", a_response["messages"][-1].content))

        b_response = agent_b.invoke({"messages": [
            ("human", f"Argue against: {topic}\n\nPrevious: {history}")]})
        history.append(("against", b_response["messages"][-1].content))

    # summarize
    return llm.invoke(f"Summarize this debate:\n{history}").content

Practical Examples¶

Example 1 — Software development team¶

def dev_team(feature_request):
    # architect designs
    architecture = architect.invoke(f"Design architecture for: {feature_request}")

    # coder implements
    code = coder.invoke(f"Implement this design:\n{architecture}")

    # reviewer checks
    review = reviewer.invoke(f"Review this code:\n{code}")

    # if issues found, coder fixes
    if "issue" in review.lower():
        code = coder.invoke(f"Fix issues:\n{review}\n\nCode:\n{code}")

    return code

Example 2 — Research pipeline¶

def research_team(topic):
    # parallel research
    with ThreadPoolExecutor(max_workers=3) as pool:
        sources = list(pool.map(
            lambda agent: agent.invoke({"messages": [("human", f"Research {topic} from {agent.source}")]})["messages"][-1].content,
            [academic_agent, news_agent, code_agent]
        ))

    # synthesize
    synthesis = synthesizer.invoke(f"Synthesize these sources:\n{sources}")
    return synthesis["messages"][-1].content

Common Patterns¶

Pattern: supervisor with handoff¶

def supervised_workflow(task):
    # supervisor assigns
    assignment = supervisor.invoke(f"Who should handle: {task}")
    agent_name = assignment.content.strip()

    # execute
    agent = agent_registry[agent_name]
    result = agent.invoke({"messages": [("human", task)]})

    # check if handoff needed
    if needs_handoff(result):
        next_agent = supervisor.invoke(f"Hand off from {agent_name}: {result}")
        result = agent_registry[next_agent.content.strip()].invoke(
            {"messages": [("human", f"Continue: {result}")]})

    return result

Pattern: agent registry¶

class AgentRegistry:
    def __init__(self):
        self.agents = {}

    def register(self, name, agent, description):
        self.agents[name] = {"agent": agent, "description": description}

    def get_descriptions(self):
        return "\n".join(f"- {k}: {v['description']}" for k, v in self.agents.items())

    def get(self, name):
        return self.agents[name]["agent"]

Pattern: message passing¶

def message_pass(from_agent, to_agent, message):
    """Structured message passing between agents."""
    result = from_agent.invoke({"messages": [("human", message)]})
    response = result["messages"][-1].content
    return to_agent.invoke({"messages": [("human", f"Message from {from_agent.name}: {response}")]})

Best Practices¶

Start with one agent — add multi-agent only when needed.
Supervisor pattern is the simplest and most common.
Specialize agents — each agent should have a clear, narrow role.
Use shared state for communication — not message passing.
Limit agent count — 2–4 agents is usually enough.
Add error handling — agents can fail; have fallbacks.
Test each agent independently before combining.
Log agent interactions — multi-agent systems are hard to debug.

Common Mistakes¶

Mistake	Fix
Too many agents	Start with 2-3; add only when needed
Agents with overlapping roles	Make each agent's role distinct
No supervisor	Add a coordinator to prevent chaos
No error handling	Add try/except and fallback agents
Ignoring communication overhead	Minimize agent-to-agent messages

Troubleshooting¶

Symptom	Cause / Fix
Agents loop forever	Add max iterations; check supervisor logic
Wrong agent handles task	Improve supervisor classification
Agents don't collaborate	Use shared state, not isolated contexts
System is too slow	Reduce agents; run in parallel where possible

ai-agents.md — single agent loop
routing.md — routing decisions
planning.md — planning strategies
memory.md — shared state
langgraph — LangGraph multi-agent
multi-agent — LangGraph patterns

Official Documentation¶

LangGraph multi-agent — https://langchain-ai.github.io/langgraph/tutorials/
CrewAI — https://docs.crewai.com/
AutoGen — https://microsoft.github.io/autogen/

Last Updated¶

2026-09-06 — Initial version.

Multi-Agent Systems¶

Table of Contents¶

Installation / Setup¶

Core Syntax¶

Multi-agent architectures¶

Supervisor pattern¶

LangGraph multi-agent¶

Handoff pattern¶

Common Operations¶

Agent communication via shared state¶

Parallel agent execution¶

Debate pattern¶

Practical Examples¶

Example 1 — Software development team¶

Example 2 — Research pipeline¶

Common Patterns¶

Pattern: supervisor with handoff¶

Pattern: agent registry¶

Pattern: message passing¶

Best Practices¶

Common Mistakes¶

Troubleshooting¶

Related Cheat Sheets¶

Official Documentation¶

Last Updated¶