先说个真实场景。业务线每个月问我最多的不是做张报表而是我有个问题你帮我看看数据。问题看着小背后是一连串动作确认口径、写 SQL、跑数、核对、画图、再解释。过去这一套流程走下来快则半小时慢则一整天而且大部分问题高度重复纯粹在消耗分析师的精力。我动手搭了一个数据智能分析 Agent把自然语言问题直接接到 SQL 查询、统计计算和图表生成上运营同学现在自己提问就能拿到结论我只在口径有争议时才介入。这篇文章把我从架构设计、技术选型到代码实现的完整过程写出来踩过的坑也一并记录适合正在入门 Agent 开发、或者想用 Agent 把重复取数工作自动化的人参考。1. 为什么数据分析场景最适合做 Agent 落地1.1 传统分析流程的瓶颈不是技术是来回沟通很多团队的数据分析工具已经不少了BI 看板、报表平台、数据门户但业务方遇到一个具体问题时依然习惯性地来问分析师。原因很简单——看板只覆盖了预设好的那些指标而真实业务问题往往是昨天大盘跌了跌在哪条渠道上个月新客的次月留存按注册渠道拆一下这种临时性、组合式的查询。它由多个步骤组成先定位表再写 SQL然后可能还要做透视、算比率、画对比图。这套流程最大的问题不是某一步有多难而是每一步之间的沟通成本。分析师听懂需求要花时间写出来发现口径理解错了又要重来等数据交付到业务手里可能已经过了大半天决策早就错过了最佳窗口。更麻烦的是这类问题里有相当一部分是重复的上周刚查过的指标这周再查一次的情况比比皆是但因为没有沉淀成可复用的能力每次都得从头走一遍。我最初想过把这些需求做成参数化报表结果发现不可行业务的提问方式五花八门过滤条件、分组维度、对比周期随意组合配置表单比写 SQL 还慢。这时候我才意识到真正该做的是一个能自己理解问题、规划步骤、操作数据的系统也就是现在我们说的 Agent。1.2 Agent 和 Text-to-SQL 有什么本质区别很多人第一次接触数据智能分析时会觉得这不就是 Text-to-SQL 吗把自然语言翻译成 SQL 的技术几年前就有了代理的差别到底在哪我用一张表把两者的区别说清楚对比维度Text-to-SQL数据智能分析 Agent处理流程单次翻译问题 → 一条 SQL → 结果多步规划拆解问题 → 多次调用工具 → 根据中间结果调整错误处理翻译错了只能重来执行结果异常时能自己换条件、换表、换口径再试图表生成需要额外程序完成可以自主决定调用绘图工具多表关联依赖一次性写对 JOIN可以分步确认表结构后再关联能力扩展只能出 SQL接入更多工具就能查数、画图、做统计、发通知核心差异在于决策权。Text-to-SQL 是一次性翻译模型没有机会看到执行结果再调整而 Agent 是一个循环模型生成工具调用、执行、把结果返回给模型、模型再决定下一步。这就像雇了一个实习生你给他一个目标他先查表结构、再写查询、看到结果不对就换个写法最后把结论整理给你——而不是你每次把问题翻译成 SQL 扔给他执行。这个差异决定了 Agent 能处理的查询复杂度上限比 Text-to-SQL 高一个量级也决定了它的实现难度不在调用模型这一步而在工具设计、循环控制和异常处理上。2. 最小闭环架构先搞懂五层设计再写代码2.1 核心由五个模块构成我在搭第一个版本之前画过很多次架构图最后发现无论怎么简化一个能干活的数据分析 Agent 都逃不开下面五个模块调度大脑LLM负责理解问题、规划步骤、生成工具调用指令。这是整个系统里唯一有智能的部分但它也是最不可控的部分所以其他模块的设计目标都是围绕限制它的出错范围展开的。工具层Tools包括 SQL 查询、表结构查看、数据透视、图表绘制等。每个工具本质上是一个函数输入输出都是结构化的文本或文件模型通过函数名和参数描述来知道这个工具是干什么的。记忆层Memory保存当前对话的上下文、前几步的结果、甚至用户长期使用中沉淀下来的指标口径。记忆决定了 Agent 是每次从零开始还是越用越懂你。执行环境Runtime真实运行 SQL 和代码的地方。可以是本地 SQLite、远程数据库也可以是沙箱化的 Python 执行器。执行环境要保证一件事——Agent 的任何操作都不会破坏底层数据。安全边界Guardrail包括 SQL 只读限制、查询超时、返回行数限制、敏感数据脱敏等。这一层不是后补的而应该从第一个版本就开始设计。这五个模块里工具层和调度层是灵魂记忆和安全边界是能不能上线用的关键执行环境反而是最简单的那部分——用现成的数据库连接和 pandas 就能撑起一个原型。2.2 ReAct 模式是 Agent 工作的底层逻辑数据分析 Agent 的调度循环背后是 ReActReasoning Acting模式。它的运行逻辑可以抽象成三步在循环里反复执行思考Thought模型面对当前状态判断现在该做什么。比如用户问哪个渠道销售额最高模型会想我应该先查一下销售表里有哪些字段。行动Action模型从工具列表里选一个生成参数。对应到代码里就是返回一个tool_call包含工具名和 JSON 格式的参数。观察Observation系统执行工具把真实结果返回给模型模型再看结果决定下一步。这个循环会一直持续到模型认为问题已经解决输出最终回答为止。我第一次理解这个模式时想到的是人类分析师的真实工作方式拿到一个陌生数据库谁也不会上来就写一个几百行的 JOIN而是先SHOW TABLES再DESC一下然后写个小查询看看数据长什么样确认无误后才写正式查询。ReAct 本质上就是把这套流程模拟出来了。在工程实现上现在的大模型 API 基本都支持 Function Calling工具调用机制。模型不再需要从文本里生成调用哪个函数这样的描述而是直接输出结构化的调用请求系统只需要按 JSON 格式解析并执行就行。这大大减少了文本解析出错的可能性也是我推荐新手直接从 Function Calling 入手而不是从提示词硬写的原因。3. 框架选型手写循环、LangChain、LlamaIndex 的取舍3.1 主流方案横评开始写代码之前我花了两天时间对比了市面上主流的 Agent 开发方案。这里直接说结论方案优点缺点适合场景手写 Function Calling 循环逻辑透明、依赖少、调试容易生态能力需要自己补工具数量少、流程固定的项目LangChain / LangGraph组件全、多 Agent 编排成熟抽象层级多、版本升级频繁、出问题不好排查复杂工作流、生产级多 Agent 系统LlamaIndex对 RAG 和数据源支持好偏知识库场景通用 Agent 能力相对弱文档问答 结构化数据结合的助理AutoGen / MetaGPT多 Agent 对话协作能力强上手曲线陡、运行开销大、行为不可控研究探索型、群体协作仿真如果你是刚开始学 Agent 开发我强烈建议看这些框架底下到底包了什么。大多数框架的核心不过就是我把工具定义传给了模型、模型返回工具调用、我再执行并把结果追加到对话里。框架帮你省掉的是这部分样板代码同时带来的是一整套抽象一旦自定义需求复杂一点抽象就变成了障碍。3.2 我为什么选择手写核心循环 轻量封装最终我的选择是核心循环手写工具层也不依赖框架总共不到 300 行代码。理由有三个。第一数据分析 Agent 的核心逻辑非常固定就是一个想一下、调一下、看一下的循环手写这套循环几乎没有成本但能获得完全的控制权——我可以精确控制 messages 数组里每一步的内容出问题直接看调用链就能定位。第二框架版本变化太快LangChain 早期版本和现在的 API 用法差别巨大网上教程经常对不上号团队接手维护的成本很高。第三安全控制需要我能在工具执行前做拦截和校验手写的话这些逻辑放在哪里、怎么绕都清清楚楚。当然这不代表框架没有用。如果你的 Agent 要处理几十个工具、多个角色协作、复杂的状态机LangGraph 这类编排框架能省很多事。我的建议是第一个版本一定手写把原理吃透之后再根据真实需求决定要不要引入框架。直接上框架的结果往往是你连模型为什么没有调用工具都排查不出来。4. 动手实现一个可以直接复用的数据分析 Agent4.1 环境准备与数据初始化我这边为了演示方便用 SQLite 做底层库保证任何人都能一键复现。先装依赖pip install pandas sqlalchemy openai matplotlib python-dotenv tabulate然后准备一张简单的销售表import sqlite3 conn sqlite3.connect(sales.db) conn.execute( CREATE TABLE IF NOT EXISTS sales ( id INTEGER PRIMARY KEY, order_date TEXT NOT NULL, channel TEXT NOT NULL, region TEXT NOT NULL, product TEXT NOT NULL, amount REAL NOT NULL, cost REAL NOT NULL ) ) conn.execute( INSERT INTO sales (order_date, channel, region, product, amount, cost) VALUES (2024-06-01, 直营, 华东, A, 1200, 700), (2024-06-02, 分销, 华南, B, 850, 500), (2024-06-03, 电商, 华北, A, 2000, 1100), (2024-07-01, 直营, 华东, B, 1500, 800) ) conn.commit() conn.close()环境搭建这一步我踩过一个小坑tabulate这个库如果不安装DataFrame.to_markdown()会直接报错而它又不在 pandas 的默认依赖里。所以工具函数里如果用到了 to_markdown务必在 requirements 里带上。4.2 工具层把数据库能力封装成模型能调用的函数工具层设计的原则是输入输出都必须是字符串。因为大模型的上下文只能传文本工具返回的结构化对象必须序列化为文本才能被模型看见。我写了四个工具import os import pandas as pd import sqlalchemy as sa import matplotlib.pyplot as plt DATABASE_URL os.getenv(DATABASE_URL, sqlite:///sales.db) engine sa.create_engine(DATABASE_URL) def run_sql(query: str) - str: 执行只读 SELECT 查询返回最多 30 行的 Markdown 表格。 query query.strip().rstrip(;) with engine.connect() as conn: df pd.read_sql_query(query, conn) if df.empty: return 查询结果为空请检查条件或换一个查询角度。 head df.head(30) text head.to_markdown(indexFalse) if len(df) 30: text f共返回 {len(df)} 行仅展示前 30 行\n text return text def list_tables() - str: 列出数据库中所有表名。 insp sa.inspect(engine) return 、.join(insp.get_table_names()) def describe_table(table_name: str) - str: 返回指定表的字段名和类型。 insp sa.inspect(engine) cols insp.get_columns(table_name) return \n.join(f- {c[name]} ({c[type]}) for c in cols) def plot_chart(sql_query: str, chart_type: str bar) - str: 根据 SQL 结果绘制图表并保存为 png。 with engine.connect() as conn: df pd.read_sql_query(sql_query, conn) filename chart.png df.plot(kindchart_type, figsize(10, 6)) plt.tight_layout() plt.savefig(filename) return f图表已生成{filename}请告诉用户查看。这里有一个非常关键的细节工具返回的错误信息也很重要。比如run_sql如果遇到表不存在或字段拼写错误SQLAlchemy 抛出的异常必须原样返回给模型模型才能根据错误信息自己修正。如果把异常吞掉或只返回执行失败四个字模型就会失去纠错的依据直接开始编造结果。所以我在工具函数外统一做了异常捕获def call_tool(name: str, args: dict) - str: try: if name run_sql: return run_sql(**args) if name list_tables: return list_tables() if name describe_table: return describe_table(**args) if name plot_chart: return plot_chart(**args) return f未知工具: {name} except Exception as e: return f工具执行出错{type(e).__name__}: {e}这条规则值得刻在脑子里工具永远不抛异常而是把异常描述作为正常返回给模型。这样整个循环才能继续跑下去。4.3 核心循环用 Function Calling 串起规划-执行-观察工具定义好之后核心循环反而最简短。我用 OpenAI 的 SDK 做演示其他支持 Function Calling 的模型接口逻辑完全一致import json import os from openai import OpenAI client OpenAI(api_keyos.getenv(OPENAI_API_KEY)) def run_agent(question: str, max_rounds: int 8) - str: messages [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: question} ] for _ in range(max_rounds): resp client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolsTOOLS, tool_choiceauto, ) msg resp.choices[0].message messages.append(msg) if not msg.tool_calls: return msg.content for tc in msg.tool_calls: args json.loads(tc.function.arguments or {}) result call_tool(tc.function.name, args) messages.append({ role: tool, tool_call_id: tc.id, content: result, }) return 已达到最大轮次建议把问题拆得更具体一些再试。TOOLS列表是工具函数的 JSON Schema 描述官方术语叫 tools 定义。每个工具都要写清楚名字、描述和参数结构这直接决定了模型能不能正确调用。描述文字很关键执行只读 SELECT 查询比查询数据好因为模型需要知道这个工具的能力边界和约束。这段代码里有三个新手最容易踩的坑第一tool_call_id必须和 assistant 消息里的 tool_call 的 id 一一对应。如果填错或漏填API 会直接报错我见过很多人卡在这一步。第二messages.append(msg)这一步不能省。模型返回的 assistant 消息里带着 tool_calls这段消息必须原样放进下一轮请求否则模型会失去它自己刚决定要调用工具这个上下文后续对话可能变得混乱。第三max_rounds 一定要设置。没有轮次上限的话如果模型陷入查了结果不满意又查一次的循环你的 token 会像漏水一样流走。我一般设 6 到 8 轮足够覆盖大多数分析场景。4.4 System Prompt 里最容易忽略的细节很多人把精力全放在代码上prompt 随便写两句就跑了结果 Agent 行为飘忽。数据分析 Agent 的 System Prompt 有几个必须写清楚的约束你是一名数据智能分析助理连接 SQLite 数据库 sales.db。 规则 1. 只能执行 SELECT 查询禁止 DDL、DML、删除等任何写操作。 2. 写 SQL 之前必须先调用 list_tables 和 describe_table 确认表名和字段名不要凭印象猜。 3. 查询结果为空时如实说明原因不要编造数据。 4. 需要对比或观察趋势时调用 plot_chart 生成图表并把图表文件名告诉用户。 5. 回答时先给结论再用 Markdown 表格展示关键数据便于用户核验。 6. 你的每一步工具调用过程对用户可见不要隐藏执行细节。其中第二条我特别强调一下让模型先查表结构再写 SQL能干掉一大半字段名幻觉。大模型训练时见过太多不同结构的销售表它非常容易把amount写成sales_amount把order_date写成date。强制它先describe_table确认一遍代价只是多一轮工具调用收益是 SQL 准确率明显提升。还有第六条让模型把执行过程展示出来。这不仅是可解释性问题也是防幻觉的手段——用户能直接看到 SQL 原文就能判断结果是否可信而不是盲目相信模型生成的自然语言总结。4.5 看一个完整提问的执行过程我自己搭好后跑的第一个问题是这样6 月各渠道销售额是多少跟 7 月比变化怎么样模型的实际执行过程大致如下第一轮模型没有直接写 SQL而是先调用了list_tables()确认数据库里有sales表。第二轮调用describe_table(sales)拿到字段列表确认了金额字段叫amount日期字段叫order_date。第三轮调用run_sql(SELECT channel, SUM(amount) as total FROM sales WHERE order_date LIKE 2024-06% GROUP BY channel)。第四轮看到结果后再对 7 月执行同样的查询。第五轮把两份结果对比后直接输出文字结论并附上一张按渠道对比的柱状图。整个过程不涉及任何我的干预模型自己完成了先探路、再执行、后总结的完整闭环。看到这个执行链路的那一刻我才真正理解了 ReAct 模式在生产中的价值——它不是炫技而是把分析师查数的思维方式编码进了系统里。5. 上线前必须处理的四个稳定性问题5.1 模型幻觉字段名识别与约束第一个问题就是前面提到的字段幻觉。模型可能把channel写成channels把amount写成total_amount。这种情况run_sql会返回语法错误或字段不存在的报错模型看到报错后通常会自己修正但未必每次都成功。我的处理办法是三层保险Prompt 强制要求先describe_table再写 SQL从源头减少猜错的概率。工具层在返回结果时把实际列名透传给模型让模型看到真实字段。在工具执行前加一层 SQL 静态检查用sqlglot解析出查询涉及的表名和列名和information_schema里的实际名单做比对不一致直接返回字段 X 不存在可用字段有 Y、Z。import sqlglot def validate_sql(sql: str) - str: 检查 SQL 引用的表名和字段名是否存在不存在则返回提示。 try: parsed sqlglot.parse_one(sql, readsqlite) except Exception: return SQL 语法错误请检查后重试。 # 这里可以遍历 parsed.find_all(sqlglot.exp.Column) 和表名做白名单校验 return OK注意第三层校验是为了兜底不是为了替代前面两层。因为模型即使按照 describe_table 返回的字段名写 SQL仍可能在复杂查询中搞错别名引用关系。5.2 Token 成本被 Schema 信息吃光企业生产库的表经常有几十个字段如果每个表都全量把字段名塞进 prompt一轮对话光 schema 就可能吃掉上万 token成本完全失控。我的优化策略是把 schema 动态裁剪第一次让模型describe_table时只返回字段名和类型不返回注释和索引等噪音信息。建立一个 schema 缓存表记录用户问题里出现过的表 这些表最常用的字段下次同类问题直接命中缓存。在 System Prompt 里不写完整 schema只写表名列表让模型按需去查这样大部分轮次里实际传入的 schema 都很短。实际跑下来单次对话的 token 消耗降低了 60% 到 70%而且准确率没有下降因为模型总是只关注自己当前要用的那张表。5.3 权限与安全边界绝不给模型写的能力数据安全是数据类 Agent 的生命线这一点没有任何妥协空间。我在数据库层面做了四件事使用独立的只读账号连接数据库GRANT SELECT权限从源头禁止任何INSERT、UPDATE、DELETE、CREATE。在 SQL 执行入口加一层关键词拦截虽然只读账号已经防住了写操作但拦截能减少无效执行浪费资源。强制给 SELECT 加LIMIT上限防止模型生成SELECT * FROM huge_table这种把数据库压垮的查询。设置查询超时时间超时直接中断。SQLite 还好生产环境的 MySQL 或 ClickHouse 上这一步必不可少。如果 Agent 未来要执行任意 Python 代码比如做复杂的数据清洗一定要放在 Docker 容器或云函数等隔离环境里。让大模型生成的代码直接跑在宿主机上等于把服务器的钥匙交给了不可控对象这不是胆小而是基本的安全常识。5.4 结果的可信度问题Agent 输出一个结果很容易难的是让用户相信这个结果。模型在总结数据时如果数据集很小它可能随口说一个不准确的数字如果查询结果为空它可能会为了完成任务而编造一个看似合理的回答。我的经验是让 Agent 暴露完整证据链。用户应该能看到每一步工具调用的 SQL 原文、执行时间、返回行数。当查询结果为空时要求模型必须原样返回查询结果为空而不是自行发挥。当结果返回的行数很少时提示模型在回答里说明本次分析基于 N 条记录样本量较少结论仅供参考。另外我强烈建议在最终回答里同时展示 Markdown 数据表而不是只展示模型的自然语言总结。用户自己扫一眼数据表就能发现模型总结里的偏差信任度会高很多。6. 从可用到好用记忆、多 Agent 和评估6.1 记忆不只是聊天记录更是口径沉淀第一个版本跑通之后我发现用户体验的瓶颈不在查询速度而在于同一个口径问题反复出现。今天用户问销售额明天问营收其实都指向同一个指标但模型每次都要重新理解。Agent 的长期记忆最实用的落地方式是把常用口径做成一个配置文件注入 System Promptmetrics: 销售额: SUM(amount) WHERE order_statuspaid 毛利: SUM(amount - cost) 转化率: COUNT(DISTINCT user_id) 且有下单行为 / 总访问用户 aliases: 营收: 销售额 利润: 毛利模型每次启动时自动加载这个文件遇到同义表达就能对应到统一口径。这比向量数据库检索方案便宜得多也稳定得多90% 的场景用一份 YAML 就能解决不需要为了记忆上一套 RAG。6.2 什么时候才值得上 Multi-Agent很多人一聊到 Agent 就想到 Multi-Agent好像分工协作才是 Agent 的完整形态。我的观点是数据分析场景里单 Agent 工具分组是第一选择只有满足下面条件才考虑多 Agent任务有明确的分阶段边界比如清洗数据和生成报告是两个独立环节。单个 Agent 的工具列表太长超过 15 个模型经常选错工具可以考虑按工具用途拆给不同 Agent。需要人工审批节点比如某类查询必须经过审核才能执行。真要多 Agent最轻量的方式是主 Agent 子 Agent结构主 Agent 负责任务拆解和汇总子 Agent 各自只负责一个子域每个子 Agent 都有自己的工具集和 prompt。我之前试过用 LangGraph 搭过一版能力很强但调试成本确实高建议先把单 Agent 稳定跑一阵子确认瓶颈之后再迁移。6.3 效果评估用一组黄金问题防止退化Agent 的行为天然不稳定同一个问题换个说法或者模型版本升级输出可能就变了。为了不让它在看似能用之后悄悄退化我维护了一份 golden set。做法是准备 30 到 50 个业务真实问题每个问题配上期望使用的工具序列和期望的答案要点。每次改动 prompt 或工具定义后把这批问题全部跑一遍看成功率和准确率是否下降。我用一个简单的表格记录问题期望工具是否成功答案是否准确备注6月各渠道销售额run_sql → plot_chart是是-本月毛利环比run_sql × 2是否环比计算有误同时监控三个指标工具调用失败率、SQL 语法错误率、达到 max_rounds 的会话占比。任何一个指标异常升高基本都能反查到最近的 prompt 修改或模型版本变化。我个人最大的体会是把 Agent 从能跑做到能稳定跑工作量一半都在工具设计和安全边界上而不是在模型调用上。不要一上来就追求复杂框架或炫酷的多 Agent 编排先用一个 300 行以内的最小闭环把真实业务问题跑通再逐步加上记忆、权限和评估。我团队里现在每天跑几十个数据问答被问到最多的已经变成这个口径到底怎么定义的而这恰恰是 Agent 帮你把隐性需求暴露出来的价值所在。