Sub-graphs
The Art of Modularity and State Boundaries

1What is a sub-graph?

Sub-graph is a core advanced feature in LangGraph:Insert a compiled StateGraph directly into another parent graph as a node (Node). From the perspective of the parent graph, this subgraph is no different from an ordinary node—it still receives State, executes logic, and returns updates; but inside it is actually a complete independent graph with its own state, nodes, and edges.

Subgraphs can be understood as "pictures within graphs" or "super nodes with complete workflows in functions". It allows you to achieve true meaning in the world of LangGraphNested schema（Nested Architecture）。

The specific way to add it is a line of code:

# 把编译好的子图作为节点加入父图
entry_builder.add_node("failure_analysis", fa_builder.compile())
entry_builder.add_node("question_summarization", qs_builder.compile())

herefa_builder.compile()What is returned is aCompiledStateGraphObject, which can be passed in like an ordinary functionadd_node, because LangGraph’s compiled graph implementsRunnable interface, has the same calling convention as ordinary Python functions.

2Why Subgraphs Are Needed: Modularity vs. Encapsulation

When you build a complex AI system, directly stacking all nodes in a large graph will cause a series of problems: logical confusion, state pollution, difficulty in testing, and inability to reuse. Subgraphs are created to solve these problems.

The three core values brought by sub-pictures

3Example scenario for this lesson: Log analysis system

To demonstrate subgraphs, the notebook builds aLog analysis system, this system does three things:

• 1

  Receive raw logs (raw_logs)

  The log contains user questions, answers, ratings, feedback and other fields. Go through one firstclean_logsNodes perform data cleaning.

• 2

  Run two subgraphs in parallel

  Subfigure A (failure analysis): Find logs with a score of 0 and generate a summary of failure modesfa_summary。
  Subfigure B (problem summary): Generate summary report for all log issuesreport, and sent to Slack.

• 3

  Aggregate results to parent graph

  The summaries and log processing records generated by each of the two subgraphs are automatically merged back into the parent graph state via overlapping keys.

First, notebook defines the data structure of the log:

from operator import add
from typing_extensions import TypedDict
from typing import List, Optional, Annotated

# 每一条日志的数据结构
class Log(TypedDict):
    id: str               # 日志 ID
    question: str         # 用户提问
    docs: Optional[List]  # 检索到的文档（可选）
    answer: str           # LLM 的回答
    grade: Optional[int]  # 评分，0 表示失败（可选）
    grader: Optional[str] # 评分者
    feedback: Optional[str] # 反馈信息

The test data contains two logs: a normal log (no score) and a failure log (a score of 0, indicating poor document retrieval quality):

# 正常日志：只有问答，无评分
question_answer = Log(
    id="1",
    question="How can I import ChatOllama?",
    answer="To import ChatOllama, use: 'from langchain_community.chat_models import ChatOllama.'",
)

# 失败日志：评分为 0，表示检索到的文档与问题不相关
question_answer_feedback = Log(
    id="2",
    question="How can I use Chroma vector store?",
    answer="To use Chroma, define: rag_chain = create_retrieval_chain(retriever, question_answer_chain).",
    grade=0,
    grader="Document Relevance Recall",
    feedback="The retrieved documents discuss vector stores in general, but not Chroma specifically",
)

raw_logs = [question_answer, question_answer_feedback]

4Define and compile subgraphs: FailureAnalysisState and QuestionSummarizationState

Each subgraph needs to: ① define its own state type; ② write node functions; ③ useStateGraphAssembly; ④ Callcompile(). The subgraph is defined in exactly the same way as a normal graph, except that it will eventually be "mounted" to the parent graph.

Failure Analysis Sub-graph

from langgraph.graph import StateGraph, START, END

# ① 子图的完整状态（包含私有字段 failures）
class FailureAnalysisState(TypedDict):
    cleaned_logs: List[Log]   # 与父图共享的输入键
    failures: List[Log]       # 子图私有：过滤出的失败日志
    fa_summary: str           # 与父图共享的输出键
    processed_logs: List[str]  # 与父图共享的输出键

