Module 10: Multi-Agent Systems

Pattern Subagents

Capsule overview

In the previous capsules you learned two ways to coordinate agents: the supervisor (a central agent that decides) and handoffs (direct transfer between agents). Both share one trait: the state flows through in full between agents. The researcher sees its research, the analyst sees the research + its analysis, the writer sees everything that came before. With every step, the context grows.

Subagents solve that problem with context isolation. A parent agent (supervisor) creates a subtask, sends it to a subagent, and the subagent works in its own space — without seeing the parent's full conversation. When it finishes, it returns only the result. The parent integrates that result and moves on. The parent's context doesn't grow with each subtask.

The fundamental difference: with handoffs, the destination agent receives the accumulated context. With subagents, the child agent receives only its instruction and works from scratch. It's the difference between "here's the entire case file" and "just tell me what you need me to look into."

The context bloat problem

To understand why subagents exist, you have to see the problem they solve:

from dotenv import load_dotenv
load_dotenv()

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

class SharedState(TypedDict):
    task: str
    agent_a_work: str
    agent_b_work: str
    agent_c_work: str
    context_size: int
    log: Annotated[list[str], operator.add]

def agent_a(state: SharedState) -> Command:
    work = "A" * 200
    size = len(state["task"]) + len(work)
    return Command(
        goto="agent_b",
        update={
            "agent_a_work": work,
            "context_size": size,
            "log": [f"agent_a: generated {len(work)} chars, total context: {size}"],
        },
    )

def agent_b(state: SharedState) -> Command:
    work = "B" * 300
    size = state["context_size"] + len(work)
    return Command(
        goto="agent_c",
        update={
            "agent_b_work": work,
            "context_size": size,
            "log": [f"agent_b: generated {len(work)} chars, total context: {size}"],
        },
    )

def agent_c(state: SharedState) -> dict:
    work = "C" * 400
    size = state["context_size"] + len(work)
    return {
        "agent_c_work": work,
        "context_size": size,
        "log": [f"agent_c: generated {len(work)} chars, total context: {size}"],
    }

builder = StateGraph(SharedState)
builder.add_node("agent_a", agent_a)
builder.add_node("agent_b", agent_b)
builder.add_node("agent_c", agent_c)

builder.add_edge(START, "agent_a")
builder.add_edge("agent_c", END)

graph = builder.compile()

result = graph.invoke({
    "task": "AI market analysis",
    "agent_a_work": "", "agent_b_work": "",
    "agent_c_work": "", "context_size": 0, "log": [],
})

print("=== Context bloat with shared state ===\n")
for entry in result["log"]:
    print(f"  {entry}")
print(f"\nFinal context: {result['context_size']} chars")
print(f"Did Agent C need to see all of it? Probably not.")
# Expected output:
# === Context bloat with shared state ===
#
#   agent_a: generated 200 chars, total context: 218
#   agent_b: generated 300 chars, total context: 518
#   agent_c: generated 400 chars, total context: 918
#
# Final context: 918 chars
# Did Agent C need to see all of it? Probably not.

With every handoff, the context grows. In this simplified example they're characters, but in a real system they're LLM messages, tool results, intermediate reasoning. With 5 agents and tools that return long documents, you easily hit thousands of tokens the last agent doesn't need.

With subagents, each agent works in a clean space. The parent's context stays focused.

How subagents work

A subagent's flow has 5 steps:

The parent writes a description of the subtask — "Research AI trends 2025"
The subagent receives ONLY that description — it doesn't see the parent's conversation or the work of other subagents
The subagent runs with its own tools and prompt — it works autonomously in its isolated context
The subagent returns a concise result — only the final output, not all of its internal reasoning
The parent integrates the result — it folds it into its own context and decides the next step

The key: the parent doesn't send its whole conversation to the subagent. And the subagent doesn't send its whole internal process back to the parent. Both keep their contexts separate.

Basic implementation: a subagent as a tool

The most direct pattern: create an agent with create_agent and wrap it as a @tool the parent agent can call:

from dotenv import load_dotenv
load_dotenv()

from langchain.tools import tool
from langchain.agents import create_agent

research_subagent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[],
    prompt="You are a specialized researcher. Answer concisely with key findings.",
)

analysis_subagent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[],
    prompt="You are a data analyst. Identify trends and risks. Be concise.",
)

@tool("research", description="Research a topic and return key findings")
def call_researcher(query: str) -> str:
    result = research_subagent.invoke({
        "messages": [{"role": "user", "content": query}]
    })
    return result["messages"][-1].content

@tool("analyze", description="Analyze information and return trends and risks")
def call_analyst(data: str) -> str:
    result = analysis_subagent.invoke({
        "messages": [{"role": "user", "content": data}]
    })
    return result["messages"][-1].content

supervisor = create_agent(
    "openai:gpt-4.1-mini",
    tools=[call_researcher, call_analyst],
    prompt=(
        "You are a research supervisor. "
        "Use the 'research' tool to research topics and "
        "'analyze' to analyze the findings. "
        "Combine the results into a concise final report."
    ),
)

