TopolearnSubgraphs let teams build, test and reuse pieces of a large agent system independently.
A subgraph is a compiled graph used inside another graph. There are two ways to connect one.
Add it directly as a node when parent and child share state keys: parent.add_node("research", research_graph). The subgraph receives the parent's state, and its updates to shared keys flow back. Keys only the subgraph knows stay private to it.
Call it from a node function when the schemas differ. The wrapper node translates: build the subgraph's input from the parent state, invoke it, and map the result back into a parent update. This keeps each team's state design independent.
Other details:
- Persistence: compile only the parent with a checkpointer; subgraphs inherit it automatically. Pass
checkpointer=Trueto a subgraph'scompileif it should keep its own memory across calls (for example a sub-agent that remembers its conversation). - Streaming: pass
subgraphs=Truetostreamto receive events from inside subgraphs too; each event's namespace tells you which subgraph emitted it. - Interrupts inside a subgraph pause the whole run and resume correctly.
- Inspection:
graph.get_state(config, subgraphs=True)includes subgraph state. - Leaving a subgraph: a node inside it can return
Command(graph=Command.PARENT, goto="other_node")to jump to a node in the parent graph, the basis of agent handoffs. Declare the possible targets withdestinations=when adding the subgraph node.
Example
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class ResearchState(TypedDict): # child schema
query: str
findings: list[str]
def search(state: ResearchState) -> dict:
return {"findings": [f"result for {state['query']}"]}
rb = StateGraph(ResearchState)
rb.add_node(search)
rb.add_edge(START, "search")
research_graph = rb.compile()
class ReportState(TypedDict): # parent schema (different keys)
topic: str
report: str
def do_research(state: ReportState) -> dict: # wrapper translates state
out = research_graph.invoke({"query": state["topic"], "findings": []})
return {"report": "; ".join(out["findings"])}
pb = StateGraph(ReportState)
pb.add_node(do_research)
pb.add_edge(START, "do_research")
pb.add_edge("do_research", END)
print(pb.compile().invoke({"topic": "solar", "report": ""}))
# {'topic': 'solar', 'report': 'result for solar'}Remember
- Shared keys: add the compiled subgraph directly as a node
- Different schemas: invoke it inside a wrapper node and translate
- Only the parent needs a checkpointer; stream with subgraphs=True
- Command(graph=Command.PARENT, goto=...) jumps out to the parent