这两年我一直在一线做AI应用工程，最深的感受是：LangGraph工作流与智能体，已经从概念性的Demo，走到了该认真掌握核心机制的时候。以前我习惯用Chain（链）把大模型调用串起来，遇到带记忆、带分支、带人工审批的流程就特别别扭。后来把整个流程重新建模成一张图，才真正解决了问题。这篇内容就围绕“用LangGraph编排复杂工作流、让智能体在可控边界内自主决策”这件事展开，适合已经会用LangChain、但觉得链式调用不够灵活的开发者，也适合刚接触LangGraph、想直接从实战入手的朋友。我把自己的图设计思路、踩过的坑、以及几套可以直接复用的写法一起列出来。

1. 为什么工作流编排要换一种思路：从Chain到Graph的转变

1.1 链式调用的三个痛点

我最早做“搜索资料→写摘要→生成报告”这类流程时，用的是LangChain的Chain，用管道符或glue函数把每一步串起来。开始觉得挺顺畅，但项目一复杂就暴露了三个问题。

第一，状态很难共享。每一步的中间结果要想办法传进下一步，如果某个环节需要同时使用前两步的结果，就得手动组装参数。步骤一多，参数传递的代码比业务逻辑还多，可读性骤降。

第二，路由基本靠ifelse。比如根据用户意图决定走检索还是走生成，或者根据上一步的结果决定是否重试，这些分支逻辑全部散落在主流程代码里。单分支还好，多分支叠加循环以后，整个控制流的走向几乎没法一眼看清。

第三，没有记忆。链式调用执行完就结束了，如果你希望系统能在多轮对话里记住刚才生成的内容，或者能暂停、恢复、回放历史状态，链式结构没有原生支持。要么自己外挂一个状态库，要么每次调用都把上下文重新塞进去，做得非常痛苦。

有一次我带一个内部知识库问答项目，需求是：收到问题后先判断是否需要查资料，需要就并行检索多个来源，然后汇总多个候选答案，再让一个人工节点确认是否合适，不合适就回退重新检索。这类”分支+循环+人工介入“的控制流，用Chain写出来我基本看不懂，更别说维护。后来把这个流程画成图形，问题一下子就清晰了。

1.2 当你需要的是图而不是链

LangGraph的核心思想不复杂：工作流是一张有向图。状态（State）是整张图上的全局数据，节点（Node）是具体的操作，边（Edge）是流转规则。节点从一个状态读到输入，执行完操作后返回状态的一部分更新。

链和图的区别可以用一个例子说明。比如处理一个客服工单：

• 链式的写法是固定走“分类→回答→结束”，中间想插入一个人工处理步骤，得重新组装整条链。
• 图式的写法是把“分类”节点连到“自动回答”节点和“人工处理”节点之间，再根据分类置信度决定走哪条边，后续还能继续加节点，不会影响已有的部分。

我整理了一个对比表，方便你直观感受两者的差异：

这并不意味着Chain完全没用。流程固定且简单，Chain反而更轻。但只要你需要“看情况决定下一步”，甚至“多步流程反复执行”，图就是更自然、更可维护的表达方式。

2. 理解LangGraph的四个核心概念：状态、节点、边、检查点

2.1 State是唯一的真相

LangGraph里，State通常是一个TypedDict。整个图执行过程中，这个Dict是唯一数据源。每个节点函数接收当前State作为参数，返回一个“要更新的字段”字典，图框架负责把返回值合并回State。

听起来很简单，但里面有一个容易被忽略的细节：字段的合并策略。

默认情况下，一个字段如果被两个节点都返回值，后执行的那个会直接覆盖前面的。这在很多场景是符合预期的，比如“最终答案”“当前分类结果”。但如果你希望某个字段是“追加模式”，比如对话消息列表，每次节点返回都往列表里加新消息，那就需要指定字段的Reducer。

我习惯这样定义：

import operator from typing import TypedDict, Annotated class AgentState(TypedDict): messages: Annotated[list, operator.add] # 追加模式 current_topic: str # 覆盖模式 search_count: int

Annotated[list, operator.add]的意思是：当多个节点都返回messages字段时，新内容追加到已有列表后面。这是LangGraph里最常用也最关键的技巧，没设置Reducer的话，你辛苦收集的历史消息很可能被最后一步覆盖掉。

2.2 节点不只是函数

理论上，节点就是一个Python函数：输入State，输出State的子集。但节点内部并不仅限于调用大模型。我在实际项目里塞过检索函数、数据库查询、用户审批接口、甚至触发外部任务后等待回调的逻辑。

