LangGraph 之 【状态及其管理（强制覆盖\\输入输出模式\\私有状态）、规约函数、StateGraph、添加节点\\边、Literal、tools_condition、ToolNode】

目录

1. State（状态）及其管理

1.1 Overwrite：强制覆盖而非合并

2.2 独立的输入/输出模式

2.3 节点间传递私有状态

2. Reducer（规约函数）

2.1 operator.add

2.2 add_messages

 ID 的生成与分配

2.3 自定义 Reducer

3.图即状态机

3.1. 初始化

3.2. 添加节点（add_node）

3.3. 添加固定\\条件边

Literal 让你的路由函数更健壮

tools_condition：智能路由哨兵

3.4. 编译（compile）

4. ToolNode：工具执行器

5. graph.stream() 输入与输出

LangGraph输入 vs LangChain输入

6. 图的可视化

7. 案例

[案例⼀] 智能快递配送系统

[案例二] 支持搜索的智能代理系统

[案例三] 基于 LangGraph 实现的代理式 RAG系统

1. State（状态）及其管理

State 必须继承（Subclass）TypedDict、BaseModel，或使用 @dataclass 装饰器 LangGraph 以此来解析状态的结构

• 三种状态定义方式对比

官方推荐使用 Annotated 为字段附加 reducer 策略以实现类型安全

# TypedDict：Reducer 直接放在 Annotated 的类型提示括号内
from typing import Annotated, TypedDict
import operator

class TypedDictState(TypedDict):
# Reducer 写在 Annotated 里，紧贴类型
messages: Annotated[list, operator.add]

# 普通字段直接写类型（无默认值，必须传参）
topic: str
final_report: str

# Pydantic BaseModel： Reducer 放在 Field(metadata={"reducer": …}) 里，且 必须配合默认值（否则 Pydantic 强制实例化会报错）
from pydantic import BaseModel, Field
import operator

class PydanticState(BaseModel):
# Reducer 藏在 Field 的 metadata 字典里
messages: list = Field(
default_factory=list,
metadata={"reducer": operator.add}
)

# 普通字段带默认值
topic: str = ""
final_report: str = ""

# dataclass：语法与 Pydantic 相似
from dataclasses import dataclass, field
import operator

@dataclass
class DataclassState:
# Reducer 藏在 field 的 metadata 字典里
messages: list = field(
default_factory=list,
metadata={"reducer": operator.add}
)

# 普通字段带默认值
topic: str = ""
final_report: str = ""

• 覆盖器、输入/输出模式与私有状态

1.1 Overwrite：强制覆盖而非合并

（1） 默认行为 vs Overwrite 默认情况下，每个状态字段都绑定了一个 Reducer 函数，用于决定如何合并节点返回的更新 例如，add_messages 会按 ID 追加或覆盖消息。 有些场景下，我们需要完全替换某个字段的值，而不是合并。此时就需要使用 Overwrite

from typing import Annotated, TypedDict
import operator
from langgraph.graph import Overwrite

class State(TypedDict):
messages: Annotated[list, operator.add]

def add_message(state: State):
# 默认行为：通过 Reducer (operator.add) 合并
return {"messages": ["first message"]}

def replace_messages(state: State):
# 绕过 Reducer，直接替换整个消息列表
return {"messages": Overwrite(["replacement message"])}

（2） 工作原理 Overwrite 是一个特殊的哨兵对象，它在 LangGraph 的 Reducer 执行流程中被特殊处理：

# 伪代码：LangGraph 内部的 Reducer 执行逻辑
def apply_update(current_value, update_value):
if isinstance(update_value, Overwrite):
# 直接返回 Overwrite 包裹的值，跳过 Reducer
return update_value.value
else:
# 正常调用 Reducer 函数
return reducer(current_value, update_value)

（3） 应用场景

# 实际示例：系统重置节点
def system_reset(state: State):
if state.get("need_reset"):
# 完全重置消息历史
return {"messages": Overwrite([SystemMessage(content="系统已重置，请重新开始")])}
return {}

2.2 独立的输入/输出模式

（1） 为什么需要独立模式？ 默认情况下，LangGraph 使用单一的状态模式，这意味着：

• 调用方可以传入状态中的所有字段。
• 最终输出会包含状态中的所有字段。

这在许多场景下是不合适的。考虑一个问答系统：

• 输入：用户的问题（字符串）
• 输出：AI 的答案（字符串）
• 内部：可能需要存储中间结果、检索上下文等