# ② 输出状态（仅包含需要"发布"给父图的键）
class FailureAnalysisOutputState(TypedDict):
    fa_summary: str
    processed_logs: List[str]

# ③ 节点：从 cleaned_logs 中筛选出有 grade 字段（即失败）的日志
def get_failures(state):
    cleaned_logs = state["cleaned_logs"]
    failures = [log for log in cleaned_logs if "grade" in log]
    return {"failures": failures}

# ④ 节点：根据失败日志生成摘要（生产中替换为 LLM 调用）
def generate_summary(state):
    failures = state["failures"]
    fa_summary = "Poor quality retrieval of Chroma documentation."
    return {
        "fa_summary": fa_summary,
        "processed_logs": [f"failure-analysis-on-log-{failure['id']}" for failure in failures]
    }

# ⑤ 构建并编译子图，指定 output_schema 限制输出
fa_builder = StateGraph(
    state_schema=FailureAnalysisState,
    output_schema=FailureAnalysisOutputState   # 关键：限制输出键
)
fa_builder.add_node("get_failures", get_failures)
fa_builder.add_node("generate_summary", generate_summary)
fa_builder.add_edge(START, "get_failures")
fa_builder.add_edge("get_failures", "generate_summary")
fa_builder.add_edge("generate_summary", END)

Question Summarization Sub-graph

# ① 子图状态（qs_summary 是私有中间字段，不会发布到父图）
class QuestionSummarizationState(TypedDict):
    cleaned_logs: List[Log]   # 与父图共享的输入键
    qs_summary: str           # 子图私有：中间摘要
    report: str               # 与父图共享的输出键
    processed_logs: List[str]  # 与父图共享的输出键

# ② 输出状态：只发布 report 和 processed_logs 给父图
class QuestionSummarizationOutputState(TypedDict):
    report: str
    processed_logs: List[str]

# ③ 节点：生成问题摘要
def generate_summary(state):
    cleaned_logs = state["cleaned_logs"]
    summary = "Questions focused on usage of ChatOllama and Chroma vector store."
    return {
        "qs_summary": summary,
        "processed_logs": [f"summary-on-log-{log['id']}" for log in cleaned_logs]
    }

# ④ 节点：把摘要发送到 Slack（此处模拟）
def send_to_slack(state):
    qs_summary = state["qs_summary"]
    report = "foo bar baz"   # 生产中：调用 Slack API
    return {"report": report}

qs_builder = StateGraph(QuestionSummarizationState, output_schema=QuestionSummarizationOutputState)
qs_builder.add_node("generate_summary", generate_summary)
qs_builder.add_node("send_to_slack", send_to_slack)
qs_builder.add_edge(START, "generate_summary")
qs_builder.add_edge("generate_summary", "send_to_slack")
qs_builder.add_edge("send_to_slack", END)

5State sharing mechanism: Overlapping Keys are a bridge for communication

The communication between the subgraph and the parent graph is the most important part of Sub-graph to understand in depth.There is only one core rule: communication is achieved through overlapping key names.

When entering the sub-picture (parent picture → sub-picture)

When the parent graph is executed to the child graph node, LangGraph will automatically compare the current state of the parent graph with the state definition of the child graph.Key-value pairs with the same nameExtract it and pass it in as the input state of the subgraph.

cleaned_logsIt is a key defined by both the parent and child graphs, so its value automatically flows from the parent graph to the child graph. And the father picture’sraw_logsThe key is not recognized by the subgraph (it is not in the State definition of the subgraph), so it will not be passed in.

When exiting the sub-picture (sub-picture → parent picture)

After the subgraph execution is completed, LangGraph will output the subgraph in the stateA key-value pair with the same name as the parent graph stateWrite back to parent graph state. A key private to the subgraph (e.g.failures、qs_summary) will not appear in the parent diagram.

Complete mapping of status keys

