基于LangGraph构建会规划的数据分析Agent实战

1. 为什么“会规划”才是数据分析 Agent 的分水岭

做过数据分析的人都有一个共识：写 SQL 不难，难的是拿到一个模糊需求之后，知道先查什么、再查什么、中间发现异常怎么调整方向。比如业务方丢过来一句“最近转化率是不是有问题”，背后可能涉及数据口径确认、时间范围对齐、维度拆解、异常归因、交叉验证等一连串动作。传统的“单轮问答式”数据分析工具，你问一句它答一句，每一步都要人来指挥，本质上是个高级计算器。而真正有价值的 Agent，应该像一个有经验的分析师：接到目标后自己拆任务、排优先级、执行、观察中间结果、根据反馈调整下一步，最后给出有依据的结论。

这就是“会规划”的含义，也是 LangGraph 在这个场景里真正能发挥价值的地方。LangGraph 是构建有状态、多步骤 Agent 的编排框架，它把整个分析流程建模成一张图：节点是具体的动作（理解需求、生成查询、执行、校验、归因），边是流转条件（成功继续、失败重试、异常转向）。相比一条直线跑到底的链式调用，图结构天然支持循环、分支和状态回传，这正是“规划—执行—反思—再规划”所需要的骨架。

这篇文章面向的是已经了解 Python、接触过大模型 API 调用、想进一步做多步骤 Agent 的开发者。我会从整体设计思路讲起，把状态设计、节点拆分、条件路由、循环控制这些关键点拆开揉碎，再给出一套可以直接参考复现的实现方案，最后把我踩过的坑和排查经验整理出来。你不需要是 LangGraph 专家，但需要有一点“把业务流程翻译成状态机”的思维准备。

2. 整体架构设计与方案选型思路

2.1 为什么选图结构而不是链式调用

先说清楚一个选型问题：为什么不用简单的 Chain（链）而要用 Graph（图）。链式调用的假设是“步骤固定、顺序确定、每步都成功”。但数据分析的真实流程根本不满足这个假设。举个最常见的场景：Agent 生成了一段查询，执行后发现返回空结果。这时候它不该直接报错结束，而应该回到“理解需求”节点，重新判断是不是过滤条件写错了，或者时间范围选得不对。这种“回到上游重来”的能力，链式结构做起来非常别扭，而图结构里加一条回边就解决了。

LangGraph 的核心抽象正好对应这个需求。它用State（状态）贯穿全程，每个节点读取状态、修改状态、返回状态增量；用Edge（边）决定下一步去哪，普通边是固定跳转，条件边是根据状态动态选择。规划能力本质上就是“根据当前状态决定下一个节点”，这在 LangGraph 里就是条件边的自然表达。

我对比过几种方案：纯 Prompt 让模型一次性输出完整分析计划，灵活但不可控，中间出错无法干预；纯代码写死流程，可控但僵硬，遇到新需求就得改代码。LangGraph 处在中间：流程骨架由代码定义，保证可控；具体决策由模型在节点内完成，保证灵活。这个平衡点对数据分析场景特别合适。

2.2 状态设计：整个 Agent 的“共享内存”

状态设计是整张图的地基，设计不好后面全是坑。我的经验是，状态字段要围绕“规划所需的最小信息集”来定，既不能太少导致节点之间信息断层，也不能太多导致状态臃肿、每次传递都浪费 token。

我最终用的状态结构大致包含这几类字段：用户目标（原始需求文本）、分析计划（当前拆解出的步骤列表）、当前步骤索引（执行到哪了）、中间结果（每步的执行产出，比如查询结果、统计值）、观察记录（对中间结果的判断，比如“结果为空”“数值异常”）、重试计数（防止死循环）、最终结论（汇总输出）。这里有个关键设计：中间结果和观察记录要分开存。结果是客观数据，观察是模型对数据的主观判断，混在一起后面做归因时会乱。

提示：状态字段尽量用可序列化的基础类型（字符串、列表、字典），不要塞复杂对象。LangGraph 在节点间传递状态时可能涉及序列化，塞了不可序列化的东西会在运行时炸掉，而且报错信息往往不直观。