我们不想把内部状态都暴露给用户！

（2） 三种模式定义 LangGraph 允许为图定义三种不同的状态模式：

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

# 1. 定义输入模式 – 只包含用户问题
class InputState(TypedDict):
question: str

# 2. 定义输出模式 – 只包含 AI 答案
class OutputState(TypedDict):
answer: str

# 3. 定义完整状态模式（内部使用）
class OverallState(InputState, OutputState):
# 可以添加内部字段
internal_context: str

def answer_node(state: InputState):
"""处理输入并生成答案"""
# 这里可以访问 question
return {
"answer": f"Answer to: {state['question']}",
"internal_context": "Some internal data" # 内部字段，不出现在输出中
}

# 构建图时指定输入输出模式
builder = StateGraph(
OverallState,
input_schema=InputState, # 输入验证
output_schema=OutputState # 输出过滤
)
builder.add_node("answer_node", answer_node)
builder.add_edge(START, "answer_node")
builder.add_edge("answer_node", END)
graph = builder.compile()

# 测试
result = graph.invoke({"question": "What is LangGraph?"})
print(result) # 输出: {'answer': 'Answer to: What is LangGraph?'}
# 注意：question 和 internal_context 都被过滤掉了，不在输出中

（3） 应用场景

2.3 节点间传递私有状态

（1） 核心概念 有时节点间需要传递临时数据，这些数据：

• 对中间逻辑很重要
• 不应该出现在最终输出中
• 只在特定节点间共享

这就是私有状态（Private State）的概念 （2） 实现机制 私有状态的实现依赖于 LangGraph 的类型系统：

• 节点通过类型注解声明自己需要接收哪些字段
• 框架会根据注解自动裁剪状态，只传递节点需要的部分
• 未在全局状态中声明的字段，可以作为私有状态在特定节点间传递

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

# 公共状态（最终输出中可见）
class OverallState(TypedDict):
final_result: str

# 节点1的私有输出
class Node1Output(TypedDict):
sensitive_data: str # 这个字段不会出现在最终状态中

# 节点2需要的输入（包含私有数据）
class Node2Input(TypedDict):
sensitive_data: str

def node_1(state: OverallState) -> Node1Output:
"""第一步：获取包含敏感信息的原始数据"""
private_data = "这是敏感信息"
print(f"Node1: 获取到敏感数据，但不会暴露给最终输出")
return {"sensitive_data": private_data}

def node_2(state: Node2Input) -> OverallState:
"""第二步：处理数据，移除敏感信息"""
print(f"Node2: 处理敏感数据: {state['sensitive_data']}")
# 处理数据，返回清理后的结果
return {"final_result": "清理后的处理结果"}

def node_3(state: OverallState) -> OverallState:
"""第三步：只看到清理后的数据"""
print(f"Node3: 只能看到最终结果: {state['final_result']}")
return {"final_result": state["final_result"] + " – 完成"}

# 构建图
builder = StateGraph(OverallState)
builder.add_sequence([node_1, node_2, node_3])
builder.add_edge(START, "node_1")
graph = builder.compile()

# 测试
response = graph.invoke({"final_result": "initial"})
print(f"\\n最终输出: {response}")

（3） 应用场景

2. Reducer（规约函数）

• 定义：本质上是一个 Python 可调用对象（Callable） 签名为 (current_value, update_value) -> new_value
• 默认行为(没绑定 Reducer): LangGraph 会执行硬覆盖（Overwrite）——用新值直接替换旧值
• 触发时机：在每一个节点（Node）执行完毕，调用 return 或 Command(update=…) 返回状态增量时，LangGraph 会遍历返回的字典，针对每个字段调用对应的 Reducer 节点返回 Command(update={"messages": [ai_msg]}, goto="next")，Reducer 同样会作用于 update 字段
• 它通过 typing.Annotated 挂载到字段上 TypedDict、Pydantic、dataclass 都完全支持 Annotated

from typing import Annotated, TypedDict
import operator

class AgentState(TypedDict):
# 没有 Reducer -> 覆盖模式
user_query: str
# 挂载 Reducer -> 合并模式
messages: Annotated[list, operator.add]

内置 Reducer

2.1 operator.add

• 适用类型：list、tuple
• 表现：执行列表拼接 current + update。
• 致命隐患：如果更新时你返回了一个巨大的历史列表(比如全量对话)，由于 + 操作会产生新列表，这会瞬间让 Checkpointer 的序列化负载翻倍，极易触发 Postgres 的 string too long 报错

