LangGraph实战：用状态图掌控Agent流程与多Agent协作

LangGraph 这类工具，最近讨论最多的就是它能不能把复杂的 Agent 流程真正变成可控制、可复用、可调试的工程代码。我的判断是：如果你已经受够了拿一堆 if/else 拼 Prompt、在主流程里到处塞状态变量、多个 Agent 一协作就乱套，LangGraph 值得投入时间。它用三个核心概念——State、Node、Edge——把任务拆成一张图，每一步改什么状态、走哪条边、失败去哪、循环到什么时候停，都能明确写出来。这篇文章面向零基础读者，我会从环境准备开始，直接带你跑通第一个最小图，再逐步拆解状态管理、条件分支、多 Agent 协作，最后补上我自己实际排查时最容易踩的坑。最值得关注的地方是：LangGraph 不只是画图框架，它的价值在于把流程控制从 Prompt 里解放出来，让大模型应用具备真正的工程结构。

1. 先弄清楚 LangGraph 到底解决什么问题

1.1 为什么要用状态图来管理大模型流程

过去写大模型应用，最常见的做法是写一个长函数：先调用模型，拿到结果，再判断结果内容，再用 if/else 决定下一步调哪个模型。这种代码在小 Demo 里很直观，但一旦任务多起来，问题立刻暴露。

第一个问题是流程被写死在代码里。今天要加一个 LLM 调用，明天要给某个步骤加人工确认，后天要调整分支条件，就要去改主函数，改完很容易影响其他环节。

第二个问题是状态管理混乱。多个模型调用之间需要共享上下文：用户输入、中间结果、重试次数、错误信息、最终答案。用全局变量、类成员、字典来回传，短时间能跑，时间长了根本不知道当前状态被谁改过。

第三个问题是多 Agent 协作没有统一调度逻辑。A Agent 的输出要交给 B Agent，B 出错要不要交给 C 处理，还是直接重跑一次？这个决策如果放在模型 Prompt 里做，结果就不稳定；如果放在代码里做，又很难覆盖所有情况。

LangGraph 解决的就是这三类问题：把流程建模成一张图，状态集中定义，节点只负责“从状态拿输入、算结果、更新状态”，边负责“下一步到底去哪”。这张图在运行前可以先检查结构是否合法，运行中每一步都能看到状态变化，出问题还能回放。

1.2 和 LangChain 到底有什么区别

很多人刚接触时会混淆 LangChain 和 LangGraph，搜索里的相关提问也非常多。

LangChain 偏重模型调用封装和工具集成。它提供统一的模型接口、Prompt 模板、输出解析器、向量库连接方式，适合快速组装“模型 + 工具 + 知识库”这类能力。但 LangChain 本身不强行规定流程结构，流程怎么走还是你自己控制。

LangGraph 更偏重任务编排。它把业务流程显式建模成图，明确每个节点是什么、节点之间怎么连接、状态如何流转。它并不替代 LangChain 的模型封装能力，反而经常配合 LangChain 使用。

我的理解是：LangChain 解决“怎么方便地调用大模型”，LangGraph 解决“多个大模型步骤之间怎么协调”。如果只是单个模型问答，或者一个简单工具调用，直接用 LangChain 就够了。一旦需要多个模型按条件执行、循环执行、并行执行，LangGraph 的优势才会体现出来。

好，概念理解到这一步就可以直接上手了。不要等把文档全看完再动手，先跑通最小代码，再回头对照概念，会快很多。

2. 搭建环境，先把最小图跑起来

2.1 安装依赖：Python 环境、包管理、版本确认

LangGraph 目前以 Python 生态为主，官方文档里也以 Python 示例最多。Node.js 用户虽然也有相关资源，但本文按 Python 路线走，对新手更友好。

先确认本机环境：

python --version pip --version

建议使用 Python 3.10 或 3.11。如果本机版本较老，先用虚拟环境隔离，不要直接改系统 Python。

创建虚拟环境并激活：

python -m venv langgraph_env # Windows langgraph_env\Scripts\activate # macOS / Linux source langgraph_env/bin/activate

安装核心依赖：

pip install langgraph langchain-openai

