基于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 典型问题速查表问题现象可能原因排查方向解决手段图执行到某节点卡住条件边返回值不在映射表里检查路由函数返回值加兜底默认分支中间结果丢失状态字段没用 reducer检查字段定义用 Annotated 加 operator.add无限重试重试计数没生效检查计数更新逻辑每步独立计数上限结论数据不准失败结果混入汇总检查结果标识带步骤索引和尝试次数结构化输出解析失败模型返回格式不稳检查 Prompt 和模型加校验重试降级token 消耗过快中间结果太大检查状态体积执行节点做聚合截断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 也是东一榔头西一棒子。所以如果你准备动手做先把状态字段和流转逻辑在纸上画清楚再写代码能省掉大量返工。

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

相关推荐

ML Kit设备端AI实战：OCR、人脸检测与条码扫描的选型与避坑

分数阶统一混沌系统MATLAB仿真：从离散化到相图与分岔图

PostgreSQL 查询全部函数信息：从 pg_proc 到综合模板

更多资讯

ETestDEV5通信协议管理字段属性配置详解与避坑指南

三菱Q系列PLC与威纶通触摸屏项目实战：从架构到调试全解析

ARM汇编语言程序设计：寄存器、指令与经典例程实战指南

大模型技术全解：从原理到应用与踩坑指南

小学二年级数学趣味题PPT课件拆解：五类题型教学与python-pptx批量改题

agent-skills：智能体能力操作系统的架构设计与工程实践

今日更新

微信云开发旅游小程序实战：LBS打卡+UGC社交全栈方案

金融AI代理平台核心拆解：从多代理协作到财报分析实战

基于大数据的智慧旅游推荐与可视化平台架构实践

本周精选

金融AI落地指南：从调研报告到技术选型与工程实践

大模型上下文成本与注意力优化技术深度解析——为何O(n²)平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测

基于GMM的语音识别系统原理与MATLAB实现详解

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？