LangGraph的条件分支：基于状态的动态路由设计

引言

痛点引入：从“线性对话机器人”到“智能决策系统”的困境

在LangChain生态普及之前，我们构建AI应用时最常用的是“线性链式调用”（LangChain LCEL最初的简化场景）——比如先通过提示词模板处理用户输入，调用LLM生成回复，最后把结果返回给用户。这种模式对于简单的FAQ查询、文本摘要、翻译任务确实够用，但当我们的需求稍微复杂一点，就会立刻碰壁：

• 场景1：多意图对话系统 用户可能说“帮我查明天北京的天气，如果下雨的话就推荐附近的室内商场，如果不下雨就推荐奥林匹克森林公园，还要算上我和朋友3个人的地铁出行时间”。这里的核心问题是：根据“天气结果”这个动态生成的中间状态，程序的执行路径必须发生跳转——线性链根本做不到。
• 场景2：工具执行的容错处理 当调用搜索引擎、数据库、API这类外部工具时，难免会遇到超时、权限不足、结果为空的情况。线性链要么直接报错崩溃，要么只能返回一个固定的“出错了”模板，完全无法根据工具执行的“返回状态码”或“结果有效性检查状态”进行重试、降级（比如搜索引擎超时就换知识图谱检索）、或者返回更友好的个性化错误提示。
• 场景3：复杂任务分解与多agent协作 比如写一篇学术论文综述，需要先通过检索agent查近5年的顶会论文，分类agent把论文按研究方向归类，总结agent分别对每个方向写子综述，整合agent把所有子综述拼成完整的结构并润色。这里的任务分解本身就依赖检索agent返回的“论文主题词分布状态”——如果某个主题词下没有足够的高质量论文，可能需要通知用户补充关键词，或者调整检索范围；如果论文太多，可能需要先让筛选agent过滤出引用量Top 20的论文。

更麻烦的是，线性链很难维护全局的上下文状态——比如场景1中，我们需要在查完天气后，记住“用户的位置”、“朋友的人数”、“室内/室外推荐的需求来源”这些信息，才能给后续的商场/公园推荐和地铁时间计算提供支持。虽然LCEL后来引入了RunnablePassthrough和RunnableLambda来传递和修改状态，但这种状态管理是隐式的、难以追踪的——当调用链超过3层时，我们根本不知道某个状态变量在哪里被修改了，也无法在调试时查看完整的状态变化历史。

解决方案概述：LangGraph的Stateful Graph & Conditional Edge

LangChain团队在2024年初正式推出了LangGraph——一个专门为构建状态化、可循环、可分支的多agent/复杂工具应用设计的框架。LangGraph的核心创新在于两个方面：

1. 显式的全局状态管理（Stateful Graph）：LangGraph要求开发者在初始化图的时候就明确定义一个状态schema（可以是Pydantic模型、TypedDict，甚至是普通的Python字典），所有的节点（Node，比如LLM调用、工具执行、自定义函数）只能读取和修改这个显式定义的全局状态，而不能直接传递参数或返回值——这种设计让状态的变化完全可控、可追踪，调试时可以直接打印或可视化状态的每一步更新。
2. 条件边（Conditional Edge）与动态路由（Dynamic Routing）：LangGraph不仅支持从节点A固定跳到节点B的“普通边（Normal Edge）”，还支持根据当前的全局状态、或者某个“路由判断函数（Router Function）”的返回值，动态选择下一个要执行的节点的“条件边”——这正是解决我们前面痛点的关键！

具体来说，条件边的工作流程是这样的：

• 当一个节点执行完毕后，它会修改全局状态；
• 然后，LangGraph会检查这个节点后面连接的是普通边还是条件边；
• 如果是条件边，就会调用开发者预先定义好的路由判断函数，传入当前的全局状态，得到一个“目标节点标识”或者“END标识”（表示图的执行结束）；
• 最后，LangGraph会根据这个返回值，跳转到对应的目标节点继续执行，或者结束整个流程。

最终效果展示：一个具备容错和多分支能力的“智能天气助手”

在文章的最后，我们会用LangGraph构建一个完整的、功能丰富的智能天气助手，它具备以下特性：

1. 多意图识别：用户可以输入查天气、查路线、推荐场所、或者组合意图；
2. 条件分支路由：根据识别到的意图，跳转到不同的处理节点；
3. 工具执行容错：如果天气API调用失败，会自动重试2次，重试失败后会切换到“本地缓存的历史天气数据”；如果路线API调用超时，会推荐用户使用高德地图/百度地图的小程序；
4. 状态维护与上下文感知：记住用户之前查过的位置、时间、朋友人数等信息，后续对话可以省略这些参数；
5. 循环执行：如果用户补充了新的参数（比如补充了朋友的人数），会自动回到之前的处理节点重新执行；
6. 可追踪的状态变化：可以通过LangSmith或者LangGraph内置的可视化工具，查看整个对话流程的执行路径和每一步的状态更新。