另一个容易忽略的点是状态更新的合并策略。LangGraph 默认对状态做覆盖式更新，但像“中间结果列表”这种字段，你希望的是追加而不是覆盖。这时候要用带 reducer 的字段定义，比如用operator.add让列表自动拼接。我一开始没注意这个，导致每执行一步前面的结果就被冲掉了，排查了半天才发现是合并策略的问题。

2.3 节点拆分粒度：粗一点还是细一点

节点拆多细，直接决定图的复杂度和可维护性。拆太细，图里几十个节点，调试时跳来跳去看不清主线；拆太粗，一个节点里塞太多逻辑，又退化成“大函数”，失去图结构的价值。

我的实践标准是：一个节点只做一件有明确输入输出的事，且这件事的成败可以被独立判断。按这个标准，我把核心流程拆成六个节点：需求理解、计划生成、步骤执行、结果观察、计划调整、结论汇总。其中“步骤执行”和“结果观察”是分开的，因为执行是“跑查询”，观察是“判断结果好不好”，两者的失败模式完全不同——执行可能因为语法错误失败，观察可能因为结果不符合预期而触发调整。分开之后，条件路由就能精准地针对不同失败类型走不同分支。

这里有个反直觉的经验：不要把“规划”和“执行”混在一个节点里。我早期图省事，让一个节点既生成计划又执行第一步，结果发现模型在生成计划时容易被具体执行细节带偏，计划质量明显下降。拆开之后，规划节点专注“想清楚要做哪几步”，执行节点专注“把当前这步做好”，各司其职，效果立竿见影。

3. 核心节点实现与关键细节拆解

3.1 需求理解节点：把模糊需求变成可执行目标

需求理解节点的任务是把用户那句模糊的话，转成结构化的分析目标。比如“最近转化率是不是有问题”，要转成“分析指定时间段内转化率的变化趋势，识别是否存在显著下降，并定位可能的原因维度”。这一步看着简单，但它决定了后面所有步骤的方向，做不好后面全白搭。

实现上，我用一个带结构化输出的 Prompt，让模型返回 JSON，包含分析对象（转化率）、时间范围（如果用户没说就默认最近 30 天，并在结论里注明假设）、关注维度（按渠道、按地区等）、预期产出（趋势判断 + 归因）。这里的关键技巧是强制模型显式列出它的假设。用户没说时间范围，模型不能默默自己定一个，而要写进“假设”字段，这样最终结论里能提醒用户“本次分析基于最近 30 天，如需调整请说明”。这个细节能极大提升结果的可信度。

注意：结构化输出一定要做校验和兜底。模型偶尔会返回缺字段或格式不对的 JSON，我一般用 Pydantic 做解析，解析失败就触发一次重试，重试还失败就降级成“让模型用自然语言描述目标”，至少保证流程不中断。

3.2 计划生成节点：让模型学会“分步走”

计划生成是整个 Agent “会规划”的核心体现。我让模型基于结构化目标，输出一个有序的步骤列表，每步包含动作类型（查询、计算、对比、归因）、具体描述、依赖的前置步骤。为什么要带依赖关系？因为有些步骤必须等前面的结果出来才能定，比如“如果整体转化率下降，再按渠道拆解”，这个“再”就是依赖。

这里有个我反复调试出来的经验：计划不要一次生成太细。一开始我让模型把每步的 SQL 都写出来，结果发现它在前面的数据还没查出来时，就凭想象写后面的查询，经常驴唇不对马嘴。后来改成“先生成粗粒度计划（3 到 5 步），每步执行时再细化”，效果好很多。这其实符合人类分析师的工作方式：先定大方向，边走边看。

计划生成节点的输出要存进状态的“分析计划”字段，同时把“当前步骤索引”置为 0。这里要处理一个边界情况：如果模型生成的计划为空或者只有一步“直接给结论”，要能识别出来并打回重生成。我的做法是加一个简单的校验：计划步骤少于 2 步就触发重试，因为稍微像样的数据分析都不可能一步到位。

3.3 步骤执行节点：把计划翻译成动作

执行节点负责把当前步骤的描述，翻译成具体动作并执行。数据分析场景里，动作主要是三类：查数据库、做计算、调外部工具。我用一个分发逻辑，根据步骤的“动作类型”字段路由到不同的执行器。