如果是用其他模型供应商，可以替换成对应的 LangChain 集成包。安装完成后，先确认一下版本：

pip show langgraph

不同版本的 API 会有细节差异，比如某些字段改名、某些边注册方式变化。我的建议是：不要盲目追求最新版，先固定一个已知能跑的版本，等第一个图跑通后再考虑升级。具体版本号以你安装时的最新稳定版为准，这一点在排查问题时非常重要。

2.2 第一个最小图：单节点、双节点、终止边

写一个最简单、没有任何实际语义的图，先验证链路能通。

先引入必要组件：

from typing import TypedDict from langgraph.graph import StateGraph, END class DemoState(TypedDict): text: str def step_one(state: DemoState): print("执行 step_one") return {"text": state["text"] + " -> step_one"} def step_two(state: DemoState): print("执行 step_two") return {"text": state["text"] + " -> step_two"} graph = StateGraph(DemoState) graph.add_node("step_one", step_one) graph.add_node("step_two", step_two) graph.set_entry_point("step_one") graph.add_edge("step_one", "step_two") graph.add_edge("step_two", END) app = graph.compile() result = app.invoke({"text": "start"}) print(result)

如果一切正常，控制台会输出：

执行 step_one 执行 step_two {'text': 'start -> step_one -> step_two'}

这段代码虽然简单，但它完整展示了 LangGraph 的基本套路：定义状态结构，编写节点函数，构建图，添加节点，连接边，编译图，然后 invoke 输入。后续所有复杂功能，都是在这个基础上扩展。

2.3 验证图和状态更新的判断标准

第一次跑通后，不要急着加功能，先确认几个关键点：

第一，invoke的返回值里，text字段是不是按你预期累积了内容。LangGraph 默认的 reducer 行为是“覆盖字段”，所以如果你在多个节点里都返回同一个字段，后一个节点会覆盖前一个节点写入的值。上面例子里能累积，是因为我在每个节点里都把旧值取出来拼接后再返回。这里很容易踩坑，后面我会专门说。

第二，看执行顺序是不是严格按边的方向走。如果发现 step_two 先执行，那要检查是否设置了错误的 entry point，或者节点名重复导致覆盖。

第三，编译阶段有没有报错。graph.compile()是结构校验，不是运行校验。如果图里存在悬空边、缺失节点、入口不合法，通常会在 compile 时报错。

注意：如果控制台只输出结果、没有打印节点里的内容，不要慌，先确认是不是用了别的运行方式。用invoke是同步执行，节点打印应该会显示。如果用了ainvoke异步方式，则要在 async 环境里运行。

3. 拆解 State、Node、Edge 三个核心组件

3.1 State：状态结构怎么设计，直接影响后续所有节点

State 是 LangGraph 里的核心数据结构，所有节点共享同一个状态对象。你可以把它理解成一张“任务单据”，上面记录了跑完这个流程需要携带的所有信息。

用 TypedDict 定义状态的好处是类型清楚、IDE 有提示、运行时也能减少字段拼写错误。

一个实际场景的状态设计示例：

from typing import TypedDict, Annotated, List def merge_messages(existing: List[str], new: List[str]) -> List[str]: return existing + new class AgentState(TypedDict): user_query: str intermediate_results: Annotated[List[str], merge_messages] error_count: int final_answer: str

这里值得强调的是Annotated。它在类型标注里附加了一个 reducer 函数，表示这个字段在多个节点更新时，不是简单覆盖，而是调用你指定的合并策略。

如果字段不加 reducer，默认行为是“后写覆盖”。这适合user_query、final_answer这类只需要保存最终值的字段。而intermediate_results、messages这类需要不断累积的字段，就应该定义一个合并函数。

实际经验是：状态字段不要设计得太大、太散。字段太多，节点函数里来回取、回传，心智负担很重；字段太少，又容易把所有信息塞进一个大字符串，后续难以结构化处理。建议按照“用户输入、中间过程、最终输出、控制信息”四类来拆分。

控制信息也很关键，比如error_count、max_retries、current_step。这类字段决定了流程要不要重试、要不要跳转，放错地方会让分支逻辑很难维护。

