企业官网建设流程全解析

很多做数据分析的团队都卡在同一个地方：单步问答型工具基本够用，一旦需求变成"先看整体分布、再对比不同分组、最后定位异常原因"这种多步骤任务，要么手动写一堆脚本，要么就靠分析师人工接力。而 LangGraph 这类有状态编排框架，恰恰是把这种"多步骤思考 + 工具调用"落成代码结构的好选择。这篇文章我用一个完整的项目实例来拆解，怎么基于 LangGraph 构建一个真正会规划的数据分析 Agent——它拿到问题后不是急着调工具，而是先把问题拆解成可执行的分析计划，再一步步执行、验证、修正，最后输出结论。内容会覆盖状态图设计、Planner / Executor / Reflector 节点的实现细节、代码执行的安全边界，以及我从 Demo 推进到生产环境时踩过的几个真实坑。

1. 分析场景里最贵的开销，其实是"不会规划"

先说个反直觉的观察：LLM 调用本身不算贵，贵的是来回试探的过程。没有规划能力的 Agent 面对一个问题，经常是用户问一句、它调一次工具、看结果不顺眼再换一个工具，来回几个轮次之后，前后文里塞满了半截结果和无效尝试，上下文越来越长，模型反而更容易迷失方向。数据分析恰恰是重灾区，因为它天然需要多步操作：取数、清洗、聚合、可视化、解读，每一步的结果都会影响下一步的选择。

我最早试着用普通 ReAct 模式直接搭分析 Agent，问题暴露得很明显。比如用户问"最近三个月的订单趋势为什么波动这么明显"，Agent 第一轮可能只去查了总订单量，发现解释不了，又去查退款率，还不够，再查客单价。三步操作之间没有统一计划，每一步都是重新猜测，最后给出来的结论没有逻辑递进，用户还得自己拼凑信息。更麻烦的是，如果某一步工具执行报错，Agent 可能随机换一条路继续走，最终结果完全不可控。

引入规划机制之后，这三个问题都有了比较直接的改善。第一，规划让问题理解前置。Agent 拿到问题后先明确"我要分析的是什么、数据有哪些、结论需要支撑几个子问题"，而不是边做边想。第二，执行路径稳定。每一步做什么、依赖上一步的什么结果，都按计划走，步骤之间可以传递中间产物，上下文不用把全部历史都塞进去。第三，可解释性大幅提升。用户能看到一个清晰的分析计划，每一步在做什么一目了然，这比一个黑箱式的"AI 一顿操作猛如虎"强太多。

这里也回答一个常见疑问：为什么不能直接用 LangChain 的普通 Chain 串几个步骤？关键在于数据分析是个不确定流程。计划第一版经常跟实际数据对不上，比如原计划要按月份分组统计，结果发现数据里根本没有月份字段，只有下单时间戳，这时候必须支持动态修订计划、回退重来，甚至加一步预处理。Chain 是固定链路，做不到这种循环和分支；而 LangGraph 的状态图天生支持条件路由和循环，这才是规划型 Agent 真正需要的底座。

2. 用 LangGraph 把分析过程改造成一张状态图

LangGraph 的核心建模方式，是把整个工作流定义成一张图：节点是业务动作，边是流转条件，而整个流程共享一份状态（State）。和传统流程图不一样的是，这份状态会持久化保存，节点可以读取它、修改它，再传给下一个节点。这种设计特别适合分析场景，因为分析的中间结果、当前执行到哪一步、哪些数据已经拿到，这些信息天然需要跨步骤共享。

2.1 状态定义：想清楚哪些信息要跨节点传递

我第一次写这个 Agent 时，状态定义得特别偷懒，直接把所有对话消息都塞进去，结果每个节点都要重新读一遍冗长的历史记录，既浪费 token 又容易出错。后来学乖了，把状态拆成两块：一块是给 LLM 看的对话消息，另一块是给流程控制看的结构化字段。

from typing import TypedDict, Annotated, List, Optional from langgraph.graph.message import add_messages class AnalysisState(TypedDict): question: str # 用户原始问题 data_desc: str # 数据集结构说明，包含表名、字段、类型 plan: List[str] # 分析计划，每一项是一个可执行动作 current_index: int # 当前执行到计划的第几步 step_outputs: List[dict] # 每一步的中间结果，防止上下文丢失 last_error: Optional[str] # 最近一次执行错误，用于反思节点决策 final_answer: str # 最终汇总结论 messages: Annotated[list, add_messages] # LLM 对话记录，自动追加

