Sub-Agents
Decompose work across multiple specialized agents with a visible delegation log.
"""LangGraph agent backing the Sub-Agents demo.Demonstrates multi-agent delegation with a visible delegation log.A top-level "supervisor" LLM orchestrates three specialized sub-agents,exposed as tools: - `research_agent` — gathers facts - `writing_agent` — drafts prose - `critique_agent` — reviews draftsEach sub-agent is a full `create_agent(...)` under the hood. Everydelegation appends an entry to the `delegations` slot in shared agentstate so the UI can render a live "delegation log" as the supervisorfans work out and collects results. This is the canonical LangGraphsub-agents-as-tools pattern, adapted to surface delegation events tothe frontend via CopilotKit's shared-state channel.This is the FastAPI variant — the graph is exported and registered in`langgraph.json`. Identical agent topology to the langgraph-pythonreference; only the server framework differs."""import uuidfrom operator import addfrom typing import Annotated, Literal, TypedDictfrom langchain.agents import AgentState as BaseAgentState, create_agentfrom langchain.tools import ToolRuntime, toolfrom langchain_core.messages import HumanMessage, ToolMessagefrom langchain_openai import ChatOpenAIfrom langgraph.types import Commandfrom copilotkit import CopilotKitMiddleware# ---------------------------------------------------------------------------# Shared state# ---------------------------------------------------------------------------class Delegation(TypedDict): id: str sub_agent: Literal["research_agent", "writing_agent", "critique_agent"] task: str status: Literal["running", "completed", "failed"] result: strclass AgentState(BaseAgentState): """Shared state. `delegations` is rendered as a live log in the UI. `delegations` uses `operator.add` as its channel reducer so concurrent tool calls within a single supervisor turn each contribute their own entry. Without a reducer, parallel `tool_calls` would each read the same snapshot and the channel would last-write-wins, silently dropping every delegation but one from the UI log. """ delegations: Annotated[list[Delegation], add]# ---------------------------------------------------------------------------# Sub-agents (real LLM agents under the hood)# ---------------------------------------------------------------------------# Each sub-agent is a full-fledged `create_agent(...)` with its own# system prompt. They don't share memory or tools with the supervisor —# the supervisor only sees their return value._sub_model = ChatOpenAI(model="gpt-4o-mini")_research_agent = create_agent( model=_sub_model, tools=[], system_prompt=( "You are a research sub-agent. Given a topic, produce a concise " "bulleted list of 3-5 key facts. No preamble, no closing." ),)_writing_agent = create_agent( model=_sub_model, tools=[], system_prompt=( "You are a writing sub-agent. Given a brief and optional source " "facts, produce a polished 1-paragraph draft. Be clear and " "concrete. No preamble." ),)_critique_agent = create_agent( model=_sub_model, tools=[], system_prompt=( "You are an editorial critique sub-agent. Given a draft, give " "2-3 crisp, actionable critiques. No preamble." ),)def _invoke_sub_agent(agent, task: str) -> str: """Run a sub-agent on `task` and return its final message content.""" result = agent.invoke({"messages": [HumanMessage(content=task)]}) messages = result.get("messages", []) if not messages: return "" return str(messages[-1].content)def _delegation_command( sub_agent: str, task: str, status: Literal["completed", "failed"], result: str, tool_call_id: str,) -> Command: """Build a Command that appends a single new delegation entry. Emits ONLY the new entry under `delegations`. The channel reducer (`operator.add` on `AgentState.delegations`) extends the existing list, so parallel tool calls within one supervisor turn each contribute their own entry instead of clobbering each other via a last-write-wins read-modify-write. """ entry: Delegation = { "id": str(uuid.uuid4()), "sub_agent": sub_agent, # type: ignore[typeddict-item] "task": task, "status": status, "result": result, } return Command( update={ "delegations": [entry], "messages": [ ToolMessage( content=result, tool_call_id=tool_call_id, ) ], } )def _delegate( sub_agent_name: str, agent, task: str, tool_call_id: str,) -> Command: """Invoke a sub-agent and turn the outcome into a Command. Wrapped in try/except so that a sub-agent LLM failure (rate limit, transport error, missing API key, etc.) is recorded as a `failed` delegation entry and surfaced to the supervisor as a ToolMessage, instead of propagating and crashing the supervisor turn. The user-facing `result` is scrubbed to the exception class name only; full details are captured server-side via the standard logging path when the exception is re-raised at the caller's discretion (we do not re-raise here — recovery is the supervisor's job). """ try: result = _invoke_sub_agent(agent, task) return _delegation_command( sub_agent_name, task, "completed", result, tool_call_id ) except Exception as exc: # noqa: BLE001 - intentional broad catch # Keep the message generic; class name only, no exception args # (which can contain prompts, keys, or other sensitive data). message = ( f"sub-agent call failed: {exc.__class__.__name__} " f"(see server logs for details)" ) return _delegation_command( sub_agent_name, task, "failed", message, tool_call_id )# ---------------------------------------------------------------------------# Supervisor tools (each tool delegates to one sub-agent)# ---------------------------------------------------------------------------# Each @tool wraps a sub-agent invocation. The supervisor LLM "calls"# these tools to delegate work; each call synchronously runs the# matching sub-agent, records the delegation into shared state, and# returns the sub-agent's output as a ToolMessage the supervisor can# read on its next step.@tooldef research_agent(task: str, runtime: ToolRuntime) -> Command: """Delegate a research task to the research sub-agent. Use for: gathering facts, background, definitions, statistics. Returns a bulleted list of key facts. """ return _delegate("research_agent", _research_agent, task, runtime.tool_call_id)@tooldef writing_agent(task: str, runtime: ToolRuntime) -> Command: """Delegate a drafting task to the writing sub-agent. Use for: producing a polished paragraph, draft, or summary. Pass relevant facts from prior research inside `task`. """ return _delegate("writing_agent", _writing_agent, task, runtime.tool_call_id)@tooldef critique_agent(task: str, runtime: ToolRuntime) -> Command: """Delegate a critique task to the critique sub-agent. Use for: reviewing a draft and suggesting concrete improvements. """ return _delegate("critique_agent", _critique_agent, task, runtime.tool_call_id)# ---------------------------------------------------------------------------# Supervisor (the graph we export)# ---------------------------------------------------------------------------graph = create_agent( model=ChatOpenAI(model="gpt-4o-mini"), tools=[research_agent, writing_agent, critique_agent], middleware=[CopilotKitMiddleware()], state_schema=AgentState, system_prompt=( "You are a supervisor agent that coordinates three specialized " "sub-agents to produce high-quality deliverables.\n\n" "Available sub-agents (call them as tools):\n" " - research_agent: gathers facts on a topic.\n" " - writing_agent: turns facts + a brief into a polished draft.\n" " - critique_agent: reviews a draft and suggests improvements.\n\n" "For most non-trivial user requests, delegate in sequence: " "research -> write -> critique. Pass the relevant facts/draft " "through the `task` argument of each tool. Keep your own " "messages short — explain the plan once, delegate, then return " "a concise summary once done. The UI shows the user a live log " "of every sub-agent delegation." ),)What is this?#
Sub-agents are the canonical multi-agent pattern: a top-level supervisor LLM orchestrates one or more specialized sub-agents by exposing each of them as a tool. The supervisor decides what to delegate, the sub-agents do their narrow job, and their results flow back up to the supervisor's next step.
This is fundamentally the same shape as tool-calling, but each "tool" is itself a full-blown agent with its own system prompt and (often) its own tools, memory, and model.
When should I use this?#
Reach for sub-agents when a task has distinct specialized sub-tasks that each benefit from their own focus:
- Research → Write → Critique pipelines, where each stage needs a different system prompt and temperature.
- Router + specialists, where one agent classifies the request and dispatches to the right expert.
- Divide-and-conquer — any problem that fits cleanly into parallel or sequential sub-problems.
The example below uses the Research → Write → Critique shape as the canonical example.
Setting up sub-agents#
Install the LangGraph Python SDK
uv add copilotkitpoetry add copilotkitpip install copilotkit --extra-index-url https://copilotkit.gateway.scarf.sh/simple/conda install copilotkit -c copilotkit-channelWire CopilotKit middleware into your graph
Sub-agents are wired as tools on a supervisor create_agent call. The
supervisor's middleware list includes CopilotKitMiddleware() so the
delegation log slot streams back to the UI through shared state.
import operator
import uuid
from typing import Annotated, Literal, TypedDict
from langchain.agents import AgentState as BaseAgentState, create_agent
from langchain.tools import ToolRuntime, tool
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langgraph.types import Command
from copilotkit import CopilotKitMiddleware
# ---------------------------------------------------------------------------
# Shared state
# ---------------------------------------------------------------------------
class Delegation(TypedDict):
id: str
sub_agent: Literal["research_agent", "writing_agent", "critique_agent"]
task: str
status: Literal["completed"]
result: str
# Cap the supervisor → critique sub-agent loop at a single iteration.
# Without this, the supervisor LLM occasionally re-calls `critique_agent`
# repeatedly on the same draft (visible as stacking 🧐 cards in the
# chat). The critic only adds value once per draft, so we hard-stop
# after `_MAX_CRITIQUE_ITERATIONS` invocations and return a no-op
# result.
_MAX_CRITIQUE_ITERATIONS = 1
class AgentState(BaseAgentState):
"""Shared state. `delegations` is rendered as a live log in the UI.
`delegations` uses an `operator.add` reducer so that concurrent
sub-agent emissions in the same supervisor step accumulate into a
single list instead of conflicting (LangGraph would otherwise raise
`INVALID_CONCURRENT_GRAPH_UPDATE` — "Can receive only one value per
step. Use an Annotated key to handle multiple values.").
"""
delegations: Annotated[list[Delegation], operator.add]
# ---------------------------------------------------------------------------
# Sub-agents (real LLM agents under the hood)
# ---------------------------------------------------------------------------
# Each sub-agent is a full-fledged `create_agent(...)` with its own
# system prompt. They don't share memory or tools with the supervisor —
# the supervisor only sees their return value.
_sub_model = ChatOpenAI(model="gpt-5.4")
_research_agent = create_agent(
model=_sub_model,
tools=[],
system_prompt=(
"You are a research sub-agent. Given a topic, produce a concise "
"bulleted list of 3-5 key facts. No preamble, no closing."
),
)
_writing_agent = create_agent(
model=_sub_model,
tools=[],
system_prompt=(
"You are a writing sub-agent. Given a brief and optional source "
"facts, produce a polished 1-paragraph draft. Be clear and "
"concrete. No preamble."
),
)
_critique_agent = create_agent(
model=_sub_model,
tools=[],
system_prompt=(
"You are an editorial critique sub-agent. Given a draft, give "
"2-3 crisp, actionable critiques. No preamble."
),
)See src/agents/subagents.py for the canonical supervisor + three
sub-agents pattern with a live delegation log.
Each sub-agent is a full create_agent(...) call with its own model,
its own system prompt, and (optionally) its own tools. They don't share
memory or tools with the supervisor; the supervisor only ever sees
what the sub-agent returns.
import uuidfrom operator import addfrom typing import Annotated, Literal, TypedDictfrom langchain.agents import AgentState as BaseAgentState, create_agentfrom langchain.tools import ToolRuntime, toolfrom langchain_core.messages import HumanMessage, ToolMessagefrom langchain_openai import ChatOpenAIfrom langgraph.types import Commandfrom copilotkit import CopilotKitMiddleware# ---------------------------------------------------------------------------# Shared state# ---------------------------------------------------------------------------class Delegation(TypedDict): id: str sub_agent: Literal["research_agent", "writing_agent", "critique_agent"] task: str status: Literal["running", "completed", "failed"] result: strclass AgentState(BaseAgentState): """Shared state. `delegations` is rendered as a live log in the UI. `delegations` uses `operator.add` as its channel reducer so concurrent tool calls within a single supervisor turn each contribute their own entry. Without a reducer, parallel `tool_calls` would each read the same snapshot and the channel would last-write-wins, silently dropping every delegation but one from the UI log. """ delegations: Annotated[list[Delegation], add]# ---------------------------------------------------------------------------# Sub-agents (real LLM agents under the hood)# ---------------------------------------------------------------------------# Each sub-agent is a full-fledged `create_agent(...)` with its own# system prompt. They don't share memory or tools with the supervisor —# the supervisor only sees their return value._sub_model = ChatOpenAI(model="gpt-4o-mini")_research_agent = create_agent( model=_sub_model, tools=[], system_prompt=( "You are a research sub-agent. Given a topic, produce a concise " "bulleted list of 3-5 key facts. No preamble, no closing." ),)_writing_agent = create_agent( model=_sub_model, tools=[], system_prompt=( "You are a writing sub-agent. Given a brief and optional source " "facts, produce a polished 1-paragraph draft. Be clear and " "concrete. No preamble." ),)_critique_agent = create_agent( model=_sub_model, tools=[], system_prompt=( "You are an editorial critique sub-agent. Given a draft, give " "2-3 crisp, actionable critiques. No preamble." ),)Keep sub-agent system prompts narrow and focused. The point of this pattern is that each one does one thing well. If a sub-agent needs to know the whole user context to do its job, that's a signal the boundary is wrong.
Exposing sub-agents as tools#
The supervisor delegates by calling tools. Each tool is a thin wrapper
around sub_agent.invoke(...) that:
- Runs the sub-agent synchronously on the supplied
taskstring. - Records the delegation into a
delegationsslot in shared agent state (so the UI can render a live log). - Returns the sub-agent's final message as a
ToolMessage, which the supervisor sees as a normal tool result on its next turn.
import uuidfrom operator import addfrom typing import Annotated, Literal, TypedDictfrom langchain.agents import AgentState as BaseAgentState, create_agentfrom langchain.tools import ToolRuntime, toolfrom langchain_core.messages import HumanMessage, ToolMessagefrom langchain_openai import ChatOpenAIfrom langgraph.types import Commandfrom copilotkit import CopilotKitMiddleware# ---------------------------------------------------------------------------# Shared state# ---------------------------------------------------------------------------class Delegation(TypedDict): id: str sub_agent: Literal["research_agent", "writing_agent", "critique_agent"] task: str status: Literal["running", "completed", "failed"] result: strclass AgentState(BaseAgentState): """Shared state. `delegations` is rendered as a live log in the UI. `delegations` uses `operator.add` as its channel reducer so concurrent tool calls within a single supervisor turn each contribute their own entry. Without a reducer, parallel `tool_calls` would each read the same snapshot and the channel would last-write-wins, silently dropping every delegation but one from the UI log. """ delegations: Annotated[list[Delegation], add]# ---------------------------------------------------------------------------# Sub-agents (real LLM agents under the hood)# ---------------------------------------------------------------------------# Each sub-agent is a full-fledged `create_agent(...)` with its own# system prompt. They don't share memory or tools with the supervisor —# the supervisor only sees their return value._sub_model = ChatOpenAI(model="gpt-4o-mini")_research_agent = create_agent( model=_sub_model, tools=[], system_prompt=( "You are a research sub-agent. Given a topic, produce a concise " "bulleted list of 3-5 key facts. No preamble, no closing." ),)_writing_agent = create_agent( model=_sub_model, tools=[], system_prompt=( "You are a writing sub-agent. Given a brief and optional source " "facts, produce a polished 1-paragraph draft. Be clear and " "concrete. No preamble." ),)_critique_agent = create_agent( model=_sub_model, tools=[], system_prompt=( "You are an editorial critique sub-agent. Given a draft, give " "2-3 crisp, actionable critiques. No preamble." ),)def _invoke_sub_agent(agent, task: str) -> str: """Run a sub-agent on `task` and return its final message content.""" result = agent.invoke({"messages": [HumanMessage(content=task)]}) messages = result.get("messages", []) if not messages: return "" return str(messages[-1].content)def _delegation_command( sub_agent: str, task: str, status: Literal["completed", "failed"], result: str, tool_call_id: str,) -> Command: """Build a Command that appends a single new delegation entry. Emits ONLY the new entry under `delegations`. The channel reducer (`operator.add` on `AgentState.delegations`) extends the existing list, so parallel tool calls within one supervisor turn each contribute their own entry instead of clobbering each other via a last-write-wins read-modify-write. """ entry: Delegation = { "id": str(uuid.uuid4()), "sub_agent": sub_agent, # type: ignore[typeddict-item] "task": task, "status": status, "result": result, } return Command( update={ "delegations": [entry], "messages": [ ToolMessage( content=result, tool_call_id=tool_call_id, ) ], } )def _delegate( sub_agent_name: str, agent, task: str, tool_call_id: str,) -> Command: """Invoke a sub-agent and turn the outcome into a Command. Wrapped in try/except so that a sub-agent LLM failure (rate limit, transport error, missing API key, etc.) is recorded as a `failed` delegation entry and surfaced to the supervisor as a ToolMessage, instead of propagating and crashing the supervisor turn. The user-facing `result` is scrubbed to the exception class name only; full details are captured server-side via the standard logging path when the exception is re-raised at the caller's discretion (we do not re-raise here — recovery is the supervisor's job). """ try: result = _invoke_sub_agent(agent, task) return _delegation_command( sub_agent_name, task, "completed", result, tool_call_id ) except Exception as exc: # noqa: BLE001 - intentional broad catch # Keep the message generic; class name only, no exception args # (which can contain prompts, keys, or other sensitive data). message = ( f"sub-agent call failed: {exc.__class__.__name__} " f"(see server logs for details)" ) return _delegation_command( sub_agent_name, task, "failed", message, tool_call_id )# ---------------------------------------------------------------------------# Supervisor tools (each tool delegates to one sub-agent)# ---------------------------------------------------------------------------# Each @tool wraps a sub-agent invocation. The supervisor LLM "calls"# these tools to delegate work; each call synchronously runs the# matching sub-agent, records the delegation into shared state, and# returns the sub-agent's output as a ToolMessage the supervisor can# read on its next step.@tooldef research_agent(task: str, runtime: ToolRuntime) -> Command: """Delegate a research task to the research sub-agent. Use for: gathering facts, background, definitions, statistics. Returns a bulleted list of key facts. """ return _delegate("research_agent", _research_agent, task, runtime.tool_call_id)@tooldef writing_agent(task: str, runtime: ToolRuntime) -> Command: """Delegate a drafting task to the writing sub-agent. Use for: producing a polished paragraph, draft, or summary. Pass relevant facts from prior research inside `task`. """ return _delegate("writing_agent", _writing_agent, task, runtime.tool_call_id)@tooldef critique_agent(task: str, runtime: ToolRuntime) -> Command: """Delegate a critique task to the critique sub-agent. Use for: reviewing a draft and suggesting concrete improvements. """ return _delegate("critique_agent", _critique_agent, task, runtime.tool_call_id)This is where CopilotKit's shared-state channel earns its keep: the
supervisor's tool calls mutate delegations as they happen, and the
frontend renders every new entry live.
Rendering a live delegation log#
On the frontend, the delegation log is just a reactive render of the
delegations slot. Subscribe with useAgent({ updates: [UseAgentUpdate.OnStateChanged, UseAgentUpdate.OnRunStatusChanged] }),
read agent.state.delegations, and render one card per entry.
/** * Live delegation log — renders the `delegations` slot of agent state. * * Each entry corresponds to one invocation of a sub-agent. The list * grows in real time as the supervisor fans work out to its children. * The parent header shows how many sub-agents have been called and * whether the supervisor is still running. */export function DelegationLog({ delegations, isRunning }: DelegationLogProps) { return ( <div data-testid="delegation-log" className="w-full h-full flex flex-col bg-white rounded-2xl shadow-sm border border-[#DBDBE5] overflow-hidden" > <div className="flex items-center justify-between px-6 py-3 border-b border-[#E9E9EF] bg-[#FAFAFC]"> <div className="flex items-center gap-3"> <span className="text-lg font-semibold text-[#010507]"> Sub-agent delegations </span> {isRunning && ( <span data-testid="supervisor-running" className="inline-flex items-center gap-1.5 px-2 py-0.5 rounded-full border border-[#BEC2FF] bg-[#BEC2FF1A] text-[#010507] text-[10px] font-semibold uppercase tracking-[0.12em]" > <span className="w-1.5 h-1.5 rounded-full bg-[#010507] animate-pulse" /> Supervisor running </span> )} </div> <span data-testid="delegation-count" className="text-xs font-mono text-[#838389]" > {delegations.length} calls </span> </div> <div className="flex-1 overflow-y-auto p-4 space-y-3"> {delegations.length === 0 ? ( <p className="text-[#838389] italic text-sm"> Ask the supervisor to complete a task. Every sub-agent it calls will appear here. </p> ) : ( delegations.map((d, idx) => { const style = SUB_AGENT_STYLE[d.sub_agent]; return ( <div key={d.id} data-testid="delegation-entry" className="border border-[#E9E9EF] rounded-xl p-3 bg-[#FAFAFC]" > <div className="flex items-center justify-between mb-2"> <div className="flex items-center gap-2"> <span className="text-xs font-mono text-[#AFAFB7]"> #{idx + 1} </span> <span className={`inline-flex items-center gap-1 px-2 py-0.5 rounded-full text-[10px] font-semibold uppercase tracking-[0.1em] border ${style.color}`} > <span>{style.emoji}</span> <span>{style.label}</span> </span> </div> <span className="text-[10px] uppercase tracking-[0.12em] font-semibold text-[#189370]"> {d.status} </span> </div> <div className="text-xs text-[#57575B] mb-2"> <span className="font-semibold text-[#010507]">Task: </span> {d.task} </div> <div className="text-sm text-[#010507] whitespace-pre-wrap bg-white rounded-lg p-2.5 border border-[#E9E9EF]"> {d.result} </div> </div> ); }) )} </div> </div> );}The result: as the supervisor fans work out to its sub-agents, the log grows in real time, giving the user visibility into a process that would otherwise be a long opaque spinner.
Related#
- Shared State — the channel that makes the delegation log live.
- State streaming — stream individual sub-agent outputs token-by-token inside each log entry.