class State(TypedDict):
messages: Annotated[list, operator.add]

2.2 add_messages

这是 langgraph.graph.message 模块里专门为 LLM 对话设计的 它不仅仅是 operator.add 的包装，而是实现了基于 ID 的去重合并：

• 旧消息 + 新消息（相同消息 ID）= 覆盖
• 旧消息 + 新消息（不同消息 ID）= 追加
• 旧消息 + RemoveMessage（指定消息 ID）= 删除指定消息

工程意义：在 ReAct Agent 的多次工具调用中，LLM 可能会因为重试而二次返回相同的 ToolMessage。这个 Reducer 能保证 Checkpointer 里只存一份，避免上下文爆炸

from langgraph.graph.message import add_messages

class MessagesState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]

 ID 的生成与分配

消息 ID 的生成和管理是自动的，但也允许你手动控制：

• 自动生成：如果在创建消息对象（如 HumanMessage、AIMessage）时没有指定 id，LangGraph 的 add_messages 函数在合并时，会调用 uuid.uuid4() 为其自动生成一个全局唯一的 ID
• 手动指定：强烈建议你主动为消息设置一个有意义的 ID，尤其是在需要精确控制消息更新（如流式输出）的场景下

# 示例
old = [HumanMessage(content="Hello", id="1")]
new = [AIMessage(content="Hi there!", id="2")]
result = add_messages(old, new)
# result 将包含两条消息，ID 分别为 "1" 和 "2"

# 示例
old = [HumanMessage(content="Hello", id="1")]
new = [HumanMessage(content="Hello again", id="1")]
result = add_messages(old, new)
# result 中只有一条 ID 为 "1" 的消息，其内容被更新为 "Hello again"

# 示例
from langchain_core.messages import RemoveMessage
old = [HumanMessage(content="Hello", id="1")]
result = add_messages(old, [RemoveMessage(id="1")])
# result 将为空列表

2.3 自定义 Reducer

当你的业务逻辑需要按时间戳去重、按特定键合并字典或保留最高优先级配置时， 必须手写 Reducer

案例：去重追加（防止 RAG 上下文重复） 假设你要往 documents 列表里加文档，但要求相同 doc_id 只保留最新的那一份

from typing import Any, List, Dict

def dedupe_docs(current: List[Dict], update: List[Dict]) -> List[Dict]:
# 注意：不要修改 current 本身（不可变原则），要构造新列表
if not current:
return update.copy()

# 构建映射：用 doc_id 作为唯一键
doc_map = {doc["doc_id"]: doc for doc in current}
for doc in update:
doc_map[doc["doc_id"]] = doc # 后面的覆盖前面的

# 返回新列表（保证 Checkpointer 能检测到变化）
return list(doc_map.values())

class State(TypedDict):
documents: Annotated[list, dedupe_docs]

3.图即状态机

LangGraph 中，图（Graph）是通过状态、节点和边实现工作流进而显式定义的，而循环（Loop）是隐式推断的。你不需要写 while True，只要通过条件边（Conditional Edges） 指回上游节点，LangGraph 就会自动形成循环，直到条件边返回 END 为止

创建工作流遵循铁三角定律：

StateGraph 核心 API 

3.1. 初始化

from langgraph.graph import StateGraph
from typing import TypedDict

# 1. 定义输⼊模式 – 只包含⽤⼾输⼊
class InputState(TypedDict):
topic: str # ⽤⼾输⼊的主题
# 2. 定义输出模式 – 只包含最终结果
class OutputState(TypedDict):
final_content: str # 最终的内容

class MyState(TypedDict):
user_input: str
counter: int

builder = StateGraph(
MyState,
input_schema=InputState,
output_schema=OutputState
)

原理：传入的 state_schema 不仅用于类型提示，更重要的是，LangGraph 会在编译时利用它来注册状态中的字段，为每个字段创建独立的数据通道（Channel）并绑定对应的 Reducer 逻辑 如果节点返回了 schema 中未定义的键，该字段不会触发 KeyError，而是被框架静默丢弃——它不会被写入状态通道，因此无法传递给后续节点（除非配合私有状态模式，通过显式声明节点的输入/输出 Schema 来传递）

3.2. 添加节点（add_node）

节点是一个接收 state 作为参数，并返回字典（dict） 或 Command 对象的函数

# 方式一：最常用，node 传字符串，action 传函数
builder.add_node("node_name", my_function)

