用 LangGraph 重构 Agent 控制流

我们从零写过一个数据分析 Agent。它的框架其实很简单模型决定下一步控制器调用工具工具返回观察结果模型继续决定下一步直到完成或触发停止条件这个循环能够让一个 CSV 分析任务跑起来。先检查字段再写 Python执行成功后生成图表最后输出结论。整个过程可以说明 Agent 的底层机制LLM 不直接执行动作它只是提出动作请求交给外层的控制器来执行。但是我们都知道真实的产品设计远比这个要复杂简单的while循环很快会变得吃力。它不只是“读表 - 写代码 - 回答”还要判断字段是否足够选择分析路径处理代码报错验证汇总口径重画图表在指标含义不清时暂停询问用户在恢复后继续执行还要保留中间状态方便调试。所有这些逻辑如果继续塞进一个循环和一堆if/else系统表面上还在跑控制流却已经变成一团隐形状态。这个时候引入 LangGraph 就显得十分必要了目的就是把 Agent 的循环、分支、重试、暂停和恢复变成一张可命名、可检查、可恢复的状态图。换句话说LangGraph 不是让 Agent 更像聊天而是让 Agent 的控制流更像一个紧密协作的工程系统。一、循环的天花板一个最小数据分析 Agent写出来的循环可能是这样for step inrange(max_steps): action ask_model(messages)if action[type]final:return action[answer] observation call_tool(action) messages.append(observation)这段代码的优点很明显直观、少依赖、容易理解。它适合教学也适合跑通最小闭环。问题出现在第二阶段。假设用户上传销售数据要求“分析过去 12 个月增长来源并生成图表”。Agent 至少需要经历这些节点inspect_dataplan_analysisprepare_datasetrun_analysis_codevalidate_resultrepair_or_replancompose_report中间还会出现其他的逻辑分支字段足够 - 继续分析字段不清 - 询问用户代码报错 - 修代码重试统计口径不一致 - 回到数据准备图表不可读 - 只重画图证据不足 - 降级结论或停止手写循环当然也能处理这些分支但是问题也很明显第一状态散落。字段概况、当前计划、已生成图表、最近错误、连续失败次数、待确认问题可能分别藏在messages、局部变量、工具返回值和日志里。系统能跑但没人能一眼看清“当前任务到底处于什么状态”。第二跳转隐形。某个if error_type KeyError决定重试当前节点某个if diff_ratio 0.01决定回到清洗节点。这些跳转写在函数深处不形成可视化、可测试的控制面。第三恢复困难。任务执行到第 5 步时需要用户确认“销售额用含税还是不含税口径”如果没有持久化状态系统只能把整段上下文重新塞回模型或者从头再跑。第四调试靠猜。线上日志里只看到模型调用、工具调用和一段最终回答很难回答它为什么跳过异常检测为什么重试了三次为什么没有暂停询问用户为什么用了旧的图表产物二、状态图心智LangGraph 里的StateGraph可以先按三个词理解State、Node、Edge。State是整张图共享的任务状态。不能简单理解是聊天历史的别名应该看作是控制器用来判断下一步的结构化事实用户目标、数据概况、当前计划、工具观察、产物路径、失败次数、最终答案。Node是一个可执行步骤。它读取当前状态做一小段明确的工作然后返回状态更新。这个步骤可以是确定性函数也可以在内部调用 LLM 或工具。Edge决定下一个节点。普通边表达固定顺序条件边表达分支、循环、重试和结束条件。把数据分析 Agent 改成状态图以后它大概长这样START - inspect_data - plan_analysis - run_code - validate_result - run_code # 可修复错误重试 - plan_analysis # 计划或口径有问题重规划 - ask_user # 需要人类确认 - compose_answer # 结果通过 - END # 达到停止条件这张图是一个可执行的控制结构每个节点叫什么、读什么状态、写什么状态、下一跳由什么条件决定都进入了程序。这会改变我们写 Agent 的方式。在裸循环里我们常常问“模型下一轮应该说什么”在状态图里我们会先问“当前状态是什么哪个节点有资格处理这个状态处理完以后下一跳应该由哪个条件决定”这种心智变化很关键。因为 Agent 的不确定性主要来自模型但系统可靠性来自控制结构。LangGraph 不会替我们保证模型永远选对分析路径它只是让每一次路径选择都有位置、有状态、有边界。三、状态建模重写数据分析 Agent第一步不是加节点而是定义状态。一个最小版本可以这样写from typing import Annotated, TypedDictimport operatorclassAnalysisState(TypedDict, totalFalse): question:str csv_path:str profile:dict plan:list[dict] current_step:str observations: Annotated[list[dict], operator.add] artifacts: Annotated[list[str], operator.add] failures:int final_answer:strquestion和csv_path是任务入口。profile保存字段、样本、缺失值和基础统计。plan保存分析步骤不只是给模型看的自然语言计划。observations记录工具结果和检查点结果。artifacts保存图表、表格、报告草稿等产物路径。failures是控制循环的刹车。final_answer只在交付节点写入。Annotated[list[dict], operator.add]这一类 reducer 表示节点返回新的观察结果时不是覆盖旧列表而是追加进去。状态图不是把所有东西都塞进一条 messages而是让不同类型的信息各归其位。节点函数也应该保持小而清楚。definspect_data(state: AnalysisState)-dict: profile inspect_csv(state[csv_path])return{profile: profile,observations:[{node:inspect_data,ok:True,columns: profile[columns],shape: profile[shape],}],}这个节点只做一件事检查数据并把压缩后的数据概况写回状态。它不顺手规划任务也不生成报告。再看执行代码节点defrun_code(state: AnalysisState)-dict: task state[plan][0] result run_python( csv_pathstate[csv_path], codetask[code],) update {observations:[{node:run_code,task_id: task[id],ok: result.ok,error_type: result.error_type,summary: result.summary,}],artifacts: result.artifacts,}ifnot result.ok: update[failures] state.get(failures,0)1return update这里也有一个边界节点可以调用工具但工具结果要被压缩成下一步决策需要的观察。不要把完整 traceback、完整 CSV、完整中间表一股脑塞回状态。状态不是垃圾箱它是控制流的工作台。四、条件边有了状态和节点接下来是图结构。下面是一个简化的 LangGraph 骨架from langgraph.graph import END, START, StateGraphbuilder StateGraph(AnalysisState)builder.add_node(inspect_data, inspect_data)builder.add_node(plan_analysis, plan_analysis)builder.add_node(run_code, run_code)builder.add_node(validate_result, validate_result)builder.add_node(clarify_metric, clarify_metric)builder.add_node(compose_answer, compose_answer)builder.add_edge(START,inspect_data)builder.add_edge(inspect_data,plan_analysis)builder.add_edge(plan_analysis,run_code)builder.add_edge(run_code,validate_result)builder.add_conditional_edges(validate_result, route_after_validation,{retry:run_code,replan:plan_analysis,clarify:clarify_metric,answer:compose_answer,stop: END,},)builder.add_edge(clarify_metric,plan_analysis)builder.add_edge(compose_answer, END)graph builder.compile()值得注意的是add_conditional_edges。普通流程编排最擅长表达“下一步固定是谁”。Agent 控制流最麻烦的是“下一步取决于刚才发生了什么”。代码执行失败时不一定回到同一个节点统计口径不一致时可能要回到数据准备图表标签挤在一起时只需要重画图连续失败超过阈值时应该停止或请求人类介入。这些判断都应该从状态里读而不是让模型在长上下文里临时猜。def route_after_validation(state: AnalysisState)-str: last state[observations][-1] failures state.get(failures,0)if failures 2:returnstopif last.get(error_type)in{KeyError,SyntaxError}:returnretryif last.get(quality_issue)metric_mismatch:returnreplanif last.get(needs_clarification):returnclarifyif last.get(passed):returnanswerreturnstop这个路由函数可以很简单甚至完全不调用模型。它做的是控制决策不是业务分析。在很多 Agent 项目里最值得从模型手里拿回来的恰恰是这种控制决策。模型可以负责生成候选代码、解释异常、写报告草稿但“同一错误是否还能重试”“失败是否影响下游产物”“是否达到停止条件”最好由状态、规则、预算和检查点共同决定。LangGraph 的条件边让这种边界更自然。我们不需要把所有逻辑写进一个巨大的 System Prompt也不需要让模型每一轮自由决定“接下来我要做什么”。图结构会把可控的部分先固定下来把真正需要模型判断的部分留在节点内部。五、暂停与恢复数据分析 Agent 并不总能一路自动跑完。用户说“分析增长来源”但数据里同时有gross_sales和net_sales。 用户说“找出异常月份”但某个月数据疑似重复导入需要确认是否剔除。 用户要求生成预测但当前样本只有 6 个月继续预测很可能误导。这些时候Agent 不应该硬着头皮继续自动化。它应该暂停把需要确认的问题交给人类然后在得到回答后从原状态继续。这就是 checkpoint 和 interrupt 的价值。from langgraph.checkpoint.memory import InMemorySaverfrom langgraph.types import Command, interruptdefclarify_metric(state: AnalysisState)-dict: metric interrupt({question:数据中同时存在 gross_sales 和 net_sales本次增长分析使用哪个口径,options:[net_sales,gross_sales],})return{metric: metric}checkpointer InMemorySaver()graph builder.compile(checkpointercheckpointer)执行时给每个任务一个稳定的thread_idconfig {configurable:{thread_id:analysis-20260705-001,}}graph.invoke({question:分析过去 12 个月增长来源,csv_path:sales_2025.csv,}, configconfig,)如果运行到interrupt图会在检查点处暂停。用户选择net_sales后再用同一个thread_id恢复graph.invoke( Command(resumenet_sales), configconfig,)教学示例里可以用内存 checkpointer生产环境则应该换成持久化存储。重点不是具体存在哪里而是暂停时要保存足够完整的图状态已经检查过哪些字段当前计划是什么哪些产物已经生成为什么需要用户确认恢复后应该回到哪个节点。这里还有一个容易忽略的细节interrupt恢复时会从调用interrupt的节点开头重新执行。因此不要在interrupt前放非幂等副作用。对数据分析 Agent 来说更稳的做法是先问清口径再执行高成本代码、写文件或生成最终报告如果确实要在暂停前写状态也要保证重复执行不会制造重复产物。这和把聊天历史重新发给模型完全不同。重新发送聊天历史是希望模型“记得刚才发生了什么”。 checkpoint 恢复是系统自己知道“刚才停在图上的哪个点状态里有什么下一条边怎么走”。对可维护性来说这个差别很大。前者依赖模型重建现场后者依赖运行时恢复现场。六、调试颗粒度LangGraph 会让控制流更清晰但不会自动让系统设计变好。状态图也可能被写乱。最常见的问题是节点颗粒度不对。如果把整个数据分析流程都放进一个agent_nodeagent_node: 检查字段、规划任务、写代码、执行工具、修复错误、生成报告那只是把while循环搬进了一个节点图上看起来干净节点内部还是黑箱。但如果把节点拆得过细parse_date_columndrop_missing_rowssort_by_monthgroup_by_categoryrename_axissave_tmp_table图会变成一张维护成本很高的微步骤地图。模型和人都很难看懂真正的控制意图。更稳的粒度是按“可检查产物”拆节点节点产物检查点inspect_data字段、样本、缺失值摘要字段是否足够回答问题plan_analysis结构化分析计划是否覆盖用户目标是否超预算run_code汇总表、图表、日志是否成功执行是否生成产物validate_result质量检查结果口径、总额、排序、证据是否通过compose_answer最终结论是否引用证据是否说明局限这个粒度的好处是每个节点都有明确输入、输出和失败含义。调试时我们可以问很具体的问题是 inspect_data 没暴露真实字段是 plan_analysis 选错了指标是 run_code 生成了错误汇总是 validate_result 没拦住口径问题是 compose_answer 夸大了证据这比一句“Agent 分析错了”有用得多。状态也要克制。不要把所有中间内容都长期保留在状态里。原始数据、完整代码日志、大表格、完整 traceback都应该被压缩、截断或转成外部产物引用。状态里保留控制流需要的信息摘要、路径、错误类型、证据索引、失败计数、下一步约束。还有一个容易踩的坑不要把条件边全部交给 LLM。某些路由确实需要模型判断比如“这份报告是否回答了用户真正的问题”。但很多路由不需要模型文件不存在 - 停止字段不存在 - 重试或澄清连续失败超过阈值 - 停止汇总总额差异超过容差 - 重规划图表文件未生成 - 重试当前节点这些条件越确定越应该用规则。状态图的价值之一就是让我们把确定性的控制逻辑从 Prompt 里拿出来放回程序。七、迁移路径不是所有 Agent 都应该一开始就上 LangGraph。如果任务只有两三步比如“读取 CSV统计销售额最高的品类画一张柱状图”手写循环更轻。为了一个简单任务引入图、检查点和持久化可能只是把复杂度提前了。LangGraph 更适合这些信号已经出现的时候信号为什么状态图更合适分支越来越多条件边比散落的if/else更清楚需要暂停恢复checkpoint 能保存运行时状态中间产物要审计节点和状态能形成可追踪链路失败恢复路线不同错误类型可以路由到不同节点多个检查点介入检查结果能直接改变下一跳调试成本上升图结构能定位跑偏节点一个现实的迁移顺序可以很小第一步保留原来的工具函数不重写业务能力第二步把 messages 里的隐含信息拆成 AnalysisState第三步把原循环里的关键阶段拆成 4-6 个节点第四步只给最危险的分支加条件边第五步给需要人类确认的位置加 interrupt第六步接入持久化 checkpointer 和 trace不要为了使用框架而重写一切。真正应该迁移的是那些已经在裸循环里变得难以观察、难以恢复、难以测试的控制流。对数据分析 Agent 来说LangGraph 最值得先接管的不是 Pandas 代码怎么写也不是报告怎么措辞而是这几件事当前分析状态在哪里哪个节点负责生成哪个产物失败后回到哪里什么时候必须停下来人类确认后如何继续最终结论引用了哪些已验证证据做到这一步Agent 不会因为用了 LangGraph 就自动更聪明。它仍然可能写错代码、选错指标、生成不合适的图。但当它出错时错误不再混在一整段对话里。我们能看到它错在哪个节点状态里缺了什么哪条边把它带到了错误路径哪个检查点本该拦住它。这才是框架真正应该提供的价值不是包装 Agent 的神秘感而是降低 Agent 控制流的维护成本。学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】