4. Pattern: Subagents

Overview

In the previous capsules you explored two multi-agent orchestration patterns. Supervisor: a central coordinator that decides which worker works. Handoffs: direct transfer between agents. Both have something in common that isn't immediately obvious: they share context. The supervisor sees everything the workers produce. In handoffs, each agent inherits the previous one's history. As the system grows, this becomes a serious problem.

Imagine a Research Agent that delegates a search to a worker. That worker needs to know what to search for — a query, maybe a couple of constraints. But in a shared-state scheme, it receives everything: the 47 previous messages, the results of earlier analyses, the supervisor's plans. 95% of that context is noise. It's not just inefficient — it eats tokens, confuses the LLM, and degrades answer quality. This is the context bloat problem.

Subagents solve this: the parent delegates a task to the child with only the necessary context. The child works in isolation — its own state, its own tools, its own flow. When it finishes, it returns a clean result to the parent. It doesn't know who called it, it doesn't know the full history, it has no access to the global state.

The Subagents Pattern

Three roles, one flow

┌─────────────────────────────────────────────────┐
│                  PARENT AGENT                   │
│                                                 │
│  1. Receives the complete task from the user    │
│  2. Decomposes it into sub-tasks                │
│  3. Prepares minimal context for each child     │
│  4. Delegates to the child(ren)                 │
│  5. Receives the result(s)                      │
│  6. Aggregates and synthesizes the final answer │
│                                                 │
│     ┌────────┐   ┌────────┐   ┌────────┐       │
│     │CHILD A │   │CHILD B │   │CHILD C │       │
│     │isolated│   │isolated│   │isolated│       │
│     │context │   │context │   │context │       │
│     └────────┘   └────────┘   └────────┘       │
└─────────────────────────────────────────────────┘

The flow, step by step:

1. The parent receives the task. "Research RAG, analyze the techniques, generate a report."
2. It decomposes. It decides it needs a researcher, an analyst, a writer.
3. It prepares minimal context. For the researcher: just the query. For the analyst: just the raw data. For the writer: just the key points.
4. Each child runs in isolation. The researcher doesn't know an analyst exists. The analyst doesn't know who found the data.
5. Each child returns a result. Structured data, not its full history.
6. The parent aggregates. It takes the results and composes the final answer.

The fundamental difference

Context Isolation

Why isolation matters

Context bloat isn't a theoretical problem. It's the #1 problem that destroys the quality of multi-agent systems in production:

Iteration 1:  Supervisor → researcher.  messages: 5.    ✓ Works well.
Iteration 2:  Supervisor → analyst.     messages: 12.   The analyst sees searches it doesn't need.
Iteration 3:  Supervisor → writer.      messages: 25.   It gets confused, loses focus.
Iteration 10: Re-research.              messages: 80+.  Context window at its limit.

What the child sees vs what the parent sees

PARENT STATE (complete):
  messages: [47 messages from the whole session]
  plan: ["research RAG", "analyze", "write"]
  iteration_count: 3
  quality_score: 0.6