这里最关键的设计是两个字段：current_index和step_outputs。前者让 Agent 知道自己执行到计划的哪一步，是流程控制的路标；后者把每步的产出物单独存成结构化数据，而不是混在对话消息里。这样执行器节点做下一步时，只需要读取上一步的结果作为输入，不用从塞满消息的上下文里"大海捞针"。

2.2 节点与边的设计：规划、执行、反思三件套

整个图我拆成了四个核心节点，外加一条反思回环。节点职责是这样划分的：

• Planner（规划节点）：根据用户问题、数据集结构生成一份有序分析计划。
• Executor（执行节点）：读取计划中的当前步骤，选择合适的工具（如 SQL 查询、代码执行）完成该步骤，并把结果写入step_outputs。
• Reflector（反思节点）：分析上一步的执行结果或错误，决定是继续下一步、修改计划，还是补一个额外的预处理步骤。
• Finisher（汇总节点）：把全部中间结果压缩成面向用户的最终结论。

边的关系就更有意思了。Planner 结束后无条件进入 Executor；Executor 执行完一步后不直接进下一步，而是先到路由函数做一个三岔判断：计划做完了就进 Finisher；报错了或者结果不符合预期就进 Reflector；否则回到 Executor 执行下一步。Reflector 那边，根据反思结果有两种出口，要么重新生成计划然后回到 Planner，要么给 Executor 补一个临时修正动作。

from langgraph.graph import StateGraph, START, END builder = StateGraph(AnalysisState) builder.add_node("planner", plan_node) builder.add_node("executor", execute_node) builder.add_node("reflector", reflect_node) builder.add_node("finisher", finish_node) builder.add_edge(START, "planner") builder.add_edge("planner", "executor") # 执行完后先反思，再决定是继续执行、修订计划、还是收尾 builder.add_conditional_edges( "executor", route_after_execution, { "continue": "executor", "revise": "reflector", "finish": "finisher", }, ) builder.add_edge("reflector", "planner") builder.add_edge("finisher", END) agent = builder.compile()

2.3 循环不是 Bug，是分析的常态

这个结构刚搭完时，团队里有人问：如果执行一直失败，不是会死循环吗？确实会，所以我在路由函数里加了一个重试次数上限。

def route_after_execution(state: AnalysisState) -> str: # 计划步骤全部完成，收尾 if state["current_index"] >= len(state["plan"]): return "finish" # 上一步执行出错，且还没超过重试上限，进反思节点 if state.get("last_error") and state["current_index"] < 3: return "revise" # 正常情况，继续执行下一步 return "continue"

但换个角度想，循环本身不是坏事。数据分析中"一步走不通、换条路再走"本来就是正常玩法。关键是循环要受控：每一步都要有退出条件，要么成功前进，要么失败重试到上限后强制收尾。我见过不少新手图便宜，把反思逻辑直接交给 LLM 自由输出"continue 还是 revise"，这非常危险，因为模型输出随机性强，很容易原地打转。正确做法是像上面代码一样，把流程控制放进确定的 Python 逻辑里，LLM 只负责提供内容（比如反思意见），不负责决定流程走向。

3. 核心实现：让 Agent 先把问题拆成可执行计划

这一节我把三个核心节点的实现逐个拆开讲，重点说清楚我是怎么拿到稳定可用的分析计划，以及计划执行不下去的时候怎么办。

3.1 Planner 节点：用结构化输出锁定分析步骤

Planner 的唯一职责，是把一句含糊的"帮我看下销售数据"变成明确的分析步骤。实现上我做了两层：第一层是给 LLM 提供足够的数据集上下文，第二层是用结构化输出约束计划格式。

from pydantic import BaseModel, Field class AnalysisPlan(BaseModel): steps: List[str] = Field(description="按顺序排列的具体分析步骤，每步必须只做一件事") data_needs: List[str] = Field(description="每个步骤所需的数据字段或数据表") risk: str = Field(description="可能出现的风险，比如数据缺失、字段语义不明确") def plan_node(state: AnalysisState) -> dict: llm = get_llm() prompt = f""" 你是一名资深数据分析师。用户的问题是：{state['question']} 可用数据说明：{state['data_desc']} 请把上述问题拆解成一个可执行的分析计划。要求： 1. 每步只做一件事，比如"统计各月份总订单量"、"对比退款率趋势"。 2. 步骤之间要有依赖关系，后一步必须能用到前一步的结果。 3. 不要生成只能靠人眼判断、无法用代码执行的步骤。 4. 预计需要的字段必须写清楚。 """ resp = llm.with_structured_output(AnalysisPlan).invoke(prompt) return {"plan": resp.steps}