# 方式二：传入 StateNode 对象（高级用法，用于子图或预配置节点）
# builder.add_node(StateNode("node_name", action=my_function))

• 节点不需要返回全量状态，只需要返回增量
• 如果节点返回的字典中，某个字段在 State 中绑定了 Reducer（如 add_messages），LangGraph 会调用 Reducer 进行合并；否则直接覆盖
• add_node 原生支持异步函数（async def），无需额外封装。LangGraph 在 invoke 时会自动识别协程并调度

3.3. 添加固定\\条件边

（1）固定边（add_edge）

from langgraph.graph import START, END
builder.add_edge(START, "preprocess")
builder.add_edge("preprocess", "llm_call")
builder.add_edge("llm_call", END)

• add_edge 只是将两个节点名称存入一个邻接表中。真正的路由发生在编译后的 _route 方法里，它会机械地取出 start_key 对应的所有 end_key。如果 start_key 有多个出边，LangGraph 默认会并行执行（如果它们互不依赖），或者按添加顺序串行（取决于 interrupt 配置）
• START 是一个特殊的虚拟节点，用于表示图的“入口管道”。使用 add_edge(START, …) 让图的拓扑结构在视觉上更统一，尤其在处理多个入口节点时优势明显

（2）条件边（add_conditional_edges）

builder.add_conditional_edges(
source: str, # 起始节点
path: Callable, # 路由函数，接收 state 返回 string/list
path_map: Optional[Union[dict, list]] = None # 映射表
)

• 当 path_map 为列表时，路由函数的返回值必须与列表中的字符串完全一致（包括大小写、空格、特殊字符），否则框架找不到节点，直接抛出 KeyError；当 path_map 是字典时：path_function 返回字典的键，对应的值才是实际要跳转的节点名。
• 在路由函数中直接通过 state["key"] 取值，如果该键尚未初始化，会触发 KeyError
• path_map 只是映射表，它不会帮你注册节点。你必须在调用 add_conditional_edges 之前，显式调用 add_node 注册所有可能跳转的目标节点，（如 END）

Literal 让你的路由函数更健壮

from typing import Literal

def should_continue(state: AgentState) -> Literal["tools", "__end__"]:
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return "__end__"

虽然 Python 运行时不会强制检查 Literal，但 mypy 或 pyright 会在静态检查时报错。更重要的是，LangGraph 在编译时如果开启了 validate=True（默认开启），会通过反射读取返回注解，提前发现路由路径是否在 path_map 中缺失，将运行时错误提前到编译时

tools_condition：智能路由哨兵

tools_condition 是 LangGraph 预置的一个条件路由函数，专门负责判断 LLM 的最新回复是否需要调用工具，从而决定工作流是继续执行工具还是结束

（1）核心职责

tools_condition 的作用是检查当前状态中的消息列表，特别是最后一条 AIMessage：

• 如果这条消息包含 tool_calls 属性（即 LLM 请求调用工具），则返回字符串 "tools"。
• 否则，返回 "__end__"，表示工作流应该结束。

其函数签名如下：

from typing import Literal, Union
from langchain_core.messages import AnyMessage
from pydantic import BaseModel

def tools_condition(
state: Union[list[AnyMessage], dict[str, Any], BaseModel],
messages_key: str = "messages"
) -> Literal["tools", "__end__"]:
…

返回值类型：Literal["tools", "__end__"]，明确限定只有两种可能的路由结果

（2）工作原理

• 如果该消息是 AIMessage 且具有非空的 tool_calls 属性 → 返回 "tools"。
• 否则 → 返回 "__end__"。

（3）典型用法：配合 ToolNode 构建 ReAct 循环

tools_condition 与 ToolNode 一起使用，形成 “LLM 决策 → 工具执行 → 回到 LLM” 的闭环

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode, tools_condition

# 假设已有 llm_node（调用 LLM 并返回 AIMessage）和 tool_node（执行工具）
builder = StateGraph(MessagesState)

builder.add_node("agent", llm_node)
builder.add_node("tools", tool_node)

builder.set_entry_point("agent")

# 关键：为 agent 节点添加条件边，使用 tools_condition 做路由
builder.add_conditional_edges(
"agent", # 源节点
tools_condition, # 路由函数
{
"tools": "tools", # 返回 "tools" 时跳转到工具节点
"__end__": END # 返回 "__end__" 时结束
}
)

# 工具执行后，必须回到 agent 节点，让 LLM 根据工具结果继续推理
builder.add_edge("tools", "agent")

graph = builder.compile()