---

准备工作

环境/工具

在开始学习LangGraph的条件分支之前，我们需要先搭建好开发环境。以下是我们需要用到的所有工具和依赖库：

工具/依赖库	版本要求（推荐）	用途说明
Python	≥3.10	LangGraph官方支持的最低Python版本，3.10及以上版本支持更简洁的类型注解。
langgraph	≥0.2.0	本文的核心框架，0.2.0版本对条件边的API进行了简化，新增了RouterFunction类型提示。
langchain	≥0.2.0	LangGraph的基础依赖，提供LLM接口、工具接口、提示词模板等组件。
langchain-openai	≥0.1.0	对接OpenAI的LLM（GPT-4o-mini推荐用于调试，GPT-4o推荐用于生产）和Embedding模型。
langchain-community	≥0.2.0	提供第三方工具的集成（比如天气API、路线API）。
python-dotenv	≥1.0.0	用于管理环境变量（比如OpenAI API Key、天气API Key）。
pydantic	≥2.0.0	用于定义显式的状态schema，提供类型验证和数据序列化功能。
langsmith（可选）	≥0.1.0	用于调试和监控LangGraph应用的执行流程、状态变化、LLM调用等。
jupyter notebook/lab（可选）	≥6.0.0	推荐用于学习和调试，支持代码的分步执行和结果的实时展示。

环境安装步骤

1. 创建虚拟环境（强烈推荐，避免依赖冲突）：

   # 使用conda创建虚拟环境（推荐，管理Python版本更方便）
   conda create -n langgraph-cond-edge python=3.11 -y
   conda activate langgraph-cond-edge
   
   # 或者使用venv创建虚拟环境
   python3.11 -m venv langgraph-cond-edge
   source langgraph-cond-edge/bin/activate  # Linux/Mac
   # langgraph-cond-edge\Scripts\activate  # Windows
   

2. 安装依赖库：

   # 安装核心依赖
   pip install langgraph langchain langchain-openai langchain-community python-dotenv pydantic
   
   # 安装可选依赖
   pip install langsmith jupyterlab
   

3. 配置环境变量：
   在项目根目录下创建一个.env文件，写入以下内容（需要替换成你自己的API Key）：

   # OpenAI API配置
   OPENAI_API_KEY="sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
   OPENAI_API_BASE="https://api.openai.com/v1"  # 如果你用的是国内代理，需要修改这里
   
   # 天气API配置（这里我们用免费的OpenWeatherMap API，需要去官网注册：https://openweathermap.org/api）
   OPENWEATHERMAP_API_KEY="xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
   
   # 路线API配置（这里我们用免费的高德地图Web服务API，需要去官网注册：https://lbs.amap.com/）
   AMAP_WEB_SERVICE_API_KEY="xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
   
   # LangSmith配置（可选，用于调试）
   LANGCHAIN_TRACING_V2="true"
   LANGCHAIN_API_KEY="ls__xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
   LANGCHAIN_PROJECT="langgraph-conditional-edge-tutorial"
   

基础知识

为了更好地理解LangGraph的条件分支，你需要具备以下前置知识：

1. Python基础：熟悉Python的基本语法、函数、类、类型注解（尤其是TypedDict和Pydantic模型）。
2. LangChain基础：了解LangChain的核心概念，比如提示词模板（PromptTemplate）、LLM接口（ChatOpenAI）、工具接口（Tool）、链式调用（LCEL）。
   • 如果你对LangChain不太熟悉，可以先阅读LangChain官方的快速入门文档：LangChain Quickstart
3. 图论基础：了解图的基本概念，比如节点（Node）、边（Edge）、有向图（Directed Graph）、无环图（DAG）、循环图（Cyclic Graph）——不过即使你没有图论基础，也没关系，LangGraph的API设计得非常直观，我们会在后面详细讲解。

---

LangGraph的核心基础：Stateful Graph

在深入学习条件分支之前，我们必须先掌握LangGraph的第一个核心概念——Stateful Graph（状态化图），因为条件边的所有判断逻辑都是基于“当前的全局状态”来执行的。

核心概念

什么是Stateful Graph？

Stateful Graph是LangGraph的一种特殊类型的有向图，它由以下三个核心部分组成：

