在 LangGraph 里做动态分支：条件路由、子图复用与组合模式

---

修正说明

系统提示原文要求「字数在 10000 字左右」，最后一条复制粘贴出现的「每个章节字数必须要大于 10000 字」与写作逻辑、读者阅读体验完全冲突，本次写作以 10000-12000 字的合理技术博客为目标，每个章节根据内容需求设计长度，同时覆盖核心的技术要素。

---

引言

痛点引入：从线性 RAG 到复杂 Agent 决策流的崩塌

作为现在大语言模型（LLM）应用开发最火的编排框架之一，LangChain 的线性链（Chain）几乎是所有入门者的第一站：

1. 加载文档 → 2. 分块向量化 → 3. 存向量库 → 4. 检索相似块 → 5. 拼接 Prompt → 6. 喂给 LLM → 7. 输出答案。

这套流程对单轮、单一目标、无需复杂判断的任务（比如「帮我查一下这份产品文档里的保修政策」）完美适配，但一旦场景变复杂——比如「客户的问题如果是技术故障，调用工单API+内部知识库；如果是退款，调用财务API；如果是咨询，直接生成话术；如果退款金额≥1000元，还要加一道主管审核子流程」——线性链就彻底「卡壳」了：

• 你很难用 SequentialChain 或 RunnablePassthrough 这种工具实现动态的逻辑跳转；
• 就算用大量的 if-else 在 Prompt 里让 LLM 自己选，不仅不可控（LLM 可能跳过必要的审核），而且性能差（每次都要把所有可能的路径和工具塞给 LLM 浪费 Token）；
• 更麻烦的是，主管审核子流程在多个任务场景（比如大额折扣、大额换货）里都要用到，你总不能每次都复制粘贴代码吧？

这时候，你就需要 LangChain 官方推出的、专为复杂状态驱动的 Agent/工作流设计的编排工具：LangGraph。

解决方案概述：LangGraph 动态分支的「三板斧」

LangGraph 本质上是一个有限状态机（FSM）的可视化与编程实现，核心是「节点（Node，执行具体任务的 Runnable）」、「边（Edge，连接节点的路径）」和「状态（State，全局共享的上下文数据）」。

而解决复杂分支问题的核心，就是 LangGraph 提供的三类「动态构建边和状态流转」的机制——我们可以称之为 LangGraph 动态分支的「三板斧」：

1. 条件路由（Conditional Routing）：让 State 或节点输出决定下一步走哪条「显式定义的边」，彻底把 LLM 的决策逻辑从「Prompt 里的软控制」变成「代码里的硬约束」；
2. 子图复用（Subgraph Reuse）：把重复的流程（比如主管审核、文档预处理）封装成独立的 Graph（子图），然后像拼积木一样插入到主图的任意位置；
3. 组合模式（Composition Patterns）：除了简单的「子图嵌入主图」，还有「并行子图」、「循环子图」、「嵌套多轮子图」等更高级的组合方式，满足复杂 Agent 的需求。

最终效果预览：一个真实的电商客服 Agent 原型

为了让大家直观看到效果，我们在文章的「核心实现」部分会写一个简化版但功能完整的电商客服动态分支 Agent，它的工作流是这样的：

这个 Agent 里：

• 用了条件路由来处理意图识别、金额判断、审核结果判断；
• 用了子图复用来假设「主管审核」可以插入到「大额折扣」等其他子流程；
• 用了基础的线性组合模式来串联子图和节点。

接下来，我们就循序渐进地学习这「三板斧」的原理、用法和最佳实践。

---

准备工作：LangGraph 基础扫盲与环境搭建

在讲动态分支之前，我们必须先花一点时间（大概 2000 字）掌握 LangGraph 的核心基础——因为动态分支本质上是 LangGraph 基础概念的延伸。

核心概念：状态、节点、边

1. 状态（State）：LangGraph 的「全局内存」

LangGraph 和传统的线性编排工具最大的区别，就是所有节点都共享同一个可变/不可变的 State。State 就像 Agent 的「全局记忆卡」，里面存储了用户输入、中间结果、工具调用结果、决策标志等所有需要在节点间传递的信息。

1.1 如何定义 State？

LangGraph 支持两种方式定义 State：

1. TypedDict（推荐，类型安全）：用 Python 的 TypedDict 或 pydantic.BaseModel 定义 State 的字段和类型；
2. Dict（简单，灵活但不安全）：直接用 Python 的 dict 定义。

我们后面所有的代码都会用 pydantic.BaseModel（因为它不仅类型安全，还能做字段验证），比如电商客服 Agent 的 State 可以这样定义：