查数据库这块，我让模型生成查询语句，但绝不直接执行。中间加了一层校验：先做语法检查，再做“危险操作”检查（比如有没有删表、改数据的语句），通过之后才真正执行。这个安全层是必须的，模型偶尔会生成一些意料之外的语句，直接跑在生产库上后果不堪设想。校验不通过就返回错误信息给观察节点，让它决定是重试还是调整计划。

计算类步骤相对简单，用 Python 的 pandas 处理查询结果就行。但要注意数据量控制：查询结果可能很大，全塞进状态会撑爆上下文。我的做法是在执行节点里做聚合和截断，只把关键统计值和少量样本行存进状态，原始大数据留在本地临时文件里，需要时再读。

# 执行节点的核心逻辑示意（简化版） def execute_step(state): step = state["plan"][state["current_step"]] if step["action_type"] == "query": sql = generate_sql(step, state["context"]) if not validate_sql(sql): return {"observation": "查询校验未通过", "last_error": "invalid_sql"} result = run_query(sql) summary = summarize(result) # 聚合 + 截断 return {"intermediate_results": [summary], "observation": "执行成功"} elif step["action_type"] == "compute": # pandas 计算逻辑 ...

3.4 结果观察节点：Agent 的“自我反思”能力

观察节点是我认为整个设计里最有价值的一环，也是“会规划”和“死板执行”的分界线。它的任务是拿到执行结果后，判断这个结果是否合理、是否推进了目标、下一步该继续还是调整。

判断逻辑我分成三层。第一层是技术性检查：结果是不是空的、有没有报错、数值是不是明显异常（比如转化率算出 500%）。第二层是业务性检查：这个结果和预期方向一致吗？比如计划里预期转化率下降，结果却上升了，那就要标记出来。第三层是进度检查：当前步骤的产出，够不够支撑最终结论？如果不够，是继续下一步还是补充新步骤。

这三层检查我都用 Prompt 让模型来做，但给它明确的判断框架，而不是让它自由发挥。实践下来，给模型一个结构化的“观察模板”（包含“结果摘要”“是否异常”“异常类型”“建议动作”几个字段）比让它自由描述要稳定得多。建议动作只有几个枚举值：继续、重试当前步、调整计划、直接汇总。这样条件路由就能直接根据这个字段做判断，不用再解析自然语言。

提示：观察节点最容易出的问题是“过度乐观”——结果明明是空的，模型却说“执行成功”。我在 Prompt 里专门加了反例说明，明确告诉它“空结果、报错、数值超出合理范围都算异常”，并且要求它先陈述事实再下判断，这个顺序能显著降低误判率。

4. 条件路由与循环控制：让 Agent 真正“活”起来

4.1 条件边的设计：根据观察结果决定去向

条件边是 LangGraph 实现动态规划的关键。在观察节点之后，我接了一个路由函数，读取状态里的“建议动作”字段，决定下一步走哪条边。映射关系是这样的：建议“继续”就走回执行节点，索引加一；建议“重试当前步”就走回执行节点，索引不变，但重试计数加一；建议“调整计划”就走回计划生成节点，带上当前的观察记录作为调整依据；建议“直接汇总”就跳到结论节点。

这个路由函数看着简单，但它是整个 Agent 智能感的来源。因为去向不是写死的，而是模型根据实际执行情况动态决定的。我实测下来，一个设计良好的路由能让 Agent 在遇到数据异常时自动多查几个维度，而不是傻乎乎地按原计划走完然后给个错误结论。

路由函数里一定要处理兜底情况：如果“建议动作”字段缺失或者是个没见过的值，默认走“继续”，并且记录一条警告。我遇到过模型返回了个拼写错误的动作值，如果没有兜底，整个图就卡住了。

4.2 循环控制：防止 Agent 陷入死循环

有循环就有死循环风险，这是图结构 Agent 必须严肃对待的问题。我见过太多 Demo 里 Agent 在“执行—观察—重试”之间无限打转，最后把 token 烧光。控制手段我用了三层。

第一层是单步重试上限：同一个步骤连续重试超过 3 次，强制跳过，标记为“未能完成”，继续下一步。第二层是总步数上限：整个分析流程执行的总步骤数（含重试和新增）超过 15 步，强制进入汇总。第三层是计划调整上限：计划被重新生成的次数超过 2 次，说明模型一直在原地打转，强制汇总当前已有结果。