result = supervisor.invoke({
    "messages": [{"role": "user", "content": "Research and analyze the state of LLMs in production"}]
})

for msg in result["messages"]:
    msg.pretty_print()
# Expected output (varies by model):
# ================================ Human Message =================================
# Research and analyze the state of LLMs in production
# ================================== Ai Message ==================================
# I'm going to research this topic and then analyze it.
# [tool call: research("Current state of LLMs in enterprise production")]
# ================================== Tool Message ================================
# LLMs in production have seen significant advances: ...
# ================================== Ai Message ==================================
# [tool call: analyze("LLMs in production have seen...")]
# ================================== Tool Message ================================
# Trends: growing adoption. Risks: cost and hallucinations...
# ================================== Ai Message ==================================
# REPORT: LLMs in production show growing adoption...

Context isolation happens naturally:

✅ The research_subagent only sees: "Research the state of LLMs in production" — it doesn't see the supervisor's full conversation
✅ The analysis_subagent only sees the findings the supervisor passes it — it doesn't see the user's original question or the supervisor's reasoning
✅ The supervisor only sees each subagent's final results — it doesn't see their internal reasoning

A subagent as a compiled subgraph

When you need more control over the subagent (custom state, multiple nodes, internal flows), you can create a compiled StateGraph and invoke it from a node in the parent graph:

from dotenv import load_dotenv
load_dotenv()

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

class SubagentState(TypedDict):
    task: str
    result: str

class ParentState(TypedDict):
    objective: str
    research_result: str
    analysis_result: str
    final_report: str
    log: Annotated[list[str], operator.add]

def do_research(state: SubagentState) -> dict:
    return {"result": f"Research on '{state['task']}': 5 sources, 3 key findings"}

def do_analysis(state: SubagentState) -> dict:
    return {"result": f"Analysis of '{state['task']}': upward trend, 2 risks"}

sub_research_builder = StateGraph(SubagentState)
sub_research_builder.add_node("work", do_research)
sub_research_builder.add_edge(START, "work")
sub_research_builder.add_edge("work", END)
research_subgraph = sub_research_builder.compile()

sub_analysis_builder = StateGraph(SubagentState)
sub_analysis_builder.add_node("work", do_analysis)
sub_analysis_builder.add_edge(START, "work")
sub_analysis_builder.add_edge("work", END)
analysis_subgraph = sub_analysis_builder.compile()

def research_node(state: ParentState) -> dict:
    sub_result = research_subgraph.invoke({"task": state["objective"], "result": ""})
    return {
        "research_result": sub_result["result"],
        "log": [f"research subagent returned: {len(sub_result['result'])} chars"],
    }

def analysis_node(state: ParentState) -> dict:
    sub_result = analysis_subgraph.invoke({"task": state["research_result"], "result": ""})
    return {
        "analysis_result": sub_result["result"],
        "log": [f"analysis subagent returned: {len(sub_result['result'])} chars"],
    }

def report_node(state: ParentState) -> dict:
    report = (
        f"REPORT\n"
        f"Objective: {state['objective']}\n"
        f"Research: {state['research_result']}\n"
        f"Analysis: {state['analysis_result']}"
    )
    return {"final_report": report, "log": ["report generated"]}

builder = StateGraph(ParentState)
builder.add_node("research", research_node)
builder.add_node("analysis", analysis_node)
builder.add_node("report", report_node)

builder.add_edge(START, "research")
builder.add_edge("research", "analysis")
builder.add_edge("analysis", "report")
builder.add_edge("report", END)

graph = builder.compile()

result = graph.invoke({
    "objective": "State of AI agents in 2025",
    "research_result": "", "analysis_result": "",
    "final_report": "", "log": [],
})

print(result["final_report"])
print(f"\nFlow:")
for entry in result["log"]:
    print(f"  → {entry}")
# Expected output:
# REPORT
# Objective: State of AI agents in 2025
# Research: Research on 'State of AI agents in 2025': 5 sources, 3 key findings
# Analysis: Analysis of 'Research on 'State of AI agents in 2025': 5 sources, 3 key findings': upward trend, 2 risks
#
# Flow:
#   → research subagent returned: 67 chars
#   → analysis subagent returned: 104 chars
#   → report generated

The isolation is explicit: each subgraph has its own SubagentState, separate from ParentState. The parent node builds the input ({"task": state["objective"]}), invokes the subgraph, and pulls out only sub_result["result"]. The subgraph never sees ParentState and the parent never sees the subgraph's internal state.

Context isolation: who sees what

To make the context isolation concrete, let's look explicitly at what information each agent has available:

from dotenv import load_dotenv
load_dotenv()

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

class SubState(TypedDict):
    instruction: str
    result: str

class ParentState(TypedDict):
    user_query: str
    conversation_history: list[str]
    internal_reasoning: str
    sub_result_1: str
    sub_result_2: str
    final_answer: str
    log: Annotated[list[str], operator.add]