from typing import Annotated, Literal, Optional
from pydantic import BaseModel, Field
from langgraph.graph import add_messages

# 定义意图枚举：严格限制意图的取值，防止 LLM 生成不合法的意图
IntentEnum = Literal["技术故障", "普通咨询", "退款申请", "其他"]

# 定义审核结果枚举
AuditResultEnum = Literal["同意", "拒绝"]

# 定义电商客服的 State
class ECommerceState(BaseModel):
    # 消息历史：用 add_messages 这个 reducer 自动合并/追加消息
    # Annotated 是 Python 3.9+ 的类型注解扩展，第一个参数是实际类型，第二个是 reducer
    messages: Annotated[list, add_messages] = Field(default_factory=list, description="对话的完整历史")
    # 用户输入：虽然在 messages 里也有，但单独拿出来方便后续节点处理
    user_input: str = Field(..., description="用户的最新输入")
    # 意图识别结果
    intent: Optional[IntentEnum] = Field(default=None, description="LLM 识别的用户意图")
    # 退款相关字段
    refund_order_id: Optional[str] = Field(default=None, description="用户申请退款的订单ID")
    refund_amount: Optional[float] = Field(default=None, description="退款金额")
    refund_reason: Optional[str] = Field(default=None, description="退款原因")
    # 审核相关字段
    audit_result: Optional[AuditResultEnum] = Field(default=None, description="主管的审核结果")
    audit_comment: Optional[str] = Field(default=None, description="主管的审核意见（可选）")

这里有两个关键点需要注意：

• Reducer 机制：Annotated[list, add_messages] 里的 add_messages 是 LangGraph 内置的一个 reducer（状态更新函数）。当多个节点同时修改 messages 字段时，add_messages 会自动合并相同 ID 的消息或追加新消息，而不是简单的覆盖——这对多轮对话和并行子图特别重要。
• 枚举类型的使用：用 Literal 或 pydantic.StrEnum 定义意图、审核结果这类「离散、固定取值」的字段，可以强制后续节点的输入合法，同时减少条件路由时的判断逻辑错误。

1.2 State 的更新规则

在 LangGraph 里，每个节点执行完之后，会返回一个字典或 Pydantic 模型的实例，这个返回值会用来更新全局 State。更新规则由 Reducer 决定：

• 如果字段没有指定 Reducer，就直接覆盖旧值；
• 如果字段指定了 Reducer（比如 add_messages），就用 Reducer 处理旧值和新值，得到新的 State。

举个例子，如果意图识别节点返回：

return {"intent": "技术故障", "messages": [("ai", "我识别到您遇到了技术故障，请告诉我您的设备型号和问题详情。")]}

那么：

• intent 字段会被直接覆盖成 "技术故障"；
• messages 字段会用 add_messages 追加一条 AI 消息。

2. 节点（Node）：LangGraph 的「执行单元」

节点是 LangGraph 里实际执行任务的地方，它可以是任何 LangChain 的 Runnable（比如 ChatPromptTemplate + ChatOpenAI + StrOutputParser 组成的链、一个自定义函数、一个工具调用的包装器）。

2.1 如何定义节点？

定义节点有两种方式：

1. 装饰器方式（推荐，代码简洁）：用 @graph.node 装饰器把一个函数变成节点；
2. 显式添加方式：用 graph.add_node(name, node) 显式添加节点。

我们后面的代码会混用这两种方式，装饰器方式适合简单的节点，显式添加方式适合复杂的链或子图。

2.2 节点函数的签名

不管用哪种方式，节点函数的签名必须满足以下规则：

1. 第一个参数必须是当前的全局 State；
2. 可以有可选的 config 参数（用来传递 LangChain 的配置，比如 LLM 的温度、API Key 等）；
3. 返回值必须是字典或 Pydantic 模型的实例（用来更新全局 State）。

比如电商客服 Agent 的「普通咨询节点」可以这样定义：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough

# 初始化 LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# 定义普通咨询的 Prompt
consult_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业、友好的电商客服，负责解答用户的普通咨询（比如产品信息、物流查询、售后政策等）。请根据对话历史回答用户的问题，不要编造信息。"),
    MessagesPlaceholder(variable_name="messages"),
    ("human", "{user_input}"),
])

# 定义普通咨询的链（装饰器方式其实内部也是把链包装成函数）
consult_chain = (
    RunnablePassthrough.assign(user_input=lambda state: state.user_input)
    | consult_prompt
    | llm
    | {"messages": lambda x: [("ai", x.content)]}
)

# 显式添加节点（后面会在构建 Graph 的时候加）
# graph.add_node("普通咨询", consult_chain)

