Subgraphs
Subgraphs let you compose larger agents from smaller, focused units. injectAgent() streams their output through the same message, state, tool-call, and custom-event signals as the parent graph.
LangGraph subgraphs are graph nodes. Deep Agents-style subagents are delegated tool calls. injectAgent() requests subgraph streams by default, and every namespaced child run appears in the subagents() signal — tool-dispatched children under their tool-call id (matched via subagentToolNames + subagent_type), plain subgraph nodes under their namespace key, named by node. A child's tokens live on its stream and never merge into the parent transcript.
How subgraph composition works
Subgraph composition starts on the agent side. Each subgraph is a fully compiled StateGraph that can be added as a node in a parent graph.
from langgraph.graph import END, START, MessagesState, StateGraph
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-5-mini")
# --- Research subgraph ---
def search_web(state: MessagesState) -> dict:
query = state["messages"][-1].content
results = web_search(query)
return {"messages": [{"role": "assistant", "content": results}]}
def summarize_results(state: MessagesState) -> dict:
response = llm.invoke(state["messages"])
return {"messages": [response]}
research_builder = StateGraph(MessagesState)
research_builder.add_node("search", search_web)
research_builder.add_node("summarize", summarize_results)
research_builder.add_edge(START, "search")
research_builder.add_edge("search", "summarize")
research_builder.add_edge("summarize", END)
research_subgraph = research_builder.compile()
# --- Analysis subgraph ---
def analyze_data(state: MessagesState) -> dict:
response = llm.invoke([
{"role": "system", "content": "Analyze the data and provide insights."},
*state["messages"],
])
return {"messages": [response]}
analysis_builder = StateGraph(MessagesState)
analysis_builder.add_node("analyze", analyze_data)
analysis_builder.add_edge(START, "analyze")
analysis_builder.add_edge("analyze", END)
analysis_subgraph = analysis_builder.compile()
# --- Parent orchestrator ---
def route_task(state: MessagesState) -> str:
last = state["messages"][-1].content.lower()
if "research" in last or "search" in last:
return "research"
return "analyze"
builder = StateGraph(MessagesState)
builder.add_node("research", research_subgraph)
builder.add_node("analyze", analysis_subgraph)
builder.add_conditional_edges(START, route_task)
builder.add_edge("research", END)
builder.add_edge("analyze", END)
graph = builder.compile()A child's streamed tokens never merge into the parent transcript — they land on the child's own stream in subagents(), keyed by the research:<uuid> namespace. What the transcript shows at settle is decided by state: because both graphs above share MessagesState, the child's message enters the parent's message list and arrives with the final values sync. Give the child its own schema (below) and it never does.
Streamed chunks from top-level side-effect nodes — a router, a title generator — are a separate concern: whitelist your conversational nodes with transcriptNodeNames.
Giving the child its own state
Adding a compiled graph as a node does not isolate state. If you want a real boundary, design one: give the child its own state schema and share only the keys you want crossing it. LangGraph passes a subgraph node through the keys the two schemas have in common.
from typing import Annotated, TypedDict
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
class ResearchState(TypedDict):
"""Child state — deliberately has no `messages` key."""
research_topic: str
research_brief: str
class OrchestratorState(TypedDict):
"""Parent state — the transcript plus the shared channel."""
messages: Annotated[list, add_messages]
research_topic: str
research_brief: str
async def research_node(state: ResearchState) -> dict:
# Receives a topic, returns a brief. No transcript access.
brief = await researcher.ainvoke(f"Topic: {state['research_topic']}")
return {"research_brief": brief.content}
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()
def route_after_orchestrate(state: OrchestratorState) -> str:
# Writing a topic is what triggers delegation.
return "research" if state.get("research_topic") else "answer"
parent = StateGraph(OrchestratorState)
parent.add_node("orchestrate", orchestrate_node)
parent.add_node("research", compiled_research) # the compiled graph IS the node
parent.add_node("answer", answer_node) # the only node that writes messages
parent.add_edge(START, "orchestrate")
parent.add_conditional_edges(
"orchestrate", route_after_orchestrate, {"research": "research", "answer": "answer"}
)
parent.add_edge("research", "answer")
parent.add_edge("answer", END)
graph = parent.compile()Because ResearchState has no messages key, the child cannot read the transcript or append to it — its brief reaches the parent through research_brief and never becomes a chat message. Pair it with transcriptNodeNames: ['answer'] so only the parent's answering node streams into messages().
OrchestratorState and PipelineState below are placeholders for your own graph's state schema — the shape your subgraph's StateGraph produces. They mirror the Python state the same way ChatState does on the State Management page. Use createAgentRef<YourState>('your-assistant-id') to create a typed ref, then pass it to both provideAgent() and injectAgent().
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 are keyed by their namespace segment and named by node; they register on their first streamed event and settle with the run.
// In a shared file (e.g. agent.ts):
// import { createAgentRef } from '@threadplane/chat';
// export const ORCHESTRATOR = createAgentRef<OrchestratorState>('orchestrator');
// Configure in app.config.ts:
// provideAgent(ORCHESTRATOR, {
// apiUrl: '...',
// subagentToolNames: ['task', 'delegate_to_researcher'],
// });
const orchestrator = injectAgent(ORCHESTRATOR);
// All subagent streams (active and completed)
const subagents = computed(() => orchestrator.subagents());
// Only active ones
const running = computed(() =>
[...orchestrator.subagents().values()].filter((subagent) =>
subagent.status() === 'pending' || subagent.status() === 'running'
)
);
const runningCount = computed(() => running().length);
// 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(`${runningCount()} subagents currently running`);
});Subagent stream details
Each SubagentStreamRef exposes its own reactive signals — status, messages, and state — 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() ?? []);Orchestrator pattern
The orchestrator pattern delegates specialised work to subagents and merges their results. Each subagent runs its own graph independently while the parent coordinates the whole.
// In a shared file (e.g. agent.ts):
// import { createAgentRef } from '@threadplane/chat';
// export const PIPELINE = createAgentRef<PipelineState>('pipeline-orchestrator');
// Configure in app.config.ts:
// provideAgent(PIPELINE, {
// apiUrl: '...',
// subagentToolNames: ['task'],
// });
const pipeline = injectAgent(PIPELINE);
// Derive a summary of all subagent states
const pipelineStatus = computed(() => {
const agents = pipeline.subagents();
const entries = [...agents.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,
};
});Subagent progress UI
Render live progress for each subagent using the signals above.
import { Component, computed, ChangeDetectionStrategy } from '@angular/core';
import { injectAgent } from '@threadplane/langgraph';
import { ORCHESTRATOR } from './agent'; // createAgentRef<OrchestratorState>('orchestrator')
@Component({
selector: 'app-subagent-progress',
templateUrl: './progress-panel.component.html',
changeDetection: ChangeDetectionStrategy.OnPush,
})
export class SubagentProgressComponent {
protected readonly orchestrator = injectAgent(ORCHESTRATOR);
subagentEntries = computed(() =>
[...this.orchestrator.subagents().entries()]
);
}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. Render a child's live output from its own stream:
// In a shared file (e.g. agent.ts):
// import { createAgentRef } from '@threadplane/chat';
// export const ORCHESTRATOR = createAgentRef<OrchestratorState>('orchestrator');
// Configure in app.config.ts:
// provideAgent(ORCHESTRATOR, {
// apiUrl: '...',
// subagentToolNames: ['task'],
// });
const orchestrator = injectAgent(ORCHESTRATOR);
// The parent's transcript — child chatter is structurally absent
const parentMessages = computed(() => orchestrator.messages());Set subagentToolNames to the tool names that spawn subagents. injectAgent() uses this to identify tool calls that create subagent streams.
Registration is skipped silently unless the tool call also carries a valid subagent_type argument: a string of 3-50 characters, starting with a letter, containing only letters, digits, _, or -. A value like qa (too short) or 2nd_pass (leading digit) produces no subagent and no error, so subagents() stays empty with nothing in the console to explain it.
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.
Always check failedAgents() before presenting final results. A completed orchestrator can still have subagents that errored — success at the top level does not guarantee all delegates succeeded.
When to use subagents vs a single agent
Use subagents when tasks are independent and can run in parallel, when each task needs its own context window, or when you want isolated error boundaries. Use a single agent for sequential reasoning, tasks that share tightly coupled state, or when latency from spawning subagents outweighs the parallelism benefit.
None of those three come from compiling a child graph. A narrow context window follows from what you pass into the child, an error boundary from how the parent handles a failed delegation, and state isolation from giving the child its own schema. Compiling buys you nested execution and a namespace; the rest is yours to design. See the decision matrix.
What's Next
Understand how injectAgent() surfaces tokens, status, and errors in real time.
Inspect earlier states and replay alternate execution paths with checkpoint history.
Write unit and integration tests for orchestrator graphs and subagent interactions.
Full reference for injectAgent() options, signals, and subagent configuration.