在LangGraph里，节点可以是同步函数，也可以是异步函数。如果整个图都用async/await风格写，执行效率会更高，尤其是节点内部有I/O等待时。我建议一个节点只做一件事，比如“调用模型生成问题”和“执行搜索”分开，这样单独调试、复用、加权限都好说。

有一条我踩过坑的规律：节点之间不要直接传递全局变量，一切数据都通过State走。如果某个节点需要前序多个节点的数据，直接从State里取；如果节点要返回多个结果，就写成多个State字段。一开始不习惯，但图复杂以后，这种约定能让你少掉很多头发。

2.3 普通边与条件边

LangGraph的边分两种。

普通边用add_edge，表示固定的流转关系：A执行完一定走到B。比如“读取输入→规范化”这种没有分支的步骤。

条件边用add_conditional_edges，第二参数是一个路由函数，接收State，返回下一步的目标节点名。所有“看情况走哪里”的逻辑都收敛在这里。

我的经验是，路由函数不要写复杂逻辑，尽量保持简单、可预测。一个典型写法：

def route_after_analysis(state) -> str: if state["has_material"] and state["search_count"] < 3: return "generate" return "search"

需要注意，add_conditional_edges的第三个参数是路径映射。如果路由函数返回的是"generate"，但映射里没有这个键，运行时直接报错。这个错误我犯过不止一次，后来把映射里的每个键都写成了常量，避免手滑。

2.4 Checkpointer：让工作流有记忆

Checkpointer是LangGraph提供的关键机制。它会像录像机一样，在执行到每个“超级步”后把State快照持久化到存储后端。等下次需要从某个点继续时，可以通过load_state恢复。

这里要引入一个概念：thread_id。可以把它理解成会话ID或Case ID。同一个图的多次运行，只要thread_id相同，State就是连续的；不同thread_id之间完全隔离。

一个实用场景：多轮对话机器人。用户的每一轮提问都使用同一个thread_id调用同一张图，对话历史自然累积在State里，不需要自己拼聊天记录。另一个场景是人机协同流程：图执行到某个节点后暂停，等人工审批后再调用恢复。Checkpointer保证暂停期间的所有数据都是落盘的，不会因为进程重启而丢失。

我在原型阶段喜欢用MemorySaver做内存版Checkpointer，进程一结束状态就没了。真正上生产环境再换成数据库版Checkpointer，比如SQLite、Postgres的实现，数据才能长期留存。

3. 手写一个“研究助手”工作流，从图设计到完整跑通

3.1 场景拆解与图设计

下面用一个具体项目来说明，我称之为“研究助手”。用户给出一个主题，系统自动拆解成子问题、逐项检索资料、汇总判断是否还有信息缺口，最后生成一份研究报告。这个场景几乎覆盖了LangGraph大多数常用能力：节点、条件循环、多字段状态、Checkpointer。

图的结构非常简单：

START → 规划子问题 → 并行检索 → 汇总分析 → 是否继续? → 是则回到检索，否则生成报告 → END

设计阶段不要急着写代码。我会先把每个状态字段列出来，再标出每个节点要读什么、写什么。这个习惯帮我避开了大量后期返工。

3.2 定义状态与节点

状态字段定义如下：

from typing import TypedDict, Annotated class ResearchState(TypedDict): topic: str questions: list[str] search_results: dict[str, str] needs_more: bool report: str loop_count: int

第一个节点是规划子问题。它调用模型，把主题拆成若干可检索的具体问题。

def plan_questions(state: ResearchState) -> dict: prompt = f"围绕「{state['topic']}」拆解出3到5个需要检索的子问题，直接输出编号列表。" response = llm.invoke(prompt) questions = [line.strip() for line in response.content.splitlines() if line.strip()] return {"questions": questions, "loop_count": state.get("loop_count", 0) + 1}

检索节点是典型的I/O密集操作。如果子问题有4个，串行检索会很慢，我用线程池并行处理：

from concurrent.futures import ThreadPoolExecutor def search_all(state: ResearchState) -> dict: results = {} with ThreadPoolExecutor(max_workers=3) as pool: for q, res in zip(state["questions"], pool.map(search_engine, state["questions"])): results[q] = res return {"search_results": results, "loop_count": state["loop_count"] + 1}

在真实项目里，search_engine可以替换成内部搜索服务的客户端，或者一个向量数据库查询函数，甚至多路检索后做个简单融合也可以。尽量把这类具体逻辑包在函数里，图节点只负责编排。

3.3 条件路由：如何防止死循环

汇总分析节点把搜索结果喂给模型，判断当前资料是否足够。如果不够且还没超过最大循环次数，就回到检索节点继续搜；否则进入生成报告节点。