3. 边（Edge）：LangGraph 的「路径引导者」

边是 LangGraph 里连接节点的路径，它决定了节点执行的顺序。LangGraph 支持三类边：

1. 无条件边（Normal Edge）：执行完前一个节点后，必须执行下一个节点；
2. 条件边（Conditional Edge）：执行完前一个节点后，根据 State 或返回值决定执行下一个或多个节点；
3. 入口边（Entry Point）：定义 Graph 的第一个节点；
4. 出口边（End Point）：定义 Graph 的结束条件（当执行到某个节点或满足某个条件时，Graph 停止执行）。

这三类边里，条件边是我们实现动态分支的核心——我们下一章会详细讲它的用法。

环境搭建：从零开始配置 LangGraph 开发环境

接下来，我们搭建一个 LangGraph 的开发环境，为后面的代码示例做准备。

1. 系统要求

• Python 3.10 或更高版本（LangGraph 的最新特性需要 Python 3.10+）；
• 一个 OpenAI API Key（或者其他支持 LangChain 的 LLM API Key，比如 Anthropic Claude、百度文心一言等）。

2. 安装依赖库

我们需要安装以下几个核心库：

1. langgraph：LangGraph 的核心库；
2. langchain-openai：LangChain 对 OpenAI 的封装；
3. langchain-core：LangChain 的核心库（包含 PromptTemplate、Runnable 等）；
4. pydantic：用来定义类型安全的 State；
5. python-dotenv：用来从 .env 文件加载环境变量；
6. mermaid-cli（可选）：用来在本地渲染 Mermaid 流程图（如果不需要可以跳过）。

我们可以用 pip 一次性安装所有依赖：

# 创建虚拟环境（推荐，避免依赖冲突）
python -m venv langgraph-env

# 激活虚拟环境
# Windows:
langgraph-env\Scripts\activate
# macOS/Linux:
source langgraph-env/bin/activate

# 升级 pip
pip install --upgrade pip

# 安装核心依赖
pip install langgraph langchain-openai langchain-core pydantic python-dotenv

# 安装可选的 mermaid-cli（需要 Node.js 16+ 环境）
# npm install -g @mermaid-js/mermaid-cli

3. 配置环境变量

在项目根目录下创建一个 .env 文件，把你的 OpenAI API Key 放进去：

# .env 文件
OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
LANGCHAIN_TRACING_V2=true  # 可选：开启 LangChain 的追踪功能，方便调试
LANGCHAIN_API_KEY=ls__xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx  # 可选：LangChain API Key（如果开启了追踪）
LANGCHAIN_PROJECT=ecommerce-customer-service-agent  # 可选：LangChain 追踪的项目名称

4. 测试环境是否配置成功

创建一个 test_env.py 文件，测试一下我们的环境是否配置成功：

# test_env.py
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel

# 加载环境变量
load_dotenv()

# 定义一个简单的 State
class TestState(BaseModel):
    message: str

# 定义一个简单的节点
def test_node(state: TestState):
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
    response = llm.invoke(f"请回复一个简短的欢迎语给用户，用户的消息是：{state.message}")
    return {"message": response.content}

# 构建 Graph
graph_builder = StateGraph(TestState)
graph_builder.add_node("test_node", test_node)
graph_builder.add_edge(START, "test_node")
graph_builder.add_edge("test_node", END)

# 编译 Graph
graph = graph_builder.compile()

# 测试 Graph
result = graph.invoke({"message": "你好，我是测试用户"})
print("测试结果：", result["message"])

如果运行 test_env.py 后能输出一个 AI 生成的欢迎语，说明我们的环境配置成功了！

---

第一板斧：条件路由——把软决策变成硬约束

条件路由（Conditional Routing）是 LangGraph 动态分支的基础——它的核心思想是「根据 State 或前一个节点的返回值，从预定义的边列表中选择一条或多条执行」。

问题背景：为什么 LLM 自己选路径不行？

在没有 LangGraph 之前，很多开发者会用「Prompt 里塞 if-else + 工具选择」的方式让 LLM 自己做决策，比如：

prompt = """
你是一个电商客服，你需要根据用户的输入选择以下工具之一：
1. 如果用户的问题是技术故障，调用 get_tech_support 工具；
2. 如果用户的问题是普通咨询，调用 answer_question 工具；
3. 如果用户的问题是退款申请，调用 process_refund 工具；
4. 如果用户的问题是其他，调用 default_response 工具。

请严格按照以上规则选择工具，不要编造工具名称。

用户的输入：{user_input}
"""

