去年年中，我的团队决定把几个稳定的业务Agent全部从LangChain迁到LangGraph。当时不少人觉得这是折腾——毕竟LangChain用得好好的，Agent也能跑。结果迁移完第一个复杂Agent之后，所有人的态度都变了：原来那种靠“链”硬拗出来的流程，在LangGraph里被拆得清清楚楚，状态流转、分支条件、人工介入全部可视化，调试效率直接翻倍。这个系列已经写到第四篇，前几篇分别讲了LangChain的起步、工具调用和Prompt工程，这次我想把LangGraph整个核心知识体系拉通一次，做一个从LangChain过渡到LangGraph的全景式梳理。不管你是刚接触Agent开发，还是已经在用LangChain写了一些简单Agent，这篇文章都适合你。我尽量不堆概念，而是结合我们实际迁移过程中踩过的坑、验证过的用法来写。1. LangChain的链式思维：能跑，但很拧巴要理解LangGraph为什么出现，得先回到LangChain。LangChain最核心的抽象是Chain，本质上是一条固定流水线：Prompt模板接LLM，LLM接输出解析器，或者再接下一个Prompt。这个抽象在文档问答、文本翻译这类线性流程里非常好用，封装干净，写起来也快。但Agent场景一出现，Chain就开始暴露短板。Agent的本质是“模型自己决定调用什么工具、按什么顺序调用”，这需要流程具有分支、循环、条件跳转的能力。用Chain来实现这种能力，做法通常是什么？我见过不少：把模型输出里的意图判断结果塞进一个Router Chain，按路由结果选择不同的子链执行；或者暴力一点，把循环直接嵌在Python函数里，在函数内部循环调用同一个Chain，直到满足退出条件。这两种做法都能跑，但维护成本很高。第一个做法，当工具数量超过五个，路由分支变成网状，代码就变成一团面条，你根本说不清某个网络请求是从哪个分支触发的。第二个做法，循环逻辑和业务逻辑混在同一个函数里，状态完全靠外部变量传递，一旦并发一上来，变量被互相覆盖，Bug惨不忍睹。我在一篇内部技术总结里写过一句话：LangChain的Chain适合“从左到右的一条路”，而Agent需要的是一张“可以走回头路的地图”。LangGraph就是把这张地图做成了正规军。LangGraph并没有推翻LangChain。它保留了LangChain的Prompt模板、模型接口、工具封装这些好用的东西，只是在“流程编排”这一层换了一套引擎。流程图上的每一个节点，可以是普通的Python函数，也可以是LangChain的Runnable；节点的输入输出，共享一个全局状态对象。你过去在LangChain里写下的Prompt、Tool定义、Model调用，在LangGraph里基本可以直接复用，真正的增量是图结构和状态管理。2. 四大基础概念：State、Node、Edge、编译LangGraph的知识体系，可以压缩成四个词：State、Node、Edge、Compile。这四个概念全部吃透，整个框架就没有秘密了。2.1 State：Agent的全局记忆与数据流转中枢很多从LangChain转过来的人，上手LangGraph第一脚踩坑就踩在State上。State就是整个图运行期间共享的一个数据对象，所有节点读它、改它、再传给它，循环往复直到图结束。State的定义方式有两种：TypedDict和Pydantic。我的建议是能用Pydantic就用Pydantic，尤其是字段较多的场景，IDE提示和运行时校验都能帮你省时间。from typing import TypedDict, Annotated import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] current_step: str tool_results: dict注意上面这个Annotated[list, operator.add]，这是LangGraph状态机制的精髓，叫做Reducer。默认情况下，节点返回的字典会覆盖State里同名字段，但如果你把某个字段声明成Annotated[list, operator.add]，那每次更新就是“追加”而不是“覆盖”。这一点在Agent里极其常用。因为Agent的典型状态就是对话消息列表，每执行一步，需要把模型回复、工具结果都追加到同一份消息列表里。如果没有Reducer，用一个新列表覆盖老列表，前一轮对话上下文立刻就丢了。我建议所有LangGraph新手都手动跑一遍Reducer生效和没生效的对比实验，眼见为实，这个机制的理解程度直接决定后面写复杂Agent的幸福感。2.2 Node：把一切操作变成图上的节点LangGraph里的节点就是普通Python函数，签名统一：接收State，返回State的更新字典。你可以把模型调用写进节点，把工具执行写进节点，把自定义逻辑写进节点，然后用add_node注册进图里。from langgraph.graph import StateGraph def call_model(state: AgentState) -> dict: response = llm.invoke(state["messages"]) return {"messages": [response]} graph = StateGraph(AgentState) graph.add_node("call_model", call_model)注意几点实操经验：节点函数里不要持有全局可变状态。同一个节点可能被并发执行多次，如果函数内部用了一个模块级字典缓存东西，数据串了极难排查。节点返回的字典只包含你要更新的字段，不需要把整个State原样返回。LangGraph会自动做合并处理，返回了多余的字段反而容易出错。节点函数默认是同步的。如果某个操作是IO密集型的（比如调用外部API），你可以定义成async def，但注意一旦图里出现异步节点，整个图要用异步接口驱动（后面会细讲）。节点是整个图的基本执行单元，它的粒度怎么切，取决于你想在什么地方能插入逻辑。我自己的习惯是：模型调用一个节点，每个工具单独一个节点，状态处理的纯函数逻辑单独一个节点。粒度分得细，后面做人工介入、排查问题都会方便很多。2.3 Edge：控制流的连接纽带光有节点没有边，图只是一堆孤立的函数。Edge定义了节点之间的流转关系。LangGraph的边分为两类：普通边和条件边。普通边就是无条件连接，一个节点执行完必定进入下一个节点，适合顺序执行的场景。条件边就厉害了，它是实现Agent自主决策的核心。条件边的路由逻辑由你定义的一个函数决定，这个函数读取State，然后返回下一个要执行的节点名字。def route_after_model(state: AgentState) -> str: last_message = state["messages"][-1] if last_message.tool_calls: return "tool_node" return "end" graph.add_conditional_edges("call_model", route_after_model, { "tool_node": "tool_node", "end": END })这里的第三参数字典是可选的，如果你不传，LangGraph默认认为返回值就是目标节点名。我建议还是写清楚映射字典，代码可读性会好很多。条件边数量不受限制，你可以给同一个节点接多个条件边，返回不同的目标。这就像地铁换乘，进站之后你根据目的地选择不同线路。Agent的思考-行动-观察循环，就是靠这条条件边撑起来的：模型判断需要调用工具就去执行工具节点，不需要就直接结束。2.4 编译：从图定义到可执行对象所有节点和边都定义好之后，还需要最后一步：编译。注意这里的编译不是传统意义的编译成机器码，LangGraph的compile()会把你的图结构封装成一个可执行对象，同时绑定检查点存储、中断配置等运行时能力。app = graph.compile() result = app.invoke({"messages": [{"role": "user", "content": "今天北京适合穿什么？"}]})编译后的对象有三种调用方式：invoke：同步执行整个图，返回最终State；stream：同步逐节点执行，每次产生输出就返回给调用方，适合看执行过程；astream：异步流式版本，配合async for使用。实际开发里，我几乎都用stream而不是invoke，原因很实际：Agent执行链路长，一旦某个节点阻塞住，你至少知道卡在哪一步。直接invoke的话，用户只能看到转圈图标，后台日志如果不完整，排查问题全靠猜。第一次在LangGraph里跑通一个两节点的图时，我心里冒出的想法是：这不就是给LangChain装了一套“状态机操作系统”吗？事实证明这个类比挺准确。3. 设计意图：从地铁换乘图到Agent的执行路线说到地铁换乘，我是真心觉得这个类比用来理解LangGraph太贴切了。你从A站上车，到B站需要换乘，这时你会看换乘标识，判断走哪条通道去几号线。对应到LangGraph里：车站就是节点；你手里持有的地铁卡余额就是State；换乘站的指示牌就是条件边；刷卡出站就是END。普通Chain只能让你一条线坐到终点，想换线？你得自己写个判断逻辑去手动“下车再买票”。LangGraph则把整个地铁网络图显式铺在眼前，你可以标注哪条线是主线、哪个换乘点可以人工介入（打断流程等人工确认）。我们在生产环境里有一个工单自动分类Agent，它需要读取工单内容，判断类型，再决定调用哪个知识库检索工具。用LangChain写的时候，分类逻辑和检索逻辑分散在三个Chain里，中间用Router串，整个执行路径不直观。改成LangGraph之后，执行路线一目了然：read_ticket节点读取工单；classify节点判断类型，分类结果写入State；条件边从classify出发，根据类型进入不同的检索工具节点；最后汇总进answer节点生成回复。这个图上墙做Review，团队里任何人都能看懂流程。这就是“把控制流变成数据结构”的威力。4. 实操一个可运行的ReAct Agent理论讲再多不如直接跑一个Demo。下面我写的例子是一个支持天气查询和计算器的信息助手Agent，完全按照ReAct模式实现，即：思考 → 调用工具 → 观察结果 → 再思考，直到答案生成。4.1 第一步：定义状态结构和工具from typing import TypedDict, Annotated, Sequence import operator from langchain_core.messages import BaseMessage, HumanMessage, ToolMessage from langgraph.graph import StateGraph, END class AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], operator.add]工具我就简单实现了两个：一个是查天气的模拟函数，一个是计算器。这里为了演示，我直接用@tool装饰器把它们声明成LangChain工具。from langchain_core.tools import tool @tool def get_weather(city: str) -> str: """获取指定城市当前天气""" weather_data = {"北京": "晴，28℃", "上海": "小雨，24℃"} return weather_data.get(city, "暂无数据") @tool def calculate(expression: str) -> str: """计算简单的四则运算表达式""" try: return str(eval(expression)) except Exception: return "表达式错误"4.2 第二步：构建图from langchain_openai import ChatOpenAI from langgraph.prebuilt import ToolExecutor tools = [get_weather, calculate] tool_executor = ToolExecutor(tools) llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) llm_with_tools = llm.bind_tools(tools) def agent_node(state: AgentState) -> dict: response = llm_with_tools.invoke(state["messages"]) return {"messages": [response]} def tool_node(state: AgentState) -> dict: last_message = state["messages"][-1] tool_calls = last_message.tool_calls outputs = [] for call in tool_calls: tool_result = tool_executor.invoke(call) outputs.append(ToolMessage(content=str(tool_result), tool_call_id=call["id"])) return {"messages": outputs} def should_continue(state: AgentState) -> str: last_message = state["messages"][-1] return "tools" if last_message.tool_calls else "end" graph = StateGraph(AgentState) graph.add_node("agent", agent_node) graph.add_node("tools", tool_node) graph.set_entry_point("agent") graph.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END}) graph.add_edge("tools", "agent") app = graph.compile()4.3 第三步：跑起来result = app.invoke({ "messages": [HumanMessage(content="北京今天多少度？顺便算一下 (23+45)*2")], }) for msg in result["messages"]: print(msg.type, ":", msg.content)执行过程简单描述一下：用户发消息，入口节点agent把消息喂给绑定了工具的LLM；模型判断这两个问题都要调工具，于是输出两个tool_calls；条件边检测到tool_calls非空，路由到tools节点；tools节点逐个执行工具，把结果包装成ToolMessage追加进State；回到agent节点，模型看到工具结果后生成最终回答；条件边发现没有新的tool_calls，走end边，图结束。这个循环就是Agent的灵魂。模型不是一次性输出答案，而是“边看结果边决定下一步”，LangGraph用一个条件边把整个循环撑了起来，代码量不多，但结构非常清晰。5. 记忆、持久化与人工介入：生产级Agent的必备能力跑通Demo只是第一步。真要放到生产环境，有三个能力跑不掉：记忆、持久化、人工介入。5.1 记忆与ThreadAgent一轮对话之后，如果你直接invoke下一轮，State是全新的，模型根本不记得刚才聊了什么。LangGraph给出的方案是Checkpoint。在编译图的时候指定一个CheckpointSaver，再给每次会话指定一个thread_id，LangGraph就会自动把每一步的State存下来。下一次用同一个thread_id继续invoke，图会从最近一次状态接着走。from langgraph.checkpoint.memory import MemorySaver saver = MemorySaver() app = graph.compile(checkpointer=saver) config = {"configurable": {"thread_id": "user_123"}} result1 = app.invoke({"messages": [HumanMessage(content="我叫小明")]}, config=config) result2 = app.invoke({"messages": [HumanMessage(content="我叫什么名字？")]}, config=config) # 模型能回答出“小明”注意MemorySaver是存内存里的，进程重启就没了。真正生产环境请用持久化方案，比如数据库存储的Checkpointer，具体根据你的技术栈选型。5.2 人工介入人工介入用的是interrupt机制，本质是在图执行到某个节点之前挂起，等人工确认后再继续。我们有个Code Review Agent就用了这个能力：模型生成代码建议之后，图会挂起，Senior工程师看一遍，点击确认或修改后，图才继续后续的消息发送。from langgraph.types import interrupt, Command def review_node(state: AgentState) -> dict: approval = interrupt({ "question": "是否批准这条自动回复？", "content": state["messages"][-1].content, }) return {"approval": approval}实现思路就是在需要打断的节点函数里调用interrupt，LangGraph会把状态存到检查点，然后把控制权交回给调用方。后续通过Command(resume=value)恢复执行。这个能力在自动化程度高的业务里非常关键。它让AI出力的同时，人在关键节点保留否决权，而不是完完全全把流程交给模型。6. 常见问题与排查技巧实录最后一块，我把这半年迁移过程中遇到的高频问题整理成清单，条条都是真金白银的经验。6.1 State字段不更新现象：节点返回了某个字段，但下一节点读取到的还是旧值。排查思路：先看目标字段是否被声明成了Reducer（比如Annotated[list, operator.add]），Reducer的语义是追加，不是覆盖，如果你期望覆盖，别给它声明Reducer；再看节点返回值里字段名是否和State定义里的完全一致，一个字母打错，LangGraph不会报错，只是静默忽略。6.2 条件边死循环现象：Agent在agent → tools之间反复跳，停不下来，甚至整个图超时。我遇到的多数情况是工具结果里包含了模型判定为“还要继续调用工具”的关键词。比如工具返回的文本里有一段示例代码，里面恰好长得很像工具调用指令。处理方法是：在工具节点里给ToolMessage增加一个标记，明确来源是工具结果，模型权重的提示词里也要强调“不要对工具返回的示例再发起调用”。另外，给条件边加一个最大步数保护。比如状态里加一个step_count，每次循环自增，超过阈值直接路由到END。这在调试和线上都是保命手段。6.3 并发修改同一状态键现象：消息列表偶尔出现重复内容，或者部分消息丢失。原因通常是Reducer机制应对并发更新时的冲突。LangGraph允许多个节点并行执行，如果两个并行节点同时往同一个字段追加内容，有可能出现竞态。解决方法是给Reducer传一个自定义合并函数，或者在你的业务逻辑里避免“多个路径同时更新同一个字段”。6.4 异步节点驱动方式错误现象：图里定义了async def节点，但用app.invoke(...)去调用，直接报错。LangGraph对异步的支持是值得顺带吐槽一下的。如果你的图里有任何异步节点，整个图必须用await app.ainvoke(...)或async for ... astream(...)来驱动。我们在一个爬虫类Agent里踩过这个坑，最后统一所有节点改成同步函数，避免混用。6.5 回调跟踪查问题时，我习惯在节点函数里临时加一行print打印关键State字段。LangGraph官网也有带界面的调试工具，可以直观看到每一步的状态变化和节点执行顺序。但注意：使用调试工具时不要在生产环境开启，会拖慢执行速度。7. 后面还能怎么扩展？LangGraph这套知识体系，打通之后扩展性非常强。方向包括：多Agent协作：不同Agent作为不同的图节点，Agent之间通过State传递任务，LangGraph天然支持这种拓扑；递归循环：处理需要反复优化的任务（比如写代码多轮测试修复），只需要调整条件边即可；子图封装：把复杂流程拆成子图，在大图里作为一个节点调用，模块化程度更高；自适应工具选择：根据用户的意图动态加载不同的工具集，工具本身不进图，工具列表放进State。我个人实际使用下来，LangGraph最大的价值不是性能，而是把“Agent会怎么做决策”这件事变得完全可解释。以前用LangChain写Agent，模型的决策路径藏在回调日志里，出了问题只能一遍遍翻输出。现在State里每一轮的工具调用、每一步的状态变化都在同一份数据里，任何一步都能回放。这个体验，用过的团队基本都回不去了。如果你做完这篇文章里的所有示例，再回去看那些用LangChain硬写出来的Agent，大概率也会产生“这个流程其实可以更清晰”的感觉。那就对了，说明你开始适应图的思维方式了。
去年年中，我的团队决定把几个稳定的业务Agent全部从LangChain迁到LangGraph。当时不少人觉得这是折腾——毕竟LangChain用得好好的，Agent也能跑。结果迁移完第一个复杂Agent之后，所有人的态度都变了：原来那种靠“链”硬拗出来的流程，在LangGraph里被拆得清清楚楚，状态流转、分支条件、人工介入全部可视化，调试效率直接翻倍。这个系列已经写到第四篇，前几篇分别讲了LangChain的起步、工具调用和Prompt工程，这次我想把LangGraph整个核心知识体系拉通一次，做一个从LangChain过渡到LangGraph的全景式梳理。不管你是刚接触Agent开发，还是已经在用LangChain写了一些简单Agent，这篇文章都适合你。我尽量不堆概念，而是结合我们实际迁移过程中踩过的坑、验证过的用法来写。1. LangChain的链式思维：能跑，但很拧巴要理解LangGraph为什么出现，得先回到LangChain。LangChain最核心的抽象是Chain，本质上是一条固定流水线：Prompt模板接LLM，LLM接输出解析器，或者再接下一个Prompt。这个抽象在文档问答、文本翻译这类线性流程里非常好用，封装干净，写起来也快。但Agent场景一出现，Chain就开始暴露短板。Agent的本质是“模型自己决定调用什么工具、按什么顺序调用”，这需要流程具有分支、循环、条件跳转的能力。用Chain来实现这种能力，做法通常是什么？我见过不少：把模型输出里的意图判断结果塞进一个Router Chain，按路由结果选择不同的子链执行；或者暴力一点，把循环直接嵌在Python函数里，在函数内部循环调用同一个Chain，直到满足退出条件。这两种做法都能跑，但维护成本很高。第一个做法，当工具数量超过五个，路由分支变成网状，代码就变成一团面条，你根本说不清某个网络请求是从哪个分支触发的。第二个做法，循环逻辑和业务逻辑混在同一个函数里，状态完全靠外部变量传递，一旦并发一上来，变量被互相覆盖，Bug惨不忍睹。我在一篇内部技术总结里写过一句话：LangChain的Chain适合“从左到右的一条路”，而Agent需要的是一张“可以走回头路的地图”。LangGraph就是把这张地图做成了正规军。LangGraph并没有推翻LangChain。它保留了LangChain的Prompt模板、模型接口、工具封装这些好用的东西，只是在“流程编排”这一层换了一套引擎。流程图上的每一个节点，可以是普通的Python函数，也可以是LangChain的Runnable；节点的输入输出，共享一个全局状态对象。你过去在LangChain里写下的Prompt、Tool定义、Model调用，在LangGraph里基本可以直接复用，真正的增量是图结构和状态管理。2. 四大基础概念：State、Node、Edge、编译LangGraph的知识体系，可以压缩成四个词：State、Node、Edge、Compile。这四个概念全部吃透，整个框架就没有秘密了。2.1 State：Agent的全局记忆与数据流转中枢很多从LangChain转过来的人，上手LangGraph第一脚踩坑就踩在State上。State就是整个图运行期间共享的一个数据对象，所有节点读它、改它、再传给它，循环往复直到图结束。State的定义方式有两种：TypedDict和Pydantic。我的建议是能用Pydantic就用Pydantic，尤其是字段较多的场景，IDE提示和运行时校验都能帮你省时间。from typing import TypedDict, Annotated import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] current_step: str tool_results: dict注意上面这个Annotated[list, operator.add]，这是LangGraph状态机制的精髓，叫做Reducer。默认情况下，节点返回的字典会覆盖State里同名字段，但如果你把某个字段声明成Annotated[list, operator.add]，那每次更新就是“追加”而不是“覆盖”。这一点在Agent里极其常用。因为Agent的典型状态就是对话消息列表，每执行一步，需要把模型回复、工具结果都追加到同一份消息列表里。如果没有Reducer，用一个新列表覆盖老列表，前一轮对话上下文立刻就丢了。我建议所有LangGraph新手都手动跑一遍Reducer生效和没生效的对比实验，眼见为实，这个机制的理解程度直接决定后面写复杂Agent的幸福感。2.2 Node：把一切操作变成图上的节点LangGraph里的节点就是普通Python函数，签名统一：接收State，返回State的更新字典。你可以把模型调用写进节点，把工具执行写进节点，把自定义逻辑写进节点，然后用add_node注册进图里。from langgraph.graph import StateGraph def call_model(state: AgentState) -> dict: response = llm.invoke(state["messages"]) return {"messages": [response]} graph = StateGraph(AgentState) graph.add_node("call_model", call_model)注意几点实操经验：节点函数里不要持有全局可变状态。同一个节点可能被并发执行多次，如果函数内部用了一个模块级字典缓存东西，数据串了极难排查。节点返回的字典只包含你要更新的字段，不需要把整个State原样返回。LangGraph会自动做合并处理，返回了多余的字段反而容易出错。节点函数默认是同步的。如果某个操作是IO密集型的（比如调用外部API），你可以定义成async def，但注意一旦图里出现异步节点，整个图要用异步接口驱动（后面会细讲）。节点是整个图的基本执行单元，它的粒度怎么切，取决于你想在什么地方能插入逻辑。我自己的习惯是：模型调用一个节点，每个工具单独一个节点，状态处理的纯函数逻辑单独一个节点。粒度分得细，后面做人工介入、排查问题都会方便很多。2.3 Edge：控制流的连接纽带光有节点没有边，图只是一堆孤立的函数。Edge定义了节点之间的流转关系。LangGraph的边分为两类：普通边和条件边。普通边就是无条件连接，一个节点执行完必定进入下一个节点，适合顺序执行的场景。条件边就厉害了，它是实现Agent自主决策的核心。条件边的路由逻辑由你定义的一个函数决定，这个函数读取State，然后返回下一个要执行的节点名字。def route_after_model(state: AgentState) -> str: last_message = state["messages"][-1] if last_message.tool_calls: return "tool_node" return "end" graph.add_conditional_edges("call_model", route_after_model, { "tool_node": "tool_node", "end": END })这里的第三参数字典是可选的，如果你不传，LangGraph默认认为返回值就是目标节点名。我建议还是写清楚映射字典，代码可读性会好很多。条件边数量不受限制，你可以给同一个节点接多个条件边，返回不同的目标。这就像地铁换乘，进站之后你根据目的地选择不同线路。Agent的思考-行动-观察循环，就是靠这条条件边撑起来的：模型判断需要调用工具就去执行工具节点，不需要就直接结束。2.4 编译：从图定义到可执行对象所有节点和边都定义好之后，还需要最后一步：编译。注意这里的编译不是传统意义的编译成机器码，LangGraph的compile()会把你的图结构封装成一个可执行对象，同时绑定检查点存储、中断配置等运行时能力。app = graph.compile() result = app.invoke({"messages": [{"role": "user", "content": "今天北京适合穿什么？"}]})编译后的对象有三种调用方式：invoke：同步执行整个图，返回最终State；stream：同步逐节点执行，每次产生输出就返回给调用方，适合看执行过程；astream：异步流式版本，配合async for使用。实际开发里，我几乎都用stream而不是invoke，原因很实际：Agent执行链路长，一旦某个节点阻塞住，你至少知道卡在哪一步。直接invoke的话，用户只能看到转圈图标，后台日志如果不完整，排查问题全靠猜。第一次在LangGraph里跑通一个两节点的图时，我心里冒出的想法是：这不就是给LangChain装了一套“状态机操作系统”吗？事实证明这个类比挺准确。3. 设计意图：从地铁换乘图到Agent的执行路线说到地铁换乘，我是真心觉得这个类比用来理解LangGraph太贴切了。你从A站上车，到B站需要换乘，这时你会看换乘标识，判断走哪条通道去几号线。对应到LangGraph里：车站就是节点；你手里持有的地铁卡余额就是State；换乘站的指示牌就是条件边；刷卡出站就是END。普通Chain只能让你一条线坐到终点，想换线？你得自己写个判断逻辑去手动“下车再买票”。LangGraph则把整个地铁网络图显式铺在眼前，你可以标注哪条线是主线、哪个换乘点可以人工介入（打断流程等人工确认）。我们在生产环境里有一个工单自动分类Agent，它需要读取工单内容，判断类型，再决定调用哪个知识库检索工具。用LangChain写的时候，分类逻辑和检索逻辑分散在三个Chain里，中间用Router串，整个执行路径不直观。改成LangGraph之后，执行路线一目了然：read_ticket节点读取工单；classify节点判断类型，分类结果写入State；条件边从classify出发，根据类型进入不同的检索工具节点；最后汇总进answer节点生成回复。这个图上墙做Review，团队里任何人都能看懂流程。这就是“把控制流变成数据结构”的威力。4. 实操一个可运行的ReAct Agent理论讲再多不如直接跑一个Demo。下面我写的例子是一个支持天气查询和计算器的信息助手Agent，完全按照ReAct模式实现，即：思考 → 调用工具 → 观察结果 → 再思考，直到答案生成。4.1 第一步：定义状态结构和工具from typing import TypedDict, Annotated, Sequence import operator from langchain_core.messages import BaseMessage, HumanMessage, ToolMessage from langgraph.graph import StateGraph, END class AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], operator.add]工具我就简单实现了两个：一个是查天气的模拟函数，一个是计算器。这里为了演示，我直接用@tool装饰器把它们声明成LangChain工具。from langchain_core.tools import tool @tool def get_weather(city: str) -> str: """获取指定城市当前天气""" weather_data = {"北京": "晴，28℃", "上海": "小雨，24℃"} return weather_data.get(city, "暂无数据") @tool def calculate(expression: str) -> str: """计算简单的四则运算表达式""" try: return str(eval(expression)) except Exception: return "表达式错误"4.2 第二步：构建图from langchain_openai import ChatOpenAI from langgraph.prebuilt import ToolExecutor tools = [get_weather, calculate] tool_executor = ToolExecutor(tools) llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) llm_with_tools = llm.bind_tools(tools) def agent_node(state: AgentState) -> dict: response = llm_with_tools.invoke(state["messages"]) return {"messages": [response]} def tool_node(state: AgentState) -> dict: last_message = state["messages"][-1] tool_calls = last_message.tool_calls outputs = [] for call in tool_calls: tool_result = tool_executor.invoke(call) outputs.append(ToolMessage(content=str(tool_result), tool_call_id=call["id"])) return {"messages": outputs} def should_continue(state: AgentState) -> str: last_message = state["messages"][-1] return "tools" if last_message.tool_calls else "end" graph = StateGraph(AgentState) graph.add_node("agent", agent_node) graph.add_node("tools", tool_node) graph.set_entry_point("agent") graph.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END}) graph.add_edge("tools", "agent") app = graph.compile()4.3 第三步：跑起来result = app.invoke({ "messages": [HumanMessage(content="北京今天多少度？顺便算一下 (23+45)*2")], }) for msg in result["messages"]: print(msg.type, ":", msg.content)执行过程简单描述一下：用户发消息，入口节点agent把消息喂给绑定了工具的LLM；模型判断这两个问题都要调工具，于是输出两个tool_calls；条件边检测到tool_calls非空，路由到tools节点；tools节点逐个执行工具，把结果包装成ToolMessage追加进State；回到agent节点，模型看到工具结果后生成最终回答；条件边发现没有新的tool_calls，走end边，图结束。这个循环就是Agent的灵魂。模型不是一次性输出答案，而是“边看结果边决定下一步”，LangGraph用一个条件边把整个循环撑了起来，代码量不多，但结构非常清晰。5. 记忆、持久化与人工介入：生产级Agent的必备能力跑通Demo只是第一步。真要放到生产环境，有三个能力跑不掉：记忆、持久化、人工介入。5.1 记忆与ThreadAgent一轮对话之后，如果你直接invoke下一轮，State是全新的，模型根本不记得刚才聊了什么。LangGraph给出的方案是Checkpoint。在编译图的时候指定一个CheckpointSaver，再给每次会话指定一个thread_id，LangGraph就会自动把每一步的State存下来。下一次用同一个thread_id继续invoke，图会从最近一次状态接着走。from langgraph.checkpoint.memory import MemorySaver saver = MemorySaver() app = graph.compile(checkpointer=saver) config = {"configurable": {"thread_id": "user_123"}} result1 = app.invoke({"messages": [HumanMessage(content="我叫小明")]}, config=config) result2 = app.invoke({"messages": [HumanMessage(content="我叫什么名字？")]}, config=config) # 模型能回答出“小明”注意MemorySaver是存内存里的，进程重启就没了。真正生产环境请用持久化方案，比如数据库存储的Checkpointer，具体根据你的技术栈选型。5.2 人工介入人工介入用的是interrupt机制，本质是在图执行到某个节点之前挂起，等人工确认后再继续。我们有个Code Review Agent就用了这个能力：模型生成代码建议之后，图会挂起，Senior工程师看一遍，点击确认或修改后，图才继续后续的消息发送。from langgraph.types import interrupt, Command def review_node(state: AgentState) -> dict: approval = interrupt({ "question": "是否批准这条自动回复？", "content": state["messages"][-1].content, }) return {"approval": approval}实现思路就是在需要打断的节点函数里调用interrupt，LangGraph会把状态存到检查点，然后把控制权交回给调用方。后续通过Command(resume=value)恢复执行。这个能力在自动化程度高的业务里非常关键。它让AI出力的同时，人在关键节点保留否决权，而不是完完全全把流程交给模型。6. 常见问题与排查技巧实录最后一块，我把这半年迁移过程中遇到的高频问题整理成清单，条条都是真金白银的经验。6.1 State字段不更新现象：节点返回了某个字段，但下一节点读取到的还是旧值。排查思路：先看目标字段是否被声明成了Reducer（比如Annotated[list, operator.add]），Reducer的语义是追加，不是覆盖，如果你期望覆盖，别给它声明Reducer；再看节点返回值里字段名是否和State定义里的完全一致，一个字母打错，LangGraph不会报错，只是静默忽略。6.2 条件边死循环现象：Agent在agent → tools之间反复跳，停不下来，甚至整个图超时。我遇到的多数情况是工具结果里包含了模型判定为“还要继续调用工具”的关键词。比如工具返回的文本里有一段示例代码，里面恰好长得很像工具调用指令。处理方法是：在工具节点里给ToolMessage增加一个标记，明确来源是工具结果，模型权重的提示词里也要强调“不要对工具返回的示例再发起调用”。另外，给条件边加一个最大步数保护。比如状态里加一个step_count，每次循环自增，超过阈值直接路由到END。这在调试和线上都是保命手段。6.3 并发修改同一状态键现象：消息列表偶尔出现重复内容，或者部分消息丢失。原因通常是Reducer机制应对并发更新时的冲突。LangGraph允许多个节点并行执行，如果两个并行节点同时往同一个字段追加内容，有可能出现竞态。解决方法是给Reducer传一个自定义合并函数，或者在你的业务逻辑里避免“多个路径同时更新同一个字段”。6.4 异步节点驱动方式错误现象：图里定义了async def节点，但用app.invoke(...)去调用，直接报错。LangGraph对异步的支持是值得顺带吐槽一下的。如果你的图里有任何异步节点，整个图必须用await app.ainvoke(...)或async for ... astream(...)来驱动。我们在一个爬虫类Agent里踩过这个坑，最后统一所有节点改成同步函数，避免混用。6.5 回调跟踪查问题时，我习惯在节点函数里临时加一行print打印关键State字段。LangGraph官网也有带界面的调试工具，可以直观看到每一步的状态变化和节点执行顺序。但注意：使用调试工具时不要在生产环境开启，会拖慢执行速度。7. 后面还能怎么扩展？LangGraph这套知识体系，打通之后扩展性非常强。方向包括：多Agent协作：不同Agent作为不同的图节点，Agent之间通过State传递任务，LangGraph天然支持这种拓扑；递归循环：处理需要反复优化的任务（比如写代码多轮测试修复），只需要调整条件边即可；子图封装：把复杂流程拆成子图，在大图里作为一个节点调用，模块化程度更高；自适应工具选择：根据用户的意图动态加载不同的工具集，工具本身不进图，工具列表放进State。我个人实际使用下来，LangGraph最大的价值不是性能，而是把“Agent会怎么做决策”这件事变得完全可解释。以前用LangChain写Agent，模型的决策路径藏在回调日志里，出了问题只能一遍遍翻输出。现在State里每一轮的工具调用、每一步的状态变化都在同一份数据里，任何一步都能回放。这个体验，用过的团队基本都回不去了。如果你做完这篇文章里的所有示例，再回去看那些用LangChain硬写出来的Agent，大概率也会产生“这个流程其实可以更清晰”的感觉。那就对了，说明你开始适应图的思维方式了。