def sub_worker(state: SubState) -> dict:
    visible_keys = list(state.keys())
    result = (
        f"Subagent working. "
        f"Sees: {visible_keys}. "
        f"Instruction: '{state['instruction']}'. "
        f"Does NOT see: user_query, conversation_history, internal_reasoning"
    )
    return {"result": result}

sub_builder = StateGraph(SubState)
sub_builder.add_node("work", sub_worker)
sub_builder.add_edge(START, "work")
sub_builder.add_edge("work", END)
subagent = sub_builder.compile()

def step_1(state: ParentState) -> dict:
    sub_result = subagent.invoke({
        "instruction": "Look up information about Python async",
        "result": "",
    })
    return {
        "sub_result_1": sub_result["result"],
        "conversation_history": state["conversation_history"] + ["step_1 completed"],
        "internal_reasoning": "I need async data to answer the user",
        "log": ["step_1: subagent invoked with isolated context"],
    }

def step_2(state: ParentState) -> dict:
    sub_result = subagent.invoke({
        "instruction": "Look up benchmarks of async vs sync in Python",
        "result": "",
    })
    return {
        "sub_result_2": sub_result["result"],
        "conversation_history": state["conversation_history"] + ["step_2 completed"],
        "log": ["step_2: second subagent invoked, also isolated"],
    }

def synthesize(state: ParentState) -> dict:
    parent_sees = {
        "user_query": state["user_query"][:30],
        "history_length": len(state["conversation_history"]),
        "reasoning": state["internal_reasoning"][:30],
        "sub_result_1": state["sub_result_1"][:40],
        "sub_result_2": state["sub_result_2"][:40],
    }
    return {
        "final_answer": f"The parent sees its whole context: {parent_sees}",
        "log": ["synthesize: the parent integrates the results"],
    }

builder = StateGraph(ParentState)
builder.add_node("step_1", step_1)
builder.add_node("step_2", step_2)
builder.add_node("synthesize", synthesize)

builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "synthesize")
builder.add_edge("synthesize", END)

graph = builder.compile()

result = graph.invoke({
    "user_query": "Explain async/await in Python to me",
    "conversation_history": [],
    "internal_reasoning": "",
    "sub_result_1": "", "sub_result_2": "",
    "final_answer": "", "log": [],
})

print("=== What the subagent sees ===")
print(f"  {result['sub_result_1']}\n")
print("=== What the parent sees ===")
print(f"  {result['final_answer']}\n")
print("=== Flow ===")
for entry in result["log"]:
    print(f"  → {entry}")
# Expected output:
# === What the subagent sees ===
#   Subagent working. Sees: ['instruction', 'result']. Instruction: 'Look up information about Python async'. Does NOT see: user_query, conversation_history, internal_reasoning
#
# === What the parent sees ===
#   The parent sees its whole context: {'user_query': 'Explain async/await in Python ', 'history_length': 2, 'reasoning': 'I need async data to answer the', 'sub_result_1': 'Subagent working. Sees: ['instruction', ', 'sub_result_2': 'Subagent working. Sees: ['instruction', '}
#
# === Flow ===
#   → step_1: subagent invoked with isolated context
#   → step_2: second subagent invoked, also isolated
#   → synthesize: the parent integrates the results

The subagent sees: ['instruction', 'result']. It doesn't see user_query, conversation_history, or internal_reasoning. The parent sees its own full state plus the subagents' results. Each one works in its own space.

Subagents in parallel

One of the most practical advantages of subagents: because they work in an isolated context, you can run several at the same time with no state conflicts:

from dotenv import load_dotenv
load_dotenv()

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

class SubState(TypedDict):
    source: str
    query: str
    result: str

class ParentState(TypedDict):
    query: str
    web_results: str
    papers_results: str
    docs_results: str
    combined_report: str
    log: Annotated[list[str], operator.add]

def search_source(state: SubState) -> dict:
    return {
        "result": f"[{state['source']}] 3 results for '{state['query']}'"
    }

sub_builder = StateGraph(SubState)
sub_builder.add_node("search", search_source)
sub_builder.add_edge(START, "search")
sub_builder.add_edge("search", END)
search_subagent = sub_builder.compile()

def parallel_research(state: ParentState) -> dict:
    sources = [
        {"source": "web", "query": state["query"], "result": ""},
        {"source": "papers", "query": state["query"], "result": ""},
        {"source": "docs", "query": state["query"], "result": ""},
    ]

    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        futures = {
            executor.submit(search_subagent.invoke, src): src["source"]
            for src in sources
        }
        results = {}
        for future in concurrent.futures.as_completed(futures):
            source_name = futures[future]
            results[source_name] = future.result()["result"]

    return {
        "web_results": results["web"],
        "papers_results": results["papers"],
        "docs_results": results["docs"],
        "log": [f"parallel_research: 3 subagents completed ({', '.join(results.keys())})"],
    }