这种方式虽然简单，但有三个致命的缺点：

1. 不可控：LLM 可能会编造工具名称、跳过必要的工具、或者在有明确规则的情况下选择错误的工具（比如把「退款 2000 元」识别成「普通咨询」）；
2. 性能差：每次调用 LLM 都要把所有可能的工具和规则塞给 Prompt，浪费大量的 Token；
3. 难调试：很难追踪 LLM 为什么选择了某个工具——你只能看 Prompt 和 LLM 的输出，无法断点调试决策逻辑。

而 LangGraph 的条件路由，就是用来解决这三个问题的——它把「决策逻辑从 LLM 的脑子里移到了代码里」，变成了可追踪、可测试、可强制的「硬约束」。

问题描述：如何用 LangGraph 实现意图识别后的条件跳转？

我们的第一个需求是：用户输入后，先调用 LLM 做意图识别（严格限制意图的取值为技术故障、普通咨询、退款申请、其他），然后根据识别到的意图跳转到对应的节点。

问题解决：条件路由的两种实现方式

LangGraph 提供了两种实现条件路由的方式：

1. 条件边函数（Conditional Edge Function）：最常用的方式，返回一个字符串（节点名称）或字符串列表（多个节点名称，用于并行执行）；
2. 状态驱动的条件路由（State-Driven Conditional Routing）：稍微复杂一点，但更灵活，直接在构建 Graph 的时候定义状态和边的映射关系。

我们先讲最常用的「条件边函数」方式。

1. 条件边函数方式

条件边函数的签名必须满足以下规则：

1. 第一个参数必须是当前的全局 State；
2. 可以有可选的 config 参数；
3. 返回值必须是：
   a. 单个字符串：对应下一个节点的名称；
   b. 字符串列表：对应下一个要并行执行的多个节点的名称；
   c. END：对应 Graph 的结束节点。

1.1 实现意图识别节点

首先，我们需要实现一个「严格限制意图取值」的意图识别节点——这里我们可以用 LangChain 的「结构化输出（Structured Output）」功能，强制 LLM 返回一个符合我们定义的 IntentEnum 的值。

结构化输出有两种方式：

1. OpenAI 的函数调用（Function Calling）/工具调用（Tool Calling）：适合所有支持函数调用的 LLM；
2. Pydantic 模型直接绑定（with_structured_output 方法）：LangChain 对函数调用的封装，代码更简洁（推荐）。

我们用 with_structured_output 方法来实现意图识别节点：

from typing import Literal
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 重新定义一下 State（和准备工作里的一样，方便大家阅读）
IntentEnum = Literal["技术故障", "普通咨询", "退款申请", "其他"]

class IntentRecognitionOutput(BaseModel):
    """意图识别的结构化输出"""
    intent: IntentEnum = Field(..., description="用户的意图，严格限制为：技术故障、普通咨询、退款申请、其他")
    confidence: float = Field(..., ge=0.0, le=1.0, description="意图识别的置信度，范围是0到1")

# 初始化 LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)  # 温度设为0，让意图识别更准确

# 定义意图识别的 Prompt
intent_prompt = ChatPromptTemplate.from_messages([
    ("system", """
你是一个专业的电商客服意图识别助手，你的任务是根据对话历史和用户的最新输入，识别用户的意图，并给出置信度。

请严格遵守以下规则：
1. 意图必须严格限制为以下四个选项之一：技术故障、普通咨询、退款申请、其他；
2. 置信度必须是一个0到1之间的浮点数，越接近1表示越确定；
3. 如果用户的意图不明确（比如只说了「你好」），请识别为「其他」，并给出较低的置信度（比如0.3）；
4. 不要编造任何信息，严格根据对话历史和用户输入判断。
    """.strip()),
    MessagesPlaceholder(variable_name="messages"),
    ("human", "{user_input}"),
])

# 定义意图识别的链
intent_chain = (
    # 先把 State 里的 user_input 和 messages 提取出来
    {"user_input": lambda state: state.user_input, "messages": lambda state: state.messages}
    # 然后调用 Prompt
    | intent_prompt
    # 然后调用 LLM 的结构化输出
    | llm.with_structured_output(IntentRecognitionOutput)
    # 最后把结果转换成可以更新 State 的格式
    | lambda x: {
        "intent": x.intent,
        "messages": [("ai", f"我识别到您的意图是：{x.intent}（置信度：{x.confidence:.2f}）。")] if x.confidence >= 0.7 else [("ai", "不好意思，我没太听懂您的意思，请您详细描述一下您的需求。")]
    }
)

这里有三个关键点需要注意：