def analyze_material(state: ResearchState) -> dict: material = "\n".join(f"{k}: {v}" for k, v in state["search_results"].items()) prompt = f"基于以下资料判断是否足够撰写报告。只回答yes或no。\n{material}" answer = llm.invoke(prompt).content.strip().lower() return {"needs_more": answer == "no"} def should_continue(state: ResearchState) -> str: if state["needs_more"] and state["loop_count"] < 4: return "retry" return "finalize"

这里的关键点是把循环上限放在路由函数里。loop_count在多个节点里递增，配合上限判断，从机制上避免模型反复要求重搜导致死循环。这种环视设计在工作流里非常重要，没有上限的循环就是一把随时会引爆的雷。

最后是生成报告节点：

def generate_report(state: ResearchState) -> dict: material = "\n".join(f"{k}: {v}" for k, v in state["search_results"].items()) report = llm.invoke(f"根据以下资料撰写结构清晰的报告：\n{material}") return {"report": report.content}

3.4 编译成可执行对象并加入记忆

接下来把节点拼成一张图。先实例化StateGraph，传入状态定义；接着添加节点和边；最后调用compile()得到可执行对象。

from langgraph.graph import StateGraph, START, END builder = StateGraph(ResearchState) builder.add_node("plan", plan_questions) builder.add_node("search", search_all) builder.add_node("analyze", analyze_material) builder.add_node("generate", generate_report) builder.add_edge(START, "plan") builder.add_edge("plan", "search") builder.add_edge("search", "analyze") builder.add_conditional_edges( "analyze", should_continue, {"retry": "search", "finalize": "generate"} ) builder.add_edge("generate", END) graph = builder.compile()

加上Checkpointer和thread_id，整个工作流就带上了多轮会话记忆：

from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() graph = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "research-case-001"}} result = graph.invoke({"topic": "LangGraph工作流设计原则"}, config) print(result["report"])

我强烈建议在compile()之前，把节点定义、边定义都放在同一个模块里。如果节点函数在多个文件分散，调试的时候很难追踪执行路径，只能靠记忆拼图，效率很低。整张图的“地图”应该一眼能看完，这是可维护性的底线。

4. 给工作流装上大脑：智能体循环与ReAct模式

4.1 从固定管道到自主决策：工具调用循环

上面这个研究助手的工作流节点是事先定好的，模型只是在节点内部做内容生成。这种设计叫“确定性编排”：流程骨架不变，模型负责具体文本。但很多人想做的智能体，是让模型自己决定“下一步该调用什么工具、调几次、什么时候结束”。这正是LangGraph里最常用的ReAct模式：思考（Reason）、行动（Act）、观察（Observe）循环。

用LangGraph实现这个循环很直观：模型节点返回结果，如果结果里有工具调用请求，就走工具执行节点；工具执行完把结果放回消息历史，再回到模型节点继续判断，直到模型给出最终回答。控制权完全交给模型，但边界由你设定。

下面是我在项目里频繁使用的一个最小实现。

4.2 实现一个带记忆的工具调用循环

先定义状态。这里的关键是消息列表用追加Reducer：

import operator from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langchain_core.messages import AIMessage, ToolMessage class AgentState(TypedDict): messages: Annotated[list, operator.add]

模型节点：

def call_model(state: AgentState) -> dict: response = llm.invoke(state["messages"]) return {"messages": [response]}

工具执行节点。这里假设tools_by_name是一个字典，键是工具名，值是可调用对象。工具调用的返回结果必须以ToolMessage的形式写回消息历史，而且要绑定对应的tool_call_id，否则一些模型服务商不认账。

def call_tool(state: AgentState) -> dict: last_ai_message = state["messages"][-1] tool_outputs = [] for tool_call in last_ai_message.tool_calls: tool = tools_by_name[tool_call["name"]] result = tool.invoke(tool_call["args"]) tool_outputs.append( ToolMessage(content=str(result), tool_call_id=tool_call["id"]) ) return {"messages": tool_outputs}

路由函数逻辑就一句话：最后一条AI消息里有没有工具调用请求？有就执行工具，没有就用最终回答作为图终态。

def route_after_model(state: AgentState) -> str: last_message = state["messages"][-1] if last_message.tool_calls: return "tools" return END

把节点和边拼起来：

builder = StateGraph(AgentState) builder.add_node("model", call_model) builder.add_node("tools", call_tool) builder.add_edge(START, "model") builder.add_conditional_edges("model", route_after_model, {"tools": "tools", END: END}) builder.add_edge("tools", "model") graph = builder.compile()

这个循环每次工具执行完，消息列表会多出一条ToolMessage，再回到模型节点，模型就知道工具返回了什么，可以据此决定下一步。我一直觉得，这是理解“智能体”概念最直观的代码：所谓智能，就是“模型根据不断更新的上下文决定动作序列”的能力。