def combine_results(state: ParentState) -> dict:
    report = (
        f"COMBINED REPORT for '{state['query']}':\n"
        f"  Web: {state['web_results']}\n"
        f"  Papers: {state['papers_results']}\n"
        f"  Docs: {state['docs_results']}"
    )
    return {"combined_report": report, "log": ["combine: report generated"]}

builder = StateGraph(ParentState)
builder.add_node("parallel_research", parallel_research)
builder.add_node("combine_results", combine_results)

builder.add_edge(START, "parallel_research")
builder.add_edge("parallel_research", "combine_results")
builder.add_edge("combine_results", END)

graph = builder.compile()

result = graph.invoke({
    "query": "RAG patterns in production",
    "web_results": "", "papers_results": "", "docs_results": "",
    "combined_report": "", "log": [],
})

print(result["combined_report"])
print(f"\nLog:")
for entry in result["log"]:
    print(f"  → {entry}")
# Expected output:
# COMBINED REPORT for 'RAG patterns in production':
#   Web: [web] 3 results for 'RAG patterns in production'
#   Papers: [papers] 3 results for 'RAG patterns in production'
#   Docs: [docs] 3 results for 'RAG patterns in production'
#
# Log:
#   → parallel_research: 3 subagents completed (web, papers, docs)
#   → combine: report generated

The 3 subagents run in parallel with ThreadPoolExecutor. Each one works in its own state (SubState), searches its assigned source, and returns its result. The parent node collects all the results and integrates them. No state conflicts, because each subagent has its own space.

Single dispatch tool: one tool for N subagents

When you have many subagents, creating one @tool per subagent scales badly. The single dispatch pattern uses a single parameterized tool that invokes the right subagent by name:

from dotenv import load_dotenv
load_dotenv()

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

class SubState(TypedDict):
    task: str
    result: str

def researcher_work(state: SubState) -> dict:
    return {"result": f"[RESEARCH] Research on: {state['task'][:40]}"}

def writer_work(state: SubState) -> dict:
    return {"result": f"[WRITER] Content generated for: {state['task'][:40]}"}

def reviewer_work(state: SubState) -> dict:
    return {"result": f"[REVIEWER] Review completed of: {state['task'][:40]}"}

def build_subagent(work_fn) -> object:
    b = StateGraph(SubState)
    b.add_node("work", work_fn)
    b.add_edge(START, "work")
    b.add_edge("work", END)
    return b.compile()

REGISTRY = {
    "researcher": build_subagent(researcher_work),
    "writer": build_subagent(writer_work),
    "reviewer": build_subagent(reviewer_work),
}

def dispatch(agent_name: str, task_description: str) -> str:
    """Invoke a subagent by name with context isolation."""
    if agent_name not in REGISTRY:
        return f"Error: agent '{agent_name}' not found. Available: {list(REGISTRY.keys())}"
    subagent = REGISTRY[agent_name]
    result = subagent.invoke({"task": task_description, "result": ""})
    return result["result"]


class OrchestratorState(TypedDict):
    objective: str
    steps_completed: list[str]
    final_output: str
    log: Annotated[list[str], operator.add]

def orchestrator(state: OrchestratorState) -> dict:
    research = dispatch("researcher", f"Research: {state['objective']}")
    writing = dispatch("writer", f"Write based on: {research}")
    review = dispatch("reviewer", f"Review: {writing}")

    return {
        "steps_completed": [research, writing, review],
        "final_output": review,
        "log": [
            f"dispatch → researcher: {research[:40]}...",
            f"dispatch → writer: {writing[:40]}...",
            f"dispatch → reviewer: {review[:40]}...",
        ],
    }

builder = StateGraph(OrchestratorState)
builder.add_node("orchestrator", orchestrator)
builder.add_edge(START, "orchestrator")
builder.add_edge("orchestrator", END)

graph = builder.compile()

result = graph.invoke({
    "objective": "Best practices for deploying AI agents",
    "steps_completed": [], "final_output": "", "log": [],
})

print(f"Final output: {result['final_output']}")
print(f"\nSteps completed:")
for step in result["steps_completed"]:
    print(f"  {step}")
print(f"\nLog:")
for entry in result["log"]:
    print(f"  → {entry}")
# Expected output:
# Final output: [REVIEWER] Review completed of: Review: [WRITER] Content generated for:
#
# Steps completed:
#   [RESEARCH] Research on: Research: Best practices for deploying A
#   [WRITER] Content generated for: Write based on: [RESEARCH] Research on:
#   [REVIEWER] Review completed of: Review: [WRITER] Content generated for:
#
# Log:
#   → dispatch → researcher: [RESEARCH] Research on: Research: Best p...
#   → dispatch → writer: [WRITER] Content generated for: Write ba...
#   → dispatch → reviewer: [REVIEWER] Review completed of: Review: ...

A single dispatch(agent_name, task) invokes any registered subagent. To add a new subagent, you just register it in REGISTRY. The orchestrator doesn't need to change.