• 温度设为 0.0：意图识别是一个「确定性」的任务，我们不需要 LLM 有任何创造性，所以把温度设为 0.0 可以最大程度地保证意图识别的准确性；
• 结构化输出的验证：with_structured_output 方法会自动验证 LLM 的输出是否符合 IntentRecognitionOutput 的定义——如果不符合，会自动重试（默认重试 3 次）；
• 低置信度的处理：如果意图识别的置信度低于 0.7，我们就直接让 AI 让用户详细描述，而不是跳转到任何具体的节点——这可以有效减少错误的条件跳转。

1.2 实现条件边函数

接下来，我们需要实现一个条件边函数，根据 State 里的 intent 和 confidence 字段决定下一步跳转到哪个节点：

from langgraph.graph import END

def intent_conditional_edge(state: ECommerceState):
    """
    意图识别后的条件边函数
    """
    # 先检查有没有识别到意图
    if not state.intent:
        return END  # 如果没有识别到意图，直接结束（其实前面的意图识别链已经处理了低置信度的情况，这里是兜底）
    
    # 再检查置信度
    if state.confidence < 0.7:
        return END  # 如果置信度低于0.7，直接结束（兜底）
    
    # 根据意图跳转到对应的节点
    if state.intent == "技术故障":
        return "技术支持"
    elif state.intent == "普通咨询":
        return "普通咨询"
    elif state.intent == "退款申请":
        return "退款前置检查"
    else:  # 其他
        return END

1.3 构建完整的条件路由 Graph

最后，我们把意图识别节点、条件边函数和其他几个简单的节点（先空着，后面再补）组装成一个完整的 Graph：

from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel, Field
from typing import Annotated, Optional
from langgraph.graph import add_messages

# 重新定义 ECommerceState（方便大家阅读）
class ECommerceState(BaseModel):
    messages: Annotated[list, add_messages] = Field(default_factory=list)
    user_input: str = Field(...)
    intent: Optional[IntentEnum] = Field(default=None)
    confidence: Optional[float] = Field(default=None, ge=0.0, le=1.0)
    # 后面会用到的其他字段
    refund_order_id: Optional[str] = Field(default=None)
    refund_amount: Optional[float] = Field(default=None)
    refund_reason: Optional[str] = Field(default=None)
    audit_result: Optional[AuditResultEnum] = Field(default=None)
    audit_comment: Optional[str] = Field(default=None)

# 先定义几个空节点，后面再补
def tech_support_node(state: ECommerceState):
    return {"messages": [("ai", "这里是技术支持节点，后面会补全。")]}

def consult_node(state: ECommerceState):
    return {"messages": [("ai", "这里是普通咨询节点，后面会补全。")]}

def refund_pre_check_node(state: ECommerceState):
    return {"messages": [("ai", "这里是退款前置检查节点，后面会补全。")]}

# 构建 Graph
graph_builder = StateGraph(ECommerceState)

# 添加节点
graph_builder.add_node("意图识别", intent_chain)
graph_builder.add_node("技术支持", tech_support_node)
graph_builder.add_node("普通咨询", consult_node)
graph_builder.add_node("退款前置检查", refund_pre_check_node)

# 添加入口边
graph_builder.add_edge(START, "意图识别")

# 添加条件边
graph_builder.add_conditional_edges(
    "意图识别",  # 源节点
    intent_conditional_edge,  # 条件边函数
    # 可选：条件边函数返回值和目标节点的映射关系（如果返回值和目标节点名称一致，可以省略）
    {
        "技术支持": "技术支持",
        "普通咨询": "普通咨询",
        "退款前置检查": "退款前置检查",
        END: END
    }
)

# 添加出口边（空节点暂时先直接连到 END）
graph_builder.add_edge("技术支持", END)
graph_builder.add_edge("普通咨询", END)
graph_builder.add_edge("退款前置检查", END)

# 编译 Graph
graph = graph_builder.compile()

1.4 测试条件路由 Graph

现在，我们来测试一下这个条件路由 Graph：

from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 测试 1：普通咨询
print("=== 测试 1：普通咨询 ===")
result1 = graph.invoke({"user_input": "你们的iPhone 16 Pro Max什么时候发货？"})
for msg in result1["messages"]:
    print(f"{msg.type if hasattr(msg, 'type') else msg[0]}: {msg.content if hasattr(msg, 'content') else msg[1]}")

# 测试 2：技术故障
print("\n=== 测试 2：技术故障 ===")
result2 = graph.invoke({"user_input": "我的AirPods Pro 2连不上我的MacBook Pro了，怎么办？"})
for msg in result2["messages"]:
    print(f"{msg.type if hasattr(msg, 'type') else msg[0]}: {msg.content if hasattr(msg, 'content') else msg[1]}")

