Subgraphs

Subgraphs let you compose larger agents from smaller, focused units. A compiled StateGraph becomes a node in a parent graph, and injectAgent() streams the whole composition through the same message, state, tool-call, and custom-event signals as a single graph. The running example is a research orchestrator: the parent decides per turn whether to enter the child at all, and this guide walks the three files that make it work.

What the demo does

The Run tab shows the prebuilt <chat> composition beside a sidebar that reports which branch the parent took. Ask a factual question and the sidebar switches to "Nested — research subgraph ran", printing the topic the parent handed to the child and the brief the child handed back, followed by the child's own stream and its status. Ask a greeting and the sidebar reports "Direct — subgraph skipped" instead, because the parent answered without entering the child.

Two welcome suggestions set both branches up. "Ask something that needs research" sends a question about LangGraph checkpointing, which routes through the child. "Ask something that does not" sends a greeting, which does not. In both cases the answer in the transcript is written by the parent: the brief renders in the sidebar and nowhere else.

How it is built

Three files carry the feature: a graph whose child is a separately compiled graph, a one-file typed agent ref, and a component that reads the two keys the parent and child share. Open the Code tab to read them in place.

The state boundary

Adding a compiled graph as a node does not isolate state by itself. The boundary is designed, and it is designed in the two state schemas: LangGraph wires a subgraph node through the keys the two schemas have in common.

class ResearchState(TypedDict):
    """Child graph state — deliberately has no `messages` key.
 
    The only keys here are the ones shared with the parent, which is exactly
    the contract LangGraph uses to pass state into and out of a subgraph node.
    """
 
    research_topic: str
    research_brief: str
 
 
class OrchestratorState(TypedDict):
    """Parent graph state — the transcript plus the shared subgraph channel."""
 
    messages: Annotated[list, add_messages]
    research_topic: str
    research_brief: str
    # Final-answer identity binds the retained boundary to one original turn.
    completed_turn_id: NotRequired[str]
    completed_answer_id: NotRequired[str]

ResearchState has no messages key, so the child can neither read the transcript nor append to it: the two keys it does share are the entire interface, a topic in and a brief out.

The child graph

The child is an ordinary graph. One node, one model call, and a compile() at the end that produces the object the parent will mount.

# ── Child: research subgraph ──────────────────────────────────────────────
 
async def research_node(state: ResearchState) -> dict:
    """Turn a topic into an internal brief. No transcript access."""
    response = await researcher.ainvoke(
        [
            SystemMessage(content=RESEARCH_PROMPT),
            HumanMessage(content=f"Topic: {state['research_topic']}"),
        ]
    )
    text = response.content
    if isinstance(text, list):  # multi-part content
        text = "".join(part.get("text", "") for part in text if isinstance(part, dict))
    return {"research_brief": str(text).strip()}
 
research_graph = StateGraph(ResearchState)
research_graph.add_node("research", research_node)
research_graph.add_edge(START, "research")
research_graph.add_edge("research", END)
compiled_research = research_graph.compile()

The module-level RESEARCH_PROMPT (not shown) restates the structural fact in words, telling the researcher that it cannot see the chat transcript and that its output is an internal brief for the parent to use.

Deciding whether to delegate

The parent's first node classifies the turn with a structured-output call and writes a topic when research is warranted. Writing that topic is the delegation: the conditional edge routes on nothing else.

async def orchestrate_node(state: OrchestratorState) -> dict:
    """Classify the request. Writing a topic is what triggers delegation.
 
    Both shared keys are reset every turn so a topic left over from an
    earlier turn in the same thread can't re-trigger the subgraph.
    """
    decision = await router.ainvoke(
        [SystemMessage(content=ROUTER_PROMPT), *state["messages"]]
    )
    topic = decision.topic.strip() if decision.needs_research else ""
    return {"research_topic": topic, "research_brief": ""}
 
def route_after_orchestrate(state: OrchestratorState) -> str:
    """The parent's decision: enter the child graph, or skip it."""
    return "research" if state.get("research_topic") else "answer"