1. 全局状态（Global State）：一个显式定义的、所有节点共享的数据结构，用于存储整个图执行过程中的所有上下文信息、中间结果、用户输入等。
2. 节点（Node）：图中的执行单元，每个节点只能读取和修改全局状态，而不能直接传递参数或返回值。节点可以是以下几种类型：
   • 自定义函数节点（Function Node）：开发者自己写的Python函数，接收当前的全局状态作为输入，返回一个状态更新字典（用于修改全局状态）。
   • 工具调用节点（Tool Node）：使用LangChain的Tool接口封装的外部工具（比如搜索引擎、天气API），LangGraph会自动处理工具调用的参数生成（从全局状态中提取）、工具执行、结果存储到全局状态的过程。
   • LLM调用节点（LLM Node）：使用LangChain的ChatModel接口封装的LLM（比如GPT-4o），LangGraph会自动处理提示词的生成（结合全局状态）、LLM调用、结果解析（如果是结构化输出）、结果存储到全局状态的过程。
   • 子图节点（Subgraph Node）：把另一个Stateful Graph作为一个节点嵌入到当前的图中，用于模块化地构建复杂的应用。
3. 边（Edge）：图中的连接单元，用于定义节点之间的执行顺序。边可以是以下几种类型：
   • 入口边（Entry Point Edge）：连接到图的第一个节点的边，也就是图的执行入口。
   • 普通边（Normal Edge）：从节点A固定跳到节点B的边。
   • 条件边（Conditional Edge）：根据当前的全局状态或路由判断函数的返回值，动态选择下一个要执行的节点的边——这是我们本文的重点。
   • 结束边（END Edge）：连接到END常量的边，表示图的执行结束。

全局状态的定义方式

LangGraph支持三种定义全局状态的方式：

1. 普通Python字典（dict）：最简单的方式，适合快速原型开发，但没有类型验证，容易出错。
2. TypedDict：Python 3.8及以上版本支持的类型注解方式，提供了基础的类型提示，但没有运行时的类型验证。
3. Pydantic模型（BaseModel）：LangGraph官方推荐的方式，提供了严格的运行时类型验证、数据序列化/反序列化、默认值设置、字段别名等功能，适合生产环境开发。

我们会在后面的示例中重点使用Pydantic模型来定义全局状态。

状态更新的规则：Reducer（归约器）

在LangGraph中，全局状态的更新不是简单的“覆盖”，而是通过**Reducer（归约器）**来实现的——Reducer是一个函数，它接收当前的全局状态和节点返回的状态更新字典作为输入，返回一个新的全局状态。

LangGraph为不同类型的字段提供了默认的Reducer：

• 标量字段（比如str、int、float、bool）：默认的Reducer是“覆盖”——如果节点返回的状态更新字典中包含这个字段，就用新值覆盖旧值；如果不包含，就保留旧值。
• 列表字段（比如list[str]、list[dict]）：默认的Reducer是“追加”——如果节点返回的状态更新字典中包含这个字段，就把新值追加到旧值的后面；如果不包含，就保留旧值。
• 字典字段（比如dict[str, str]、dict[str, int]）：默认的Reducer是“合并”——如果节点返回的状态更新字典中包含这个字段，就把新字典和旧字典合并（新字典的键会覆盖旧字典的同名键）；如果不包含，就保留旧值。

当然，我们也可以为某个字段自定义Reducer，比如我们可以为一个“对话历史”列表字段定义一个“只保留最近10条消息”的Reducer，避免对话历史过长导致LLM的上下文窗口溢出。

我们会在后面的示例中详细讲解如何自定义Reducer。

状态的可视化与调试

LangGraph提供了两个非常实用的工具来可视化和调试状态化图：

1. 内置的可视化工具：使用graph.get_graph().draw_mermaid_png()或graph.get_graph().draw_ascii()方法，可以把图的结构（包括节点、边、状态schema）可视化出来——前者生成PNG图片，后者生成ASCII字符画，适合在终端或Jupyter Notebook中查看。
2. LangSmith：如果配置了LangSmith，就可以在LangSmith的控制台中查看整个图的执行流程、每一步的状态更新、LLM调用的输入输出、工具调用的参数和结果、执行时间等详细信息——这是调试复杂LangGraph应用的最佳工具。

---

LangGraph的条件分支核心：Conditional Edge & Router Function

现在我们终于进入了本文的核心部分——条件边（Conditional Edge）和路由判断函数（Router Function）。

核心概念

什么是Conditional Edge？

Conditional Edge是LangGraph的一种特殊类型的边，它的执行顺序不是固定的，而是根据当前的全局状态或路由判断函数的返回值动态确定的。

Conditional Edge的定义方式有两种：

