LangChain状态机思维核心技巧与实战完全解析：从零掌握Agent教程第16讲

2026-06-08阅读 0热度 0

其他

上一节我们列出了LangGraph的基本节点类型——Node、Edge、State、Checkpoint、Interrupt，但这些零件光知道名字可没法把机器转起来。今天我们先打住，聊聊这些零件到底怎么“组装”，以及背后的设计理念。其实，光知道有哪些零件可不够，关键得先明白怎么“组装”它们。简单复习一下几个核心概念： * **Node**：本质是一个函数，可以是LLM调用、工具调用，或者一个判断逻辑。 * **Edge**：定义了节点间的跳转规则，分普通边和条件边两种。 * **State**：共享状态，用来存储情绪、记忆、亲密度等全局信息。它支持Reducer机制，能自动合并不同节点对同一字段的更新。 * **Checkpoint**：自动持久化，支持内存、SQLite或Postgres，遇到崩溃可以恢复，还能时光倒流做调试。 * **Interrupt**：原生支持Human-in-the-loop，允许用户在运行中途纠正、审批或回滚。有了这些，我们就能进入正题了。 ## 一、状态机思维在LangChain的早期版本（包括LCEL）里，绝大多数工作流本质上是一个**有向无环图（DAG）**，或者一条简单的线性流水线。数据像流水一样，从A流到B再到C，想“回头”几乎不可能。 LangGraph最核心的突破，就是它支持**带环的有向图**。相比古早的链式版本，灵活性强了不止一个档次。想象一下，假设我们有一个业务，理想状态下希望LLM依次执行三个步骤： 1. 调用工具查询数据 2. 格式化数据 3. 打印结果 ![Agent教程16：认识LangChain(中)，状态机思维](http://img.318050.com/uploads/20260608/17808728436a25f68bbc6a6339872121.webp) 你发现了，这就是早期的链式思维。一旦LLM因为随机性在某一步出错，整个流程就轰然倒塌。事实证明，链式操作救不了Agent开发。一种更稳健、更擅长应对随机性的设计哲学被LangGraph选中了——**状态机**。 ![Agent教程16：认识LangChain(中)，状态机思维](http://img.318050.com/uploads/20260608/17808728446a25f68c37670058286602.webp) 所谓**状态**，可以理解为一组能被所有节点读取和更新的变量。对状态的使用，通常遵循这样的规则： * 在Node中**更新**状态。 * 在选择连线时**读取**状态。以上图为例，如果格式化失败了，那`state`就被标记为`'fail'`。到了选择连线的环节，因为状态已经等于`'fail'`，所以程序会跳进格式化自循环的那条连线，重试一次。正式生产时，我们还可以引入`fail_count`这样的计数器，来限制自循环的上限次数。 ## 二、定义状态状态的定义方式非常直接，用TypedDict和注解就能搞定： ```python from typing import TypedDict, Annotated from langchain_core.messages import BaseMessage import operator class AgentState(TypedDict): input_text: str messages: Annotated[list[BaseMessage], operator.add] result: dict status: str # 状态标识 "success" 或 "error" ``` 这个定义很好理解：`AgentState`继承自`TypedDict`，有4个基本属性。其中比较难理解的是这一行： ```python messages: Annotated[list[BaseMessage], operator.add] ``` 简单来说，这是Python的注解写法。它规定了`messages`的类型是`list[BaseMessage]`，并且指明其更新逻辑只能是**add**（追加），而不是**覆盖**。后续LangGraph在操作这个属性时，必须遵守以下方式： ```python state["messages"] = operator.add(旧的messages列表, 新返回的messages列表) ``` 这样就能保证老的聊天历史不会被覆盖。如果你希望`messages`只保留最新的5条，可以自己写一个自定义合并逻辑（Reducer）： ```python # 1. 自己写一个合并逻辑（Reducer） def keep_last_5_messages(old_messages: list, new_messages: list) -> list: """自定义 Reducer 函数：合并消息，但只保留最后 5 条""" # 先把旧的和新的加起来 combined = old_messages + new_messages # 然后切片，只取最后 5 个元素 return combined[-5:] # 2. 把自定义函数塞进 Annotated 里！ class AgentState(TypedDict): input_text: str # 告诉 LangGraph，更新 messages 时用这个函数 messages: Annotated[list[BaseMessage], keep_last_5_messages] status: str ``` ## 三、构建节点 Node 开篇我们就说了，Node的本质就是一个函数。假如我们要实现一个最简单的例子： ![Agent教程16：认识LangChain(中)，状态机思维](http://img.318050.com/uploads/20260608/17808728446a25f68cb79c6541911291.webp) 那我们至少需要构建三个函数： * `node_a_input`：返回一段文本。 * `node_b_llm`：用LLM格式化文本，但有概率失败。 * `node_c_print`：打印状态中的`result`值。在LangGraph里，一个节点Node大概长这样： ```python def node_a_input(state: AgentState): """节点A：负责接收初始输入""" text = "张三18岁。" print(f"\n▶ [Node A] 初始化输入: {text}") # 将初始数据写到状态黑板上 return {"input_text": text} ``` 它可以读取入参`state`上的状态值，它的返回值则可以更新`state`的最新信息。非常好理解。根据这个模板，我们可以写出核心的`node_b_llm`逻辑： ![Agent教程16：认识LangChain(中)，状态机思维](http://img.318050.com/uploads/20260608/17808728456a25f68d37896350680860.webp) ## 四、构建有向图接下来，我们把刚才定义的这些节点“组装”成一张有向图： ```python from langgraph.graph import StateGraph, START, END workflow = StateGraph(AgentState) # 添加所有节点 workflow.add_node("node_a", node_a_input) workflow.add_node("node_b", node_b_llm) workflow.add_node("node_c", node_c_print) # 定义流转规则 workflow.add_edge(START, "node_a") workflow.add_edge("node_a", "node_b") def router_edge(state: AgentState): """条件边：读取 status 字段决定去向""" if state["status"] == "error": print("↳ [Edge 路由] 发现错误状态，打回 Node B 重试！") return "retry" else: print("↳ [Edge 路由] 状态成功，放行到 Node C。") return "continue" # 添加条件边：离开 node_b 时，用 router_edge 进行判断 workflow.add_conditional_edges( "node_b", router_edge, { "retry": "node_b", # 如果返回 "retry"，跳回 node_b "continue": "node_c" # 如果返回 "continue"，跳到 node_c } ) workflow.add_edge("node_c", END) ``` 其他代码都非常好理解，无需多言。唯一需要解释的是条件边的部分： ```python workflow.add_conditional_edges( "node_b", router_edge, { "retry": "node_b", "continue": "node_c" } ) ``` 这段代码的含义是：对`node_b`的出口增加择线逻辑`router_edge`。如果它返回`retry`，就回到`node_b`进行自循环；如果返回`continue`，就跳转到`node_c`。而`router_edge`本身也是一个和Node类似的方法，可以读取并消费`state`，完成状态机的流转判断。 ## 五、执行脚本在本课程的完整demo项目中，你可以拿到本节课我们讲解的demo源码。执行以下命令： ```bash python lesson_16\02_state_demo.py ``` ![Agent教程16：认识LangChain(中)，状态机思维](http://img.318050.com/uploads/20260608/17808728456a25f68dcde48111221121.webp) 哪怕第一次出错了，整个程序依然可以自动完成重试，并输出期望中的结构。没错，这就是新版本LangChain的核心思维转变： * **忘记链式** * **拥抱状态机思维** 你需要不停地维护一组全局状态，直到你放弃任务或者完成任务。 ## 六、小结本章我们学习了新版LangChain和LangGraph的状态机思维。记住，核心思想很简单：**维护一组全局状态，直到任务完成或者你主动放弃**。接下来，我们还需要了解一下LangChain的其他基本能力，敬请期待。 ![Agent教程16：认识LangChain(中)，状态机思维](http://img.318050.com/uploads/20260608/17808728466a25f68e3a6d5535446154.webp)

LangChain状态机思维核心技巧与实战完全解析：从零掌握Agent教程第16讲

相关阅读

最新教程

最新资讯