Both shared keys are reset on every turn, so a topic left over from an earlier turn in the same thread cannot re-trigger the child.

Writing the answer

One node writes to the transcript. It reads the brief out of state when there is one, folds it into the system context, and streams the user-facing turn.

async def answer_node(state: OrchestratorState) -> dict:
    """The only node that writes to the transcript.
 
    `transcriptNodeNames: ['answer']` on the Angular side mirrors this:
    the router's and the subgraph's tokens never reach the chat UI.
    """
    system_prompt = (PROMPTS_DIR / "subgraphs.md").read_text()
    brief = state.get("research_brief") or ""
    context = (
        [SystemMessage(content=f"Internal brief returned by the research subgraph:\n{brief}")]
        if brief
        else []
    )
    response = await llm.ainvoke(
        [SystemMessage(content=system_prompt), *context, *state["messages"]]
    )
    question = next(
        (message for message in reversed(state["messages"])
         if getattr(message, "type", None) == "human"),
        None,
    )
    human_id = getattr(question, "id", None)
    answer_id = getattr(response, "id", None)
    completion = {}
    if (isinstance(human_id, str) and human_id
            and isinstance(answer_id, str) and answer_id
            and human_id != answer_id):
        completion = {
            "completed_turn_id": human_id,
            "completed_answer_id": answer_id,
        }
    return {"messages": [response], **completion}

The brief arrives here as context rather than as a chat message, so what the user reads is the parent's own prose.

Adding the child as a node

The compiled child graph is passed straight to add_node. There is no wrapper function, and that is what makes it a subgraph rather than an inline helper call.

parent_graph = StateGraph(OrchestratorState)
parent_graph.add_node("orchestrate", orchestrate_node)
# The compiled child graph IS the node — no wrapper function. This is what
# makes it a subgraph rather than an inline helper call.
parent_graph.add_node("research", compiled_research)
parent_graph.add_node("answer", answer_node)
parent_graph.add_edge(START, "orchestrate")
parent_graph.add_conditional_edges(
    "orchestrate",
    route_after_orchestrate,
    {"research": "research", "answer": "answer"},
)
parent_graph.add_edge("research", "answer")
parent_graph.add_edge("answer", END)
return parent_graph.compile()

The child runs as its own graph with its own step sequence, and LangGraph emits its stream events under a research:<uuid> namespace rather than flattening them into the parent's.

The typed agent ref

The Angular side declares the parent's state once and hands it to a ref that both the provider and the component use. Because SubgraphsState names the two shared keys, agent.value() is typed at every read site.

import { createAgentRef } from '@threadplane/chat';
 
/**
 * Parent graph state for the subgraphs example.
 *
 * `research_topic` and `research_brief` are the two keys the parent shares
 * with the compiled child graph — writing a topic is what routes execution
 * into the subgraph, and the brief is what comes back out. The child's own
 * state has no `messages` key, which is why nothing it produces reaches the
 * transcript.
 */
export interface SubgraphsState {
  messages: unknown[];
  research_topic: string;
  research_brief: string;
}
 
/**
 * Typed DI handle for the subgraphs agent.
 * Wire with `provideAgent(SUBGRAPHS_AGENT, () => ...)` and inject with
 * `injectAgent(SUBGRAPHS_AGENT)` to get `LangGraphAgent<SubgraphsState>`.
 */
export const SUBGRAPHS_AGENT = createAgentRef<SubgraphsState>('subgraphs');

The application config registers that ref with provideAgent().

The running example registers the provider through a factory that reads its connection from the host that serves the demo, which is how the same build runs locally and in production. Your own application passes the two connection values as literals, and keeps transcriptNodeNames exactly as the example sets it:

provideAgent(SUBGRAPHS_AGENT, {
  apiUrl: 'https://your-deployment.langgraph.app',
  assistantId: 'subgraphs',
  transcriptNodeNames: ['answer'],
}),

assistantId must match the graph name in langgraph.json. transcriptNodeNames whitelists the top-level nodes whose message tokens belong in the chat transcript, which here means the answer node alone. The parent's other node runs a structured-output call of its own, and the whitelist is what keeps that traffic out of the chat.

