资讯动态

LangGraph实战:用图编排搞定有状态AI Agent工作流

发布时间:2026/10/23 5:17:15 来源:尧图企业网站定制
做AI Agent应用的这段时间我最大的感受是写单个大模型调用很简单但一旦涉及多步骤、要保存中间状态、要让人工介入确认代码就会迅速失控。无论是回调嵌套、全局变量管理还是自己写状态机最终都会掉进“能跑但不敢改”的泥潭。这个问题困扰了我很久直到我认真过了一遍LangGraph才真正感受到“以图为中心”的开发方式有多顺手。这篇内容我会直接把我在项目中使用LangGraph的完整经验拆开来讲从原理、核心概念到可复用的工程实现一条线说清楚。LangGraph是一个专注于构建有状态、可编排AI工作流的应用开发框架。它不是一个“大而全”的全家桶而是一个把Agent开发从“脚本串联”提升到“图结构编排”的底层支撑。它的核心价值在于把复杂的多步骤AI应用拆解成节点和边的图由框架来负责状态流转、分支控制、持久化和人工介入开发者只需要专注于每个节点内部到底做什么。老实说这个框架适合两类人一类是写过多轮Agent但被复杂逻辑折磨过的开发者另一类是打算做生产级AI应用、不想只停留在demo层面的技术决策者。1. 为什么是LangGraphAI应用编排的痛点与解法1.1 传统开发范式在大模型应用上的尴尬传统的后端开发我们习惯用函数调用来组织逻辑A函数调用B函数B函数返回结果整个流程是确定的、线性的、可预期的。但大模型应用本质上是一个“不确定的流程”同一个prompt模型可能决定调用工具可能决定直接回答也可能反问澄清。如果你用传统的函数调用来写这个“不确定性”会让代码充斥着大量的if-else判断甚至需要自己设计一个复杂的状态机来管理各种分支。我之前做一个自动写作Agent时就踩过这个坑。整个流程需要“分析主题—生成大纲—逐段写作—整体润色—人工确认—最终输出”。用普通Python代码实现我需要维护一个全局变量来存中间产物每一步都要检查上一步的状态是否完整还要设计一个“下一步是什么”的判断逻辑。前两周写起来还行到了后面要加入“用户中途修改大纲”这个分支时代码几乎无法维护。每一次状态变化都要同步修改多处逻辑改着改着就不知道哪里会有遗漏。LangGraph对这个问题给出的答案非常简单直接把流程重新定义为一张图。每个环节是一个节点节点与节点之间的连线是边状态是全局唯一的、流动的数据。开发者不需要再关心“这个函数什么时候被调用”只需要设立好节点和边的拓扑关系框架自然会按照逻辑去执行。这种思维方式对传统开发者来说需要一点点适应时间但一旦适应了你会觉得之前那些复杂的流程控制其实本质都是图的问题。1.2 图结构带来的关键优势图结构所带来的第一优势是可视化思维。你可以把整个应用直接用图的形式画出来先做什么、再做什么、什么条件下走哪条路、哪里需要等待用户反馈一目了然。这对团队协作尤其重要。以前评审一个Agent的逻辑需要读几百行代码现在直接把图画出来大家看一眼就知道改哪里。第二优势是可控性。在LangGraph中你可以随时在一个节点执行结束后暂停整个图的运行等待外部输入然后再恢复执行。这是普通函数调用链做不到的。举个例子在生成完一篇文章后你希望用户确认是否继续发布。在LangGraph里这只是一个边上的判断条件节点执行完自动停住等用户给了确认再往下走。这个能力对生产级应用来说太关键了。第三优势是持久化。图运行时每走一步都会把状态保存下来。进程崩了、机器重启了恢复的时候可以从断点继续跑而不是从头再来。这对长耗时任务尤其重要。你做一个月度报告生成已经跑到第7个节点了这时候服务器重启传统脚本基本就要重跑而在LangGraph里直接从上次停下的节点继续就好。2. 核心概念拆解把LangGraph的几根支柱掰开说2.1 状态State整个图流转的数据中枢LangGraph里的State不是某个局部变量它是贯穿整个图运行的数据中枢。你可以把它理解为一张“画布”每个节点在这张画布上填充、修改、覆盖内容然后传给下一个节点。在代码层面State通常定义为一个TypedDict或Pydantic模型。我重点提醒一点定义State时要设计好哪些字段是“只读快照”哪些字段是“追加记录”。比如对话消息列表就应该设计成追加式的每次节点往里加消息而不是覆盖掉之前的消息。LangGraph提供了相应的机制来实现这种追加操作用起来很顺手。我自己第一次写的时候没注意把历史消息字段定义成了普通字符串结果每跑一步前面的对话历史就丢了排查了好久才发现是State设计问题。State是图结构的核心约束点它决定了节点之间能传什么数据、不能传什么数据。所以建议在项目开始前先把State字段完整地列出来哪怕后面要加也比中途乱改要好。一个建议是宁可字段多几个也不要事后频繁改State的结构因为所有节点都依赖它。2.2 节点Node与边Edge流程的最小执行单元与连接器节点是LangGraph中的最小执行单元。每个节点接收State作为输入执行自己的逻辑返回更新后的State片段。节点的逻辑可以是任何Python代码调用大模型、调用外部API、访问数据库、执行本地脚本都行。节点本身不关心别的节点是怎么实现的它只关心输入输出的State契约。边决定状态在节点间的流转方向。最简单的是普通边表示固定顺序执行复杂一点的是条件边根据当前State的内容决定下一步走哪个节点。这是LangGraph实现“大模型决定下一步”的关键机制。举个例子你有一个节点负责“意图识别”根据识别结果条件边可以选择走“调用搜索工具”节点也可以走“直接回答”节点。大模型的输出在这里起到了路由决策的作用。我当时写第一个Agent时最兴奋的点就是条件边。以前要自己写一个router把大模型的输出映射到具体的函数现在直接声明一条条件边通过一个映射函数就可以实现。代码量少了很多逻辑也更直观了。不过条件边有三个坑要提醒一下映射函数一定要有默认分支防止大模型输出了意外的结果导致图抛异常。条件边的判断条件要放在State的共享字段里不要放在节点内部局部变量里不然其他节点看不到。条件分支的命名要语义化方便在调试时快速看懂走向。2.3 节点间协作为什么需要“汇总节点”和“分支节点”在图结构里不是所有流程都是简单的一条线走到底。实际项目里经常出现“多个节点并行执行最后汇总结果”的情况。LangGraph对这种情况有很优雅的支持你可以在图中定义多个分支框架会自动并行执行互不依赖的节点然后在某个节点历史上汇合。我第一次用这个能力是在做一个多源资料收集Agent。需要同时去几个不同的数据源抓取信息然后汇总成一份报告。按传统写法这些抓取逻辑要么串行执行导致速度慢要么自己开线程池然后再处理结果。在LangGraph里只需要把几个抓取节点设计成互不依赖的分支最后接一个汇总节点即可。整个过程不需要手动管并发框架帮你调度。当然汇总节点要自己处理“部分分支失败”的情况。我比较推荐的做法是每个分支节点不轻易抛异常而是把异常信息写入State的error字段里汇总节点读取时自己判断哪些分支数据可信。这样既不会让整个图垮掉也给后续重试留了空间。3. 实操过程从零搭建一个可落地的Agent工作流3.1 环境准备与项目初始化在动手之前先把环境准备好。LangGraph是基于Python的Python版本建议3.10及以上。安装主框架和配套的LangChain生态组件我的实践是用pyproject.toml管理依赖核心依赖如下面的代码块所示。LangGraph本身是框架核心LangChain的模型封装层用起来方便配上LangGraph的checkpoint存储。[tool.poetry.dependencies] python ^3.11 langgraph ^0.2.0 langchain-core ^0.3.0 langchain-openai ^0.2.0 langgraph-checkpoint-sqlite ^0.1.0安装完成后我建议先做一个最小可用图验证环境通畅再开始复杂逻辑。这里给一个最简单的例子两个节点先问候再告别中间通过State传递消息。这个过程能帮你确认安装没问题也能直观感受State是怎么在节点间流动的。from typing import TypedDict from langgraph.graph import StateGraph, END class SimpleState(TypedDict): messages: list[str] def greet_node(state: SimpleState) - SimpleState: return {messages: state[messages] [你好我是助手]} def bye_node(state: SimpleState) - SimpleState: return {messages: state[messages] [再见期待下次对话]} graph StateGraph(SimpleState) graph.add_node(greet, greet_node) graph.add_node(bye, bye_node) graph.add_edge(greet, bye) graph.set_entry_point(greet) graph.add_edge(bye, END) app graph.compile() result app.invoke({messages: []}) print(result[messages])这段代码跑通之后你对LangGraph的核心流转就会有一个直观的感觉。重点观察一下每个节点接收的state和它返回的state之间的关系理解“每次节点返回的部分字段会被合并到全局State中”这件事。这是LangGraph运作的基础机制。3.2 实现一个带工具调用和人工确认的完整流程最小示例跑通后我直接进入真实场景设计一个“智能客服工单处理”工作流。这个流程做的事情是用户描述问题Agent判断是否需要查数据库查询后给出方案最后请用户确认是否解决。这个例子涵盖了工具调用、条件分支和人工介入三个核心场景。先定义State结构。这里需要几个字段user_request用户原始输入。intent意图识别结果是“咨询”还是“投诉”。db_result数据库查询结果。final_answer最终回复内容。needs_confirmation是否需要用户确认。class TicketState(TypedDict): user_request: str intent: str db_result: str final_answer: str needs_confirmation: bool然后是节点函数。意图识别节点调用大模型判断意图数据库查询节点模拟查库方案生成节点根据查询结果生成回复人工确认在这里我简化成一个条件分支。def analyze_intent(state: TicketState) - TicketState: prompt f判断以下用户请求的类型只能回答咨询或投诉\n{state[user_request]} response llm.invoke(prompt) return {intent: response.content.strip()} def query_database(state: TicketState) - TicketState: if state[intent] 咨询: return {db_result: 常见问题库未命中转人工处理} return {db_result: 投诉工单已创建编号为20250801} def generate_answer(state: TicketState) - TicketState: if state[db_result].startswith(常见问题): return { final_answer: 抱歉未能从知识库找到有效方案已为你转接人工。, needs_confirmation: False } return { final_answer: f已处理完毕工单编号{state[db_result]}, needs_confirmation: True } def route_after_generate(state: TicketState): if state[needs_confirmation]: return confirm return end这里把人工确认设计成了一种特殊节点在真实场景中这个节点会通过某种阻塞机制等待用户输入在示例里就简化为打印确认信息。把节点和边组合起来就能得到一个完整图。graph StateGraph(TicketState) graph.add_node(analyze, analyze_intent) graph.add_node(query, query_database) graph.add_node(generate, generate_answer) graph.add_node(confirm, confirm_node) graph.set_entry_point(analyze) graph.add_edge(analyze, query) graph.add_edge(query, generate) graph.add_conditional_edges(generate, route_after_generate, {confirm: confirm, end: END}) graph.add_edge(confirm, END) app graph.compile()走一遍这个流程你就能发现LangGraph把“意图分析—分支路由—人工介入”这三件在传统代码里很麻烦的事情全部压缩成了图上的节点与边。整个逻辑线特别清楚跟画的架构图几乎一一对应。这是LangGraph能落地的核心原因代码即架构图架构图即代码。3.3 持久化配置与断点续跑的正确姿势在演示代码里我们用的是内存State一跑完就没了。到了生产环境必须有持久化LangGraph为此提供了checkpoint机制我实际项目里用的是基于SQLite的存储方式。配好持久化之后图每次执行节点的中间状态都会被记录即使触发中断也能从断点恢复。from langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string(checkpoints.db) as saver: app graph.compile(checkpointersaver) config {configurable: {thread_id: ticket-1001}} result app.invoke(initial_state, config)这里有两个细节要注意。thread_id是关键它标识一条独立的会话流。同一个图应用可以处理多个用户的多个工单每个工单就是一条thread_id状态互不干扰。另外持久化不止是“保存结果”它还支持你在任意节点暂停。比如当你希望节点“generate_answer”执行完后停住等人工确认再继续时可以用interrupt机制。这样系统可以在任何节点间暂停把控制权交给外部等外部返回结果后再恢复。这个能力在实际项目中非常实用尤其是需要人工审核的场景我用它实现过“AI生成候选人评价后HR确认再发送邮件”的完整流程。4. 常见问题与排查技巧实录4.1 状态更新异常与数据丢失排查思路我说一个自己实际踩过的坑。第一次使用LangGraph时我发现某个节点更新State后下一个节点读到的还是旧值。查了很久最后发现问题出在返回的数据结构上。如果节点返回的是一个嵌套很深的字典并且没有使用深拷贝部分字段在合并时可能因为引用关系出现意外覆盖。尤其当State里包含复杂对象时建议在定义State时显式指定哪些字段需要做追加合并哪些字段是做覆盖更新的。这个设计一开始想清楚能少踩很多坑。另外当你发现图运行结果不对不要直接去猜逻辑。先用LangGraph内置的调试模式把每一步的State输出打印出来对比每个节点执行前后的State差异。我一般是把State打到一个JSON文件里然后逐步检查哪一步的数据不符合预期。这比在代码里到处打印要高效得多。4.2 条件分支不生效的常见原因和定位方法条件边不生效是我见过最多的问题。几个高频原因第一映射函数的返回值没匹配到任何分支名又没设默认分支导致图直接报错解决办法是加一个fallback分支第二映射函数读的字段和节点实际写入的字段名不一致比如节点返回的是db_result但映射函数读的是dbResult拼写不一致自然匹配不上第三条件的优先级没理清多个分支条件重叠导致总是走同一条路。定位这类问题方法还是看State。写一个临时节点把映射函数输入的关键字段打印出来基本一眼就能看出来是哪里的问题。千万别凭感觉去改分支逻辑要相信数据。判断条件的字段里如果是空值再检查上游节点是否真的给这个字段赋了值。4.3 并发执行时的共享状态污染并行分支是LangGraph的强项但用不好也会出问题。最大的隐患是多个分支同时写同一个State字段产生竞态。我的做法是给不同分支设计独立的State字段分支结束前不互相交叉。比如“资料收集分支A”就只写data_source_a“资料收集分支B”就只写data_source_b汇总节点再读取这两个字段做整合。这样做不会产生写冲突而且调试时也一眼就能看清每个分支做了什么。等图运行稳定之后再考虑哪些字段可以合并优化初期保守一些永远没错。5. 长期维护这类框架应用的几个建议5.1 一定要把图结构文档化任何一个用了LangGraph的项目必须有当前版本的图结构文档。我在项目里固定维护一张图的说明文档包含节点清单、边清单、State字段定义、每个节点的输入输出示例。版本迭代时这张图和代码同步更新。不要觉得代码本身可读就够了图一复杂光靠读代码去推理关系效率太低了。尤其团队有新人加入时给新人一张图他能很快上手。5.2 为每个节点写清晰的命名规范节点命名这件事看起来不重要实际影响巨大。我见过不少把节点命名为node1、node2的项目调试时看到图里的条件分支完全不知道这个节点是干什么的。我自己固定了一套命名规范动词开头的短句比如parse_user_input、query_knowledge_base、generate_final_response。同时给节点设置简短的描述信息这样在调试时路径日志一目了然。命名和描述花不了多少时间但在排查和维护时节省的时间是成倍的。5.3 预留容错与重试机制的接口图里的任何一个节点都可能失败外部API会超时模型会返回异常格式数据库会暂时不可用。我在设计节点时会要求每个节点做到两件事第一在节点内部捕获异常并把异常信息写入State的error字段第二为可重试的节点设计重试机制。LangGraph本身支持有限次数的重跑但节点的内部容错逻辑还得自己写。比如调用大模型时可以加超时和重试调用外部API时可以做指数退避。把这些机制都封装成统一函数在节点里统一调用代码会非常干净。生产环境里失败不是例外而是常态。图的优雅之处在于它可以把系统的一部分失败隔离开来不会因为一个节点的异常导致整个应用崩溃。但前提是你自己设计了容错机制框架只负责编排不负责魔法。写在最后的个人体会我在实际项目中用了LangGraph一段时间之后最大的感受是它真正改变了我设计AI应用的方式。以前我思考的是“这个功能怎么实现”现在我会先想“这个流程怎么画成一张图”。画好图实现细节反而变得简单了。如果你正在做一个多步骤、有状态、需要人工反馈的AI应用LangGraph是眼下最值得投入时间研究的框架之一。它不算完美但它的图式思维确实代表了AI应用开发从“脚本时代”走向“编排时代”的关键一步。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