这里有个非常关键的细节：必须用结构化输出，而不是让 LLM 自由写一大段计划文本。自由文本的问题在于下一步执行时解析困难，模型可能生成"第一步先探索一下，第二步再看看"，这种描述没法映射到具体的工具调用。我踩过这个坑之后直接换了 Pydantic 结构化输出，强制 LLM 输出一个步骤列表，每一步都是完整的指令，比如"对 order_date 按月聚合，计算每月的订单量和总金额"，拿到这种字符串，执行器才知道该让工具做什么。

步骤粒度也很讲究。太粗会导致执行器一个动作干太多事情，出错后无法定位；太细则计划本身就要耗掉大量 token。我实践下来比较合适的粒度是：一个步骤对应一次工具调用，比如查询、聚合、绘图各算一步，这样一方面出错范围小，另一方面反思节点也能准确指出"哪一步出了问题"。

3.2 Executor 节点：按步执行并保留中间产物

Executor 是整个图里干活最多的节点。它的输入是plan和current_index，先确定当前要执行哪一步，然后根据步骤内容选择合适的工具。工具选择逻辑我放到下一节详细讲，这里只看节点骨架：

def execute_node(state: AnalysisState) -> dict: idx = state["current_index"] step = state["plan"][idx] try: result = call_tool_for_step(step, state["step_outputs"]) new_outputs = list(state["step_outputs"]) new_outputs.append({"step": step, "result": result}) return { "step_outputs": new_outputs, "current_index": idx + 1, "last_error": None, } except Exception as e: return { "last_error": f"第 {idx + 1} 步执行失败: {step}，错误信息: {str(e)}", "current_index": idx + 1, }

值得注意的细节是执行完一步之后我立刻把current_index加一，但返回错误时也加一。为什么要这样？因为在路由函数里，我是通过last_error判断进不进反思分支的，如果出错卡在同一个索引不放，整个图就会卡死在该步骤反复重试。让索引先往前走，再由反思节点决定是回到 Planner 重新调整计划，还是具体修复这一步，流程会更顺。

另一个细节是step_outputs我用的是不可变追加方式：新建一个 list、把旧结果拷出来再加上新的。LangGraph 的状态更新支持多种合并策略，追加型数据用这种手动复制方式最稳，避免多个节点并发更新导致状态冲突。这里也可以改成Annotated[list, operator.add]自动合并，但我个人更喜欢显式控制，出错时更容易排查。

3.3 Reflector 节点：计划与实际脱节时怎么办

Reflector 是整个架构里"会规划"的体现——它负责判断原计划是否需要修正。触发场景主要有三种：上一步执行报错（比如字段不存在、数据类型不对）、执行结果与预期偏差过大（比如查出来的数据量为 0）、以及拿到了出乎意料的中间结果（比如某个维度数据突然暴涨，需要追加深入分析）。

def reflect_node(state: AnalysisState) -> dict: llm = get_llm() prompt = f""" 原始问题：{state['question']} 当前计划：{state['plan']} 已经执行的结果：{state['step_outputs']} 最新错误或异常：{state['last_error']} 请判断：继续按原计划执行，还是需要修改计划？如需修改，请给出新的步骤列表。 """ resp = llm.with_structured_output(AnalysisPlan).invoke(prompt) # 如果模型给出的新计划与旧计划实质相同，则不再修改 if resp.steps == state["plan"]: return {"last_error": None, "plan": state["plan"]} return {"plan": resp.steps, "last_error": None}

这里有一个我在实际运行中总结的经验：Reflector 不能只靠last_error触发。数据异常但没报错的情况同样常见。比如某一步查出来的日期都是未来时间，代码执行没任何问题，但结果明显不符合逻辑。所以我强烈建议在 Executor 返回结果后加一层"结果合理性校验"：数值型结果看看范围是否合理，时间型看看是否在当前区间，字段类型转换是否成功。有异常就在last_error里带一句描述，这样 Reflector 就能判断并介入。这类轻量校验逻辑成本极低，但能挡住大量无效的后续步骤。

4. 工具与代码安全：分析 Agent 的执行边界