3.2 Node：节点函数怎么改状态值

Node 的本质是一个普通函数：接收当前状态，返回一个字典。返回的字典会被 LangGraph 合并进下一次状态。

一个常见误区是：在节点函数里直接修改传入的 state 参数，然后不返回任何内容。LangGraph 的运行机制并不保证这种修改会持续生效，更稳妥的方式是返回一个包含要更新字段的字典。

示例：

def call_llm_node(state: AgentState): # 这里真正调用模型，省略具体请求 answer = f"processed: {state['user_query']}" return {"final_answer": answer} def check_error_node(state: AgentState): # 根据当前状态判断是否要重试 error_count = state.get("error_count", 0) if "fail" in state.get("final_answer", ""): return {"error_count": error_count + 1} else: return {"error_count": error_count}

第二个节点即使不需要更新字段，也要返回一个空字典或者原样返回。因为节点函数可以只读取状态不更新，此时返回空字典即可：

def read_only_node(state: AgentState): print(state["user_query"]) return {}

节点命名要简短有意义。LangGraph 允许自定义字符串节点名，这一步不要省。后面排查复杂流程图时，日志里出现的就是节点名，名字起得好，几分钟就能定位问题；名字随意，查到自己都头大。

3.3 Edge：普通边和条件边的本质区别

Edge 决定执行顺序。LangGraph 里最常用的是普通边和条件边。

普通边用graph.add_edge(from_node, to_node)，表示固定跳转。比如：

graph.add_edge("step_one", "step_two")

这句话的意思是：step_one执行结束后，无条件进入step_two。

条件边用add_conditional_edges，表示根据状态或外部结果动态选择下一个节点。比如：

def route(state: AgentState): if state.get("error_count", 0) >= 3: return "give_up" return "retry" graph.add_conditional_edges( "check_error_node", route, { "give_up": "give_up", "retry": "retry", } )

这里route是路由函数，返回值是字符串，字典的 key 必须包含所有可能返回值，字典的 value 是实际要跳转的节点名。字典也可以省略，LangGraph 支持用路由函数直接返回节点名，但显式写映射表更清楚，也能在编译期检查。

普通边的优势是简单直接；条件边的优势是动态。实际流程永远是两者结合：主干用普通边，分支点用条件边，边界处用 END 终止。

4. 手把手实现条件分支与循环控制

4.1 设计一个真实的业务分支场景

条件分支最常见的场景是：模型生成结果后，需要判断“要不要重试”。

这里给一个典型例子：一个任务流程包含意图识别、任务执行、结果检查、重试或终出口。

我要先定义完整状态：

from typing import TypedDict class TaskState(TypedDict): task_input: str intent: str execution_result: str retry_count: int max_retries: int

再定义节点：

def recognize_intent(state: TaskState): # 识别意图，省略真实模型调用 result = "search" return {"intent": result} def execute_task(state: TaskState): # 执行具体任务 return {"execution_result": "success" if state["retry_count"] < 1 else "fail"} def inspect_result(state: TaskState): # 检查结果 if "fail" in state["execution_result"]: return {"retry_count": state["retry_count"] + 1} return {"retry_count": state["retry_count"]} def handle_failure(state: TaskState): # 达到重试上限，走兜底处理 print("达到最大重试次数，转人工或兜底") def finish_success(state: TaskState): print("任务成功完成")

这里retry_count和max_retries就是控制重试次数的关键字段。路由函数应该读取这两个字段，而不是靠模型 Prompt 去决定。

4.2 conditional_edges 的路由函数和映射表

路由函数核心逻辑：

def should_retry(state: TaskState): if state["retry_count"] >= state["max_retries"]: return "give_up" if "fail" in state.get("execution_result", ""): return "retry" return "success" graph = StateGraph(TaskState) graph.add_node("recognize_intent", recognize_intent) graph.add_node("execute_task", execute_task) graph.add_node("inspect_result", inspect_result) graph.add_node("handle_failure", handle_failure) graph.add_node("finish_success", finish_success) graph.set_entry_point("recognize_intent") graph.add_edge("recognize_intent", "execute_task") graph.add_edge("execute_task", "inspect_result") graph.add_conditional_edges( "inspect_result", should_retry, { "retry": "execute_task", "success": "finish_success", "give_up": "handle_failure", } ) graph.add_edge("handle_failure", END) graph.add_edge("finish_success", END) app = graph.compile()