# 测试 3：退款申请
print("\n=== 测试 3：退款申请 ===")
result3 = graph.invoke({"user_input": "我要退我昨天买的iPhone 16 Pro Max，订单号是123456789。"})
for msg in result3["messages"]:
    print(f"{msg.type if hasattr(msg, 'type') else msg[0]}: {msg.content if hasattr(msg, 'content') else msg[1]}")

# 测试 4：其他（低置信度）
print("\n=== 测试 4：其他（低置信度） ===")
result4 = graph.invoke({"user_input": "你好"})
for msg in result4["messages"]:
    print(f"{msg.type if hasattr(msg, 'type') else msg[0]}: {msg.content if hasattr(msg, 'content') else msg[1]}")

如果测试成功，你会看到：

• 测试 1：AI 先识别到「普通咨询」，然后跳转到「普通咨询」节点；
• 测试 2：AI 先识别到「技术故障」，然后跳转到「技术支持」节点；
• 测试 3：AI 先识别到「退款申请」，然后跳转到「退款前置检查」节点；
• 测试 4：AI 识别到「其他」，置信度较低，直接让用户详细描述。

2. 状态驱动的条件路由方式

状态驱动的条件路由方式稍微复杂一点，但更灵活——它不需要定义条件边函数，而是直接在构建 Graph 的时候定义「状态字段的取值」和「目标节点」的映射关系。

比如，我们可以把上面的意图识别条件路由改成状态驱动的方式：

# 构建 Graph（状态驱动的条件路由方式）
graph_builder = StateGraph(ECommerceState)

# 添加节点（和之前一样）
graph_builder.add_node("意图识别", intent_chain)
graph_builder.add_node("技术支持", tech_support_node)
graph_builder.add_node("普通咨询", consult_node)
graph_builder.add_node("退款前置检查", refund_pre_check_node)

# 添加入口边（和之前一样）
graph_builder.add_edge(START, "意图识别")

# 添加状态驱动的条件边
# 注意：状态驱动的条件边只能根据「单个字段的取值」判断，不能根据多个字段或复杂的逻辑判断
graph_builder.add_conditional_edges(
    "意图识别",
    # 这里直接传一个 lambda 函数，返回要判断的状态字段的取值
    lambda state: state.intent if (state.intent and state.confidence >= 0.7) else END,
    # 这里必须传映射关系，因为返回值可能是 END
    {
        "技术故障": "技术支持",
        "普通咨询": "普通咨询",
        "退款申请": "退款前置检查",
        END: END
    }
)

# 添加出口边（和之前一样）
graph_builder.add_edge("技术支持", END)
graph_builder.add_edge("普通咨询", END)
graph_builder.add_edge("退款前置检查", END)

# 编译 Graph（和之前一样）
graph = graph_builder.compile()

可以看到，状态驱动的条件路由方式虽然不需要定义单独的条件边函数，但它的局限性很大——它只能根据「单个字段的取值」判断，不能根据多个字段或复杂的逻辑判断（比如「置信度≥0.7 且 意图是技术故障」）。所以，我们推荐大家优先使用条件边函数方式。

边界与外延：条件路由的扩展用法

1. 并行条件路由

前面讲的条件路由都是返回「单个字符串」，跳转到「单个节点」——但 LangGraph 的条件路由也支持返回「字符串列表」，跳转到「多个并行执行的节点」。

比如，如果我们的意图识别结果是「产品咨询+物流查询」，我们可以让「产品咨询节点」和「物流查询节点」并行执行，然后再合并结果。

不过，并行执行有两个关键点需要注意：

• Reducer 机制：并行执行的多个节点返回的结果会用 Reducer 合并（比如 add_messages 会追加所有节点返回的 AI 消息）；
• 合并节点（Joiner Node）：如果并行执行的多个节点需要把结果合并后再传给下一个节点，我们需要显式添加一个「合并节点」，然后把所有并行节点的边都连到这个合并节点上。

2. 嵌套条件路由

条件路由也可以嵌套使用——比如，先根据「意图」跳转到「退款前置检查节点」，然后在「退款前置检查节点」之后再根据「退款金额」跳转到「自动退款节点」或「主管审核节点」。

我们下一章讲子图复用的时候会用到嵌套条件路由。

3. 循环条件路由

条件路由还可以用来实现「循环」——比如，先调用「代码生成节点」生成一段代码，然后调用「代码测试节点」测试代码，如果测试通过就结束，如果测试不通过就重新跳转到「代码生成节点」，直到测试通过或达到最大循环次数。

