我们从零写过一个数据分析 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, total=False): 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:str

question和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_path=state["csv_path"], code=task["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:return"stop"if last.get("error_type")in{"KeyError","SyntaxError"}:return"retry"if last.get("quality_issue")=="metric_mismatch":return"replan"if last.get("needs_clarification"):return"clarify"if last.get("passed"):return"answer"return"stop"

这个路由函数可以很简单，甚至完全不调用模型。它做的是控制决策，不是业务分析。

在很多 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(checkpointer=checkpointer)

执行时给每个任务一个稳定的thread_id：

config ={"configurable":{"thread_id":"analysis-20260705-001",}}graph.invoke({"question":"分析过去 12 个月增长来源","csv_path":"sales_2025.csv",}, config=config,)

如果运行到interrupt，图会在检查点处暂停。用户选择net_sales后，再用同一个thread_id恢复：

graph.invoke( Command(resume="net_sales"), config=config,)

教学示例里可以用内存 checkpointer，生产环境则应该换成持久化存储。重点不是具体存在哪里，而是暂停时要保存足够完整的图状态：已经检查过哪些字段，当前计划是什么，哪些产物已经生成，为什么需要用户确认，恢复后应该回到哪个节点。

这里还有一个容易忽略的细节：interrupt恢复时，会从调用interrupt的节点开头重新执行。因此，不要在interrupt前放非幂等副作用。对数据分析 Agent 来说，更稳的做法是先问清口径，再执行高成本代码、写文件或生成最终报告；如果确实要在暂停前写状态，也要保证重复执行不会制造重复产物。

这和把聊天历史重新发给模型完全不同。

重新发送聊天历史，是希望模型“记得刚才发生了什么”。 checkpoint 恢复，是系统自己知道“刚才停在图上的哪个点，状态里有什么，下一条边怎么走”。

对可维护性来说，这个差别很大。前者依赖模型重建现场，后者依赖运行时恢复现场。

六、调试颗粒度

LangGraph 会让控制流更清晰，但不会自动让系统设计变好。状态图也可能被写乱。

最常见的问题，是节点颗粒度不对。

如果把整个数据分析流程都放进一个agent_node：

agent_node: 检查字段、规划任务、写代码、执行工具、修复错误、生成报告

那只是把while循环搬进了一个节点，图上看起来干净，节点内部还是黑箱。

但如果把节点拆得过细：

parse_date_columndrop_missing_rowssort_by_monthgroup_by_categoryrename_axissave_tmp_table

图会变成一张维护成本很高的微步骤地图。模型和人都很难看懂真正的控制意图。

更稳的粒度，是按“可检查产物”拆节点：

这个粒度的好处是，每个节点都有明确输入、输出和失败含义。调试时我们可以问很具体的问题：

是 inspect_data 没暴露真实字段？是 plan_analysis 选错了指标？是 run_code 生成了错误汇总？是 validate_result 没拦住口径问题？是 compose_answer 夸大了证据？

这比一句“Agent 分析错了”有用得多。

状态也要克制。不要把所有中间内容都长期保留在状态里。原始数据、完整代码日志、大表格、完整 traceback，都应该被压缩、截断或转成外部产物引用。状态里保留控制流需要的信息：摘要、路径、错误类型、证据索引、失败计数、下一步约束。

还有一个容易踩的坑：不要把条件边全部交给 LLM。

某些路由确实需要模型判断，比如“这份报告是否回答了用户真正的问题”。但很多路由不需要模型：

文件不存在 -> 停止字段不存在 -> 重试或澄清连续失败超过阈值 -> 停止汇总总额差异超过容差 -> 重规划图表文件未生成 -> 重试当前节点

这些条件越确定，越应该用规则。状态图的价值之一，就是让我们把确定性的控制逻辑从 Prompt 里拿出来，放回程序。

七、迁移路径

不是所有 Agent 都应该一开始就上 LangGraph。

如果任务只有两三步，比如“读取 CSV，统计销售额最高的品类，画一张柱状图”，手写循环更轻。为了一个简单任务引入图、检查点和持久化，可能只是把复杂度提前了。

LangGraph 更适合这些信号已经出现的时候：

一个现实的迁移顺序可以很小：

第一步：保留原来的工具函数，不重写业务能力第二步：把 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%免费】