这里最重要的设计是：路由函数不依赖大模型的输出解析，而是读取状态里的结构化字段。这样无论模型输出变成什么样，只要我们在节点函数里把结果判定写成结构化字段，分支就能稳定执行。

4.3 循环检测和终止条件怎么设置

如果inspect_result判断为 retry，就会跳回execute_task，从而形成循环。这在 LangGraph 里是允许的，但必须确保循环能在有限次数内退出。

终止条件通常有几种：

1. 计数器达到上限，如retry_count >= max_retries。
2. 状态字段满足某个条件，如final_answer != ""。
3. 时间或成本预算，如已经消耗多少轮模型调用。

如果出现“最大递归深度超限”或者图进入无限循环，优先检查这三项：计数器有没有在节点函数里正确更新，路由分支里是否每个节点都有到达 END 的路径，状态字段名是否写错。

另一个容易忽略的点是：如果重试节点同时改写了输入参数，循环后的状态可能和第一次执行时不一致，导致错误越来越复杂。我的建议是重试时尽量只修改控制字段，不要随意改写核心输入，这样每次循环的起点才可控。

5. 多 Agent 协作的几种落地模式

5.1 主从模式（Supervisor）的本质分工

多 Agent 协作最常用的不是“让 Agent 自由聊天”，而是由一个主控 Agent 统一调度，把任务分发给多个专业 Agent。这种模式在 LangGraph 里非常自然。

一个简单的主从结构是：

• Supervisor 节点负责理解用户需求，决定下一步把任务交给哪个 Agent。
• Worker Agent 节点负责具体执行，返回结果。
• Supervisor 根据 Worker 的返回结果判断任务是否完成，或者继续分发下一个任务。

举个最小例子：

from typing import TypedDict, Literal class TeamState(TypedDict): user_request: str assigned_to: str worker_result: str finish: bool def supervisor_node(state: TeamState): # 根据请求分发任务，可以调模型判断，这里用规则模拟 if "数学" in state["user_request"]: return {"assigned_to": "math_agent"} if "写作" in state["user_request"]: return {"assigned_to": "writing_agent"} return {"assigned_to": "general_agent"} def math_agent(state: TeamState): return {"worker_result": "math result", "finish": True} def writing_agent(state: TeamState): return {"worker_result": "writing result", "finish": True} def general_agent(state: TeamState): return {"worker_result": "general result", "finish": True} def supervisor_route(state: TeamState): return state["assigned_to"]

在这个结构里，Supervisor 不一定是要复杂的大模型 prompt，也可以是一个规则映射函数。关键点是：任务分发的决策写入assigned_to字段，后面的路由通过这个字段执行。这样后续想换成模型决策，只需要改写supervisor_node，不需要改整个图结构。

5.2 把 Subagent 当作特殊 Tool 来调用

现在很多 LangGraph 多 Agent 实践里，会强调一个观点：Subagent 本质上可以看作另一种 Tool 调用。不要把多个 Agent 当作完全平行的“智能体聊天室”，而是把子任务封装成工具，主 Agent 只决定“要不要调用、传什么参数、怎么检查返回结果”。

这种设计的好处是结构清晰：主 Agent 仍然是调用方，保持单一的决策焦点；子 Agent 是执行方，输出结构化结果。排查问题时，你可以把子 Agent 当作普通工具调用来验证，降低不确定性。

在 LangGraph 里实现一个 Subagent 工具的方式有很多，比如在节点里直接把子 Agent 的函数作为普通 Python 函数调用。不需要强行把子 Agent 也实现成一张独立图。除非你确实需要子 Agent 内部也具备循环、分支和持久化能力，否则先用函数调用更快。

5.3 并行分支和状态汇总

有些任务可以拆成多个互不依赖的子任务并行执行，LangGraph 支持从同一个节点出发连接多个后续节点。很多帖子提到“并行分支”，实际做起来要注意状态原子性。