这三层限制都写在状态里，每次路由前检查。超过限制时，不是简单报错，而是带着已有结果进入汇总节点，让模型基于不完整的信息给出“阶段性结论”，并明确说明哪些部分没能完成。这样即使流程被截断，用户也能拿到有价值的东西，而不是一个冷冰冰的错误。

注意：重试计数要区分“同一步骤的重试”和“全局重试”。我一开始只用一个全局计数器，结果发现某一步重试两次后，后面正常步骤也被误判超限。后来改成每个步骤独立计数，用字典存，问题才解决。

4.3 状态在循环中的正确传递

循环场景下状态传递特别容易出 bug，因为同一个节点会被多次执行，每次执行都应该基于最新的状态。这里的关键是明确哪些字段是累积的、哪些是覆盖的。中间结果列表是累积的，每次执行追加；当前步骤索引是覆盖的，每次更新；观察记录也是累积的，保留历史方便后面归因。

我踩过的一个坑是：重试当前步时，上一次失败的结果还留在“中间结果”里，导致后面汇总时把失败结果也当有效数据用了。解决办法是在重试时，先把上一次该步骤产生的结果标记为“已废弃”，或者干脆在状态里给每条结果带上“步骤索引 + 尝试次数”的标识，汇总时只取每个步骤最后一次成功的结果。这个细节不做，最终结论的数据准确性就没法保证。

5. 实操全流程与可复现方案

5.1 环境准备与依赖安装

先把环境搭起来。核心依赖就几个：LangGraph 本体、一个大模型 SDK、数据处理用的 pandas、数据库连接库（按你实际用的数据库选）。我用的是虚拟环境隔离，避免版本冲突。

python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install langgraph langchain-core pandas # 数据库驱动按需安装，比如 pip install psycopg2-binary

模型这块，我用的是支持结构化输出的通用大模型接口，通过环境变量配置密钥，代码里不硬编码。这里提醒一句：不同模型对结构化输出的支持程度差异很大，选一个稳定支持 JSON 模式或者函数调用的模型，能省掉大量解析容错的代码。

5.2 状态定义与图构建

状态用 TypedDict 定义，带 reducer 的字段用 Annotated 标注。图构建分三步：初始化 StateGraph、添加节点、添加边。普通边用 add_edge，条件边用 add_conditional_edges 并传入路由函数。

from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, END class AnalysisState(TypedDict): goal: str structured_goal: dict plan: list current_step: int intermediate_results: Annotated[list, operator.add] observations: Annotated[list, operator.add] retry_counts: dict total_steps: int final_conclusion: str graph = StateGraph(AnalysisState) graph.add_node("understand", understand_node) graph.add_node("plan", plan_node) graph.add_node("execute", execute_node) graph.add_node("observe", observe_node) graph.add_node("summarize", summarize_node) graph.set_entry_point("understand") graph.add_edge("understand", "plan") graph.add_edge("plan", "execute") graph.add_edge("execute", "observe") graph.add_conditional_edges("observe", route_after_observe, { "continue": "execute", "replan": "plan", "summarize": "summarize", }) graph.add_edge("summarize", END) app = graph.compile()

这段代码是整个骨架，节点函数的具体实现按前面几节讲的逻辑填充。编译之后用app.invoke(初始状态)就能跑起来。

5.3 一次完整运行的现场记录

我拿一个模拟场景跑了一遍：目标是“分析某产品最近一个月的转化率变化”。需求理解节点输出结构化目标，假设了“最近 30 天”“整体转化率”。计划节点生成四步：查整体转化率趋势、按渠道拆解、按地区拆解、归因分析。

执行第一步，查询返回了整体趋势数据，观察节点判断“趋势下降明显，建议继续”。第二步按渠道拆解，发现某个渠道下降特别严重，观察节点标记“发现异常渠道，建议继续深挖”。第三步按地区拆解时，查询返回空结果，观察节点判断“结果为空，可能是地区字段口径问题，建议重试当前步”。重试时调整了字段名，拿到结果。第四步归因，模型结合前面结果给出“主要受某渠道影响”的结论。最后汇总节点输出完整报告，包含数据依据和假设说明。