3.4. 编译（compile）

# 单机版
app = builder.compile()

# 生产版（带持久化）
from langgraph.checkpoint.postgres import PostgresSaver
with PostgresSaver.from_conn_string("postgresql://…") as checkpointer:
app = builder.compile(checkpointer=checkpointer)

compile 做了三件极其重要的事：

CompiledStateGraph 实现了 Runnable 接口，可以异步调用、流式传输、批处理和运行 编译后得到的 CompiledStateGraph 对象才是真正用于运行的。它提供了丰富的执行和交互方法：

• .stream 和 .astream 的默认行为是按节点输出，若需要 LLM 逐字生成效果，请使用 astream_events
• .get_state 和 .update_state 是 Human-in-the-loop 的基石，配合 interrupt 机制可实现断点审批、回滚等高级流程控制
• 所有方法均支持传入 config 参数，用于配置 thread_id、checkpoint_id、recursion_limit 等运行时选项

4. ToolNode：工具执行器

ToolNode 是 LangGraph 框架中的一个预置节点，是"工具执行器"或 "工具调度中心"，它 的核心职责是：当大语言模型（LLM）决定调用某个工具时，负责执行该工具，并将执行结果返回给图的状态（State）

• 工作原理（三步走）

Step 1：接收请求（读取状态） ToolNode 会检查输入的状态（State），特别是其中的 messages 列表：

• 工具名称（name）
• 调用参数（args）
• 调用 ID（id，用于关联请求与响应）

Step 2：执行工具（并行调度） 一旦发现有工具调用请求，ToolNode 会根据请求中的工具名称，去调用你预先注册好的对应工具函数：

# 注册示例
from langgraph.prebuilt import ToolNode
from langchain_community.tools import DuckDuckGoSearchRun

tools = [DuckDuckGoSearchRun()]
tool_node = ToolNode(tools)

关键特性：如果 tool_calls 中包含多个请求，ToolNode 会并行执行它们，显著提高执行效率

Step 3：返回结果（更新状态） 工具执行完成后，ToolNode 会将每个工具的执行结果封装成一个 ToolMessage：

# ToolMessage 结构示例
ToolMessage(
content="搜索结果的文本内容…",
tool_call_id="call_abc123" # 与请求中的 id 对应
)

最后，它会将这些 ToolMessage 以列表形式作为状态更新的一部分输出，追加到 messages 列表中

注意事项:

5. graph.stream() 输入与输出

• 输入：必须传入字典，且字典的键必须与你在 StateGraph 中定义的 State 字段完全匹配

对于最常见的 MessagesState，其定义如下：

class MessagesState(TypedDict):
messages: Annotated[list, add_messages]

因此，唯一合法的输入格式是：{"messages": [消息对象列表]}

• 输出：.stream() 返回一个生成器（Generator），每次迭代产生一个 chunk（数据块），代表图执行过程中的一个步骤（step）

Chunk 结构

每个 chunk 是一个字典：

• 键（Key）：当前执行完毕的节点名称（字符串）
• 值（Value）：该节点返回的状态更新（也是一个字典，包含该节点对 State 的增量修改）

执行粒度

• 顺序执行：如果图是线性或带条件边的，每次迭代通常只包含一个节点的更新。
• 并行分支：如果图中存在并行执行的分支（如使用 Send API 实现 Map-Reduce），一个 chunk 可能包含多个节点的更新

LangGraph输入 vs LangChain输入

• LangChain Runnable 的一般规则

Runnable 是 LangChain 的标准接口，.stream() 的输入类型完全由链的结构决定

LangGraph StateGraph 的特殊约束 StateGraph 虽然也实现了 Runnable 接口，但其输入被严格限定为字典：

• 唯一性：StateGraph 的 .stream() 和 .invoke() 只能接受字典输入，不存在其他形式
• 字段约束：字典的键必须与 State 中声明的字段完全一致。未在 State 中定义的键会被静默忽略（默认行为），但不会写入状态

6. 图的可视化

Mermaid 是一种使用文本描述生成流程图、饼状图、甘特图等图表的轻量级标记语言，能够帮助开发者以简单、直观的方式创建各类图表 Mermaid 在线绘图工具：https://www.jyshare.com/front-end/9729/

使用步骤： 1. 获取图的 Mermaid 表示 编译后的 CompiledStateGraph 对象提供了 get_graph() 方法以此获取编译后图结构的元数据

# 获取图结构
graph_repr = agent.get_graph(xray=True) # xray=True 显示详细视图（包含子图内部结构）