并行分支的状态更新机制是：多个分支各自产生返回值，然后在交汇点合并。如果多个分支同时更新同一个普通字段，会存在冲突。所以并行分支里，我一般把每个分支的结果写入独立的字段，最后用一个汇总节点把它们拼起来。

示例状态设计：

from typing import TypedDict, Annotated, List def merge_result(left: List[str], right: List[str]) -> List[str]: return left + right class ParallelState(TypedDict): input_text: str branch_a_result: Annotated[List[str], merge_result] branch_b_result: Annotated[List[str], merge_result] final_output: str

两个分支节点分别写branch_a_result、branch_b_result，最后汇总节点读取这两个字段生成final_output。这里用 Annotated 是为了防止某分支被多次调用时覆盖掉之前写入的内容。

并行并不是越多越好。模型调用本身耗时，多个并行分支同时跑，会同时占用多个资源。如果你用的是本地模型，显存和内存就很容易被打满；如果是 API 调用，还要注意并发限制和成本。不要一上来就开 10 个并行分支，先在两个分支上验证数据不互相污染，再逐步增加。

6. 状态持久化、长期记忆与工程化要点

6.1 需要持久化时，引入 Checkpointer

LangGraph 的默认运行方式是把状态保存在内存里，invoke 结束后状态就没了。如果需要在任务中途恢复、记录历史会话、断电恢复，就需要引入 checkpointer。

从搜索来看，大家对长期记忆的关注度确实很高。长期记忆分成两类：短期任务状态和跨会话记忆。短期任务状态是“当前任务进行到哪一步”，长期记忆是“用户上次说过什么偏好”。

短期任务状态用 checkpointer 保存，可以在中断后从某个节点恢复。长期记忆更常见的做法是把关键用户偏好写入数据库或向量库，在需要时重新加载到状态里，而不是把整个历史都塞给模型。

以 SQLite 为 Checkpointer 的示例思路（不同版本 API 可能有差异，以官方文档为准）：

from langgraph.checkpoint.memory import MemorySaver saver = MemorySaver() app = graph.compile(checkpointer=saver) config = {"configurable": {"thread_id": "user_session_123"}} app.invoke({"task_input": "hello"}, config=config)

thread_id是关键参数。同一 session 的多次 invoke 会共享 checkpointer 保存的状态，这可以用于带记忆的对话流程。

6.2 批量任务时，先关注输入命名、失败重试和日志

很多人从单条任务切到批量任务时，会直接写一个 for 循环不断 invoke。这样能跑通，但不适合生产环境。

批量任务至少要处理三个问题：输入如何组织、输出如何命名、失败如何重试。如果你有 1000 条输入，中间第 500 条因为 API 超时失败，你是整批重跑，还是单独重跑这一条？我建议先把批次切小，每批 50 条左右，保留每条任务对应的输入文件路径和输出文件路径，失败时记录日志，再单独拉起重试。

内存也要注意。如果状态里放了很大的文本或图片内容，且每次都追加到列表字段里，那么随着任务越来越多，内存占用会持续上涨。批量任务建议每批跑完后手动清理不需要的历史字段，或者在状态设计时就避免无限累积大对象。

6.3 什么情况下不建议强行上 LangGraph

并不是所有 LLM 应用都需要 LangGraph。如果你的任务只是“用户输入一个文本，调一次模型，返回结果”，直接写函数调用更简单。如果任务只有三个以内步骤，而且几乎没有分支，也可以不用图框架。

LangGraph 的真正价值在复杂流程、多 Agent 协作、需要明确控制循环和状态恢复的场景。如果流程不复杂，强行引入图结构反而会让代码变长、学习成本变高。

我一般建议的准入标准是：步骤数大于 5、有多条分支、有循环重试、有多种终止条件、需要多人协作维护流程逻辑。满足三条以上，LangGraph 才会明显带来收益。

7. 常见报错和排查顺序

7.1 修改状态值但下一个节点读不到

这是新手最容易遇到的问题。现象是：节点 A 返回了{"final_answer": "xxx"}，节点 B 读取state["final_answer"]时却拿不到，或者拿到的是旧值。