CHILD STATE (researcher — only what's needed):
  messages: [HumanMessage("Find papers about RAG published in 2024-2025")]

CHILD STATE (analyst — only what's needed):
  messages: [HumanMessage("Analyze these 3 papers: [the researcher's data]")]

The researcher doesn't know there's a plan, doesn't know the quality_score, doesn't know it's iteration 3. It only knows: "find papers about RAG".

Three concrete benefits

1. Consistent quality. Each child operates with a clean context window. It doesn't matter if the parent is 100 iterations in — the child's quality doesn't degrade.

2. Controlled costs. Without isolation, every LLM call sends the whole history. With subagents, each child sends only its minimal context. In a system with 4 agents and 10 iterations, the difference can be 10x in tokens.

3. Clean debugging. When something fails, you know exactly what the child saw. You don't have to trace which part of 80 messages caused the confusion.

The cost of isolation

• The parent decides which context to give. If it omits relevant information, the child produces poor results.
• There's no serendipity. In shared state, one worker can discover useful information another exploits. With isolation, only what the parent passes explicitly flows through.
• More responsibility on the parent. Decomposing well, selecting the right context, aggregating coherently.

Implementation with LangGraph

Child agents

In M4 (capsule 07) you learned subgraphs. Here's the key difference: the child does not share state keys with the parent automatically. The parent explicitly controls what goes in and what comes out.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool

model = init_chat_model("openai:gpt-4.1-mini")

@tool
def search_web(query: str) -> str:
    """Search the web for information about a topic."""
    return f"Results for '{query}': RAG techniques include naive RAG, advanced RAG with re-ranking, and modular RAG."

class ResearcherState(TypedDict):
    messages: Annotated[list, add_messages]

research_tools = [search_web]
research_model = model.bind_tools(research_tools)
tools_by_name = {t.name: t for t in research_tools}

def researcher_reason(state: ResearcherState) -> dict:
    response = research_model.invoke(
        [SystemMessage(content="You are a researcher. Search for information. When you have enough, answer directly.")]
        + state["messages"]
    )
    return {"messages": [response]}

def researcher_tools(state: ResearcherState) -> dict:
    last = state["messages"][-1]
    return {"messages": [
        ToolMessage(content=str(tools_by_name[tc["name"]].invoke(tc["args"])), tool_call_id=tc["id"])
        for tc in last.tool_calls
    ]}

def researcher_route(state: ResearcherState) -> str:
    last = state["messages"][-1]
    return "tools" if hasattr(last, "tool_calls") and last.tool_calls else "done"

rg = StateGraph(ResearcherState)
rg.add_node("reason", researcher_reason)
rg.add_node("tools", researcher_tools)
rg.add_edge(START, "reason")
rg.add_conditional_edges("reason", researcher_route, {"tools": "tools", "done": END})
rg.add_edge("tools", "reason")
researcher_agent = rg.compile()

class AnalystState(TypedDict):
    messages: Annotated[list, add_messages]

def analyst_node(state: AnalystState) -> dict:
    return {"messages": [model.invoke(
        [SystemMessage(content="You are an analyst. Extract the key points, strengths/weaknesses, and a recommendation.")]
        + state["messages"]
    )]}

ag = StateGraph(AnalystState)
ag.add_node("analyze", analyst_node)
ag.add_edge(START, "analyze")
ag.add_edge("analyze", END)
analyst_agent = ag.compile()

The parent with controlled context

Here's the central piece. The parent uses wrapper functions that control exactly what context each child receives:

class OrchestratorState(TypedDict):
    messages: Annotated[list, add_messages]
    research_task: str
    research_result: str
    analysis_result: str

def decompose_task(state: OrchestratorState) -> dict:
    response = model.invoke([
        SystemMessage(content="Break this request into a concrete research task. ONLY the task."),
        HumanMessage(content=state["messages"][-1].content)
    ])
    return {"research_task": response.content}

def delegate_to_researcher(state: OrchestratorState) -> dict:
    result = researcher_agent.invoke({
        "messages": [HumanMessage(content=state["research_task"])]
    })
    return {"research_result": result["messages"][-1].content}

def delegate_to_analyst(state: OrchestratorState) -> dict:
    result = analyst_agent.invoke({
        "messages": [HumanMessage(content=f"Analyze this data:\n\n{state['research_result']}")]
    })
    return {"analysis_result": result["messages"][-1].content}

def synthesize(state: OrchestratorState) -> dict:
    response = model.invoke([
        SystemMessage(content="Generate a final answer based on the research and the analysis."),
        HumanMessage(content=f"Research:\n{state['research_result']}\n\nAnalysis:\n{state['analysis_result']}")
    ])
    return {"messages": [response]}

orchestrator = StateGraph(OrchestratorState)
orchestrator.add_node("decompose", decompose_task)
orchestrator.add_node("research", delegate_to_researcher)
orchestrator.add_node("analyze", delegate_to_analyst)
orchestrator.add_node("synthesize", synthesize)
orchestrator.add_edge(START, "decompose")
orchestrator.add_edge("decompose", "research")
orchestrator.add_edge("research", "analyze")
orchestrator.add_edge("analyze", "synthesize")
orchestrator.add_edge("synthesize", END)
agent = orchestrator.compile()

result = agent.invoke({
    "messages": [HumanMessage(content="What are the most effective RAG techniques?")],
    "research_task": "", "research_result": "", "analysis_result": "",
})
print(result["messages"][-1].content)

Look at delegate_to_researcher. It doesn't pass state to the child. It extracts research_task and creates a fresh HumanMessage. The researcher doesn't know an analyst exists, that there's an analysis_result, or how many iterations the parent has run.

Designing the Parent↔Child Interface

What to pass to the child

The golden rule: the minimum needed for the child to do its job unambiguously.

Two input strategies

# Strategy 1: A single message with the context embedded
result = child.invoke({"messages": [HumanMessage(content=f"Task: {task}\nContext: {ctx}")]})

# Strategy 2: System + Human (preferable with complex system prompts)
result = child.invoke({"messages": [
    SystemMessage(content=f"Context: {ctx}. Constraints: {constraints}"),
    HumanMessage(content=task)
]})

Structured returns and metadata

For rich results, parse the child's output with structured output. For debugging, capture metadata:

from pydantic import BaseModel
import time

class ResearchResult(BaseModel):
    summary: str
    sources: list[str]
    confidence: float

def delegate_with_structure(state: OrchestratorState) -> dict:
    start = time.time()
    result = researcher_agent.invoke({"messages": [HumanMessage(content=state["research_task"])]})
    elapsed = time.time() - start
    raw = result["messages"][-1].content

    parsed = model.with_structured_output(ResearchResult).invoke(
        f"Extract structured information:\n\n{raw}"
    )
    return {
        "research_result": parsed.summary,
        "research_confidence": parsed.confidence,
        "research_metadata": {"duration_s": round(elapsed, 2), "steps": len(result["messages"])},
    }

Parallel Subagents

When to parallelize

If two sub-tasks are independent — one's result doesn't depend on the other's — run their children in parallel:

SEQUENTIAL: ──researcher──▶──analyst──▶──writer──▶    = 15 sec
PARALLEL:   ──researcher──▶─┐
            ──analyst────▶──┤──writer──▶              = 10 sec

Implementation with asyncio

import asyncio

async def parallel_delegates(state: OrchestratorState) -> dict:
    research_result, analysis_result = await asyncio.gather(
        researcher_agent.ainvoke({"messages": [HumanMessage(content=state["research_task"])]}),
        analyst_agent.ainvoke({"messages": [HumanMessage(content=state.get("preliminary_data", ""))]}),
    )
    return {
        "research_result": research_result["messages"][-1].content,
        "analysis_result": analysis_result["messages"][-1].content,
    }

Dynamic fan-out

The parent decomposes into N sub-tasks and launches N children with asyncio.gather. The LLM generates the sub-tasks, and an independent child gets created for each one. The results get combined at the end. The same parallel_delegates pattern applies, but with a dynamic list instead of a fixed one.

Considerations

results = await asyncio.gather(*tasks, return_exceptions=True)
for i, r in enumerate(results):
    if isinstance(r, Exception):
        outputs[f"child_{i}"] = f"[ERROR]: {r}"
    else:
        outputs[f"child_{i}"] = r["messages"][-1].content

Subagents vs Supervisor vs Handoffs

When to use each

• Supervisor: you need dynamic re-routing; workers need to see other workers' results
• Handoffs: a linear A → B → C flow; low latency is the priority
• Subagents: independent sub-tasks; context bloat is a problem; isolated testing; 5+ agents

Composing patterns

In practice, systems combine patterns:

Supervisor (decides what to run)
    ├── Subagent: Research (isolated, reason-act loop)
    ├── Subagent: Analysis (isolated, internal handoffs)
    │       ├── extract → classify → evaluate
    └── Subagent: Writer (isolated)

Connection to the Project

In this module's project (capsule 08), you expand the Research Agent to 4 agents. Subagents are central:

• Supervisor as the parent that decomposes the task
• Researcher as a subagent — it receives only the query, returns data
• Analyst as a subagent — it receives only the data, returns an analysis
• Writer as a subagent — it receives only the key points, returns a report

Each subagent has its own StateGraph, its own tools (the researcher with web search via MCP, the writer with filesystem via MCP), and its own internal flow.

Troubleshooting

Problem 1: The child produces generic results

Cause: The context you pass it is too vague.

# Vague — the child doesn't know what to search for
result = researcher_agent.invoke({"messages": [HumanMessage(content="Research RAG")]})

# Specific — clear direction
result = researcher_agent.invoke({"messages": [HumanMessage(content=(
    "Find the 3 most effective RAG techniques according to 2024-2025 papers. "
    "Focus on re-ranking, hybrid search, query decomposition."
))]})

Problem 2: The parent loses information from the child

Cause: You only extract messages[-1].content, which can be a summary that loses detail.

Solution: Extract more of the output when you need to preserve detail:

all_content = "\n\n".join([
    msg.content for msg in result["messages"]
    if hasattr(msg, "content") and msg.content and not hasattr(msg, "tool_calls")
])

Problem 3: Parallel subagents hit rate limits

Solution: A semaphore to cap the concurrency:

semaphore = asyncio.Semaphore(2)

async def rate_limited_invoke(agent, input_data):
    async with semaphore:
        return await agent.ainvoke(input_data)

Problem 4: The child failed but the parent treats it as a success

Cause: There's no validation of the child's output.

Solution: Validate before using:

output = result["messages"][-1].content
if len(output) < 50 or output.lower().startswith("i'm sorry"):
    return {"research_result": "", "research_valid": False}
return {"research_result": output, "research_valid": True}

Problem 5: The subagents take too long

Solution: A timeout in the parent:

try:
    result = await asyncio.wait_for(
        researcher_agent.ainvoke({"messages": [HumanMessage(content=task)]}),
        timeout=60.0
    )
except asyncio.TimeoutError:
    return {"research_result": "[TIMEOUT] Didn't finish in 60s", "research_valid": False}

Exercises

Exercise 1: A child with minimal context (Easy)

Create a summarizer_agent (a node that summarizes text). The parent has long_text, conversation_history, and user_preferences. Delegate to the summarizer passing ONLY long_text.

class SummarizerState(TypedDict):
    messages: Annotated[list, add_messages]

def summarize_node(state: SummarizerState) -> dict:
    return {"messages": [model.invoke(
        [SystemMessage(content="Summarize this in at most 3 key points.")] + state["messages"]
    )]}

sg = StateGraph(SummarizerState)
sg.add_node("summarize", summarize_node)
sg.add_edge(START, "summarize")
sg.add_edge("summarize", END)
summarizer_agent = sg.compile()

class ParentState(TypedDict):
    messages: Annotated[list, add_messages]
    long_text: str
    conversation_history: list[str]
    user_preferences: dict
    summary: str

def delegate_to_summarizer(state: ParentState) -> dict:
    result = summarizer_agent.invoke({"messages": [HumanMessage(content=state["long_text"])]})
    return {"summary": result["messages"][-1].content}

parent = StateGraph(ParentState)
parent.add_node("collect", lambda s: {"long_text": "RAG combines search with generation. It includes re-ranking and self-RAG."})
parent.add_node("delegate", delegate_to_summarizer)
parent.add_edge(START, "collect")
parent.add_edge("collect", "delegate")
parent.add_edge("delegate", END)
agent = parent.compile()

result = agent.invoke({
    "messages": [], "long_text": "",
    "conversation_history": ["msg1", "msg2"], "user_preferences": {"lang": "en"}, "summary": "",
})
print(result["summary"])

Exercise 2: Two sequential subagents (Medium)

Create a fact_checker and a rewriter. The parent first delegates to the fact_checker (it takes text, returns corrections), then to the rewriter (it takes text + corrections, returns corrected text). Each child gets only its own context.

def make_agent(system_prompt):
    class S(TypedDict):
        messages: Annotated[list, add_messages]
    def node(state: S) -> dict:
        return {"messages": [model.invoke([SystemMessage(content=system_prompt)] + state["messages"])]}
    g = StateGraph(S)
    g.add_node("work", node)
    g.add_edge(START, "work")
    g.add_edge("work", END)
    return g.compile()

fact_checker = make_agent("Identify incorrect claims. Format: 'CLAIM → CORRECTION'")
rewriter = make_agent("Rewrite the text applying the corrections. Keep the tone.")

class EditorState(TypedDict):
    messages: Annotated[list, add_messages]
    original_text: str
    corrections: str
    final_text: str

def check(state: EditorState) -> dict:
    r = fact_checker.invoke({"messages": [HumanMessage(content=state["original_text"])]})
    return {"corrections": r["messages"][-1].content}

def rewrite(state: EditorState) -> dict:
    r = rewriter.invoke({"messages": [HumanMessage(
        content=f"Text:\n{state['original_text']}\n\nCorrections:\n{state['corrections']}"
    )]})
    return {"final_text": r["messages"][-1].content}

parent = StateGraph(EditorState)
parent.add_node("check", check)
parent.add_node("rewrite", rewrite)
parent.add_edge(START, "check")
parent.add_edge("check", "rewrite")
parent.add_edge("rewrite", END)
editor = parent.compile()

result = editor.invoke({
    "messages": [], "corrections": "", "final_text": "",
    "original_text": "Python was created in 1995 by James Gosling.",
})
print(result["final_text"])

Exercise 3: Parallel subagents with aggregation (Medium)

Create a pros_agent, a cons_agent, and an examples_agent. The parent delegates the same question to all three in parallel with asyncio.gather and aggregates the results into a synthesis.

import asyncio

pros_agent = make_agent("List ONLY advantages. 5 points maximum.")
cons_agent = make_agent("List ONLY disadvantages. 5 points maximum.")
examples_agent = make_agent("Give 3 concrete real-world examples.")

class DebateState(TypedDict):
    messages: Annotated[list, add_messages]
    topic: str
    pros: str
    cons: str
    examples: str

async def parallel_work(state: DebateState) -> dict:
    msg = {"messages": [HumanMessage(content=state["topic"])]}
    p, c, e = await asyncio.gather(
        pros_agent.ainvoke(msg), cons_agent.ainvoke(msg), examples_agent.ainvoke(msg)
    )
    return {
        "pros": p["messages"][-1].content,
        "cons": c["messages"][-1].content,
        "examples": e["messages"][-1].content,
    }

def aggregate(state: DebateState) -> dict:
    response = model.invoke([
        SystemMessage(content="Generate a balanced analysis combining the pros, cons and examples."),
        HumanMessage(content=f"Advantages:\n{state['pros']}\n\nDisadvantages:\n{state['cons']}\n\nExamples:\n{state['examples']}")
    ])
    return {"messages": [response]}

parent = StateGraph(DebateState)
parent.add_node("parallel", parallel_work)
parent.add_node("aggregate", aggregate)
parent.add_edge(START, "parallel")
parent.add_edge("parallel", "aggregate")
parent.add_edge("aggregate", END)
debater = parent.compile()

result = debater.invoke({
    "messages": [], "topic": "Multi-agent systems in production",
    "pros": "", "cons": "", "examples": "",
})
print(result["messages"][-1].content)

Exercise 4: Delegation with retry and validation (Hard)

Implement delegate_with_retry: invoke the child, validate the result (≥100 chars, doesn't start with "I'm sorry"), and retry with an improved prompt if it fails. 3 attempts maximum.

worker = make_agent("Complete the task with detail and precision.")

def validate_output(output: str) -> tuple[bool, str]:
    if len(output) < 100:
        return False, "The answer is too short"
    if any(output.lower().startswith(p) for p in ["i'm sorry", "i can't", "as a model"]):
        return False, "The agent refused the task"
    return True, ""

def delegate_with_retry(agent, task: str, max_retries: int = 3) -> dict:
    current_task = task
    for attempt in range(1, max_retries + 1):
        result = agent.invoke({"messages": [HumanMessage(content=current_task)]})
        output = result["messages"][-1].content
        is_valid, reason = validate_output(output)
        if is_valid:
            return {"result": output, "attempts": attempt, "success": True}
        current_task = f"The previous attempt was insufficient ({reason}). Original task: {task}\nBe more complete."
    return {"result": f"[FAILED] {max_retries} attempts.", "attempts": max_retries, "success": False}

out = delegate_with_retry(worker, "Explain the 3 main advanced RAG techniques")
print(f"Success: {out['success']}, Attempts: {out['attempts']}")

Exercise 5: A complete supervisor + subagents (Hard)

Build a supervisor that decomposes a question into 2 sub-questions, delegates each one to a researcher_subagent with isolated context (a reason-act loop, max 2 iterations), and produces a final answer.

import asyncio

@tool
def search_knowledge(query: str) -> str:
    """Search the knowledge base."""
    return f"Data about '{query}': relevant information found."

class RState(TypedDict):
    messages: Annotated[list, add_messages]
    iterations: int

model_t = model.bind_tools([search_knowledge])
t_map = {search_knowledge.name: search_knowledge}

def r_reason(state: RState) -> dict:
    return {
        "messages": [model_t.invoke(
            [SystemMessage(content="Research this. Answer when you have enough.")] + state["messages"]
        )], "iterations": state.get("iterations", 0) + 1,
    }

def r_tools(state: RState) -> dict:
    last = state["messages"][-1]
    return {"messages": [
        ToolMessage(content=str(t_map[tc["name"]].invoke(tc["args"])), tool_call_id=tc["id"])
        for tc in last.tool_calls
    ]}

def r_route(state: RState) -> str:
    if state.get("iterations", 0) >= 2: return "done"
    last = state["messages"][-1]
    return "tools" if hasattr(last, "tool_calls") and last.tool_calls else "done"

rsg = StateGraph(RState)
rsg.add_node("reason", r_reason)
rsg.add_node("tools", r_tools)
rsg.add_edge(START, "reason")
rsg.add_conditional_edges("reason", r_route, {"tools": "tools", "done": END})
rsg.add_edge("tools", "reason")
researcher_subagent = rsg.compile()

class SupervisorState(TypedDict):
    messages: Annotated[list, add_messages]
    sub_questions: list[str]
    sub_results: list[str]

def decompose(state: SupervisorState) -> dict:
    response = model.invoke([
        SystemMessage(content="Break this into 2 independent sub-questions. One per line."),
        state["messages"][-1],
    ])
    return {"sub_questions": [q.strip() for q in response.content.strip().split("\n") if q.strip()][:2]}

async def delegate_parallel(state: SupervisorState) -> dict:
    results = await asyncio.gather(*[
        researcher_subagent.ainvoke({"messages": [HumanMessage(content=q)], "iterations": 0})
        for q in state["sub_questions"]
    ], return_exceptions=True)
    return {"sub_results": [
        r["messages"][-1].content if not isinstance(r, Exception) else f"[ERROR]: {r}"
        for r in results
    ]}

def synthesize_final(state: SupervisorState) -> dict:
    parts = "\n\n".join([f"### {q}\n{r}" for q, r in zip(state["sub_questions"], state["sub_results"])])
    return {"messages": [model.invoke([
        SystemMessage(content="Synthesize a complete answer."), HumanMessage(content=parts),
    ])]}

sup = StateGraph(SupervisorState)
sup.add_node("decompose", decompose)
sup.add_node("delegate", delegate_parallel)
sup.add_node("synthesize", synthesize_final)
sup.add_edge(START, "decompose")
sup.add_edge("decompose", "delegate")
sup.add_edge("delegate", "synthesize")
sup.add_edge("synthesize", END)
supervisor = sup.compile()

result = supervisor.invoke({
    "messages": [HumanMessage(content="How do AI agents work and what tools do they use?")],
    "sub_questions": [], "sub_results": [],
})
print(result["messages"][-1].content[:300])

Summary

In this capsule you learned:

• The subagents pattern solves context bloat: the parent delegates with minimal context, each child runs in isolation, and returns only its result
• Context isolation keeps quality constant — the child doesn't see the parent's history, doesn't know who called it, doesn't know the global state. It doesn't matter how many iterations the system has run
• Implementation with LangGraph: each child is a compiled subgraph. The parent uses wrapper functions that explicitly control input and output — no automatic shared state
• The parent↔child interface is the key design decision: what context to give the child (the minimum) and what to extract from its result (the final output + optional metadata)
• Parallel subagents with asyncio.gather reduce latency when the children are independent. Dynamic fan-out allows N children for N sub-tasks
• Subagents vs Supervisor vs Handoffs: supervisor for dynamic coordination, handoffs for linear pipelines, subagents for decomposable tasks with isolation. In practice, they combine
• Debugging is easier than with other patterns — each child has explicit input and output, reproducible in isolation

Next capsule: The Router pattern — classify the user's input and direct it to the right specialized agent. Deterministic vs LLM-based routers, and how to combine a router with subagents for systems that scale.

Additional Resources

1. LangGraph Multi-Agent — Subgraph Pattern — Official documentation on subagents as subgraphs in multi-agent systems
2. LangGraph Subgraphs — How-to Guide — A tutorial for creating and composing subgraphs with state mapping
3. LangGraph Multi-Agent Tutorial — A supervisor-with-workers tutorial that includes subagent patterns
4. Multi-Agent Architectures (LangChain Blog) — Orchestration patterns and when to use each
5. Python asyncio.gather — The official asyncio documentation for parallel subagent execution