LangGraph实战：用图状态机编排复杂Agent工作流

如果你用 LangChain 写过几个 Demo应该能感受到那种“Chain 很好用但 Agent 不好写”的落差。Chain 的优点是线性、确定、好调试但真正的 Agent 需要根据上下文决定下一步做什么可能需要循环调用工具可能需要在多轮对话里保持记忆甚至需要有人在中间批准某个操作。这些需求如果用普通代码堆很快就会变成一团乱麻。LangGraph 就是用来收拾这团乱麻的。我的判断是LangGraph 不是 LangChain 的“下一个版本”而是大模型应用开发里一次编排方式的升级——从“链式调用”走向“图状状态机”。2026 年再看这波 Agent 应用生产环境里那些能稳定跑起来的多步任务、工具调用、人工审核环节底层大概率都有一张图在支撑。这篇文章会把 LangGraph 的核心概念、条件路由、循环控制、子图、并行分支、持久化记忆一次讲清楚所有代码都是最小可运行示例不依赖真实大模型 API Key 也能跑通。如果你正处于“会写 LangChain 链但不会写真正 Agent 工作流”的阶段这篇文章可以帮你少踩很多弯路。1. 为什么大家都在讨论 LangGraphLangChain 链式调用到底卡在哪先解决一个高频问题LangGraph 和 LangChain 到底有什么区别简单理解LangChain 提供的是“组件库 链式组装”能力。你可以在一条链上串联 Prompt、模型、输出解析器、检索器但它本质上是线性的A 到 BB 到 C执行完就结束。如果你想表达“如果这一步结果不满足条件就回退到上一步重新处理”链式表达会很吃力因为你不得不在代码里插入 if/else、while 循环把状态变量在外面传来传去。LangGraph 改变的是抽象层级。它把整个 Agent 流程建模成一张有向图每个节点Node是一段处理逻辑本质是普通函数每条边Edge定义节点之间的流转方向整张图共享一个状态对象State节点通过读取和更新 State 来协作边可以是条件边ConditionalEdge根据当前 State 动态决定下一步走向图允许成环因此循环、重试、多轮工具调用都可以自然表达。举个现实场景你让 Agent 帮你查天气。大模型本身不知道天气数据它需要先判断“这问题需要调用天气工具”调用完拿到结果再判断“结果够不够是否需要继续追问城市”。这中间有判断、有循环、有工具调用用 LangChain 的线性 Chain 写会越写越别扭但用 LangGraph 画一张图就非常直观。两者的关系并不是“二选一”。LangGraph 依然会复用 LangChain 生态里的模型封装、消息类型、Tool 组件只是把流程控制权从“链”交给了“图”。这也是为什么搜索里 LangGraph 和 LangChain 总是同时出现。下面用一个对比表快速感受差异能力LangChain 链式LangGraph 图式流程结构线性一条路走到底有向图支持环条件分支需要额外路由逻辑ConditionalEdge循环重试手动实现困难边可以指回前序节点状态管理外部变量或 Chain 内传递内置 State全局共享多轮记忆需要额外组件Checkpointer thread_id人工介入支持较弱interrupt 等机制适合场景固定流程、RAG、简单问答复杂 Agent、工具循环、审批流刚开始接触时不要陷入“LangGraph 必须配 LangChain 才叫完整”的误区。LangGraph 并不强制依赖 LangChain 的全部组件你甚至可以只用纯 Python 函数定义节点。下面我们从最核心的 4 个概念开始。2. 30 秒吃透核心概念State、Node、Edge、ConditionalEdgeLangGraph 的命名很容易理解难的是把它们组合起来。我们先做一次“通俗类比”。把整张图想象成一家快递分拣仓库State仓库里的共享记事板。所有工作人员都可以看记事板、往上面记录新信息下一位工作人员根据记事板决定自己做什么。Node一个个具体工位。例如“抓取快递”“判断地址”“称重计算”。在代码里Node 就是一个普通函数它接收当前 State返回要写入 State 的新内容。Edge工位之间的传送带。比如“抓取快递”完成后强制送到“判断地址”。没有歧义固定走向。ConditionalEdge分拣员。他看一眼记事板上的地址然后决定把快递推到哪条传送带。Checkpointer仓库里的监控存档系统。每完成一个工位就把记事板内容存一份这样即使中途断电也能从上次进度恢复。翻译成代码概念就是下面这张表概念技术定义一句话理解State用 TypedDict 声明的结构化数据模型所有节点共享的上下文Node(state) - partial_state的普通函数一个干活的步骤Edgeadd_edge(A, B)A 做完后强制走 BConditionalEdgeadd_conditional_edges(A, router, mapping)A 做完后按 router 结果选下一步Compilegraph.compile()把图编译成可调用对象Invokeapp.invoke(input_state)触发一次完整运行这里要特别强调 State 的设计State 字段越清晰后面写节点、调路由、查问题就越省力。很多 LangGraph 新手第一个坑就是 State 里字段随意命名结果在多个节点之间传递时不知道某个值到底被谁改过。3. 环境准备不需要大模型 API Key 也能跑通示例安装 LangGraph 很简单建议使用虚拟环境隔离依赖python -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate pip install langgraph langchain-corePython 版本建议使用 3.9 及以上。langchain-core提供消息类型等基础类型后面的记忆示例会用到。如果只想跑最基础的图逻辑实际只需要langgraph一个包就可以。为了降低学习门槛本文核心示例不会调用真实大模型而是用普通函数模拟 LLM 的判断结果。这样做有两个好处不需要配置任何 API Key全流程本地可跑可以更清晰看到 LangGraph 的编排逻辑而不是把注意力放在 Prompt 上。如果你已经配置好OPENAI_API_KEY后续可以直接把模拟函数替换成ChatOpenAI实现流程结构不需要改变。安装完成后先验证一下版本python -c import langgraph; print(langgraph.__version__)能输出版本号说明环境正常。注意不要安装到系统级 Python避免依赖冲突。4. 第一个 LangGraph 程序Hello StateGraph我们从一个最小示例开始完整展示“定义 State、编写 Node、组装图、编译执行”的完整流程。# 文件路径demo01_hello_graph.py import operator from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END class MyState(TypedDict): count: int messages: Annotated[list[str], operator.add] def increment_node(state: MyState) - dict: # 节点函数接收当前 state返回需要更新的字段 return {count: state[count] 1, messages: [increment]} def show_node(state: MyState) - dict: return {messages: [show]} # 1. 创建图 graph StateGraph(MyState) # 2. 添加节点 graph.add_node(increment, increment_node) graph.add_node(show, show_node) # 3. 添加边入口 - increment - show - 出口 graph.add_edge(START, increment) graph.add_edge(increment, show) graph.add_edge(show, END) # 4. 编译图 app graph.compile() # 5. 执行 result app.invoke({count: 0, messages: []}) print(result)预期输出{count: 1, messages: [increment, show]}这个例子虽然简单但包含了 LangGraph 的核心机制TypedDict定义了 State 结构count是普通 int 字段messages使用Annotated[list[str], operator.add]声明了合并策略increment_node返回{count: ..., messages: [increment]}LangGraph 会将返回值合并回 Statemessages因为声明了operator.add所以两个节点返回的列表会叠加最终是[increment, show]如果messages没有声明 reducer默认会被后写入的值覆盖最终只会保留[show]。这里第一次出现了“节点如何改变 State 状态值”的关键点节点函数不需要返回整个 State只需要返回需要更新的字段。LangGraph 会自动做合并。5. State 更新机制详解节点里到底怎么改状态继续深入“LangGraph 如何在节点函数改变 State 状态值”这个问题。这是新手最容易忽略的部分但也是 LangGraph 最核心的数据流设计。LangGraph 的 State 有两条更新规则5.1 默认规则字段级覆盖如果字段没有声明 reducer节点返回该字段时会直接用新值覆盖旧值。# 文件路径demo02_default_state.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class DefaultState(TypedDict): text: str count: int def node_a(state: DefaultState) - dict: return {text: 来自A, count: state[count] 1} def node_b(state: DefaultState) - dict: return {text: 来自B} graph StateGraph(DefaultState) graph.add_node(A, node_a) graph.add_node(B, node_b) graph.add_edge(START, A) graph.add_edge(A, B) graph.add_edge(B, END) app graph.compile() print(app.invoke({text: 初始, count: 0}))输出结果{text: 来自B, count: 1}text字段最终是来自B因为node_b覆盖了node_a的写入。count是 1因为node_a增加了它而node_b没有修改count。这种“谁最后写谁生效”的规则适合存单一状态值例如当前路由决策、任务是否完成。5.2 高级规则Annotated reducer当多个节点需要往同一个列表字段追加内容时默认覆盖会丢数据。这时需要给字段声明一个 reducer告诉 LangGraph“新值来了该怎么和旧值合并”。import operator from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END class CombineState(TypedDict): logs: Annotated[list[str], operator.add] def node_a(state: CombineState) - dict: return {logs: [A执行]} def node_b(state: CombineState) - dict: return {logs: [B执行]} graph StateGraph(CombineState) graph.add_node(A, node_a) graph.add_node(B, node_b) graph.add_edge(START, A) graph.add_edge(A, B) graph.add_edge(B, END) app graph.compile() print(app.invoke({logs: []}))输出{logs: [A执行, B执行]}这里Annotated[list[str], operator.add]表示每当有节点返回logs就把新列表与旧列表相加。这就是为什么在 LangGraph 中消息列表通常用Annotated[list, add_messages]作为 reducer——它会把新的用户消息、助手消息持续累积成完整对话历史。在实际项目中设计 State 的字段时就要想清楚这个字段是“单一值”还是“需要积累的列表”。如果选错会出现很隐蔽的数据丢失 bug。6. 条件路由实战ConditionalEdge 深度解析条件路由是 Agent 从“固定流程”走向“自主决策”的第一步。LangGraph 里最常用的 API 是add_conditional_edges。它的核心思路是定义一个路由函数函数读取当前 State返回一个标识add_conditional_edges根据这个标识选择下一个节点。路由函数支持两种返回风格返回字符串配合字典映射返回整数索引配合列表映射。我们用一个“天气助手”示例来演示。假设 Agent 需要判断用户是不是在问天气如果是在问天气走天气节点否则走闲聊节点。# 文件路径demo03_conditional.py from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END class RouteState(TypedDict): user_input: str need_weather: bool response: str def analyze_node(state: RouteState) - dict: # 模拟大模型判断是否属于天气问题 text state[user_input] if 天气 in text or weather in text.lower(): return {need_weather: True} return {need_weather: False} def weather_node(state: RouteState) - dict: return {response: 今天天气晴气温 24~30 度适合出行。} def chat_node(state: RouteState) - dict: return {response: f你正在和我聊{state[user_input]}} def route_after_analyze(state: RouteState) - Literal[weather, chat]: return weather if state[need_weather] else chat graph StateGraph(RouteState) graph.add_node(analyze, analyze_node) graph.add_node(weather, weather_node) graph.add_node(chat, chat_node) graph.add_edge(START, analyze) graph.add_conditional_edges( analyze, route_after_analyze, { weather: weather, chat: chat, }, ) graph.add_edge(weather, END) graph.add_edge(chat, END) app graph.compile() print(app.invoke({user_input: 今天北京天气怎么样, need_weather: False, response: })) print(app.invoke({user_input: 你好呀, need_weather: False, response: }))运行结果{user_input: 今天北京天气怎么样, need_weather: True, response: 今天天气晴气温 24~30 度适合出行。} {user_input: 你好呀, need_weather: False, response: 你正在和我聊你好呀}这段代码里add_conditional_edges的第三个参数是字典key 是路由函数返回的字符串value 是实际要跳转的节点名。这样做的好处是路由函数只负责“决策”图负责“跳转”职责清晰。如果路由函数返回的是整数下标第 3 个参数可以换成列表def route_with_index(state: RouteState) - int: return 0 if state[need_weather] else 1 graph.add_conditional_edges( analyze, route_with