With create_agent, the pattern is identical but the subagents are LLM-backed agents:

from langchain.tools import tool
from langchain.agents import create_agent

SUBAGENTS = {
    "researcher": create_agent("openai:gpt-4.1-mini", prompt="You are a researcher..."),
    "writer": create_agent("openai:gpt-4.1-mini", prompt="You are a writer..."),
}

@tool
def task(agent_name: str, description: str) -> str:
    """Launch a subagent for a specific task."""
    agent = SUBAGENTS[agent_name]
    result = agent.invoke({"messages": [{"role": "user", "content": description}]})
    return result["messages"][-1].content

When to use subagents vs handoffs

Criterion	Subagents	Handoffs
Context	Isolated — each subagent only sees its task	Shared — context flows between agents
Context bloat	No — the parent only sees final results	Yes — it grows with every handoff
Parallel execution	Natural — subagents are independent	Hard — the flow is sequential
User interaction	Not direct — it goes through the parent	Direct — each agent can interact
Coordination	Centralized in the parent	Distributed among agents
Ideal case	Many independent subtasks	Sequential flow with shared context
Example	"Search 3 sources in parallel"	"Research → analyze → write"

Practical rule:

✅ Use subagents when the subtasks are independent and don't need to see each other
✅ Use handoffs when the flow is sequential and each agent needs the previous one's context
✅ Combine both: handoffs for the main flow, subagents for each agent's internal subtasks

Error handling and timeouts in subagents

Subagents can fail: the LLM doesn't respond, a tool errors out, or the subagent gets stuck in an infinite loop. The parent needs to handle those cases:

from dotenv import load_dotenv
load_dotenv()

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

class SubState(TypedDict):
    task: str
    result: str

class ParentState(TypedDict):
    query: str
    results: dict
    errors: dict
    final_report: str
    log: Annotated[list[str], operator.add]

def fast_agent(state: SubState) -> dict:
    return {"result": f"Fast result for: {state['task'][:30]}"}

def slow_agent(state: SubState) -> dict:
    time.sleep(0.5)
    return {"result": f"Slow result for: {state['task'][:30]}"}

def failing_agent(state: SubState) -> dict:
    raise ValueError(f"Error processing: {state['task'][:20]}")

fast_sub = StateGraph(SubState)
fast_sub.add_node("work", fast_agent)
fast_sub.add_edge(START, "work")
fast_sub.add_edge("work", END)
fast_subagent = fast_sub.compile()

slow_sub = StateGraph(SubState)
slow_sub.add_node("work", slow_agent)
slow_sub.add_edge(START, "work")
slow_sub.add_edge("work", END)
slow_subagent = slow_sub.compile()

fail_sub = StateGraph(SubState)
fail_sub.add_node("work", failing_agent)
fail_sub.add_edge(START, "work")
fail_sub.add_edge("work", END)
failing_subagent = fail_sub.compile()

AGENTS = {
    "fast": fast_subagent,
    "slow": slow_subagent,
    "failing": failing_subagent,
}

def safe_invoke(agent_name: str, task: str, timeout: float = 2.0) -> dict:
    """Invoke a subagent with a timeout and error handling."""
    agent = AGENTS[agent_name]
    try:
        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
            future = executor.submit(agent.invoke, {"task": task, "result": ""})
            result = future.result(timeout=timeout)
            return {"success": True, "result": result["result"]}
    except concurrent.futures.TimeoutError:
        return {"success": False, "result": f"TIMEOUT: {agent_name} exceeded {timeout}s"}
    except Exception as e:
        return {"success": False, "result": f"ERROR in {agent_name}: {str(e)[:50]}"}

def orchestrate(state: ParentState) -> dict:
    agents_to_run = ["fast", "slow", "failing"]
    results = {}
    errors = {}

    for name in agents_to_run:
        outcome = safe_invoke(name, state["query"])
        if outcome["success"]:
            results[name] = outcome["result"]
        else:
            errors[name] = outcome["result"]

    report_parts = [f"Query: {state['query']}"]
    if results:
        report_parts.append(f"Succeeded ({len(results)}): {list(results.keys())}")
    if errors:
        report_parts.append(f"Failed ({len(errors)}): {list(errors.keys())}")

    return {
        "results": results,
        "errors": errors,
        "final_report": " | ".join(report_parts),
        "log": [
            f"succeeded: {list(results.keys())}",
            f"errors: {list(errors.keys())}",
        ],
    }

builder = StateGraph(ParentState)
builder.add_node("orchestrate", orchestrate)
builder.add_edge(START, "orchestrate")
builder.add_edge("orchestrate", END)

graph = builder.compile()

result = graph.invoke({
    "query": "State of AI agents",
    "results": {}, "errors": {},
    "final_report": "", "log": [],
})

print(f"Report: {result['final_report']}")
print(f"\nSuccessful results:")
for name, res in result["results"].items():
    print(f"  ✅ {name}: {res}")