6Add child graph to parent graph: add_node with parallel execution

After understanding the state sharing mechanism, adding the subgraph to the parent graph is very intuitive. The key steps are: define the parent graph state, write the parent graph node, and useadd_nodeMount the compiled subgraph.

Parent graph state definition

from operator import add
from typing import Annotated

# 父图的完整状态
class EntryGraphState(TypedDict):
    raw_logs: List[Log]      # 输入：原始日志
    cleaned_logs: List[Log]  # 清洗后的日志（传入两个子图）
    fa_summary: str          # 来自失败分析子图
    report: str              # 来自问题总结子图
    # 两个子图都会输出 processed_logs，用 add Reducer 合并列表
    processed_logs: Annotated[List[str], add]

Clean node of parent graph

# 父图中的普通节点：对原始日志做清洗
def clean_logs(state):
    raw_logs = state["raw_logs"]
    cleaned_logs = raw_logs   # 生产中：做实际清洗处理
    return {"cleaned_logs": cleaned_logs}

Assemble the parent graph: mount the child graph as a node

entry_builder = StateGraph(EntryGraphState)

# 挂载普通节点
entry_builder.add_node("clean_logs", clean_logs)

# 挂载子图节点：compile() 返回的 CompiledGraph 直接传入
entry_builder.add_node("question_summarization", qs_builder.compile())
entry_builder.add_node("failure_analysis", fa_builder.compile())

# 定义执行顺序：先清洗，然后两个子图并行执行
entry_builder.add_edge(START, "clean_logs")
entry_builder.add_edge("clean_logs", "failure_analysis")       # 并行分支 1
entry_builder.add_edge("clean_logs", "question_summarization")   # 并行分支 2
entry_builder.add_edge("failure_analysis", END)
entry_builder.add_edge("question_summarization", END)

graph = entry_builder.compile()

A node hastwo exit edgesPoint to different nodes, LangGraph willAutomatic parallel executionthese two nodes. hereclean_logsAfter completion,failure_analysisandquestion_summarizationwill run at the same time, and will be summarized when they finally reach END.

Visualize nested structures: xray=1

LangGraph provides a particularly useful debugging parameter: when callingget_graph()when passed inxray=1, you can "see through" the internal structure of the sub-picture and draw the entire nested diagram completely.

# xray=1 表示"透视"：显示子图的内部节点，而不只是显示一个黑盒
display(Image(graph.get_graph(xray=1).draw_mermaid_png()))

Not addedxrayWhen , the subgraph is only displayed as a rectangular node; addedxray=1, you can see the detailed connection relationships of all nodes and edges within the subgraph, which is very suitable for debugging multi-layer nested complex graphs.

7Output Schema: Precisely control the return of subgraphs

in subgraphoutput_schemaParameters are a very important design decision. They determine which keys the subgraph "releases" to the parent graph, effectively implementing the state encapsulation of the subgraph.

Problems when output_schema is not used

If not specifiedoutput_schema, the subgraph will convert itscomplete stateAs output - includes all private intermediate fields.

The problem is: when two subgraphs are run in parallel, they both havecleaned_logsfields, and the values are the same (both are read from the parent graph). Both sub-picturescleaned_logsWriting back to the parent picture will cause conflicts, and you need to give the parent picturecleaned_logsAdd a reducer (e.g.operator.add) to merge - but this adds the same log list twice!

# 不用 output_schema 时，父图需要给 cleaned_logs 加 Reducer
class EntryGraphState(TypedDict):
    raw_logs: List[Log]
    # 因为两个子图都会"回写" cleaned_logs，需要 Reducer
    cleaned_logs: Annotated[List[Log], add]  # 会产生重复数据！
    fa_summary: str
    report: str
    processed_logs: Annotated[List[str], add]

Elegant solution using output_schema

By defining for each subgraphoutput_schema, precisely controlling which keys it publishes to the parent graph, completely avoidingcleaned_logsConflict issues:

# 失败分析子图只发布这两个键
class FailureAnalysisOutputState(TypedDict):
    fa_summary: str            # 父图有，回写到父图
    processed_logs: List[str]  # 父图有，回写到父图
    # cleaned_logs 和 failures 不在这里 → 不会回写父图

# 问题总结子图只发布这两个键
class QuestionSummarizationOutputState(TypedDict):
    report: str               # 父图有，回写到父图
    processed_logs: List[str]  # 父图有，回写到父图
    # cleaned_logs 和 qs_summary 不在这里

# 于是父图的 cleaned_logs 不需要 Reducer 了！
class EntryGraphState(TypedDict):
    raw_logs: List[Log]
    cleaned_logs: List[Log]    # 普通字段即可，无需 Reducer
    fa_summary: str
    report: str
    processed_logs: Annotated[List[str], add]  # 仍然需要，因为两个子图都输出它

Comparison of two methods

8Complete running demonstration and result interpretation

Run all the code together and observe the final output:

# 准备测试数据
question_answer = Log(
    id="1",
    question="How can I import ChatOllama?",
    answer="To import ChatOllama, use: 'from langchain_community.chat_models import ChatOllama.'",
)
question_answer_feedback = Log(
    id="2",
    question="How can I use Chroma vector store?",
    answer="To use Chroma, define: rag_chain = create_retrieval_chain(...).",
    grade=0,
    grader="Document Relevance Recall",
    feedback="The retrieved documents discuss vector stores in general, but not Chroma specifically",
)
raw_logs = [question_answer, question_answer_feedback]

# 运行图，传入原始日志
result = graph.invoke({"raw_logs": raw_logs})

actual output result

{
  'raw_logs': [
    {'id': '1', 'question': 'How can I import ChatOllama?', ...},
    {'id': '2', 'question': 'How can I use Chroma vector store?', ..., 'grade': 0, ...}
  ],
  'cleaned_logs': [
    {'id': '1', ...},  # 清洗节点处理后（本例等于原始数据）
    {'id': '2', ..., 'grade': 0}
  ],
  'fa_summary': 'Poor quality retrieval of Chroma documentation.',
                # ↑ 来自失败分析子图
  'report': 'foo bar baz',
            # ↑ 来自问题总结子图
  'processed_logs': [
    'failure-analysis-on-log-2',  # 来自子图 A（只处理了有 grade 的日志）
    'summary-on-log-1',           # 来自子图 B（处理了所有日志）
    'summary-on-log-2'
  ]
}

Interpret field by field

Summary: Complete steps for building a subgraph system

• 1

  Design state boundaries

  Make it clear which keys are shared by the parent graph and the child graph (overlapping keys), and which are private intermediate states of the child graph. Draw a "state flow diagram".

• 2

  Define the State and OutputState of each subgraph

  use TypedDictDefine the complete state, then defineOutputStatePrecisely control the keys published to the outside world to avoid parallel conflicts.

• 3

  Implement and test each subgraph independently

  use subgraph.compile().invoke({...})Test whether the logic of each sub-diagram is correct individually, and then integrate it into the parent diagram.

• 4

  Define the parent graph state and add reducer as needed

  Analyze which keys will be written by multiple parallel subgraphs, and addAnnotated[..., add]Wait for reducer.

• 5

  Use add_node to mount the compiled subgraph

  callbuilder.add_node("name", subgraph.compile()), connect the edges, compile the parent graph, and run.

# 另一种方式：把子图包在函数里，手动做状态映射
compiled_fa = fa_builder.compile()

def run_failure_analysis(state):
    # 手动提取需要传给子图的字段
    sub_input = {"cleaned_logs": state["cleaned_logs"]}
    # 调用子图
    sub_output = compiled_fa.invoke(sub_input)
    # 手动把子图输出映射回父图状态
    return {
        "fa_summary": sub_output["fa_summary"],
        "processed_logs": sub_output["processed_logs"]
    }

# 使用方式与普通节点完全一样
entry_builder.add_node("failure_analysis", run_failure_analysis)