把一个编译好的图作为节点嵌入父图——理解子图如何通信、如何封装复杂逻辑，
构建真正可扩展的多 Agent 系统。
子图（Sub-graph）是 LangGraph 中一个核心的高级特性：把一个已经编译好的 StateGraph 作为节点（Node）直接插入另一个父图中。从父图的角度来看，这个子图和普通节点没有任何区别——它依然接收 State、执行逻辑、返回更新；但它内部其实是一张完整的独立图，有自己的状态、节点和边。
可以把子图理解为"图中图"或者"函数中有完整工作流的超级节点"。它让你在 LangGraph 的世界里实现真正意义上的嵌套架构（Nested Architecture）。
子图 = 一个调用 compile() 后的 StateGraph 实例，它可以被作为节点直接传入父图的 add_node()，从而成为父图执行流程中的一个步骤。
具体添加方式就是一行代码：
# 把编译好的子图作为节点加入父图
entry_builder.add_node("failure_analysis", fa_builder.compile())
entry_builder.add_node("question_summarization", qs_builder.compile())这里 fa_builder.compile() 返回的是一个 CompiledStateGraph 对象，它和普通函数一样可以被传入 add_node，因为 LangGraph 的编译图实现了 Runnable 接口，与普通 Python 函数具有相同的调用约定。
当你构建复杂的 AI 系统时，直接把所有节点堆在一张大图里会带来一系列问题：逻辑混乱、状态污染、难以测试、无法复用。子图正是为了解决这些问题而生。
在真实的生产系统中，一个"研究助手"可能由多个专门的 Agent 子图构成：搜索子图、总结子图、事实核查子图、格式化子图……每个子图专注一件事，通过父图协调它们的执行顺序和数据流转。子图是构建可维护多 Agent 系统的关键抽象。
| 维度 | 不用子图（所有节点在一张图） | 使用子图 |
|---|---|---|
| 状态设计 | 所有中间变量都暴露在父图状态中 | 子图的私有状态完全隔离 |
| 可测试性 | 必须运行整个大图才能测试某段逻辑 | 子图可以单独 invoke() 测试 |
| 代码组织 | 几十个节点混在一起，难以阅读 | 每个子图是独立文件/函数，结构清晰 |
| 并行执行 | 手动协调，容易出错 | 多个子图节点天然支持并行 |
为了演示子图，notebook 构建了一个日志分析系统，这个系统做三件事：
日志包含用户问题、答案、评分、反馈等字段。先经过一个 clean_logs 节点做数据清洗。
子图 A（失败分析）：找出评分为 0 的日志，生成失败模式摘要 fa_summary。
子图 B（问题总结）：对所有日志的问题生成总结报告 report，并发送到 Slack。
两个子图各自产生的摘要和日志处理记录，通过重叠键自动合并回父图状态。
首先，notebook 定义了日志的数据结构：
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] # 反馈信息测试数据包含两条日志：一条正常日志（无评分），一条失败日志（评分为 0，表示文档检索质量差）：
# 正常日志：只有问答，无评分
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]notebook 中的 generate_summary 等函数直接返回硬编码字符串（如 "Poor quality retrieval of Chroma documentation."），是为了专注于演示子图架构，省去了真实 LLM 调用。在生产中，这些节点里会调用真实的 AI 模型来生成总结。
每个子图都需要：① 定义自己的状态类型；② 编写节点函数；③ 用 StateGraph 组装；④ 调用 compile()。子图定义方式和普通图完全相同，只是最终它会被"挂载"到父图上。
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)# ① 子图状态（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)两个子图都拥有 cleaned_logs（输入）和 processed_logs（输出）这两个与父图共享的键。但它们各自的私有中间状态（failures、qs_summary）互不干扰，也不会出现在父图的状态中。
子图和父图之间的通信是 Sub-graph 最需要深入理解的部分。核心规则只有一条：通信通过重叠的键名（Key Name）来实现。
当父图执行到子图节点时，LangGraph 会自动把父图当前状态中与子图状态定义中同名的键值对提取出来，作为子图的输入状态传入。
cleaned_logs 是父图和子图共同定义的键，因此它的值从父图自动流入子图。而父图的 raw_logs 等键，子图不认识（不在子图的 State 定义中），所以不会传入。
子图执行完成后，LangGraph 会把子图输出状态中与父图状态同名的键值对写回父图状态。子图私有的键（如 failures、qs_summary）不会出现在父图中。
两个子图并行运行时，都会输出 processed_logs 键。如果父图状态中该键没有 Reducer，两个并行子图的输出会产生冲突（谁的覆盖谁？）。解决方案有两种：
方案 A：给父图的 processed_logs 加 Annotated[List, add] Reducer，把两个子图的结果列表合并。
方案 B（notebook 采用）：为每个子图设置 output_schema，让不同子图输出不同的键（如 fa_summary vs report），避免冲突。但 processed_logs 两个子图都输出，所以父图仍然需要 Reducer。
| 键名 | 父图状态 | 失败分析子图 | 问题总结子图 | 说明 |
|---|---|---|---|---|
raw_logs |
有 | 无 | 无 | 父图私有，子图不可见 |
cleaned_logs |
有 | 有（输入） | 有（输入） | 重叠键：父图 → 两个子图 |
failures |
无 | 有（私有） | 无 | 子图 A 私有中间状态 |
qs_summary |
无 | 无 | 有（私有） | 子图 B 私有中间状态 |
fa_summary |
有 | 有（输出） | 无 | 重叠键：子图 A → 父图 |
report |
有 | 无 | 有（输出） | 重叠键：子图 B → 父图 |
processed_logs |
有（带 Reducer） | 有（输出） | 有（输出） | 两个子图都输出，父图用 add 合并 |
理解了状态共享机制之后，把子图加入父图就非常直观了。关键步骤是：定义父图状态、编写父图节点、用 add_node 把编译好的子图挂载上去。
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]因为两个子图只是读取 cleaned_logs，并不修改或输出它（output_schema 里没有这个键）。只有当多个并行节点都往同一个键写入时，才需要 Reducer。
# 父图中的普通节点：对原始日志做清洗
def clean_logs(state):
raw_logs = state["raw_logs"]
cleaned_logs = raw_logs # 生产中：做实际清洗处理
return {"cleaned_logs": cleaned_logs}