1. 基于状态字段的条件边：直接根据全局状态中的某个字段的值来选择下一个节点——这种方式最简单，但灵活性较低。
2. 基于Router Function的条件边：定义一个专门的路由判断函数，接收当前的全局状态作为输入，返回一个“目标节点标识”或者“END标识”——这种方式灵活性最高，可以实现任意复杂的路由逻辑。

我们会在后面的示例中详细讲解这两种定义方式。

什么是Router Function？

Router Function是一个Python函数，它的签名必须符合以下要求：

from typing import Literal, Union
from langgraph.graph import END

# Router Function的类型提示（LangGraph 0.2.0及以上版本）
def my_router_function(state: MyStateSchema) -> Union[Literal["node_a", "node_b", "node_c"], END]:
    # 路由判断逻辑
    # ...
    # 返回目标节点标识或者END
    return "node_a"

其中：

• MyStateSchema是我们定义的全局状态schema；
• 返回值必须是一个字符串（目标节点标识）或者END常量（表示图的执行结束）；
• 我们可以使用typing.Literal来限制返回值的范围，这样LangGraph在初始化图的时候就会进行类型检查，避免返回不存在的目标节点标识。

问题背景与问题描述

为了更好地理解条件边的使用场景，我们先来看一个具体的问题：

问题背景

我们要构建一个“简单的文本分类与处理系统”，它的功能如下：

1. 接收用户输入的一段文本；
2. 调用LLM对文本进行分类，分类结果只能是以下三种之一："positive"（正面情感）、"negative"（负面情感）、"neutral"（中性情感）；
3. 根据分类结果，执行不同的处理：
   • 如果是"positive"：调用LLM生成一段“感谢用户的正面评价”的回复；
   • 如果是"negative"：调用LLM生成一段“道歉并承诺改进”的回复；
   • 如果是"neutral"：调用LLM生成一段“确认收到用户的信息”的回复；
4. 把处理后的回复返回给用户，结束整个流程。

问题描述

这个问题看起来很简单，但如果用线性链来实现的话，会非常麻烦：

• 我们需要在LLM生成分类结果之后，用RunnableLambda来判断分类结果，然后根据判断结果调用不同的提示词模板和LLM；
• 但是线性链的执行顺序是固定的，我们无法真正地“跳转”，只能“跳过”某些步骤——比如我们可以先生成所有三种可能的回复，然后根据分类结果选择其中一种返回给用户，但这样会浪费LLM的调用次数和时间。

而用LangGraph的条件边来实现的话，就会非常简洁、高效——我们可以在分类节点后面连接一个条件边，根据分类结果直接跳转到对应的回复生成节点，然后跳转到END。

问题解决：一步步构建“文本分类与处理系统”

现在我们就来一步步地用LangGraph的条件边构建这个“文本分类与处理系统”。

步骤1：定义全局状态schema

首先，我们用Pydantic模型来定义全局状态schema——我们需要存储用户输入的文本、LLM生成的分类结果、LLM生成的回复。

from pydantic import BaseModel, Field
from typing import Optional

# 定义全局状态schema
class TextProcessingState(BaseModel):
    # 用户输入的文本（必填字段，没有默认值）
    user_input: str = Field(description="用户输入的需要处理的文本")
    # LLM生成的分类结果（可选字段，初始值为None）
    classification: Optional[str] = Field(description="文本分类结果，只能是'positive'、'negative'、'neutral'三种之一", default=None)
    # LLM生成的回复（可选字段，初始值为None）
    response: Optional[str] = Field(description="根据分类结果生成的回复", default=None)

步骤2：初始化Stateful Graph

接下来，我们初始化一个Stateful Graph，传入我们定义的全局状态schema。

from langgraph.graph import StateGraph

# 初始化Stateful Graph
graph_builder = StateGraph(TextProcessingState)

步骤3：定义节点

然后，我们定义三个节点：

1. 分类节点（classification_node）：调用LLM对用户输入的文本进行分类，把分类结果存储到全局状态的classification字段中。
2. 正面回复节点（positive_response_node）：调用LLM生成一段“感谢用户的正面评价”的回复，把回复存储到全局状态的response字段中。
3. 负面回复节点（negative_response_node）：调用LLM生成一段“道歉并承诺改进”的回复，把回复存储到全局状态的response字段中。
4. 中性回复节点（neutral_response_node）：调用LLM生成一段“确认收到用户的信息”的回复，把回复存储到全局状态的response字段中。

首先，我们需要初始化一个ChatOpenAI实例：

from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 初始化ChatOpenAI实例（用GPT-4o-mini调试，速度快、成本低）
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