print(f"\nErrors:")
for name, err in result["errors"].items():
    print(f"  ❌ {name}: {err}")
# Expected output:
# Report: Query: State of AI agents | Succeeded (2): ['fast', 'slow'] | Failed (1): ['failing']
#
# Successful results:
#   ✅ fast: Fast result for: State of AI agents
#   ✅ slow: Slow result for: State of AI agents
#
# Errors:
#   ❌ failing: ERROR in failing: Error processing: State of AI agents

The safe_invoke function wraps each subagent with:

✅ Timeout: If the subagent doesn't respond within N seconds, it returns an error
✅ Try/except: It catches any exception from the subagent
✅ Structured result: {"success": bool, "result": str} — the parent knows what worked and what didn't

The parent decides what to do with the errors: retry, use partial results, or abort.

Troubleshooting

Problem 1: "The subagent returns too much information"

Symptom: The subagent's result includes all of its internal reasoning, tool calls, and intermediate messages. The parent's context grows as if it were a handoff.

Cause: You're returning the subagent's entire state instead of just the final result.

Fix: Pull out only the content of the last message:

result = subagent.invoke({"messages": [{"role": "user", "content": task}]})
return result["messages"][-1].content

Problem 2: "Parallel subagents cause state errors"

Symptom: When you run subagents in parallel, some fail or return corrupted results.

Cause: The subagents share some mutable resource (a global variable, a file, a connection).

Fix: Make sure each subagent is genuinely independent. Each invocation must create its own state:

result = subagent.invoke({"task": task, "result": ""})

Problem 3: "The subagent doesn't have the tools it needs"

Symptom: The subagent can't complete its task because it's missing tools.

Cause: Each subagent has its own set of tools. It does not inherit the parent's tools.

Fix: Define the subagent's tools when you create it:

research_subagent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[web_search, paper_search],
    prompt="You are a researcher with access to web and paper search.",
)

Problem 4: "I can't debug what the subagent did"

Symptom: The subagent returns a wrong result but you don't know what happened internally.

Cause: By design, context isolation hides the subagent's internal details.

Fix: Add logging inside the subagent, or use LangSmith to trace the full execution. You can also return extra metadata alongside the result:

return {
    "result": final_answer,
    "metadata": {"steps_taken": 3, "tools_used": ["web_search"]}
}

Exercises

Exercise 1: Basic subagent with isolated context (Easy)

Build a simple parent-child system. The parent has a query and a secret_context (internal information). The subagent only receives the query and returns a result. Verify that the subagent CANNOT see secret_context.

See solution

from dotenv import load_dotenv
load_dotenv()

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

class SubState(TypedDict):
    query: str
    result: str

class ParentState(TypedDict):
    user_query: str
    secret_context: str
    sub_result: str
    log: Annotated[list[str], operator.add]

def subagent_work(state: SubState) -> dict:
    visible = list(state.keys())
    return {"result": f"Subagent sees keys: {visible}. Processed: '{state['query'][:30]}'"}

sub_builder = StateGraph(SubState)
sub_builder.add_node("work", subagent_work)
sub_builder.add_edge(START, "work")
sub_builder.add_edge("work", END)
subagent = sub_builder.compile()

def parent_node(state: ParentState) -> dict:
    result = subagent.invoke({"query": state["user_query"], "result": ""})
    return {
        "sub_result": result["result"],
        "log": [f"subagent returned: {result['result'][:50]}..."],
    }

builder = StateGraph(ParentState)
builder.add_node("parent", parent_node)
builder.add_edge(START, "parent")
builder.add_edge("parent", END)

graph = builder.compile()

result = graph.invoke({
    "user_query": "What is machine learning?",
    "secret_context": "CONFIDENTIAL: internal budget $500k",
    "sub_result": "", "log": [],
})

print(f"Result: {result['sub_result']}")
print(f"Secret context intact: {result['secret_context'][:20]}...")
# Expected output:
# Result: Subagent sees keys: ['query', 'result']. Processed: 'What is machine learning?'
# Secret context intact: CONFIDENTIAL: intern...

Exercise 2: Three subagents in parallel (Medium)

Build a system where the parent launches 3 subagents in parallel: news_agent (searches news), social_agent (searches social media), and academic_agent (searches papers). Use ThreadPoolExecutor to run them simultaneously. Combine the results into a summary.

See solution

from dotenv import load_dotenv
load_dotenv()

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

class SubState(TypedDict):
    query: str
    source_type: str
    result: str

class ParentState(TypedDict):
    topic: str
    news: str
    social: str
    academic: str
    summary: str
    log: Annotated[list[str], operator.add]

def search_source(state: SubState) -> dict:
    return {"result": f"[{state['source_type'].upper()}] 5 results for '{state['query'][:25]}...'"}

sub_builder = StateGraph(SubState)
sub_builder.add_node("search", search_source)
sub_builder.add_edge(START, "search")
sub_builder.add_edge("search", END)
search_sub = sub_builder.compile()