会规划是一回事，能不能安全执行又是另一回事。数据分析 Agent 不是只读接口，它可能要生成 Python 代码跑聚合分析，可能要查数据库，还可能要把结果绘图。这些动作如果全裸奔，生产环境根本不敢上。

4.1 分析工具的注册与选择逻辑

我给 Agent 维护了一个工具注册表，每个工具描述里包含：能做什么、输入参数结构、适合的场景，以及一个匹配函数。执行器拿到计划步骤后，先做规则匹配，再让 LLM 做二次确认。

class AnalysisTool(BaseModel): name: str description: str input_schema: dict run_func: Callable suitable_keywords: List[str] TOOL_REGISTRY = [ AnalysisTool( name="sql_query", description="对关系型数据库执行只读 SQL 查询，返回 DataFrame", input_schema={"query": "string"}, suitable_keywords=["统计", "查询", "聚合", "筛选", "按", "比"], ), AnalysisTool( name="python_exec", description="执行数据分析 Python 代码，可使用 pandas、numpy，返回文本/表格结果", input_schema={"code": "string"}, suitable_keywords=["计算", "处理", "清洗", "透视", "分析"], ), AnalysisTool( name="chart_render", description="根据数据生成图表，返回图片地址", input_schema={"chart_type": "string", "data_snippet": "json"}, suitable_keywords=["可视化", "趋势", "分布", "图", "对比"], ), ]

工具调度不是一个纯 LLM 函数调用那么简单的，因为执行器节点的上下文里已经有整个分析计划的步骤信息，如果再走一层 LLM 的 function calling，反而会多一次网络往返，而且在步骤指令明确时完全没必要。所以我用了"规则初筛 + LLM 兜底"的策略：如果步骤文本命中了某个工具的suitable_keywords，直接选它；如果多个工具都命中，就交给 LLM 挑一个最合理的；如果全部未命中，才走完整 LLM 工具选择流程。这套混合方案在速度和准确率之间平衡得不错。

4.2 代码生成和沙箱执行配置

数据分析绕不开 Python 执行，这是 Agent 能力上限所在，也是安全风险最高的一环。我给python_exec工具的运行环境做了三个层面的约束：

• 依赖白名单：只预装 pandas、numpy、matplotlib、scikit-learn 这类数据分析常用库，不在名单里的 import 一律拦截。
• 资源限制：容器内存上限 512MB，CPU 配额 1 核，单次执行超时 30 秒强制 kill。
• 网络隔离：执行环境没有出网能力，文件系统只允许读挂载的数据目录，写操作全部重定向到临时目录。

def run_code_in_sandbox(code: str, data_path: str) -> str: precheck = validate_imports(code, allowed_modules={"pandas", "numpy", "matplotlib", "sklearn", "scipy"}) if not precheck: raise SecurityError(f"包含未授权的模块: {precheck}") # 后续通过容器接口执行，挂载只读数据目录 result = sandbox_client.execute( code=code, mounts=[{"path": data_path, "read_only": True}], memory_mb=512, cpu_shares=1, timeout_sec=30, ) return result.text

这里我要强调：禁 import 白名单只是第一道防线，不是全部。即便只允许 pandas 和 numpy，模型也能写出一些非常消耗资源的代码，比如超大矩阵运算。所以一定要有执行超时和资源上限，而且要在生成代码之前就在提示词里明确告知 LLM"只能使用白名单库、数据集大小、执行时限"，让模型在写代码阶段就尽量避免危险操作。经验数据告诉我，加了这层提示之后，生成的代码质量反而更高，因为模型不再盲目尝试各种花式实现。

4.3 失败重试与中断恢复机制

分析流程跑到一半失败了，不能每次都要全部重来。LangGraph 的 checkpointer 机制在这里特别有用：每个节点执行完，状态都会被持久化保存，用户甚至可以回滚到任意历史节点重新执行。

from langgraph.checkpoint import MemoryCheckpointSaver checkpointer = MemoryCheckpointSaver() agent = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "analysis-001"}} result = agent.invoke(initial_state, config=config)

生产环境我更推荐用专门的存储后端保存检查点，比如数据库或对象存储。好处是用户看到分析结果不满意，或者中间某一步执行失败，可以直接从那个步骤继续跑，不用从头再来。这在数据分析场景下价值非常大，因为分析计划前面的步骤往往跑了很久，全部重算既浪费钱也浪费时间。