整个过程走了 6 个节点、触发了 1 次重试，总步数 5 步，在限制范围内。这个流程跑通之后，我换了个完全不同的分析目标（“分析用户留存”），Agent 也能自动生成合适的计划并执行，说明这套架构的通用性是够的。

5.4 关键参数的选择依据

几个参数值得单独说说。重试上限设为 3，是因为实测下来，同一个步骤重试超过 3 次基本就是模型能力边界了，再试也是浪费。总步数上限设为 15，是按“一个中等复杂度分析通常 5 到 8 步，留一倍余量应对重试和新增”来定的。计划调整上限设为 2，因为调整两次还调不好，说明模型对目标的理解有根本偏差，不如先出阶段性结论让人来介入。

这些数字不是拍脑袋定的，是我在十几个不同分析任务上跑下来，观察分布之后取的偏保守值。你可以根据自己的任务复杂度调整，但建议先保守后放宽，因为死循环的代价比“少查一个维度”高得多。

6. 常见问题与排查技巧实录

6.1 典型问题速查表

6.2 三个我踩过的深坑

第一个坑是状态字段的 reducer 用错。我一开始给“当前步骤索引”也加了 operator.add，结果每次更新索引都在累加，执行到第三步时索引变成了 6，直接越界。后来才明白，只有需要累积的列表类字段才用 reducer，标量字段必须用覆盖式更新。这个错误很隐蔽，因为图能跑起来，只是行为诡异。

第二个坑是观察节点的 Prompt 太宽松。早期我让模型“判断结果是否合理”，结果它对明显异常的结果也说“合理”。后来我把判断标准写死成几条硬规则（空结果算异常、报错算异常、数值超出预设范围算异常），再让模型在这几条规则基础上做补充判断，准确率大幅提升。给模型明确的判断框架，比给它自由发挥空间更可靠。

第三个坑是汇总节点信息过载。所有中间结果和观察记录都塞给汇总节点，导致它抓不住重点，结论写得又长又散。后来我在汇总前加了一个“信息压缩”步骤，只把每个步骤的最终有效结果和关键观察传进去，结论质量明显变好。这印证了一个道理：Agent 的每个节点都应该做信息过滤，而不是无脑传递。

6.3 提升规划质量的几个实操技巧

除了修 bug，还有一些能显著提升 Agent 规划质量的经验。在计划生成的 Prompt 里给几个高质量示例，比单纯描述要求有效得多，模型会模仿示例的拆解粒度。让观察节点输出“置信度”字段，汇总时对低置信度的结论明确标注，用户看到“此结论置信度较低”会比看到一个斩钉截铁的错结论体验好得多。给计划步骤加上预估耗时或复杂度，路由时可以优先执行高价值步骤，在步数受限时保证核心分析先完成。

还有一个我最近在试的方向：把历史分析记录作为上下文喂给规划节点。同一个用户之前做过类似分析，这次规划时参考上次的步骤，能明显减少重复试错。这个思路本质上是给 Agent 加“经验记忆”，对固定业务场景特别有用。

7. 后续可扩展的方向

这套骨架跑通之后，扩展空间其实很大。我目前在做的一个方向是多 Agent 协作：把“规划”和“执行”拆成两个独立的 Agent，规划 Agent 专注拆解和调整，执行 Agent 专注把单步做到极致，两者通过状态通信。这样每个 Agent 的 Prompt 可以更聚焦，整体效果比单 Agent 全包要好。

另一个方向是引入人工确认节点。在关键步骤（比如要执行一个影响面较大的查询）之前，暂停图执行，等人工确认后再继续。LangGraph 支持中断和恢复，这个能力用在数据分析场景特别合适，既保留了自动化效率，又在关键节点留了人工兜底。

最后分享一个我在实际使用中的体会：Agent 的规划能力，七分靠状态设计，三分靠 Prompt。很多人把精力全花在调 Prompt 上，却忽略了状态结构才是决定 Agent 能不能“记住上下文、做出正确决策”的根本。状态设计清楚了，Prompt 稍微粗糙一点，Agent 也能跑得不错；状态设计混乱，Prompt 写得再花哨，Agent 也是东一榔头西一棒子。所以如果你准备动手做，先把状态字段和流转逻辑在纸上画清楚，再写代码，能省掉大量返工。