def parallel_search(state: ParentState) -> dict:
    configs = [
        {"query": state["topic"], "source_type": "news", "result": ""},
        {"query": state["topic"], "source_type": "social", "result": ""},
        {"query": state["topic"], "source_type": "academic", "result": ""},
    ]

    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
        futures = {pool.submit(search_sub.invoke, c): c["source_type"] for c in configs}
        results = {}
        for f in concurrent.futures.as_completed(futures):
            name = futures[f]
            results[name] = f.result()["result"]

    return {
        "news": results["news"],
        "social": results["social"],
        "academic": results["academic"],
        "log": [f"parallel: {len(results)} sources completed"],
    }

def summarize(state: ParentState) -> dict:
    summary = (
        f"SUMMARY on '{state['topic']}':\n"
        f"  {state['news']}\n"
        f"  {state['social']}\n"
        f"  {state['academic']}"
    )
    return {"summary": summary, "log": ["summary generated"]}

builder = StateGraph(ParentState)
builder.add_node("search", parallel_search)
builder.add_node("summarize", summarize)
builder.add_edge(START, "search")
builder.add_edge("search", "summarize")
builder.add_edge("summarize", END)

graph = builder.compile()

result = graph.invoke({
    "topic": "LangGraph multi-agent patterns",
    "news": "", "social": "", "academic": "",
    "summary": "", "log": [],
})

print(result["summary"])
print(f"\nLog: {result['log']}")
# Expected output:
# SUMMARY on 'LangGraph multi-agent patterns':
#   [NEWS] 5 results for 'LangGraph multi-agent pat...'
#   [SOCIAL] 5 results for 'LangGraph multi-agent pat...'
#   [ACADEMIC] 5 results for 'LangGraph multi-agent pat...'
#
# Log: ['parallel: 3 sources completed', 'summary generated']

Exercise 3: Registry with dispatch and error handling (Medium)

Build a registry of 3 subagents (translator, summarizer, formatter). Implement a safe_dispatch function that invokes the subagent by name with try/except. The orchestrator must call all 3 in sequence, handling individual errors without aborting the whole flow.

See solution

from dotenv import load_dotenv
load_dotenv()

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

class SubState(TypedDict):
    input_text: str
    result: str

def translate(state: SubState) -> dict:
    return {"result": f"[FR] Translated: {state['input_text'][:30]}..."}

def summarize(state: SubState) -> dict:
    return {"result": f"[SUMMARY] Key points from: {state['input_text'][:30]}..."}

def format_output(state: SubState) -> dict:
    raise ValueError("Formatter temporarily unavailable")

def build_sub(fn):
    b = StateGraph(SubState)
    b.add_node("work", fn)
    b.add_edge(START, "work")
    b.add_edge("work", END)
    return b.compile()

REGISTRY = {
    "translator": build_sub(translate),
    "summarizer": build_sub(summarize),
    "formatter": build_sub(format_output),
}

def safe_dispatch(name: str, text: str) -> dict:
    try:
        result = REGISTRY[name].invoke({"input_text": text, "result": ""})
        return {"success": True, "agent": name, "result": result["result"]}
    except Exception as e:
        return {"success": False, "agent": name, "result": f"ERROR: {str(e)[:40]}"}

class OrcState(TypedDict):
    text: str
    results: list[dict]
    log: Annotated[list[str], operator.add]

def orchestrate(state: OrcState) -> dict:
    pipeline = ["translator", "summarizer", "formatter"]
    results = []
    current = state["text"]

    for agent_name in pipeline:
        outcome = safe_dispatch(agent_name, current)
        results.append(outcome)
        if outcome["success"]:
            current = outcome["result"]

    successes = [r["agent"] for r in results if r["success"]]
    failures = [r["agent"] for r in results if not r["success"]]

    return {
        "results": results,
        "log": [
            f"completed: {successes}",
            f"failed: {failures}",
        ],
    }

builder = StateGraph(OrcState)
builder.add_node("orchestrate", orchestrate)
builder.add_edge(START, "orchestrate")
builder.add_edge("orchestrate", END)

graph = builder.compile()

result = graph.invoke({"text": "AI agents are transforming the industry", "results": [], "log": []})

for r in result["results"]:
    status = "✅" if r["success"] else "❌"
    print(f"  {status} {r['agent']}: {r['result'][:50]}...")
print(f"\nLog: {result['log']}")
# Expected output:
#   ✅ translator: [FR] Translated: AI agents are transforming the...
#   ✅ summarizer: [SUMMARY] Key points from: [FR] Translated: AI age...
#   ❌ formatter: ERROR: Formatter temporarily unavailable...
#
# Log: ['completed: ['translator', 'summarizer']', "failed: ['formatter']"]

Exercise 4: Subagents with result aggregation (Advanced)

Build a "voting" system where 3 subagents analyze the same text independently and return a classification (positive, negative, neutral). The parent aggregates the votes and decides the final classification by majority. Run the subagents in parallel.