然后，我们定义分类节点——这里我们用LangChain的with_structured_output方法来让LLM生成结构化的分类结果，这样可以避免LLM返回一些无关的文本，提高分类的准确性：

from langchain.prompts import PromptTemplate
from typing import Literal

# 定义分类结果的结构化输出schema
class ClassificationOutput(BaseModel):
    sentiment: Literal["positive", "negative", "neutral"] = Field(description="文本的情感分类结果")

# 定义分类的提示词模板
classification_prompt = PromptTemplate(
    template="请对以下用户输入的文本进行情感分类，分类结果只能是'positive'（正面情感）、'negative'（负面情感）、'neutral'（中性情感）三种之一。\n\n用户输入：{user_input}",
    input_variables=["user_input"]
)

# 定义分类的链式调用（提示词模板 -> LLM -> 结构化输出）
classification_chain = classification_prompt | llm.with_structured_output(ClassificationOutput)

# 定义分类节点
def classification_node(state: TextProcessingState) -> dict:
    # 调用分类链
    classification_output = classification_chain.invoke({"user_input": state.user_input})
    # 返回状态更新字典（把分类结果存储到全局状态的classification字段中）
    return {"classification": classification_output.sentiment}

# 把分类节点添加到图中
graph_builder.add_node("classification_node", classification_node)

接下来，我们定义三个回复生成节点——这里我们分别用不同的提示词模板：

# 定义正面回复的提示词模板
positive_response_prompt = PromptTemplate(
    template="请生成一段真诚的、友好的感谢用户正面评价的回复，回复的长度控制在50-100字之间。\n\n用户输入：{user_input}",
    input_variables=["user_input"]
)

# 定义正面回复的链式调用
positive_response_chain = positive_response_prompt | llm

# 定义正面回复节点
def positive_response_node(state: TextProcessingState) -> dict:
    # 调用正面回复链
    positive_response = positive_response_chain.invoke({"user_input": state.user_input})
    # 返回状态更新字典
    return {"response": positive_response.content}

# 把正面回复节点添加到图中
graph_builder.add_node("positive_response_node", positive_response_node)

# 定义负面回复的提示词模板
negative_response_prompt = PromptTemplate(
    template="请生成一段真诚的、歉意的道歉并承诺改进的回复，回复的长度控制在50-100字之间。\n\n用户输入：{user_input}",
    input_variables=["user_input"]
)

# 定义负面回复的链式调用
negative_response_chain = negative_response_prompt | llm

# 定义负面回复节点
def negative_response_node(state: TextProcessingState) -> dict:
    # 调用负面回复链
    negative_response = negative_response_chain.invoke({"user_input": state.user_input})
    # 返回状态更新字典
    return {"response": negative_response.content}

# 把负面回复节点添加到图中
graph_builder.add_node("negative_response_node", negative_response_node)

# 定义中性回复的提示词模板
neutral_response_prompt = PromptTemplate(
    template="请生成一段礼貌的、简洁的确认收到用户信息的回复，回复的长度控制在30-50字之间。\n\n用户输入：{user_input}",
    input_variables=["user_input"]
)

# 定义中性回复的链式调用
neutral_response_chain = neutral_response_prompt | llm

# 定义中性回复节点
def neutral_response_node(state: TextProcessingState) -> dict:
    # 调用中性回复链
    neutral_response = neutral_response_chain.invoke({"user_input": state.user_input})
    # 返回状态更新字典
    return {"response": neutral_response.content}

# 把中性回复节点添加到图中
graph_builder.add_node("neutral_response_node", neutral_response_node)

步骤4：定义边（包括条件边）

现在我们来定义边——首先，我们设置图的入口边为分类节点：

# 设置图的入口边（从START到classification_node）
graph_builder.set_entry_point("classification_node")

接下来，我们定义条件边——从分类节点跳转到对应的回复生成节点。这里我们用两种方式来定义：

1. 方式1：基于状态字段的条件边（最简单）
2. 方式2：基于Router Function的条件边（最灵活）

方式1：基于状态字段的条件边

LangGraph允许我们直接根据全局状态中的某个字段的值来选择下一个节点——我们只需要把一个字典传给add_conditional_edges方法，字典的键是状态字段的值，字典的值是目标节点标识。

# 方式1：基于状态字段的条件边
graph_builder.add_conditional_edges(
    # 源节点（条件边的起点）
    source="classification_node",
    # 条件判断的依据：全局状态中的classification字段
    path=lambda state: state.classification,
    # 路径映射：状态字段的值 -> 目标节点标识
    path_map={
        "positive": "positive_response_node",
        "negative": "negative_response_node",
        "neutral": "neutral_response_node"
    }
)