中断恢复还有一个妙用：可以做审批流程。比如分析 Agent 在执行到某一步需要更新数据表或触发外部系统时，可以通过interrupt机制暂停图的执行，等人工确认后从暂停点继续。我的经验是，数据分析过程中涉及"写入动作"的环节，哪怕只是写回一个临时表，都建议保留人工介入点。

5. 从 Demo 到生产：几个躲不掉的工程细节

代码跑通了 Demo，不代表能上生产。这一节列几个我实际部署时踩过的坑和对应的解决方案。

5.1 会话隔离和并发控制

一开始我是共享一个 thread 跑所有用户的请求，结果发现两个用户的分析任务互相污染——用户 A 的数据结果串到了用户 B 的上下文里。LangGraph 的 thread_id 就是为这个设计的，每个用户会话独立一个 id，checkpoint 也按 id 隔离。

但仅有 thread_id 还不够。数据分析任务经常要挂后台跑，然后前端轮询进度。我后来用异步接口加任务表的方式解决：每次请求创建一个任务记录，带上 thread_id、用户 id、参数，后端用异步 worker 轮流消费任务。这样并发控制、超时、重试都在任务层统一管理，而不是散落在 Agent 逻辑里。

5.2 成本控制：如何避免计划无限膨胀

规划型 Agent 最容易踩的成本坑，是 Plan 越修越复杂。Reflector 节点如果频繁触发，每次都会调用 LLM 重新生成计划，这本身就是一笔不小开销。更糟糕的是，有些版本的计划比原计划多了三四个新步骤，整个执行链进一步拉长，推理时间成倍上涨。

我后来干脆引入"计划步数上限"：步骤数超过八步就强制收拢，要求 LLM 合并同类步骤；反思触发次数超过三次就停止修订，直接让 Finisher 基于已有中间结果给一个"带局限性说明"的结论。宁可结论不完美，也不能让一次分析跑到天荒地老。另外建议在计划生成前就告诉 LLM 当前数据集的大致行数、字段数量，明确提示"该数据集规模较小，三步以内能完成分析"，这种显式的复杂度约束非常有效。

5.3 可观测性与复盘

你会很惊讶，Agent 出错时想排查到底是哪一步错了、为什么错，有多困难。我刚开始所有节点只有日志输出，出问题后完全靠猜。后面做了两个改造：

一是给每个节点返回时附带一个trace_info字段，记录该节点的输入摘要、输出摘要、消耗 token 数、耗时。这样每次分析任务结束，都能拿到一个完整的执行追踪，节点之间传递了什么、哪一步耗时最久、哪一步掉了链子，一目了然。

二是对中间结果做采样存档。step_outputs里经常是很大的 DataFrame，全存下来不现实，但我会存每一批结果的 schema、行数、前几行样例。出问题时，能从这些样例里很快定位是数据本身的问题，还是后续步骤解析的问题。

我甚至做了个简单的统计面板，记录 Planner 生成计划后实际执行的成功率。跑了两周左右的数据后发现，计划步骤中"数据清洗"类的步骤成功率最低，通常是因为字段名猜想与实际 schema 不符。这个现象直接推动我把数据集的字段信息前置传给 Planner，让模型在做计划时就充分考虑字段是否真实存在，计划成功率从此提高了不少。

6. 一些个人体会：什么场景真正适合这种架构

写了这么多，最后聊聊取舍。LangGraph 构建的分析 Agent 不是银弹，我测试下来它最适配的场景有三个特征：问题不是单步指令能解决的；中间过程需要根据数据反馈动态调整；结论必须基于多个中间结果综合得出。典型如经营分析、用户行为归因、异常波动探查这类任务。反过来，如果只是固定报表查询、固定口径计算，传统代码或普通问答机器人完全能解决，根本不需要上图和 Agent，徒增复杂度和成本。

还有一点建议是：先别急着搭多 Agent 架构。很多团队一上来就规划一个"数据分析精英小分队"，什么数据工程师 Agent、可视化 Agent、报告撰写 Agent 各来一个。我的实践感受是，初期单 Agent + 良好状态设计就够了；当单 Agent 的规划逻辑复杂到状态里塞了十几类字段，才考虑按职责拆成子 Agent，每个子 Agent 保持独立的 LangGraph 图，用主图做编排。这种演进路线试错成本最低，也最容易定位问题。

如果你也想照着搭一版，建议第一版只在 Planner 和 Executor 之间加一条最简单的条件边，跑通"计划-执行-循环"的闭环，跑稳之后再逐步引入 Reflector 和中断恢复。我自己就是这么迭代过来的，诚实的建议给你参考。