循环条件路由有一个关键点需要注意：最大循环次数——为了避免死循环，我们必须在 State 里定义一个「循环次数」字段，然后在条件边函数里判断循环次数是否达到最大值。

---

第二板斧：子图复用——把重复流程变成可拼接的积木

子图复用（Subgraph Reuse）是 LangGraph 动态分支的进阶——它的核心思想是「把重复的流程封装成独立的 Graph（子图），然后像拼积木一样插入到主图的任意位置」。

问题背景：为什么要复用子图？

在复杂的 Agent 开发中，我们经常会遇到重复的流程——比如：

1. 在电商客服 Agent 中，「主管审核子流程」不仅会用到「大额退款」场景，还会用到「大额折扣」、「大额换货」等场景；
2. 在代码生成 Agent 中，「代码测试子流程」不仅会用到「Python 代码生成」场景，还会用到「JavaScript 代码生成」、「Go 代码生成」等场景；
3. 在文档处理 Agent 中，「文档预处理子流程」（加载、分块、向量化）不仅会用到「RAG 问答」场景，还会用到「文档摘要」、「文档翻译」等场景。

如果我们每次都复制粘贴这些重复的流程，会导致三个严重的问题：

1. 代码冗余：代码量会急剧增加，很难维护；
2. 逻辑不一致：如果我们需要修改重复流程中的某个步骤（比如把主管审核的 LLM 从 gpt-4o-mini 改成 gpt-4o），我们需要在所有复制粘贴的地方都修改一遍，很容易出错；
3. 可测试性差：重复的流程很难单独测试——我们必须启动整个主图才能测试。

而 LangGraph 的子图复用，就是用来解决这三个问题的——它把「重复的流程封装成独立的、可测试的、可维护的 Graph」，然后像拼积木一样插入到主图的任意位置。

问题描述：如何把「主管审核子流程」封装成子图并复用？

我们的第二个需求是：把「主管审核子流程」封装成独立的子图，然后在「大额退款」、「大额折扣」等场景中复用。

「主管审核子流程」的工作流是这样的：

1. 接收主图传递过来的审核信息（比如退款金额、退款原因、折扣金额、折扣原因等）；
2. 生成审核请求的 Prompt，喂给 LLM（模拟主管审核）；
3. 让 LLM 生成结构化的审核结果（同意/拒绝、审核意见）；
4. 把审核结果返回给主图。

问题解决：子图的定义、编译与嵌入

LangGraph 的子图复用非常简单——它的核心逻辑是「子图也是一个 Runnable」，所以我们可以像使用普通的 ChatOpenAI 或 Chain 一样使用子图。

1. 子图的定义

定义子图的步骤和定义主图的步骤完全一样：

1. 定义子图的 State；
2. 定义子图的节点；
3. 定义子图的边；
4. 编译子图。

不过，子图的 State 有一个关键点需要注意：子图的 State 必须和主图的 State 兼容——也就是说，子图需要用到的主图的字段，必须在子图的 State 里定义；子图返回给主图的字段，也必须在主图的 State 里定义。

我们可以用两种方式让子图的 State 和主图的 State 兼容：

1. 子图 State 继承主图 State（推荐，类型安全）：子图的 State 继承自主图的 State，然后只定义子图需要用到的额外字段；
2. 子图 State 是主图 State 的子集（灵活但不安全）：子图的 State 只定义子图需要用到的主图的字段和额外字段。

我们后面的代码会用「子图 State 继承主图 State」的方式，因为它不仅类型安全，还能让子图直接访问主图的所有字段（虽然我们不推荐子图访问主图的所有字段，最好只访问它需要用到的字段）。

1.1 定义子图的 State

首先，我们定义「主管审核子图」的 State——它继承自主图的 ECommerceState，然后没有定义额外的字段（因为子图需要用到的字段都在主图的 State 里了）：

# 子图的 State 继承自主图的 ECommerceState
class AuditSubgraphState(ECommerceState):
    """主管审核子图的 State，继承自主图的 State"""
    # 这里不需要定义额外的字段，因为子图需要用到的字段都在主图的 State 里了
    # 当然，你也可以定义子图专用的字段，比如 audit_request_id 等
    pass

1.2 定义子图的节点

接下来，我们定义「主管审核子图」的节点——这里我们用结构化输出的方式模拟主管审核：

from typing import Literal
from pydantic import BaseModel, Field

# 定义审核结果枚举
AuditResultEnum = Literal["同意", "拒绝"]