参数说明：

• xray=True 或 xray=1：开启详细视图模式，会递归展开子图（Subgraph）的内部节点，便于调试复杂工作流
• xray=False（默认）：仅显示顶层节点和边，子图会被折叠为一个节点

2. 根据元数据可生成流程图代码，复制粘贴至在线绘图的输入即可获取可视图

# 使用生成的语言在线打印
print(agent.get_graph(xray=True).draw_mermaid())

3. 或者生成 PNG 图片（二进制数据）并保存与显示

import matplotlib.pyplot as plt
import matplotlib.image as mpimg

try:
# 通过 draw_mermaid_png() 方法将图结构渲染为 PNG 图片的二进制数据：
mermaid_code = agent.get_graph(xray=True).draw_mermaid_png()

# 保存文件（可选）
with open("../jpg/graph1.jpg", "wb") as f:
f.write(mermaid_code)

# 使用 matplotlib 显示图像
img = mpimg.imread("../jpg/graph1.jpg")
plt.imshow(img)
plt.axis('off') # 关闭坐标轴
plt.show() # 弹出窗口显示图片

except Exception as e:
print(f"An error occurred: {e}")

7. 案例

[案例⼀] 智能快递配送系统

from typing import TypedDict, Annotated
from operator import add

from langgraph.constants import START, END
from langgraph.graph import StateGraph

# 1.定义包裹状态
class PackageState(TypedDict):
# 包裹基本信息
package_id: str # 包裹id
origin: str # 始发站
destination: str # ⽬的地

# 配送状态
status: str # "待揽收", "运输中", "派送中", "已签收"
history: Annotated[list[str], add]
total_distance: Annotated[int, add]

# 配送详情
priority: str # "普通", "加急"

# 2. 实例化 StateGraph
delivery = StateGraph(PackageState)

# 3. 定义各个配送站点
def receive_package(state: PackageState):
"""揽收站"""
return {
"status": "已揽收",
"history": [f"在{state['origin']}处揽收"]
}

def sort_package(state: PackageState):
"""分拣中⼼：根据⽬的地分拣"""
destination = state["destination"]
if "北京" in destination:
next_station = "北京分拣中⼼"
elif "上海" in destination:
next_station = "上海分拣中⼼"
else:
next_station = "其他地区分拣中⼼"

return {
"status": "已分拣",
"history": [f"分拣⾄{next_station}"]
}

def final_delivery(state: PackageState):
"""派送站"""

return {
"status": "已签收",
"history": [f"已送达{state['destination']}"]
}

# 新增节点
def standard_delivery(state: PackageState):
"""标准配送"""
return {
"status": "运输中",
"history": ["标准陆运"],
"total_distance": 500
}
def express_delivery(state: PackageState):
"""加急配送"""
return {
"status": "加急运输",
"history": ["空运加急"],
"total_distance": 800
}

# 4.添加节点
delivery.add_node("揽收站", receive_package)
delivery.add_node("分拣中心", sort_package)
delivery.add_node("派送站", final_delivery)
delivery.add_node("加急配送", express_delivery)
delivery.add_node("标准配送", standard_delivery)

# 5. 设计配送路线
delivery.add_edge(START, "揽收站")
delivery.add_edge("揽收站", "分拣中心")

def select_delivery(state: PackageState):
"""智能路由决策 – 根据包裹特性选择路线"""
if state["priority"] == "加急":
return "加急配送"
else:
return "标准配送"

delivery.add_conditional_edges(
"分拣中心", # source：起始节点。退出此节点时，将运⾏此条件边
select_delivery, # path：确定下⼀个或多个节点的可调⽤对象
["加急配送", "标准配送"] # path_map
)

delivery.add_edge("标准配送", "派送站")
delivery.add_edge("加急配送", "派送站")
delivery.add_edge("派送站", END)

# 7. 编译系统
delivery_system = delivery.compile()

# 8. 测试配送
test_packages = [
{
"package_id": "P001",
"origin": "北京",
"destination": "上海",
"priority": "普通",
"history": [],
"total_distance": 0
},
{
"package_id": "P002",
"origin": "⼴州",
"destination": "乌鲁⽊⻬",
"priority": "加急",
"history": [],
"total_distance": 0
}
]

for package in test_packages:
print(f"\\n配送包裹: {package['package_id']}")
result = delivery_system.invoke(package)
print("最终状态:", result["status"])
print("配送历史:", result["history"])
print("总⾥程:", result["total_distance"])