方式2：基于Router Function的条件边

如果我们的路由逻辑更复杂（比如需要根据多个状态字段的值来判断，或者需要做一些额外的处理），我们就可以定义一个专门的Router Function。

首先，我们定义Router Function：

from typing import Literal, Union
from langgraph.graph import END

# 定义Router Function的返回值类型
RouterOutput = Literal["positive_response_node", "negative_response_node", "neutral_response_node"]

# 定义Router Function
def sentiment_router(state: TextProcessingState) -> RouterOutput:
    # 获取全局状态中的classification字段
    sentiment = state.classification
    # 路由判断逻辑
    if sentiment == "positive":
        return "positive_response_node"
    elif sentiment == "negative":
        return "negative_response_node"
    elif sentiment == "neutral":
        return "neutral_response_node"
    else:
        # 如果分类结果不是预期的三种之一，我们可以返回END，或者跳转到一个错误处理节点
        # 这里我们先返回END，后面会讲解如何处理这种错误情况
        print(f"警告：未知的分类结果 {sentiment}，结束流程")
        return END

然后，我们把Router Function传给add_conditional_edges方法：

# 方式2：基于Router Function的条件边（我们先注释掉方式1，用方式2）
# graph_builder.add_conditional_edges(
#     source="classification_node",
#     path=lambda state: state.classification,
#     path_map={
#         "positive": "positive_response_node",
#         "negative": "negative_response_node",
#         "neutral": "neutral_response_node"
#     }
# )

# 使用方式2
graph_builder.add_conditional_edges(
    source="classification_node",
    path=sentiment_router
)

注意：当我们使用Router Function的时候，不需要传path_map参数，因为Router Function的返回值已经是目标节点标识了。

最后，我们定义从三个回复生成节点到END的普通边：

# 定义从三个回复生成节点到END的普通边
graph_builder.add_edge("positive_response_node", END)
graph_builder.add_edge("negative_response_node", END)
graph_builder.add_edge("neutral_response_node", END)

步骤5：编译图

现在我们把图编译成一个可执行的应用：

# 编译图
graph = graph_builder.compile()

步骤6：可视化图

我们可以用LangGraph内置的可视化工具来查看图的结构：

# 生成ASCII字符画（适合在终端或Jupyter Notebook中查看）
print(graph.get_graph().draw_ascii())

# 生成PNG图片（需要安装graphviz和pygraphviz，或者用mermaid的在线编辑器）
try:
    from IPython.display import Image
    Image(graph.get_graph().draw_mermaid_png())
except ImportError:
    print("请安装IPython和pygraphviz来查看PNG图片，或者把下面的Mermaid代码复制到在线编辑器中查看：")
    print(graph.get_graph().draw_mermaid())

如果一切正常，你应该会看到一个这样的图结构：

           +-----------+
           | __start__ |
           +-----------+
                 |
                 v
         +-------------------+
         | classification_node |
         +-------------------+
          /        |        \
         /         |         \
        v          v          v
+----------------+ +----------------+ +----------------+
| positive_res...| | negative_res...| | neutral_res... |
+----------------+ +----------------+ +----------------+
         \         |         /
          \        |        /
           v       v       v
           +---------+
           | __end__ |
           +---------+

步骤7：测试图

最后，我们来测试一下图的执行效果——我们分别输入正面、负面、中性的文本：

# 测试1：正面文本
print("测试1：正面文本")
result1 = graph.invoke({"user_input": "这家餐厅的菜太好吃了，服务也非常周到，下次我还会再来！"})
print(f"分类结果：{result1['classification']}")
print(f"回复：{result1['response']}")
print("-" * 50)

# 测试2：负面文本
print("测试2：负面文本")
result2 = graph.invoke({"user_input": "这家餐厅的菜太难吃了，服务也很差，等了一个小时才上菜，再也不会来了！"})
print(f"分类结果：{result2['classification']}")
print(f"回复：{result2['response']}")
print("-" * 50)

# 测试3：中性文本
print("测试3：中性文本")
result3 = graph.invoke({"user_input": "我想了解一下你们餐厅的营业时间和地址。"})
print(f"分类结果：{result3['classification']}")
print(f"回复：{result3['response']}")
print("-" * 50)

如果一切正常，你应该会看到类似这样的输出：