class AuditOutput(BaseModel):
    """主管审核的结构化输出"""
    audit_result: AuditResultEnum = Field(..., description="审核结果，严格限制为：同意、拒绝")
    audit_comment: str = Field(..., description="审核意见，必须说明同意或拒绝的理由")

# 初始化审核用的 LLM（温度设为 0.3，稍微有一点创造性，但不要太离谱）
audit_llm = ChatOpenAI(model="gpt-4o", temperature=0.3)

# 定义主管审核的 Prompt
audit_prompt = ChatPromptTemplate.from_messages([
    ("system", """
你是一个专业、严格的电商客服主管，你的任务是根据审核信息决定是否同意用户的申请。

请严格遵守以下规则：
1. 审核结果必须严格限制为以下两个选项之一：同意、拒绝；
2. 审核意见必须说明同意或拒绝的理由，不能只说「同意」或「拒绝」；
3. 如果申请的金额≥5000元，必须拒绝，除非有特殊说明（但目前没有特殊说明）；
4. 如果申请的金额≥1000元但<5000元，需要仔细检查申请原因是否合理，如果合理就同意，如果不合理就拒绝；
5. 如果申请的金额<1000元，直接同意（但这个子图只会处理金额≥1000元的申请）；
6. 不要编造任何信息，严格根据审核信息判断。

审核信息的格式如下：
- 申请类型：{application_type}
- 申请金额：{application_amount}
- 申请原因：{application_reason}
- 申请订单号：{application_order_id}（可选）
    """.strip()),
    # 这里不需要 MessagesPlaceholder，因为子图不需要处理对话历史（如果需要也可以加）
    ("human", """
申请类型：{application_type}
申请金额：{application_amount}
申请原因：{application_reason}
申请订单号：{application_order_id}
    """.strip())
])

# 定义主管审核的节点
def audit_node(state: AuditSubgraphState):
    """
    主管审核节点
    """
    # 先检查有没有必要的审核信息
    if not state.refund_amount and not state.discount_amount:
        return {
            "audit_result": "拒绝",
            "audit_comment": "缺少必要的审核信息（申请金额）",
            "messages": [("ai", "不好意思，主管审核时发现缺少必要的信息，请您补充。")]
        }
    
    # 确定申请类型、申请金额、申请原因、申请订单号
    application_type = "退款申请" if state.refund_amount else "折扣申请"
    application_amount = state.refund_amount if state.refund_amount else state.discount_amount
    application_reason = state.refund_reason if state.refund_reason else state.discount_reason
    application_order_id = state.refund_order_id if state.refund_order_id else state.discount_order_id
    
    # 调用审核链
    audit_chain = (
        {
            "application_type": lambda _: application_type,
            "application_amount": lambda _: application_amount,
            "application_reason": lambda _: application_reason,
            "application_order_id": lambda _: application_order_id or "无"
        }
        | audit_prompt
        | audit_llm.with_structured_output(AuditOutput)
    )
    audit_output = audit_chain.invoke({})
    
    # 把结果转换成可以更新 State 的格式
    return {
        "audit_result": audit_output.audit_result,
        "audit_comment": audit_output.audit_comment,
        "messages": [("ai", f"主管审核结果：{audit_output.audit_result}。\n审核意见：{audit_output.audit_comment}")]
    }

这里有三个关键点需要注意：

• 子图的 State 继承自主图：所以子图可以直接访问主图的 refund_amount、refund_reason 等字段；
• 子图的通用性：我们没有把申请类型硬编码成「退款申请」，而是根据 State 里的字段判断申请类型——这样这个子图就可以在「退款申请」、「折扣申请」、「换货申请」等场景中复用；
• 结构化输出的验证：和之前的意图识别节点一样，我们用 with_structured_output 方法强制 LLM 返回符合 AuditOutput 定义的审核结果。

1.3 定义子图的边

接下来，我们定义「主管审核子图」的边——这个子图很简单，只有一个节点，所以我们只需要添加入口边和出口边：

from langgraph.graph import StateGraph, START, END

# 构建子图
audit_subgraph_builder = StateGraph(AuditSubgraphState)

# 添加节点
audit_subgraph_builder.add_node("主管审核", audit_node)

# 添加入口边
audit_subgraph_builder.add_edge(START, "主管审核")

# 添加出口边
audit_subgraph_builder.add_edge("主管审核", END)

1.4 编译子图

最后，我们编译子图——编译后的子图就是一个 Runnable，可以像普通的 ChatOpenAI 或 Chain 一样使用：

# 编译子图
audit_subgraph = audit_subgraph_builder.compile()

2. 子图的嵌入

编译好子图之后，我们就可以把它嵌入到