Reading the boundary from Angular

injectAgent(SUBGRAPHS_AGENT) returns LangGraphAgent<SubgraphsState>, so agent.value() is a typed signal over the parent graph's live state as its values events arrive. The two shared keys are read straight off it, and a non-empty topic doubles as the "did the run nest?" signal because it is exactly what the conditional edge routes on.

/**
 * Typed agent — `agent.value()` is `Signal<SubgraphsState>`, the parent
 * graph's live state as LangGraph streams `values` events.
 */
protected readonly agent = injectAgent(SUBGRAPHS_AGENT);
 
/** The topic the parent handed to the child graph this turn, if any. */
protected readonly topic = computed(() => this.agent.value()?.research_topic ?? '');
 
/** The brief the child graph handed back. */
protected readonly brief = computed(() => this.agent.value()?.research_brief ?? '');
 
/**
 * A non-empty topic is exactly what the parent's conditional edge routes on,
 * so it doubles as the UI's "did we nest?" signal.
 */
protected readonly delegated = computed(() => this.topic().length > 0);

The sidebar renders topic() and brief() under the route line, so watching those two fields is watching the state boundary itself.

The same child, seen as a stream

agent.value() is one view of the child. agent.subagents() is the other: a Map of every namespaced child run, which for a plain subgraph node is keyed by the namespace segment and named by the node. No configuration turns it on — the entry appears on the child's first streamed event and settles with the run.

/**
 * The same child, seen as a stream. Plain subgraph children appear in
 * `subagents()` keyed by their namespace segment; `name` is the node name
 * and `status` settles with the run.
 */
protected readonly childStreams = computed(() =>
  [...this.agent.subagents().entries()].map(([id, ref]) => ({
    id,
    name: ref.name,
    status: ref.status(),
  })),
);

status() and name come off each Subagent entry in the subagents() map, which is why the mapped objects carry a called signal rather than the entry itself.

Tracking delegated subagent execution

The subagents() signal contains a Map of active child streams. Tool-dispatched children — Deep Agents' default task tool or your own delegation tools — are keyed by tool-call id and named by their subagent_type. Plain subgraph nodes, as in the example above, are keyed by their namespace segment and named by node; they register on their first streamed event and settle with the run.

Nested delegation — a subagent that itself dispatches a delegation tool — surfaces as its own entry too, keyed by its namespace path (truncated at the innermost delegation segment) and treated like a plain subgraph stream. Each level of delegation gets its own stream; the map stays flat, so there is no parent/child linking between the entries.

Tool-dispatched children need one option the example does not set, because the example has no delegation tool:

// app.config.ts
provideAgent(ORCHESTRATOR, {
  apiUrl: 'https://your-deployment.langgraph.app',
  assistantId: 'orchestrator',
  subagentToolNames: ['task', 'delegate_to_researcher'],
}),

ORCHESTRATOR and PIPELINE here stand for agent refs created the same way as SUBGRAPHS_AGENT.

With that in place, the same lookups work on either flavor:

const orchestrator = injectAgent(ORCHESTRATOR);
 
// Only the active ones
const running = computed(() =>
  [...orchestrator.subagents().values()].filter(
    (subagent) => subagent.status() === 'pending' || subagent.status() === 'running'
  )
);
 
// Lookup helpers for common UI paths
const specific = computed(() => orchestrator.getSubagent('research-tool-call-id'));
const researchers = computed(() => orchestrator.getSubagentsByType('researcher'));
 
// React to count changes
effect(() => {
  console.log(`${running().length} subagents currently running`);
});

Subagent stream details

Each SubagentStreamRef exposes its own reactive signals — status, messages, and values — so you can surface granular progress in your UI.

// Access a specific subagent by its tool call ID
const researchAgent = computed(() =>
  orchestrator.getSubagent('research-tool-call-id')
);
 
// Or get the subagents spawned by a specific AI message with tool calls
const messageAgents = computed(() => {
  const message = selectedAiMessage();
  return message ? orchestrator.getSubagentsByMessage(message) : [];
});
 
