1. 为什么“会规划”才是数据分析 Agent 的分水岭1.1 从“能回答”到“会拆解”的认知跃迁过去一年我陆陆续续帮几个团队做过数据分析类的智能助手从最简单的“查一下上个月销售额”到“帮我分析一下为什么华东区复购率掉了”中间踩的坑几乎能写一本书。最开始大家的思路都很朴素把数据库表结构塞进提示词让模型生成 SQL执行完把结果丢回去让它总结。这套流程在单轮、单表、口径明确的场景下确实能跑通但只要问题稍微复杂一点比如“对比一下新老客的客单价变化并找出贡献最大的三个品类”模型就开始胡言乱语——要么漏掉时间窗口要么把两个指标的口径搞混要么干脆编一个不存在的字段。问题的根子不在于模型不够聪明而在于我们把一个多步骤的规划问题当成了单次的生成问题。人类分析师接到这类需求时脑子里会自动跑一个流程先明确口径再拆解成几个子问题然后逐个取数、验证、交叉比对最后才组织结论。这个“先想清楚再动手”的过程就是规划。而 LangGraph 这类框架真正带来的价值是让我们能把这种规划能力显式地建模成一张状态图而不是指望模型在一次前向传播里把所有事情都想明白。我个人的判断是数据分析 Agent 的能力上限取决于它的规划层设计而不是它调用的模型有多强。一个会规划的 Agent用中等规模的模型也能把复杂任务拆得清清楚楚一个不会规划的 Agent就算接上最强的模型遇到多跳问题照样翻车。这篇文章我就把这套东西从头到尾拆一遍包括状态怎么设计、节点怎么切、循环怎么控、错误怎么兜全部是我实际跑过之后沉淀下来的方案。1.2 这篇文章适合谁能带走什么如果你正在做下面这几件事这篇内容应该能帮你省下不少试错时间已经用 LangChain 或类似工具搭过一个能查数的问答机器人但发现它处理不了复杂分析需求想引入 Agent 机制但不确定该用 ReAct 那种自由发挥的模式还是该用更结构化的图编排已经在用 LangGraph但图越画越乱节点之间状态传递经常出问题不知道怎么收敛。我会用一个模拟的电商数据分析场景贯穿全文所有表名、字段名、指标口径都是虚构的但结构完全对标真实业务。读完之后你应该能拿到三样东西一套可直接复用的状态定义和节点划分方案、一份完整的规划循环实现代码、以及一份我踩过的坑和对应排查手法的清单。不需要你有很深的图论基础但至少要写过一点 Python知道什么是状态机。2. 整体架构设计把“规划”拆成可执行的图2.1 为什么不用 ReAct而选状态图先说选型。很多人第一反应是用 ReAct 模式——让模型自己决定下一步调什么工具循环直到它觉得可以回答了。我早期也这么干过结论是在数据分析场景下纯 ReAct 的不可控性太高。原因有三个。第一数据分析对口径一致性要求极高。ReAct 每一轮都是模型临时决定这一轮它可能按“支付时间”过滤下一轮又按“下单时间”过滤最后两个数字对不上你还得回头查是哪一步跑偏了。第二ReAct 的循环没有明确的终止条件模型有时候会陷入“再查一次确认一下”的死循环token 烧得飞快。第三也是最要命的ReAct 的中间过程是黑盒出了问题你很难定位到底是规划错了、取数错了还是总结错了。状态图的好处在于它把整个流程显式地画出来。每个节点干什么、节点之间怎么流转、什么条件下走哪条边全都是你写死的。模型只在需要它发挥的地方比如拆解问题、生成 SQL、组织结论介入其余的控制逻辑由代码保证。这样一来可控性、可调试性、可复现性都上了一个台阶。代价是你得先把流程想清楚不能偷懒。2.2 核心状态对象的设计状态是整个图的血液设计得好后面一路顺畅设计得差后面处处别扭。我最终收敛下来的状态结构大概是这样的用 TypedDict 示意from typing import TypedDict, Annotated, List import operator class AnalysisState(TypedDict): question: str # 用户原始问题 intent: str # 意图分类结果 plan: List[dict] # 规划出的子任务列表 current_step: int # 当前执行到第几步 step_results: Annotated[List[dict], operator.add] # 各步结果累积 schema_context: str # 相关表结构 final_answer: str # 最终结论 error_log: Annotated[List[str], operator.add] # 错误记录 retry_count: int # 当前步骤重试次数这里有几个设计决策值得展开说。plan 用列表而不是单个字符串是因为规划出来的子任务需要结构化每个子任务至少包含“要算什么”“用哪张表”“依赖哪一步的结果”这几个字段后面执行节点才能按图索骥。step_results 用 operator.add 做累积这是 LangGraph 里处理并行或循环写入的标准做法避免后一步覆盖前一步。retry_count 单独拎出来是因为重试逻辑要按步骤粒度控制不能全局共用一个计数器否则第一步失败了重试三次第二步就没机会了。提示状态字段宁少勿多但每个字段的语义必须唯一且明确。我见过有人把“当前 SQL”和“历史 SQL”混在一个字段里结果循环的时候状态越滚越大最后直接把上下文撑爆。2.3 节点划分与流转逻辑整个图我切成了六个核心节点外加两个条件边。节点分别是意图识别、任务规划、步骤执行、结果校验、结论生成、错误处理。流转逻辑用一句话概括识别意图后进入规划规划完进入执行循环每执行一步就校验一次校验通过就推进下一步全部走完生成结论任何一步校验失败就进错误处理错误处理决定是重试还是降级。这里最关键的是执行循环的设计。它不是简单的 for 循环而是一个带条件边的子图执行节点跑完边上的条件函数判断“还有没有下一步”有就回到执行节点没有就流向结论生成。校验失败时条件函数会判断 retry_count 是否超限没超就回到执行节点重试当前步超了就流向错误处理。这种设计的妙处在于规划是一次性的执行是迭代的。规划阶段模型只需要把大问题拆成有序的子任务不用关心每个子任务具体怎么算执行阶段模型只聚焦当前这一步上下文干净出错概率大幅降低。这比让模型在一次调用里既规划又执行要稳得多。3. 核心节点实现与关键细节3.1 意图识别别让模型做它不擅长的事意图识别这个节点我的建议是能规则化就规则化规则覆盖不了的再交给模型。原因很简单意图分类是个封闭集合问题而模型在封闭集合上的表现反而不如几条精准的规则稳定。我的做法是先用关键词和正则做一层粗筛。比如问题里出现“为什么”“原因”“下降”“上涨”这类词大概率是归因分析出现“对比”“相比”“差异”是 comparative 分析出现“趋势”“走势”“变化”是时序分析其余归为通用查询。粗筛命中就直接返回命中不了再走一次模型分类提示词里把候选类别和判定标准写清楚。def classify_intent(state: AnalysisState) - dict: q state[question] rules [ (r为什么|原因|导致|下降|上涨, attribution), (r对比|相比|差异|区别, comparison), (r趋势|走势|变化|增长, trend), ] for pattern, intent in rules: if re.search(pattern, q): return {intent: intent} # 规则未命中走模型兜底 intent llm_classify(q) return {intent: intent}这么做的收益很直接大部分常见问题根本不消耗模型调用响应快、成本低、结果稳定。只有那些表述模糊的问题才需要模型介入。我实测下来规则能覆盖大概七成的真实提问剩下三成里模型再判错的也就百分之几。3.2 任务规划把大问题拆成有依赖的子任务规划节点是整个 Agent 的大脑也是最考验提示词工程的地方。我的核心思路是让模型输出结构化的子任务列表每个子任务必须包含明确的输入、输出和依赖关系。提示词里我会给一个固定的输出格式比如 JSON 数组每个元素包含step_id、description、depends_on、expected_output四个字段。depends_on是关键它让子任务之间形成有向无环图执行节点就能据此判断某一步能不能跑——依赖的结果还没出来就得等着。举个例子用户问“对比一下新老客的客单价变化并找出贡献最大的三个品类”。规划出来大概是这样的step_iddescriptiondepends_onexpected_output1按用户首单时间划分新老客无用户分层表2计算各月新老客客单价1月度客单价序列3计算各品类对新老客客单价差异的贡献1品类贡献度排名4取贡献度前三的品类3Top3 品类列表注意 step 3 依赖的是 step 1 而不是 step 2因为贡献度计算只需要用户分层不需要月度序列。这种依赖关系的精确表达能避免很多无谓的串行等待。规划提示词里我会明确要求模型“只列出必要的依赖不要为了保险把所有前置步骤都写上”。注意规划出来的子任务数量要设上限我一般卡在 6 步以内。超过 6 步通常意味着问题本身太宽泛或者模型在过度拆解。这时候更好的做法是让 Agent 反问用户澄清需求而不是硬着头皮跑一个十几步的流程。3.3 步骤执行一次只干一件事执行节点的设计原则是上下文隔离。每执行一步我只把这一步的描述、相关的表结构、以及它依赖的前置步骤结果喂给模型其余历史一概不带。这样做的好处是模型注意力集中生成的 SQL 或计算逻辑准确率高。执行节点内部其实还分两小步先生成执行方案SQL 或 Python 代码再实际执行。生成方案时我会把 schema_context 里相关的表结构带上并且明确告诉模型“只能用这些字段不要臆造”。执行环节用 try-except 包住任何异常都写进 error_log 并触发校验失败。def execute_step(state: AnalysisState) - dict: step state[plan][state[current_step]] deps [state[step_results][i] for i in step[depends_on]] prompt build_exec_prompt(step, deps, state[schema_context]) try: sql llm_generate_sql(prompt) result run_sql(sql) return {step_results: [{step_id: step[step_id], data: result}]} except Exception as e: return {error_log: [fstep {step[step_id]} failed: {e}], retry_count: state[retry_count] 1}这里有个细节step_results 的写入顺序必须和 plan 的顺序对应否则后面按索引取依赖结果会错位。我一开始用字典存结果并行执行时顺序乱了排查了半天。后来改成列表加显式 step_id 校验才彻底解决。3.4 结果校验给每一步加一道质检校验节点是我认为最容易被忽略、但价值最高的部分。很多人搭 Agent 只关心“能不能跑出结果”不关心“结果对不对”最后交付的东西看着像模像样一核对全是错的。我的校验分三层。第一层是结构校验检查返回结果是不是空的、字段是不是齐全、数值类型对不对。第二层是口径校验比如客单价必须在合理区间内复购率不能超过 100%这些用简单的断言就能挡住大部分低级错误。第三层是逻辑校验这一步需要模型介入把当前步骤的结果和它的预期输出做比对判断是否回答了这一步该回答的问题。def validate_step(state: AnalysisState) - dict: latest state[step_results][-1] step state[plan][state[current_step]] # 结构校验 if not latest[data]: return {error_log: [empty result]} # 口径校验 if 客单价 in step[description] and latest[data][0][value] 0: return {error_log: [negative aov]} # 逻辑校验交给模型 verdict llm_validate(step, latest) if not verdict[pass]: return {error_log: [verdict[reason]]} return {}校验不通过时状态里的 error_log 会累积条件边据此决定重试还是降级。重试时我会把上一次的错误信息带进执行提示词让模型知道上次错在哪避免它重复同样的错误。这个反馈机制对提升重试成功率非常关键实测能把二次成功率从三成提到七成以上。4. 完整实操流程与参数调优4.1 环境准备与依赖安装先把环境搭起来。我用的是 Python 3.11LangGraph 的版本建议锁在较新的稳定版因为它迭代很快不同版本 API 差异不小。pip install langgraph langchain langchain-openai pandas sqlalchemy数据库我用的 SQLite 做演示真实场景换成任何支持 SQLAlchemy 的库都一样。建两张模拟表订单表和用户表。订单表包含订单号、用户 ID、品类、金额、下单时间、支付时间用户表包含用户 ID、首单时间、注册渠道。数据我随机造了几千条够跑通流程就行。提示造模拟数据时一定要故意埋几个“脏数据”比如金额为负、时间为空、品类缺失。这些脏数据是检验你校验层是否有效的试金石干净的测试数据会让你误以为系统很稳。4.2 图的组装与编译节点函数写完之后用 StateGraph 把它们串起来。核心代码如下from langgraph.graph import StateGraph, END builder StateGraph(AnalysisState) builder.add_node(classify, classify_intent) builder.add_node(plan, plan_tasks) builder.add_node(execute, execute_step) builder.add_node(validate, validate_step) builder.add_node(finalize, generate_answer) builder.add_node(handle_error, handle_error) builder.set_entry_point(classify) builder.add_edge(classify, plan) builder.add_edge(plan, execute) builder.add_edge(execute, validate) builder.add_conditional_edges( validate, route_after_validate, {next: execute, retry: execute, done: finalize, fail: handle_error} ) builder.add_conditional_edges( handle_error, route_after_error, {retry: execute, give_up: finalize} ) builder.add_edge(finalize, END) graph builder.compile()route_after_validate这个条件函数是控制流的核心它要判断三件事当前步校验是否通过、是否还有下一步、重试次数是否超限。逻辑写清楚之后整个图的走向就完全确定了。def route_after_validate(state): if state[error_log] and state[retry_count] 3: return fail if state[error_log]: return retry if state[current_step] 1 len(state[plan]): return next return done4.3 参数选择与调优记录几个关键参数我调了不少轮记录一下结论。最大重试次数设 3超过 3 次基本说明这一步的规划本身有问题再重试也是浪费不如降级处理。规划子任务上限设 6前面说过原因。执行节点的温度设 0生成 SQL 这种确定性任务不需要创造力温度越低越稳。校验节点的温度可以设 0.2稍微留一点灵活性避免它过于死板地把合理结果判为失败。还有一个容易被忽略的参数是上下文窗口的分配。执行节点每次只带当前步骤和依赖结果但如果依赖结果本身很大比如一个几百行的明细表直接塞进去会挤占生成 SQL 的空间。我的做法是对依赖结果做摘要压缩只保留 schema 和几行样本需要明细的时候再单独查。这个技巧让我的 token 消耗降了将近一半。5. 常见问题与排查技巧实录5.1 规划跑偏子任务拆得莫名其妙最常见的症状是模型把简单问题拆得过于复杂或者拆出来的子任务之间逻辑不连贯。我遇到过一次用户问“上个月销售额多少”模型拆成了“确定上个月范围→查订单→过滤时间→求和→验证”五步干了一件一步就能干完的事。排查思路是先看规划提示词里有没有给足约束。我后来在提示词里加了一句“如果问题可以用单步解决就不要拆成多步”情况明显好转。另外给几个 few-shot 示例效果立竿见影尤其是那种“简单问题一步搞定”和“复杂问题拆四步”的对比示例模型看了之后拆解的粒度会合理很多。5.2 依赖错乱后一步拿不到前一步的结果这个问题的根源通常是 step_results 的索引和 plan 的索引对不上。我踩过的坑是某一步重试之后结果被追加到了列表末尾而不是替换原来的位置导致后面按索引取依赖时取到了错误的数据。解决办法是用 step_id 做键而不是用列表索引取依赖时按 step_id 查找而不是按位置。虽然前面我说用列表但列表里每个元素都带 step_id查找时遍历匹配即可。这个改动之后重试导致的数据错位问题彻底消失。5.3 校验误判合理结果被判为失败校验层太严会误杀太松会漏放这个度需要调。我遇到过模型把“客单价 0 元”判为合理的情况因为新客首单可能用全额优惠券也遇到过把正常的负增长判为异常的情况。我的经验是结构校验和口径校验用硬规则逻辑校验用软判断。硬规则只挡那些绝对不可能的情况比如空结果、字段缺失、数值超出物理范围。逻辑校验则让模型给出理由如果理由站不住脚人工复核时再调整提示词。另外把每次误判的案例收集起来定期回灌到校验提示词里校验准确率会随着使用逐步提升。5.4 循环不终止Agent 卡在重试里出不来这是最危险的问题一旦发生就是无限烧 token。我的防护有三道重试计数器按步骤独立避免全局计数被某一步耗尽总步数上限整个图跑超过 20 个节点就强制终止超时熔断单次执行超过 30 秒直接判失败。if state.get(total_nodes_visited, 0) 20: return {final_answer: 分析流程超时请简化问题后重试}这三道防线加上之后我再没遇到过跑飞的情况。建议你在图编译之后专门写几个“恶意”测试用例比如故意让某一步永远失败验证熔断机制是否生效。5.5 常见问题速查表症状可能原因排查动作解决方向规划步骤过多提示词缺约束检查规划提示词加 few-shot 和步数上限依赖取错数据索引错位打印 step_results改用 step_id 查找校验误杀规则过严看误判案例硬规则放宽软判断加理由循环不终止无熔断看节点访问数加重试上限和总步数上限SQL 字段臆造schema 不全检查 schema_context补全相关表结构结论与数据不符总结节点上下文缺失看 finalize 输入把关键结果显式传入6. 我实际跑下来的一些体会这套东西我在两个不同业务场景里落地过一个是电商分析一个是运营指标监控。最大的感受是规划层的质量决定了整个 Agent 的天花板而校验层决定了它的下限。很多人把精力全花在提示词调优上想让模型一次就规划对这方向其实有点偏。更务实的做法是接受规划会出错然后用校验和重试机制把它兜住。另外一个体会是不要追求全自动。我现在的做法是Agent 跑完规划之后会先把计划展示给用户确认用户点个头再执行。这一步看似多余实际上省掉了大量因为理解偏差导致的无效计算。等规划准确率稳定到很高之后再考虑去掉这个确认环节。最后分享一个小技巧把每次运行的完整状态快照存下来包括问题、规划、每步结果、校验记录。这些快照是调优的黄金素材比任何凭空想象的测试用例都有价值。我攒了两百多条快照之后规划提示词的迭代速度直接翻倍因为你能清楚地看到模型在哪些类型的问题上反复犯错针对性改就行了。这套方案后续还能往几个方向扩展一是把单 Agent 拆成多 Agent规划、执行、校验各由一个专门的 Agent 负责通过消息传递协作二是引入缓存层把常见问题的规划结果缓存起来相似问题直接复用三是把校验规则做成可配置的让业务方自己维护口径断言不用每次改代码。这些我还在陆续试有新的心得再单独写。