[案例二] 支持搜索的智能代理系统

from typing import TypedDict, Annotated

from langchain.chat_models import init_chat_model
from langchain.messages import AnyMessage
import operator

from langchain_core.messages import SystemMessage, ToolMessage, HumanMessage
from langchain_tavily import TavilySearch
from langgraph.constants import START, END
from langgraph.graph import StateGraph

# 1.定义工具和模型
search = TavilySearch(max_results=4)
tools = [search]

model = init_chat_model(model="deepseek-v4-flash", model_provider="deepseek")
model_with_tools = model.bind_tools(tools)

# 2.定义状态
class MessageState(TypedDict):
# 类型: list[AnyMessage] – 任意消息对象的列表
# 合并策略: operator.add – 使⽤加法操作符进⾏状态合并
# 效果: 当状态更新时，新的消息会追加到现有列表中，⽽不是替换
messages: Annotated[list[AnyMessage], operator.add]
# 类型: int – 整数值
# ⽤途: 跟踪LLM（⼤语⾔模型）的调⽤次数
llm_calls: int

# 3.定义模型节点
def llm_call(state: dict):
"LLM 决定是否需要工具调用"
return {
"llm_calls" : state.get("llm_calls", 0) + 1,
"messages" : [model_with_tools.invoke(
[ SystemMessage(content="你是⼀个乐于助⼈的助⼿，⽀持调⽤⼯具进⾏搜索。") ]
+ state.get("messages", [])
)]
}

# 4.定义工具节点
tools_by_name = {tool.name : tool for tool in tools}
def tool_node(state: dict):
"""执行工具调用"""
result = []
for tool_call in state["messages"][-1].tool_calls:
tool = tools_by_name[tool_call["name"]]
observation = tool.invoke(tool_call["args"])
result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))

return {
"messages" : result
}

def should_continue(state: dict):
"""根据LLM是否调⽤⼯具来决定是应该继续循环（路由到⼯具节点）还是停⽌循环（END）"""
if state["messages"][-1].tool_calls:
return "tool_node"
else:
return END

agent_builder = StateGraph(MessageState)
agent_builder.add_node(llm_call)
agent_builder.add_node(tool_node)

agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges(
"llm_call", # source：起始节点。退出此节点时，将运⾏此条件边。
should_continue, # path：确定下⼀个或多个节点的可调⽤对象。
["tool_node", END] # path_map：可选，将路径映射到节点名称
)
agent_builder.add_edge("tool_node", "llm_call")

# 编译图
agent = agent_builder.compile()

# 使用生成的语言在线打印
# print(agent.get_graph(xray=True).draw_mermaid())

# 测试
# messages = agent.invoke({
# "messages": [HumanMessage(content="今天重庆市垫江县的天⽓如何？")]
# })
# print(f"调⽤ LLM 总次数：{messages['llm_calls']}次")
# for m in messages["messages"]:
# m.pretty_print()

for chunk in agent.stream({"messages": [HumanMessage(content="今天西安的天⽓如何？")]}):
print(chunk)

[案例三] 基于 LangGraph 实现的代理式 RAG系统

• 实际设计

from langchain_community.document_loaders import UnstructuredMarkdownLoader
from langchain_core.messages import HumanMessage
from langchain_core.tools import create_retriever_tool
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langgraph.constants import START, END
from langgraph.graph import MessagesState, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition
from pydantic import BaseModel, Field
from typing import Literal

# 聊天模型与嵌入模型
model = ChatOpenAI(model="gpt-4o-mini")
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

# 加载文档列表
paths = [
"../Docs/markdown/C++Review.md",
"../Docs/markdown/Markdown 资源集.md",
]
docs = [UnstructuredMarkdownLoader(path).load() for path in paths]
docs_list = [item for sublist in docs for item in sublist]
# from_tiktoken_encoder：使⽤ tiktoken 编码器来计算⻓度的⽂本分割器
text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
chunk_size=500,
chunk_overlap=50,
model_name="gpt-4",
)
doc_splits = text_splitter.split_documents(docs_list)

# 使⽤内存中向量存储和 OpenAI 嵌⼊
vector_store = InMemoryVectorStore.from_documents(
documents=doc_splits,
embedding=embeddings,
)

# 使⽤ LangChain 的预构建 create_retriever_tool 创建检索器⼯具：
retriever = vector_store.as_retriever()
retriever_tool = create_retriever_tool(
retriever=retriever,
name="retrieve_dfq",
description="搜索并返回有关⽐特就业课的信息。",
)

