LangGraph实战:用图编排构建可控的AI Agent工作流
不是标题党也不打算让你看完就收藏吃灰。LangGraph这个框架近两年在AI应用开发圈里热度一直很高但很多人误以为它只是LangChain生态里的一个小工具或者觉得“直接用LangChain的Agent不就够了为什么要多学一个图框架”。直到你开始做真正复杂的多步骤任务、需要持久化记忆、或者要让AI在几个工具之间来回决策的时候链式调用那套东西就会明显卡住。这篇把我自己从Chain迁移到Graph的完整思考、核心概念拆解、以及一个可直接改来用的实操示例全部摊开讲适合正在做Agent开发、准备把AI能力落到真实业务里的程序员参考。1. 为什么从Chain走到Graph链式编排解决不了的三个场景1.1 线性流程的天花板很多人在LangChain里最早接触的是Chain比如LLMChain、SequentialChain。它的思路很直观把Prompt模板、模型调用、输出解析等组件串成一个流水线上一个环节的输出就是下一个环节的输入。处理“给一段文本先翻译再总结”这种固定顺序的任务时链式模式完全够用。但真实业务里没有那么多顺滑的直线。我举几个自己实际遇到过的场景第一工具调用的循环。Agent需要判断用户问题是否需要调用查询工具查完数据库后发现结果还不够还需要再查一次或者需要调用第二个工具。这种“查询—判断—再查询—再判断”的动作天然是一个环而链式结构不支持回头路。用Chain硬写只能靠一个Agent自己反复调用工具流程不透明中途想插入一个人工确认步骤更是无从下手。第二分支逻辑。用户输入的内容先要判断是闲聊、查单、退换货还是投诉不同意图走完全不同的处理流程。链式结构当然也能通过if-else在代码层面分叉但分叉多了之后业务逻辑散落在各个回调函数里整条链路变得很难梳理也很难可视化地排查“用户走到哪一步了”。第三状态持久化。聊天机器人需要记住对话历史多轮任务中需要在多个节点之间共享中间结果。链式调用里每个环节都是独立的函数调用状态要么塞在context里一层层传要么自己维护外部存储代码里到处是胶水逻辑。这三个场景归纳起来就一句话你需要的不再是一条线而是一张可以决定走向的图。LangGraph的核心思路就是让开发者用图的思维方式来编排AI工作流节点表示动作边表示流转关系并且支持条件分支和循环。1.2 图框架带来的思维转变我最早看到LangGraph时第一反应是“这不就是状态机吗”。想深一层之后发现它确实是一个面向LLM应用的状态机但比传统状态机多了一个关键能力每个状态流转都可能触发一次模型调用而模型的下一次输出又会决定往哪条边走。这个思维转变很重要。以前写Agent你的思维模式是“调用一个大模型让它自己决定下一步”LangGraph则是“把大模型的决策能力嵌入到流程图中每一步做什么是开发者预先设计的模型只在关键的判断节点上做选择”。前者是黑盒后者是可观测、可控制、可断点续跑的白盒。我自己切换到图模式之后最大的感受是调试从“猜Agent在想什么”变成了“看它走到了哪个节点根据条件结果判断为什么走错了”。这种确定性在开发阶段和线上运维阶段都是巨大的效率提升。2. 核心概念拆解看懂LangGraph的设计语言2.1 State整个图共享的状态对象LangGraph里最核心的概念是State它是一份贯穿整个图的数据结构图里每个节点都能读取和修改它。你可以把State理解为一张在多个工人之间传递的“流转单”每个工人在上面填自己的处理结果下一个人继续在此基础上操作。实际操作中State通常用TypedDict类定义比如from typing import TypedDict, Annotated, List class WorkflowState(TypedDict): question: str history: List[str] tool_results: List[str] intent: str final_answer: str我在项目里的经验是State字段设计要克制能用局部变量传的参数就不要放进State。因为所有写入State的数据最终都会被Checkpointer持久化保存塞太多临时变量会让每次存储的数据量变大恢复状态时也可能带上无用的脏数据。2.2 Node把动作封装成节点Node是图里执行实际工作的单元它本质上就是一个接收State、返回State更新的函数。下面是LangGraph最常见的节点定义方式from langgraph.graph import StateGraph def classify_node(state: WorkflowState) - dict: # 调用模型判断意图 intent call_llm_for_intent(state[question]) # 只返回需要更新的字段 return {intent: intent} def handle_order_node(state: WorkflowState) - dict: # 处理订单查询工具调用 result call_order_api(state[question]) return {tool_results: state[tool_results] [result]}需要特别注意的是LangGraph节点返回的字典只会更新State中对应的字段。也就是上面classify_node只更新intent字段其他字段原样保留。这个设计很实用但也带来了一个常见坑——如果你把某个字段误写成覆盖比如对历史列表用 new_list而不是扩展就会把之前节点写入的数据冲掉。2.3 Edge定义节点之间的流转关系边用来定义节点之间的连接。LangGraph提供三种核心边类型普通边固定从A到B。条件边根据一个路由函数的结果决定从当前节点走向哪个后续节点。入口边定义整个图从哪里开始执行。条件边是最常用的因为它正好对应“让模型在关键岔路口做判断”的场景。一个典型的路由函数长这样from langgraph.graph import START, END def route_by_intent(state: WorkflowState) - str: if state[intent] order: return handle_order elif state[intent] after_sale: return handle_after_sale else: return small_talk graph.add_conditional_edges(classify, route_by_intent, { handle_order: handle_order, handle_after_sale: handle_after_sale, small_talk: small_talk, })条件边让我可以在图里画出真实的业务判断逻辑而不是把一堆判断塞进Agent的系统提示词里。AI负责理解用户意图图的控制逻辑负责决策往哪走责任边界清晰。2.4 Checkpointer给整个流程加上记忆Checkpointer是LangGraph里我觉得价值被严重低估的一个特性。它的作用是把每一次节点执行后的State快照保存下来保存位置可以选内存、文件系统或外部数据库。有了Checkpointer你至少获得三种能力断点续跑流程执行到一半崩溃了可以从最近一个保存点恢复不用从头再来。人工介入在执行到某个敏感节点时暂停等人审批确认后再继续。多轮会话记忆把同一用户的多次请求关联到同一个会话ID每次新请求都从历史State恢复上下文。实际使用中不要每执行一步就持久化一次那样开销太大。LangGraph允许你配置在执行到特定节点时save我一般只在关键业务节点前设置持久化把IO成本控制在合理范围内。3. 从零搭建一个可复现的LangGraph Agent3.1 定义一个够用但不过度设计的场景我拿一个最常见的“智能客服流程”来做示例。场景要求是收到用户提问后先做意图识别如果是查询类问题调用一个模拟的查询工具查询完成后再次判断是否还需要补充信息整个流程支持多轮对话记忆。这个场景覆盖了LangGraph最核心的四种能力状态管理、节点封装、条件路由和记忆持久化。看懂这套结构其他复杂Agent基本就是在这个骨架上加节点和边。3.2 完整代码一个带分支和记忆的查询Agentimport json from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver # 1. 定义全局状态 class AgentState(TypedDict): question: str history: Annotated[List[str], operator.add] # 使用add reducer避免覆盖 need_tool: bool tool_result: str answer: str # 2. 定义工具函数 def fake_order_query(order_id: str) - str: # 模拟订单查询 return f订单{order_id}状态为已发货预计3天内送达 # 3. 定义各个节点 def intent_node(state: AgentState) - dict: # 实际项目中这里会调用LLM判断意图 question state[question] need_tool 订单 in question or 查 in question return {need_tool: need_tool} def tool_node(state: AgentState) - dict: # 从问题中简单提取一个订单号 order_id A12345 result fake_order_query(order_id) return {tool_result: result} def answer_node(state: AgentState) - dict: if state.get(tool_result): answer f查询结果{state[tool_result]} else: answer 我现在可以帮你解答一般性问题具体订单问题请提供订单号。 return {answer: answer, history: [fQ: {state[question]}, fA: {answer}]} # 4. 条件路由函数 def should_use_tool(state: AgentState) - str: return tool if state[need_tool] else answer # 5. 构建图 graph StateGraph(AgentState) graph.add_node(intent, intent_node) graph.add_node(tool, tool_node) graph.add_node(answer, answer_node) graph.add_edge(START, intent) graph.add_conditional_edges( intent, should_use_tool, {tool: tool, answer: answer} ) graph.add_edge(tool, answer) graph.add_edge(answer, END) # 6. 编译并启用内存检查点 checkpointer MemorySaver() app graph.compile(checkpointercheckpointer) # 7. 执行 config {configurable: {thread_id: user-001}} result app.invoke( {question: 帮我查一下订单A12345到哪里了, history: []}, configconfig ) print(result[answer]) # 第二次问带上thread_id后历史自动关联 result2 app.invoke( {question: 那这个订单什么时候发货的呢, history: []}, configconfig ) print(result2[answer])3.3 这段代码里的几个设计要点先说为什么用Annotated[List[str], operator.add]。这是LangGraph里一个容易忽略但很重要的机制。默认情况下节点返回什么State字段就会被覆盖为什么。如果history字段不允许追加每次节点返回时都会用[new_item]替换掉之前的所有历史。加上operator.add这个reducer之后LangGraph会自动把返回的列表元素追加到原有列表尾部而不是替换整个列表。这就是状态更新策略的配置理解了它就能避免很多“为什么运行几次之后历史只有最后一次数据”的诡异问题。再说thread_id。它相当于对话会话的ID同一个thread_id的多次invoke会被LangGraph认为是同一个会话。第一次调用时State里的history是空的执行完answer节点时history被写入了一条问答记录。第二次调用时因为指定了相同的thread_idLangGraph会先从checkpointer里恢复history字段之后所有节点读到的state里就带上了上一轮的上下文。这个机制让我们不需要自己管理数据库里的对话记录框架层就把活儿干了。最后说条件路由的返回映射。should_use_tool返回的是逻辑分支名称tool或answer第三个参数里再定义分支名和实际节点的映射关系。这个映射是可选的如果路由函数返回的名字和节点名一致可以不写映射。4. 工具选型LangGraph、AutoGen、CrewAI到底怎么选我经常被问到一个问题现在做多Agent编排有好几个框架LangGraph、AutoGen、CrewAI各有各的说法到底学哪个。我的答案一向很简单**如果你的应用核心是“可控的业务流程LLM决策”选LangGraph如果你的应用核心是“多个Agent自由对话协作”可以考虑AutoGen如果你需要高度抽象的多Agent角色扮演可以看看CrewAI。**但三个框架都学一遍没必要真要上手干活LangGraph带来的长期收益最大。为什么会这么说因为LangGraph本质上是图执行引擎控制逻辑在开发者手里模型的自由度被约束在节点内部。这种模式放到生产环境有巨大的优势流程可观测、出错了可以看路径回放、中途可以暂停人工介入。这些都是企业级应用最关心的特性。AutoGen的核心强调Agent之间的对话让多个Agent互相发消息来完成任务。这种方式在做研究原型、探索式场景时很灵活但对话过程天然不可控Agent之间聊着聊着跑偏了是常有的事。你很难在AutoGen里精准地表达“必须先完成A步骤然后才有条件地执行B步骤”。CrewAI把Agent封装成带有角色和目标的“员工”用团队协作的方式组织概念很友好上手也快。但封装的层次比较高一旦需要细粒度控制某个流程节点或者需要自定义状态持久化策略你会发现自己被框架抽象限制住了还得绕回底层去改。我的建议是学LangGraph因为它是离“图的本质”最近的框架。图、节点、状态这些概念本身就是软件工程的基础知识学会了之后就算未来不用LangGraph迁移到任何别的编排框架都很快。而其他框架的Agent抽象则不太通用换个工具可能就要推翻重学。5. 常见问题与踩坑实录5.1 状态覆盖Date 的隐性坑我最早写LangGraph时遇到过一类非常隐蔽的Bug节点A返回了字段a和b节点B返回了字段b和c按理说最后State里应该同时有a、b、c三个字段。但实际运行时a字段有时候存在有时候不存在。排查后发现是节点返回的问题。LangGraph在节点执行结束后会用节点返回的字典更新State这个更新逻辑对每个字段单独判断。如果前一个节点返回了a和b后一个节点只返回了b和c那a是保留还是消失答案是**a还是会保留因为更新的粒度是字段级不是节点级。**但如果后一个节点返回的是{a: None}那a字段就会被覆盖成None。所以我的习惯是节点返回值里只写本节点真正要改动的字段其他字段一概不返回。宁可多写几个专门处理字段合并的小工具函数也不要在节点里去手动操作整个State。5.2 Checkpointer反序列化异常使用持久化检查点时不少人会遇到恢复State时类型不对的问题。比如之前State里存的是一个对象实例恢复后变成了字典或者字段缺失。这多半是因为State里存了非JSON序列化的数据比如自定义类实例、函数对象、数据库连接。经验法则State里只放能序列化的数据。复杂对象放ID或引用需要时在节点里重新加载。我见过有人把Pandas DataFrame直接放进State看起来很方便但恢复时会报错最后只能改成存取文件路径。LangGraph的持久化设计偏向于让State保持轻量你要顺着它的设计来不要硬塞重对象。5.3 并发更新同一个thread_idLangGraph支持同一thread_id的多次invoke但如果你在同一个thread_id上并发发起多个请求可能会互相覆盖状态。比如用户双击提交按钮两个请求同时从同一个State出发后写入的那个会把先写入的覆盖掉。解决方式有两个一是业务层保证同一个会话串行处理最简单的做法是加锁或队列二是给每次请求用不同的thread_id恢复时再用历史版本管理。实际项目中我用得最多的是第一种因为用户操作天然是串行的接口层做个防重就够了。5.4 调试技巧让图跑得明明白白LangGraph提供了get_state和update_state这样的接口可以在图执行过程中手动检查状态、甚至手动修改状态后再继续执行。我强烈建议在开发阶段把每一步的State都打印出来路径调试比直接看最终输出直观得多# 模拟执行到中途检查当前state current_state app.get_state(config) print(current_state.values) # 手动修改state后继续 from langgraph.types import Command app.invoke(Command(resume{need_tool: False}), configconfig)这一段是调试利器尤其是排查“为什么它走错分支”的时候直接看关键字段在什么节点被改掉比反反复复调Prompt高效得多。另外LangGraph的draw_mermaid_png方法可以生成流程图节点关系一眼就能看懂强烈建议画出来贴在文档里有人问流程时直接甩图。6. 一些个人经验和最后一句话如果让我给刚接触LangGraph的人一个上手路线我会这么说先别急着写Agent去读一遍StateGraph的文档然后用一个只有两个节点的图跑通invoke再加入条件边再加入checkpointer最后再放工具调用。每一步都动手改代码、看状态变化比看十篇教程都实在。我自己在实际开发中的体会是LangGraph的图模式不只是技术选择更是一种思维方式。它逼着你在动手写代码之前先把流程想清楚——哪些环节需要模型判断哪些环节是固定逻辑哪些环节允许人工介入。这个前置设计的过程反而帮你节省了大量后期调试的时间。最后分享一个实用小技巧定义节点函数时建议统一把函数第一行写成打印当前节点名和关键输入这样不管最后逻辑多复杂你都能通过日志快速定位问题。这个习惯我沿用到现在每次排查线上问题时都受益。如果你正在准备把LLM应用做成真正能被业务使用的系统LangGraph值得花时间研究。它不是万能的但至少把“可控”、“可恢复”、“可观测”这三个生产级需求第一次这么清晰地统一到了一套API里。