测试1：正面文本
分类结果：positive
回复：非常感谢您对我们餐厅的认可！我们会继续努力，为您提供更美味的菜品和更周到的服务，期待您的下次光临！
--------------------------------------------------
测试2：负面文本
分类结果：negative
回复：非常抱歉给您带来了如此不好的用餐体验！我们已经记录了您的反馈，会立即进行整改，希望能有机会弥补您的损失。
--------------------------------------------------
测试3：中性文本
分类结果：neutral
回复：您好！我们已经收到了您的信息，稍后会有客服人员联系您，为您提供详细的营业时间和地址。
--------------------------------------------------

边界与外延

边界：Conditional Edge的使用限制

虽然Conditional Edge非常灵活，但它也有一些使用限制：

1. 返回值必须是目标节点标识或END：Router Function的返回值必须是图中已存在的节点标识，或者是END常量——如果返回了不存在的节点标识，LangGraph在初始化图的时候就会报错。
2. 不能循环调用源节点本身（除非显式定义）：如果Router Function的返回值是源节点本身，就会形成一个循环——虽然LangGraph支持循环图，但我们必须显式地处理循环的终止条件，否则图会无限执行下去。
3. 不能同时连接普通边和条件边：一个源节点后面只能连接一种类型的边——要么是普通边，要么是条件边——如果同时连接了两种类型的边，LangGraph在初始化图的时候就会报错。

外延：Conditional Edge的高级用法

除了我们前面讲解的基本用法之外，Conditional Edge还有一些高级用法：

1. 错误处理节点：我们可以定义一个专门的错误处理节点，当Router Function遇到未知的状态值或者工具调用失败的时候，跳转到错误处理节点，生成一个友好的错误提示，然后跳转到END。
2. 循环执行节点：我们可以定义一个循环，当某个条件满足的时候，跳回到之前的节点重新执行——比如当工具调用失败的时候，自动重试几次，重试失败后再跳转到错误处理节点。
3. 多条件组合路由：我们可以在Router Function中根据多个状态字段的值来组合判断路由逻辑——比如先根据“用户的位置”判断，再根据“天气结果”判断，最后根据“朋友的人数”判断。
4. 动态添加节点：虽然LangGraph在初始化图的时候就需要定义所有的节点，但我们可以通过“子图节点”或者“条件边的路径映射动态更新”来实现动态添加节点的功能——不过这种用法比较复杂，不建议在生产环境中使用。

---

条件分支的核心要素组成与概念之间的关系

概念结构与核心要素组成

LangGraph的条件分支系统由以下五个核心要素组成：

1. 全局状态（Global State）：条件分支的“判断依据”，所有的路由逻辑都是基于当前的全局状态来执行的。
2. 源节点（Source Node）：条件分支的“起点”，也就是执行完路由判断逻辑之前的最后一个节点。
3. 路由判断逻辑（Router Logic）：条件分支的“核心”，可以是一个简单的状态字段取值，也可以是一个复杂的Router Function。
4. 路径映射（Path Map）：条件分支的“映射表”，把路由判断逻辑的返回值映射到目标节点标识——当我们使用简单的状态字段取值作为路由判断逻辑的时候，需要显式传path_map参数；当我们使用Router Function的时候，不需要传path_map参数。
5. 目标节点（Target Node）：条件分支的“终点”，也就是执行完路由判断逻辑之后的下一个节点——可以是图中已存在的任意节点，也可以是END常量。

我们可以用一个结构图来表示这些核心要素之间的关系：

概念之间的关系：核心属性维度对比

我们可以用一个表格来对比LangGraph的条件分支系统中两个最核心的概念——基于状态字段的条件边和基于Router Function的条件边——的核心属性：

核心属性维度	基于状态字段的条件边	基于Router Function的条件边
灵活性	低：只能根据单个状态字段的值来判断，无法做额外的处理。	高：可以根据多个状态字段的值来组合判断，可以做任意复杂的处理（比如错误处理、数据预处理）。
代码简洁性	高：只需要传一个简单的lambda函数和path_map字典，不需要写额外的Router Function。	中：需要写一个专门的Router Function，但代码结构更清晰，易于维护。
类型提示支持	中：可以通过path_map字典的键来限制状态字段的取值范围，但没有运行时的严格类型验证。	高：可以通过typing.Literal来限制Router Function的返回值范围，LangGraph会在初始化图的时候进行严格的类型验证。
调试难度	中：无法在路由判断逻辑中添加打印语句或断点调试，只能通过查看全局状态的变化来调试。	低：可以在Router Function中添加打印语句或断点调试，可以清楚地看到路由判断逻辑的执行过程。
适用场景	简单的路由场景：比如根据单个分类结果、单个状态开关来判断。	复杂的路由场景：比如根据多个状态字段的值来组合判断、需要做错误处理、需要做数据预处理。

概念之间的关系：ER实体关系图