4.3 我在这个环节踩过的坑

这个工具循环看起来简单，但在真实项目里翻车风险不少。我整理了三个最经典的。

第一个坑是忽略tool_call_id。有次我图省事，工具返回的ToolMessage没有绑定id，结果模型后续消息死活对不上工具结果，表现为模型反复重试同一个工具、输出逻辑混乱。排查了很久才定位。工具调用的每次返回都必须带上对应的tool_call_id，这是消息历史的关联键。

第二个坑是模型并行返回多次工具调用。现在主流模型服务商都支持一次返回多个工具调用，但有些人写的工具节点只处理第一个，剩下的直接丢弃。结果就是模型规划好的行动只执行了一半，最后答案错得莫名其妙。处理方式是在工具节点里用循环遍历tool_calls列表，每个都执行、每结果都收集，再考虑要不要用asyncio.gather并发执行，加快速度。

第三个坑是工具参数描述不清晰。模型不会“猜”你的工具要什么参数，全靠函数签名和文档字符串。如果参数名写input、data这种含糊词，模型生成参数时会瞎填。我现在写工具函数时，参数名一定自带语义，比如file_path、timeout_seconds，并在docstring里写清取值范围和示例。这个投入对最终效果的影响，比调模型温度参数大得多。

5. 让图可观测、可调试、可人工接管

5.1 用debug开关与get_state查看状态

图跑不起来或者结果不对，第一步不是改代码，而是看状态。LangGraph里有一个简单但常被忽略的调试方法：调用时打开debug参数。

result = graph.invoke(input_data, config, debug=True)

该开关会把每一步执行前的状态、正在运行的节点、节点返回的增量一并打印出来。对于追踪“某个字段为什么一直没变、某条边为什么走了这个分支”这类问题非常有效。

另一个常用操作是graph.get_state(config)。拿到thread_id对应当前快照，可以知道现在执行到哪里、State里存了什么。我在排查多轮对话问题时，经常用它单独检查某个会话的状态，而不必等整张图跑完。

5.2 用interrupt机制把“人”放回循环里

很多人把LangGraph当作纯自动流程，忽略了它的一个强大能力：工作流可以在任意节点暂停，等待人工决定。这在审批、复查、异常兜底场景里极其好用。

以研究助手为例，如果生成报告前需要人工确认检索材料是否可靠，可以在编译时指定interrupt_before=["generate"]：

graph = builder.compile(checkpointer=checkpointer, interrupt_before=["generate"])

首次调用会在进入generate节点前暂停，外部系统此时可以检查State内容，把它展示给负责人。人工确认后，通过graph.update_state更新某个审批字段，再调用一次graph.invoke(None, config)让图继续往下走。

这个模式的工程价值很大。真正的生产系统很少是完全无人值守的，总有一些节点需要人工判断。用LangGraph自带的暂停/恢复能力，比自己做外部状态机要省力得多，也天然与Checkpointer的快照机制兼容。

5.3 并发与性能：并行检索子问题

回到研究助手那个例子，我做过一次性能优化：把检索节点从串行改成并发。在4个线程并发的情况下，整个工作流从原来25秒左右降到10秒左右。这个提升非常可观，因为LLM生成时间占比反而下降了。

用ThreadPoolExecutor能解决问题，但如果追求更高的并发控制，LangGraph也支持把分支分发成多个“并行继续执行”的任务。这在图里叫Send API，适合“把一个节点输出分发到多个处理节点”的场景。我在实际项目中用得不是特别多，因为大部分I/O密集问题用线程池和异步节点就够解决了。

我的建议是先测量瓶颈再优化。不要一开始就用复杂的并行结构，先用串行版本跑通，再观察哪个节点耗时最长。多数情况下，卡片式的并发工具调用就是最大瓶颈，把那一处优化掉就行了。

结尾的一点体会

把LangGraph用于生产项目之后，我最大的体会是：工作流的核心不在代码，而在于“把流程想清楚”。画图、定义状态、决定每一条边的走向，这些前置设计做扎实了，后面的开发顺很多。反之，如果你连状态字段都没想明白，就急着写节点函数，一定会返工。

如果让我给后来者一个建议，那就是从小图开始。哪怕是一个只有三个节点的小闭环，也先完整跑一遍，加上Checkpointer和条件边，感受一下状态如何流转。等这套心智模型建立起来，再上多节点、带人工介入、带并发分支的复杂工作流，你会发现自己对AI应用项目控制力的提升是肉眼可见的。实践之后如果遇到不好归因的疑难问题，记得回到状态里找答案——大多数诡异现象，往回看某个字段是怎么变的，比往前猜模型动机要可靠得多。