// Track its progress
const researchStatus = computed(() => researchAgent()?.status());
const researchMessages = computed(() => researchAgent()?.messages() ?? []);

How child streams get matched to tool calls

A child graph invoked inside a @tool body streams under a tools:<uuid> namespace — and that uuid is a checkpoint id, assigned independently of the tool-call id. Nothing on the wire links the two. injectAgent() bridges the gap in three tiers, in order of preference:

1. A server-announced binding (exact, works under any concurrency — see below)
2. The description ladder: the child's first human message is compared against each pending tool call's description argument — exact match, then substring in either direction
3. A positional fallback that fires only when exactly one tool child is outstanding — with several in flight, arrival order is not dispatch order, and guessing would cross-wire the cards, so ambiguous streams stay buffered instead

Tier 3 covers the common sequential shape (one dispatch per assistant turn). For parallel fan-out, or a delegation tool whose argument is not named description, announce the binding from the server — the tool body is the one place both halves are known:

from typing import Annotated
 
from langchain_core.runnables import RunnableConfig
from langchain_core.tools import InjectedToolCallId, tool
from threadplane.middleware.langgraph import announce_subagent
 
 
@tool
async def task(
    description: str,
    tool_call_id: Annotated[str, InjectedToolCallId] = None,
    config: RunnableConfig = None,
) -> str:
    announce_subagent(config, tool_call_id)  # one line — before invoking the child
    result = await child_graph.ainvoke({"messages": [("user", description)]})
    return result["messages"][-1].content

announce_subagent (threadplane-middleware ≥ 0.0.2) emits one custom event pairing the config's checkpoint_ns with the injected tool-call id. injectAgent() consumes it, attributes the stream exactly — replaying any chunks that arrived before the announcement — and never lets it override an established mapping. It returns False instead of raising when anything it needs is unavailable (outside a run, no namespace, no id), so it needs no guarding.

Orchestrator pattern

The example delegates one kind of work to one child. The same shape scales to several: each child runs its own graph while the parent coordinates, and a derived summary over subagents() gives you the fan-out at a glance.

const pipeline = injectAgent(PIPELINE);
 
// Derive a summary of all subagent states
const pipelineStatus = computed(() => {
  const entries = [...pipeline.subagents().entries()];
 
  return {
    total: entries.length,
    pending: entries.filter(([, a]) => a.status() === 'pending').length,
    running: entries.filter(([, a]) => a.status() === 'running').length,
    done: entries.filter(([, a]) => a.status() === 'complete').length,
    failed: entries.filter(([, a]) => a.status() === 'error').length,
  };
});

Child messages and the parent transcript

Child messages never appear in the parent's messages() signal — a namespaced stream belongs to its child, and messages() is the parent's transcript. That is a classification rule, not a heuristic: any event carrying a namespace is child content.

What the transcript shows once the run settles is decided by state instead. A child that shares the parent's messages key writes into the parent's message list, and that list arrives with the authoritative values sync at the end of the run. The example's child has no messages key, so nothing it produces can ever land there. Render a child's live output from its own stream, as the example does with subagents().

Streamed chunks from top-level side-effect nodes — a router, a title generator — are a separate concern, and the one transcriptNodeNames exists for.

Error handling per subagent

Each subagent exposes its own status() signal. A failure changes that subagent's status to 'error' without necessarily stopping sibling delegates.

// Collect all failed subagents reactively
const failedAgents = computed(() =>
  [...orchestrator.subagents().entries()].filter(
    ([, agent]) => agent.status() === 'error'
  )
);
 
// One effect over the derived list — it re-runs as subagents appear and fail.
effect(() => {
  for (const [id] of failedAgents()) {
    console.error(`Subagent ${id} failed`);
    // Retry, surface to user, or fall back gracefully
  }
});

Derive the list first, then react to it. Looping over a subagents() snapshot to create one effect() per entry does not work: the read happens outside a reactive context so it never re-runs, subagents that appear later never get an effect, and effect() needs an injection context.

When to use subagents vs a single agent

What's Next