最近接了一个数据分析自动化的活儿客户要求把一堆Excel、数据库报表交给AI让它自己拆问题、自己查数据、生成结论。刚开始我直接用单轮模型调用效果一言难尽模型经常跳过关键筛选条件或者问了十几轮仍原地打转。后来把项目改成用LangGraph做一个会规划的数据分析Agent才真正解决了“会想、会干、会检查”的问题。这篇文章把核心思路、完整代码和避坑经验分享出来适合想用LangGraph做落地分析工具或者在LLM应用里引入规划能力的开发者。1. 为什么数据分析Agent需要“规划”能力1.1 直接调用模型分析数据的三个大坑很多教程教的是“把问题丢给大模型再让它调几个工具函数”。这种模式做Demo很爽做真实数据分析项目时会立刻翻车。我踩过的坑主要有三个。第一漏掉前置条件。比如用户问“上个月华东区销售额环比变化”模型可能直接去统计总销售额完全忘记先按区域过滤数据。问题描述里明明有“华东区”但模型在自由对话中经常把它当背景信息忽略掉。原因是单轮调用没有给模型一个清晰的“步骤约束”它只能凭语感猜。第二目标漂移。真实分析任务往往是多步的先看数据质量再做清洗然后聚合最后对比。如果让模型一步步自由发挥它在第三步可能已经忘了一开始的业务问题甚至自作主张换了指标。比如问“哪个品类利润最高”模型答着答着开始输出“建议多备货”这已经不是数据分析而是变成写营销方案了。第三没有自纠能力。传统工具调用链是一次性的中间任何一步出错SQL语法错、字段名不对、结果为空整条链路就断了。模型不会回头检查自己的思路只会反复报错。需要外部机制替它兜底让它能“停下来重新想”。所以纯靠“一个模型 一堆工具函数”不是真正的Agent只是披着Agent外衣的对话补全。1.2 LangGraph的图式编排把“临时工”变成“项目经理”LangGraph给我的第一感觉是它把写Agent变成画流程图。以前手写Agent循环无非是while True: 模型调用 - 解析动作 - 执行 - 再调所有逻辑挤在一起调试全靠print。LangGraph的核心概念是状态、节点和边状态是流动的数据包节点是每个处理单元边决定下一步走向。你可以在节点之间加条件判断也可以让某些节点并行执行。我一般这样比喻不用LangGraph时你是在雇一个“临时工”给他一堆工具让他自己摸索用了LangGraph你是在搭建一条流水线每个节点是一个岗位岗位之间用工序单状态衔接。临时工可能干到一半跑去干别的流水线不会因为每一步都明确。回到数据分析Agent流水线的好处特别明显。我可以把“理解问题”“拆解步骤”“查询数据”“检查结论”拆成独立节点。每个节点只负责一件事模型不会因为上下文太长而“精神涣散”。更重要的是LangGraph支持checkpoint持久化节点执行到一半断电了可以从上次断点继续跑这在真实业务里价值极大。1.3 方案选型Plan-Execute 反思而不是纯ReAct行业内做Agent常提ReAct模式也就是“推理-行动”交替进行。ReAct在问答型任务上表现不错但在数据分析上我强烈建议别直接用。原因是数据分析天然适合“先出计划再执行”而不是走一步看一步。我做过对比同一个分析任务用ReAct风格让模型边想边做它会频繁修改查询条件光是“找出各省份销售额”这类步骤就反复执行了三四遍先让它写计划再执行不仅调用次数减少一半结果也更稳定。原因不复杂数据查询是有代价的频繁试错既费时间又费钱。所以我选的架构是Plan-and-Execute 反思Planner根据用户问题生成一份结构化的步骤清单。Executor按清单一步步执行每个步骤调用对应的数据工具。Reflector检查执行结果是否回答原始问题如果发现计划不合理可以重新规划。人工确认关键节点支持人审防止Agent在错误思路上越走越远。这个架构里规划是核心。一份好的计划规定了“先查什么、再算什么、最后怎么呈现”后续执行节点只需要老老实实按步骤走。规划一旦清晰Agent的表现就从“乱拳打死老师傅”变成了“按图施工”。2. 核心细节解析规划-执行-反思闭环2.1 状态定义用显式字段控制数据流在LangGraph里状态由TypedDict或Pydantic BaseModel定义。我建议每个字段都起得直白一些不要把一切都塞进messages。原因很简单如果规划结果、执行中间值全部混在消息列表里后面的节点要自己去翻历史费token又容易出错。下面是我的状态定义示例from typing import TypedDict, List, Optional from langgraph.graph import MessagesState class AnalysisState(MessagesState): user_question: str # 原始问题 plan: Optional[List[str]] None # 计划步骤列表 current_step: int 0 # 当前执行到第几步 observations: List[str] [] # 每步执行后收集到的观察结果 final_answer: Optional[str] None # 最终回答 need_replan: bool False # 是否需要重新规划 error_info: Optional[str] None # 最近一次错误信息MessagesState自带messages字段用来记录对话历史和Agent的思考轨迹。但真正控制流程的是上面的业务字段。比如current_step可以让我知道执行到哪一步observations用来累积查询结果摘要need_replan用来告诉条件边“下一步走重新规划还是结束”。用显式字段还有一个好处调试时直接打印state所有关键信息一目了然。注意不要图省事只用一个results: dict存所有中间结果。数据分析会涉及多张表、多次聚合不同步骤的数据结构差异很大全部塞进一个dict会让类型检查形同虚设。2.2 规划节点让模型先生成结构化Step Plan规划节点是整个Agent的大脑。它的输入是用户问题输出是一份结构化计划。为了让计划可解析、可校验我用了Pydantic而不是让模型输出纯文本。from pydantic import BaseModel, Field from langchain_core.prompts import ChatPromptTemplate class StepPlan(BaseModel): steps: List[str] Field(description数据分析步骤按顺序排列每步是一个清晰的动作指令) planner_prompt ChatPromptTemplate.from_messages([ (system, 你是一名资深数据分析师。请把用户问题拆解为可执行的数据分析步骤。步骤必须具体包含需要查询的字段、过滤条件和统计口径。), (human, 用户问题{question}) ]) def plan_node(state: AnalysisState): chain planner_prompt | llm.with_structured_output(StepPlan) plan chain.invoke({question: state[user_question]}).steps return { plan: plan, current_step: 0, observations: [], need_replan: False }这个设计的核心是with_structured_output它强制模型输出符合StepPlan结构的JSON而不是一段“我会先...然后...”的废话。模型一旦被结构化约束输出的步骤通常质量高很多。这里我分享一个小技巧步骤数量不要太多。我一般会做二次约束如果模型生成超过8个步骤就让它合并同类项。步骤越多执行链越长出错的概率也越大。真实项目里5个步骤左右最舒服既覆盖完整流程又不会让Agent迷路。2.3 执行节点与工具调用数据操作的“方向盘”执行节点是真正接触数据的地方。它根据current_step从计划里取出当前步骤描述然后决定调用哪个工具。工具本身是普通Python函数我用tool包装成BaseTool让模型可以按名称调用。from langchain_core.tools import tool tool def query_sales_data(region: str 全部, start_date: str 2024-01-01, end_date: str 2024-12-31) - str: 按区域和日期范围查询销售额明细返回前20行摘要。 # 这里是伪代码实际项目会连数据库或读取DataFrame df sales_table[(sales_table[region] region) if region ! 全部 else True] df df[(df[date] start_date) (df[date] end_date)] return df.head(20).to_json(orientrecords, force_asciiFalse) tool def calc_growth_rate(base_value: float, current_value: float) - str: 计算增长率返回百分比字符串。 if base_value 0: return 基数不为0无法计算增长率 return f{(current_value - base_value) / base_value * 100:.1f}% tool def plot_sales_trend(region: str 全部) - str: 生成指定区域销售额趋势图返回图片路径。 # 实际项目会调用matplotlib/plotly生成图这里省略绘图细节 return f已生成趋势图sales_trend_{region}.png执行节点拿到步骤字符串后先交给一个轻量级调度函数判断该调用哪个工具。我的判断逻辑很简单步骤里包含“查询”“明细”就调query_sales_data包含“增长”就调calc_growth_rate包含“趋势图”“可视化”就调plot_sales_trend。这个函数不需要很复杂因为规划节点已经给出了明确的动作语义。def execute_step(state: AnalysisState): if state[current_step] len(state[plan]): return {need_replan: False} step state[plan][state[current_step]] try: result route_step_to_tool(step) new_observations state[observations] [f执行第{state[current_step] 1}步{step}\n结果{result}] return { observations: new_observations, current_step: state[current_step] 1, } except Exception as e: return { error_info: str(e), need_replan: True }这里有个非常关键的细节工具函数的返回值不要直接返回完整DataFrame。我见过很多人把整个表传给模型结果几千行数据把上下文窗口撑爆。正确做法是只返回“能支撑下一步判断的摘要”比如前20行、聚合结果、条数统计。数据细节留在数据库里模型只需要知道“查到了什么、大概长什么样”。2.4 反思节点与条件边坏计划如何被纠正执行完计划里的所有步骤后不能直接让模型编答案。必须先经过反思节点它决定是收工还是回头重干。我主要检查三点执行过程中有没有异常或空结果当前观察是否覆盖了原始问题里的全部指标每个步骤的结果之间是否逻辑一致检查逻辑写成一个评估函数def check_completion(state: AnalysisState) - bool: if state.get(error_info): return False if not state.get(plan): return False if state[current_step] len(state[plan]): return False # 如果计划已经全部执行且没有错误就认为可以生成答案 return True然后把它接到LangGraph的条件边上def should_continue(state: AnalysisState) - str: if state.get(need_replan): return replan if not check_completion(state): return replan return finalize反思节点的角色是“质检员”。它发现执行结果和原始问题对不上时会回到plan_node重新生成计划。为了防止无限重规划我会加一个replan_count计数器超过2次就强制终止并报告“该任务无法自动完成”。实际操作中我发现很多任务根本不需要重规划。真正需要重规划的往往是工具调用异常比如字段拼写错误。所以我在error_info里带上具体错误信息让重新规划时能根据错误修正计划。2.5 生产化设计持久化与人工确认本地跑通Demo不难难在拿到生产环境用。LangGraph有两个特性非常实用checkpointer和interrupt。checkpointer负责保存每个节点执行后的状态快照。这样执行到一半进程崩溃恢复时不用从头开始。代码里只需要在编译图时传入一个MemorySaver或数据库持久化实例。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() app graph.compile(checkpointercheckpointer)interrupt则可以在关键节点暂停执行等待人工响应。数据分析Agent经常需要人确认口径比如“是否允许查询全表”这时候就可以在工具调用前打断等人在前端点“允许/拒绝”后再继续。from langgraph.types import interrupt def approval_node(state: AnalysisState): decision interrupt({ question: 是否同意以下数据查询口径, step: state[plan][state[current_step]] }) if decision ! allow: raise ValueError(用户拒绝该步骤) return {approved: True}人工确认看起来很繁琐但在真实业务里特别必要。因为大模型的计划再漂亮也不能完全替代业务人员的判断。尤其是“删除脏数据”“跨表关联”这类危险操作强制人审能避免很多事故。3. 实操过程从零搭建一个会规划的数据分析Agent3.1 环境准备与依赖选型先装环境。我用的Python版本是3.10以上太老的版本对Pydantic v2不友好。核心依赖如下pip install langgraph langchain langchain-openai pandas pydantic如果你不用OpenAI的模型服务也可以换成其他LangChain支持的模型封装。建议模型选支持with_structured_output或tool_calling能力的否则解析StepPlan时会很痛苦。提示langchain-openai只是一个桥接包底层还是调用远程模型API。本地开发时可以先用接口兼容的国产或开源模型部署再切主模型。注意不同模型对with_structured_output的支持程度不一样购买前先看文档。3.2 实现数据工具集我做一个简化版零售数据分析场景造两张模拟DataFrame一张销售明细表一张区域信息表。工具函数是真实项目中最低配的版本但足够跑通整个流程。import pandas as pd import numpy as np # 模拟数据实际项目里这里换成数据库连接或数据仓库API np.random.seed(42) sales_df pd.DataFrame({ order_id: [fSO{1000 i} for i in range(100)], region: np.random.choice([华东, 华北, 华南, 西南], 100), category: np.random.choice([数码, 服饰, 食品, 家居], 100), amount: np.random.uniform(100, 5000, 100).round(2), date: pd.date_range(2024-01-01, periods100) })工具函数需要遵循同一个原则输入参数尽量少返回内容尽量精。我不会把整个DataFrame传给模型而是传聚合后的摘要。tool def query_sales_amount(region: str 全部, start_date: str 2024-01-01, end_date: str 2024-12-31) - str: 按区域和日期查询销售额返回总和、订单笔数、均值。 df sales_df.copy() if region ! 全部: df df[df[region] region] df df[(df[date] start_date) (df[date] end_date)] return { 销售额总和: float(df[amount].sum()), 订单笔数: int(len(df)), 客单价: float(df[amount].mean().round(2)) } tool def query_category_sales(region: str 全部) - str: 查询各品类销售额汇总。 df sales_df.copy() if region ! 全部: df df[df[region] region] grouped df.groupby(category)[amount].sum().sort_values(ascendingFalse) return grouped.to_dict()我不建议让Agent直接生成SQL去执行。原因有两点一是模型生成的SQL错误率不低二是动态拼接SQL容易造成SQL注入。更稳妥的方式是定义一组“偏分析型”工具把查询范围限制在白名单内。模型想查数据只能通过工具函数不能自由发挥。3.3 用LangGraph编排规划、执行、反思节点准备工作做完接下来就把节点连成图。先实例化一个StateGraphfrom langgraph.graph import StateGraph, START, END graph StateGraph(AnalysisState) graph.add_node(planner, plan_node) graph.add_node(executor, execute_step) graph.add_node(reflector, reflect_node) graph.add_node(finalizer, final_answer_node) graph.add_edge(START, planner) graph.add_edge(planner, executor) graph.add_edge(executor, reflector) graph.add_conditional_edges( reflector, should_continue, { replan: planner, finalize: finalizer } ) graph.add_edge(finalizer, END)反射节点reflect_node负责判断是否需要重规划。它的代码很简单大部分逻辑已经在check_completion里def reflect_node(state: AnalysisState): completed check_completion(state) return {need_replan: not completed}最后汇总答案的final_answer_node接收observations让模型用“数据分析师”的口吻输出结论。它会引用每个步骤的结果而不是凭空编造。def final_answer_node(state: AnalysisState): prompt ChatPromptTemplate.from_messages([ (system, 你是一名严谨的数据分析师请基于所有观察结果回答用户问题。不引用不存在的数据。), (human, 用户问题{question}\n分析过程{observations}) ]) resp (prompt | llm).invoke({ question: state[user_question], observations: \n.join(state[observations]) }) return {final_answer: resp.content}3.4 跑通一个完整示例从问题到分析结论图编译完成后就可以输入自然语言问题。我用一个常见业务问题测试“对比华东和华北两个区域的总销售额哪个更高各自订单量是多少”调用方式app graph.compile(checkpointerMemorySaver()) config {configurable: {thread_id: demo-001}} result app.invoke({ messages: [HumanMessage(content对比华东和华北两个区域的总销售额哪个更高各自订单量是多少)], user_question: 对比华东和华北两个区域的总销售额哪个更高各自订单量是多少 }, config) print(result[final_answer])如果我在planner节点后面加一个打印函数会看到类似这样的日志规划器生成计划 1. 查询华东区域销售额和订单量 2. 查询华北区域销售额和订单量 3. 对比两区域指标并给出结论 执行第1步查询华东区域销售额和订单量 工具返回{销售额总和: 148365.5, 订单笔数: 27} 执行第2步查询华北区域销售额和订单量 工具返回{销售额总和: 132580.25, 订单笔数: 24} 反思节点检查计划已全部执行结果覆盖问题中的两个区域。 生成最终答案华东区域总销售额更高为148365.5订单量27笔华北区域总销售额132580.25订单量24笔。这个例子里规划、执行、反思的顺序非常清楚。模型并没有自己去拼接SQL每一步都是工具函数完成模型的职责是“判断用哪个工具、解读结果”。这也是这个架构最稳的原因把不稳定的模型限制在“做决策”和“写结论”把稳定精确的计算交给代码。3.5 关键参数调优temperature、max_iterations与预算控制参数设置直接影响Agent的可靠性和成本。我常用的几个参数如下参数建议值说明temperature0 ~ 0.2规划节点为0结论节点可以到0.2。温度太高会让计划发散。max_iterations5 ~ 8防止重规划死循环。超过次数直接报错。max_tokens512左右规划节点输出很短给太多token反而容易生成废话。replan_count最多2次超过2次重规划说明任务本身有问题。我建议temperature设置为0。因为数据分析看重的是确定性同一个问题如果两次跑出不同结论业务方根本不敢用。这里有个矛盾点温度低了模型会显得“机械”但对数据分析Agent来说“机械”反而是优点。预算控制上可以做一个简单的价格估算每次工具调用都会消耗一次模型调用和可能的函数返回token。一个典型任务大约需要4-6次LLM调用。如果每次调用消耗约3000 token那么单任务的调用成本是可以接受的。真正贵的是重规划每重来一次都会多花2次调用。所以控制好反思条件和输出长度比选哪个模型更能省钱。4. 常见问题与排查技巧实录4.1 规划经常跑偏未按步骤执行表现模型生成的计划是“1. 查数据 2. 分析数据 3. 给结论”步骤之间没有明确的字段和口径。这种“假大空”计划会导致执行节点不知道调用哪个工具。解决办法在planner_prompt里加入“禁止出现模糊动词”规则。例如“查询”只能和具体表名、区域名连用不能只说“查数据”。还可以做一次正则校验计划里如果出现“可能”“大概”这类词就强制重新生成。我在实际项目里还试过让模型先输出“我计划怎么做”的自由文本再让另一个模型把它转成结构化计划。效果略有提升但成本翻倍。现在更推荐直接用with_structured_output一步到位。4.2 工具返回数据过大上下文爆炸表现工具函数返回了完整DataFrame转成的JSON几百行甚至几千行模型还在继续调用token瞬间用完。解决办法所有工具返回值都要做“摘要化”。常用策略包括只返回head(20)并在返回值里注明总行数。先做聚合把明细变成汇总指标。返回统计描述而非原始记录均值、中位数、最大值、最小值。我见过最极端的案例是有人把一张百万行表上传给模型做“分析”结果API直接报上下文溢出。其实只要让模型知道“这个表长什么样”再用工具函数完成计算根本不需要加载全表。经验如果工具函数返回的内容一眼看不完说明设计有问题。工具函数的价值不是“喂数据”而是“给答案”。4.3 执行环节抛异常表不存在、字段名错误表现工具函数内部执行SQL或pandas操作时异常error_info里有报错堆栈Agent陷入“重规划-再报错-再重规划”的循环。解决办法要做两层防御。第一层工具函数里做好参数校验比如日期格式不对直接返回“参数错误日期格式应为YYYY-MM-DD”而不是抛一个很底层的异常。第二层在execute_step里捕获异常后把简洁的错误信息附给重规划节点让模型看到具体原因。我在重规划提示里会加一句话“上一次执行失败原因如下xxx。请生成一个新计划避免同样的错误。”这样第二次规划往往会避开错误的字段名或表名。4.4 反思节点导致死循环怎么办表现计划执行完反思节点发现“结果不够严谨”回到规划节点重新生成计划重新执行完还是不满意又回规划节点……图看起来在死循环里打转。解决办法限制重规划次数。我使用一个replan_count字段进planner节点时检查def plan_node(state: AnalysisState): if state.get(replan_count, 0) 2: return {plan: state.get(plan), need_replan: False, error_info: 自动重规划次数已达上限} # 正常生成新计划这里需要注意的是replan_count要在重新规划时递增而不是每次规划都递增。否则即使第一次规划成功也会浪费次数。正确逻辑是在should_continue里当返回replan时replan_count 1。4.5 人工审批卡住或超时表现用interrupt阻断后前端迟迟没有返回结果或者用户点了“允许”但状态没有恢复。解决办法理解LangGraph的中断机制。interrupt会暂停当前节点并把图的状态和中断点保存到checkpointer里。要恢复运行时需要再次调用同一个thread_id并传入用户输入。前端需要保留thread_id否则找不到历史状态。超时方面建议在业务层加一个审批超时时间。比如超过10分钟未处理自动走“拒绝”分支。因为Agent不能一直占着线程资源等待人工操作。我在项目中通常用一个定时任务自动取消超时审批。实操心得人工审批节点不要放在“规划之后”和“执行之前”那样太保守。更好的位置是“工具调用之前”每次工具调用都有可能触发审批。这样可以兼顾灵活性和安全性。4.6 成本居高不下优化重规划与prompt长度表现一个简单问题跑了十几步日志里全是重复查询和反思成本直线上升。优化思路是这样的为每个工具调用设置“结果缓存”。如果同一参数下查询过直接返回缓存不再调用模型。反思节点优先做规则检查而不是每次都用LLM判断。比如只有错误信息不为空时才触发重规划不要每次都让模型“评估一下”。observations只保留最近几步的摘要不用把整个历史都塞给结论节点。因为最后生成答案时只需要最新最相关的信息。我在项目中还做了一个小技巧给每个工具调用打标签统计“这个工具被模型误调用了几次”。如果某个工具频繁被误调用说明它的描述和命名有问题。比如calc_growth_rate经常被用于非增长率场景那我就会在工具描述里补充“仅用于同指标跨期对比”。最后再分享一点个人体会这套LangGraph方案我在不同项目里用了两三个月最大的体会是Agent能不能落地不在于模型有多聪明而在于流程有多“钝”。聪明模型反而容易自由发挥你必须把每一步都钉死让它没有太多跑偏空间。很多人一开始学LangGraph会追求“花哨的图结构”加一堆并行节点和复杂条件边。但真实的数据分析任务往往没那么复杂一条“规划-执行-反思”的主链路加上两三条条件分支已经覆盖了90%的需求。先跑通再优化比一开始就设计一个十节点大图更靠谱。如果你也想搭一个会规划的数据分析Agent我的建议是先抄上面的最小实现把一套可见可查的链路跑明白。等你对状态流转、断点恢复、工具调用这些机制足够熟练再往里面加并行分析、人工审批、数据权限这些东西。到那时候你会发现自己其实已经不只是在调Agent而是在做一套带SOP的数据分析流水线了。