See solution

from dotenv import load_dotenv
load_dotenv()

import concurrent.futures
from collections import Counter
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class SubState(TypedDict):
    text: str
    analyst_id: int
    classification: str

class ParentState(TypedDict):
    text: str
    votes: list[str]
    final_classification: str
    confidence: float
    log: Annotated[list[str], operator.add]

ANALYST_BIASES = {
    1: ["innovation", "growth", "opportunity"],
    2: ["risk", "cost", "problem"],
    3: [],
}

def classify(state: SubState) -> dict:
    text_lower = state["text"].lower()
    analyst_id = state["analyst_id"]
    # These keywords are matched against the user's text, which is in English.
    # They are DATA, not identifiers: they must stay in sync with the sample texts.
    positive_kw = ["good", "success", "growth", "innovation", "improvement"]
    negative_kw = ["bad", "failure", "risk", "problem", "cost"]

    bias = ANALYST_BIASES.get(analyst_id, [])
    pos_score = sum(1 for w in positive_kw + bias if w in text_lower)
    neg_score = sum(1 for w in negative_kw + bias if w in text_lower)

    # The classification IS an internal enum: the parent aggregates it with Counter
    if pos_score > neg_score:
        return {"classification": "positive"}
    elif neg_score > pos_score:
        return {"classification": "negative"}
    return {"classification": "neutral"}

sub_builder = StateGraph(SubState)
sub_builder.add_node("classify", classify)
sub_builder.add_edge(START, "classify")
sub_builder.add_edge("classify", END)
classifier = sub_builder.compile()

def parallel_classify(state: ParentState) -> dict:
    configs = [
        {"text": state["text"], "analyst_id": i, "classification": ""}
        for i in range(1, 4)
    ]

    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
        futures = [pool.submit(classifier.invoke, c) for c in configs]
        votes = [f.result()["classification"] for f in futures]

    counts = Counter(votes)
    winner = counts.most_common(1)[0]
    confidence = winner[1] / len(votes)

    return {
        "votes": votes,
        "final_classification": winner[0],
        "confidence": confidence,
        "log": [
            f"votes: {votes}",
            f"result: {winner[0]} ({confidence:.0%} confidence)",
        ],
    }

builder = StateGraph(ParentState)
builder.add_node("classify", parallel_classify)
builder.add_edge(START, "classify")
builder.add_edge("classify", END)

graph = builder.compile()

texts = [
    "Great innovation in AI, 40% growth and continuous improvement",
    "High risk and cost, the problem persists with no clear solution",
    "The market remains stable with no significant changes",
]

for text in texts:
    result = graph.invoke({
        "text": text, "votes": [],
        "final_classification": "", "confidence": 0.0, "log": [],
    })
    print(f"Text: {text[:50]}...")
    print(f"  Votes: {result['votes']}")
    print(f"  Classification: {result['final_classification']} ({result['confidence']:.0%})\n")
# Expected output:
# Text: Great innovation in AI, 40% growth and continuous ...
#   Votes: ['positive', 'positive', 'positive']
#   Classification: positive (100%)
#
# Text: High risk and cost, the problem persists with no c...
#   Votes: ['negative', 'negative', 'negative']
#   Classification: negative (100%)
#
# Text: The market remains stable with no significant chan...
#   Votes: ['neutral', 'neutral', 'neutral']
#   Classification: neutral (100%)

Summary

In this capsule you learned:

Subagents work in an isolated context — they receive only the description of their task, not the parent's full conversation. That prevents the context bloat you get with handoffs and shared state
The basic pattern is wrapping an agent as a @tool — create_agent builds the subagent, @tool exposes it to the parent. The parent calls the tool, the subagent runs in its own context, and returns only the result
Compiled subgraphs give you more control — for subagents with complex internal flows, create a standalone StateGraph with its own SubState and invoke it from a parent node
Parallel execution comes naturally — because subagents have an isolated context, you can run several at once with ThreadPoolExecutor and no state conflicts
The single dispatch pattern scales better — one dispatch(agent_name, task) backed by a registry replaces N individual tools when you have many subagents
Error handling is the parent's job — use safe_invoke with try/except and timeouts so one failing subagent doesn't take the whole system down
Subagents vs handoffs: use subagents when you need context isolation and parallel execution; use handoffs when you need shared context in a sequential flow

Next capsule: State design in multi-agent — how to design the state shared between agents, handle conflicts, and decide what information they share vs what they keep isolated.

Further resources

LangChain — Subagents — Official documentation of the subagents pattern
LangGraph — Use Subgraphs — How to use compiled subgraphs as nodes
LangGraph — Functional API — @task and @entrypoint for subagents with automatic isolation
Context Engineering for Agents — Designing the flow of context between agents
LangGraph — Multi-Agent Systems — Overview of multi-agent patterns

Module 10 — LangChain & LangGraph: From Chains to Agents

← PreviousPattern Handoffs Next →Pattern Router