排查顺序：

1. 先看节点 B 是不是真的执行在节点 A 后面。检查边的连接有没有写错。
2. 再看节点 A 返回的 key 是否和状态类型定义里的 key 完全一致。一个字母拼错就会导致新字段被写进状态，但节点 B 读的是另一个字段。
3. 看节点 A 返回的值类型是否符合 Annotated 合并函数的要求。比如合并函数期望 List，你却返回了字符串，可能报错。

如果字段需要累加，确认有没有定义 reducer。没有定义 reducer 的字段，第二次写入会覆盖第一次写入。

7.2 条件路由总走错分支

条件路由走错，绝大多数是路由函数写错了，不是图本身有问题。

排查顺序：

1. 在路由函数里打印当前状态的关键字段，确认值到底是什么。
2. 检查映射表的 key 是否覆盖路由函数所有返回值。如果返回了映射表里没有的值，运行时会报错或走默认行为。
3. 检查路由函数的返回值类型。LangGraph 允许返回节点名，但如果你混用了“返回节点名”和“返回 key 再映射”，很容易混乱。我建议固定用映射表方式，统一可查。

有一个隐蔽问题：路由函数里用了state.get("some_field", default)，但字段名写错时，它不会报错，只会返回默认值，结果分支被判定到错误路径。检查字段名要仔细。

7.3 图编译失败和循环异常

编译失败通常包含：节点不存在、边指向不存在的节点、入口节点未设置、映射表 value 指向不存在的节点。这类问题看报错信息基本能定位。

循环异常通常分两种：循环次数超预期，或者进入死循环。先确认重试字段是否被正确递增。再确认终止条件判断的字段是否在循环节点里被正确更新。最后检查边连接：如果重试路径回到了一个会再次触发同样条件的节点，而且没有更新控制字段，就会死循环。

7.4 依赖版本不一致导致 API 变化

LangGraph 的 API 还在快速迭代，有些零基础教程里的写法在最新版里可能已经不推荐。遇到报错时，不要只盯着业务逻辑，也要看看版本。

pip show langgraph查看当前版本；pip freeze里看 langchain 相关包版本。如果项目是团队协作，建议在requirements.txt里锁版本，尤其是生产项目。

我的习惯是：文档里复制代码时，先看官方文档标注的最低版本要求。很多看起来是“代码写错”的问题，实际上是包版本差异。

8. 从零到可维护，LangGraph 的正确落地姿势

回到开头的问题。LangGraph 最值得学习的不是每个 API 长什么样，而是它逼着你想清楚一件事：你的业务流程到底有几条分支、每个分支什么时候结束、状态到底怎么流转。

我的建议是先别急着写上层业务，先用 30 分钟跑通本文里的最小图，再逐个加入条件分支、循环、并行、checkpointer。整个过程都围绕同一个简单的业务样例：一个文本处理任务，做了意图识别，尝试执行，结果不行就重试，最多三次，成功进入收尾，失败进入兜底。

跑通之后，你再决定要不要上多 Agent。多 Agent 不要一开始就铺开。先用一个 Supervisor 节点配两个 Worker Agent，跑一批真实输入，观察每次路由结果是否符合预期。确认稳定后，再增加 Agent 数量，再考虑把 Subagent 封装成 Tool 调用。

最后说说长期维护。LangGraph 项目如果只靠散落的节点函数，时间长了还是会乱。建议把节点函数按业务模块拆分到不同文件，图的组装单独放一个文件，状态类型单独放一个文件。每个节点函数尽量只做一件事，输出结构化字段，不要依赖隐式全局变量。这样就算三个月后再回来改流程，只要打开图组装文件，重新捋一遍边和分支，就能很快恢复上下文。

搜了很多资料和热词之后，我发现真正让 LangGraph 出圈的不是炫酷的 Demo，而是它提供了一种“把不确定性管起来”的思路。模型输出可以不稳定，但图的骨架是稳定的。你在 Edge 上判断、在 State 上记录、在 Node 里执行，这种结构越早建立，后面做复杂 Agent 应用就越不容易翻车。