# 步骤二："设计"⼯作流程节点"
def generate_query_or_respond(state: MessagesState):
"""调⽤模型以基于当前状态⽣成响应。
给定问题，它将决定使⽤检索⼯具检索，或者简单地响应⽤⼾。"""
response = model.bind_tools([retriever_tool]).invoke(state["messages"])
return { "messages" : [response] }

# 节点2：检索器⼯具节点 retrieve
retrieve_node = ToolNode([retriever_tool])

# 节点3：问题优化节点 rewrite_question
# 相邻的字符串字面量（由空白字符分隔，包括换行和缩进）会被自动拼接成一个字符串
REWRITE_PROMPT = (
"查看输⼊并尝试推断潜在的语义意图/含义。\\n"
"这是最初的问题："
"\\n ——- \\n"
"{question}"
"\\n ——- \\n"
"提出⼀个改进后的问题："
)

def rewrite_question(state: MessagesState):
"""重写原始⽤⼾问题"""

question = state["messages"][0].content
prompt = REWRITE_PROMPT.format(question=question)
response = model.bind_tools([retriever_tool]).invoke(prompt)

return { "messages" : [{"role": "user", "content": response.content}] }

# 节点4：答案⽣成节点 generate_answer
# ⽣成答案
GENERATE_PROMPT = (
"你是负责回答问题的助⼿。 "
"使⽤以下检索到的上下⽂⽚段来回答问题。 "
"如果你不知道答案，就说你不知道。 "
"最多只⽤三句话，回答要简明扼要。\\n"
"Question: {question} \\n"
"Context: {context}"
)

def generate_answer(state: MessagesState):
"""⽣成答案"""
# 最原始问题
question = state["messages"][0].content
# 最新问题的检索结果（保证参考答案准确）
context = state["messages"][-1].content
prompt = GENERATE_PROMPT.format(question=question, context=context)
response = model.invoke([HumanMessage(content=prompt)])
return {"messages": [response]}

# 步骤三：组装"⼯作流⽔线"
workflow = StateGraph(MessagesState)
workflow.add_node(generate_query_or_respond)
workflow.add_node("retrieve", ToolNode([retriever_tool]))
workflow.add_node(rewrite_question)
workflow.add_node(generate_answer)
workflow.add_edge(START, "generate_query_or_respond")

# 条件边1：LLM 决策是否需要进⾏知识库检索
workflow.add_conditional_edges(
"generate_query_or_respond",
tools_condition,
{
"tools" : "retriever",
"__end__" : END
}
)
# 条件边2：检测【检索到的⽂档】是否与【问题】相关
GRADE_PROMPT = (
"你是⼀个评分员，评估检索到的⽂档与⽤⼾问题的相关性。 \\n "
"以下是检索到的⽂档： \\n\\n {context} \\n\\n"
"以下是⽤⼾的问题： {question} \\n"
"如果⽂档包含与⽤⼾问题相关的关键字或语义，则将其评为相关。 \\n"
"给出⼀个⼆元分数“yes”或“no”，以表明该⽂档是否与问题相关。"
)

class GradeDocuments(BaseModel):
"""使⽤⼆值评分进⾏相关性检查"""
score: str = Field(description="相关性评分：如果相关则为“yes”，如果不相关则为“no”")

def grade_documents(state: MessagesState) -> Literal["generate_answer", "rewrite_question"]:
"""确定检索到的⽂档是否与问题相关"""
question = state["messages"][0].content
context = state["messages"][-1].content
prompt = GRADE_PROMPT.format(question=question, context=context)
response = (model.with_structured_output(GradeDocuments).invoke(
[{"role": "user", "content": prompt}])
)
score = response.score
if score == "yes":
return "generate_answer"
else:
return "rewrite_question"

workflow.add_conditional_edges(
"retrieve",
# 评估代理决策
grade_documents,
["generate_answer", "rewrite_question"],
)

workflow.add_edge("generate_answer", END)
workflow.add_edge("rewrite_question", "generate_query_or_respond")
graph = workflow.compile()

# print(graph.get_graph(xray=True).draw_mermaid())
# 运⾏ RAG
for chunk in graph.stream({"messages": [HumanMessage(content="C++开发⽅向的项⽬列表")]}):
for node, update in chunk.items():
print(f"由节点 {node} 更新消息:")
if node != "rewrite_question":
update["messages"][-1].pretty_print()
print("\\n\\n")