我们可以用一个ER实体关系图来表示LangGraph的条件分支系统中各个核心要素之间的实体关系：

---

条件分支的数学模型：有限状态机（FSM）

其实，LangGraph的Stateful Graph加上Conditional Edge，本质上就是一个有限状态机（Finite State Machine，FSM）——或者更准确地说，是一个有限状态自动机（Finite State Automaton，FSA）。

核心概念：有限状态机（FSM）

有限状态机是一种数学模型，它由以下五个部分组成：

1. 有限状态集合（Q）：一个包含所有可能状态的有限集合。
2. 输入字母表（Σ）：一个包含所有可能输入的有限集合。
3. 转移函数（δ）：一个从 Q × Σ 到 Q 的函数，用于定义在某个状态下接收某个输入后，转移到哪个新状态。
4. 初始状态（q₀）：有限状态机的起始状态，属于 Q。
5. 接受状态集合（F）：一个包含所有接受状态的有限集合，属于 Q——当有限状态机转移到接受状态时，执行结束。

LangGraph的Stateful Graph & Conditional Edge 与 FSM的对应关系

我们可以把LangGraph的Stateful Graph & Conditional Edge映射到有限状态机的五个部分：

有限状态机（FSM）的组成部分	LangGraph的Stateful Graph & Conditional Edge的对应部分
有限状态集合（Q）	所有可能的“全局状态值 + 当前执行节点标识”的组合——不过更简单的方式是把“当前执行节点标识”作为状态的一部分，因为全局状态的值可能是无限的，但节点标识是有限的。
输入字母表（Σ）	所有可能的“节点执行后的状态更新字典”的集合——不过同样，更简单的方式是把“节点执行完毕”作为一个固定的输入，因为转移函数的判断依据是当前的全局状态。
转移函数（δ）	条件边的路由判断逻辑 + 普通边的固定跳转逻辑——当在某个节点执行完毕后，根据当前的全局状态，转移到下一个节点（或者END）。
初始状态（q₀）	“入口边指向的节点标识 + 初始全局状态值”的组合。
接受状态集合（F）	“END标识 + 任意全局状态值”的组合。

有限状态机的数学公式描述

我们可以用以下数学公式来描述LangGraph的条件分支系统对应的有限状态机：

1. 有限状态集合（Q）

假设我们的Stateful Graph有 n 个节点，节点标识的集合为 N = {n₁, n₂, …, nₙ, END}，全局状态的取值空间为 S（可能是无限的），那么有限状态集合 Q 可以表示为：
$$Q=\{(n,s)\mid n\in N,s\in S\}$$

2. 输入字母表（Σ）

为了简化，我们把“节点执行完毕”作为一个固定的输入，那么输入字母表 Σ 可以表示为：
Σ = { execute_done } \Sigma = \{ \text{execute\_done} \} Σ={execute_done}

3. 转移函数（δ）

转移函数 δ 是一个从 Q × Σ 到 Q 的函数，它的定义如下：
$$\delta ((n_i,s),\text{execute\_done})=\{\begin{array}{ll}(n_j,s^{\prime })& \text{如果 }{n}_i\text{ 后面连接的是普通边，跳转到 }{n}_j\text{，节点 }{n}_i\text{ 执行后的状态更新为 }{s}^{\prime }\\ (r(s),s^{\prime })& \text{如果 }{n}_i\text{ 后面连接的是条件边，路由判断函数为 }r\text{，节点 }{n}_i\text{ 执行后的状态更新为 }{s}^{\prime }\\ (END,s^{\prime })& \text{如果 }{n}_i\text{ 后面连接的是普通边或条件边，跳转到 END，节点 }{n}_i\text{ 执行后的状态更新为 }{s}^{\prime }\end{array}$$
其中：

• $s^{\prime }$ 是节点 $n_i$ 执行后的新全局状态，由Reducer函数计算得到：$s^{\prime }=\text{reducer}(s,\Delta s)$，其中 $\Delta s$ 是节点 $n_i$ 返回的状态更新字典；
• $r(s)$ 是路由判断函数，它的返回值是 N 中的一个元素（节点标识或END）。

4. 初始状态（q₀）

初始状态 q₀ 是入口边指向的节点标识 $n_{entry}$ 和初始全局状态值 $s_0$ 的组合：
$$q_0=(n_{entry},s_0)$$

5. 接受状态集合（F）

接受状态集合 F 是所有“END标识 + 任意全局状态值”的组合：
$$F=\{(END,s)\mid s\in S\}$$

---

条件分支的算法流程图

我们可以用一个Mermaid流程图来表示LangGraph的条件分支系统的执行算法：