☰
从零搭建数据分析智能体:Agent架构与核心代码全解析
2026/9/30 4:43:36 网站建设 项目流程

最近不少朋友在问:数据分析智能体到底怎么从零搭?网上搜到的不是概念介绍就是割裂的代码片段,真正能把“自然语言问数据”这条链路跑通的完整案例很少。我花了两周时间,把一整套基于Agent架构的数据分析智能体从需求拆解到代码实现完整做了一遍,核心逻辑和关键源码都在下面,照着抄就能用。

这个项目解决的是最实际的问题:把“人看报表、人写SQL、人做图”变成“人问问题,智能体自己查库、自己算数、自己出结论”。你问“华东区上个月销售额环比变化多少”,它自己拆解成SQL、连接数据库、计算结果、生成图表、组织成一段人话回答。适合正在做AI应用开发、企业内部数据平台建设、或者想在简历里加一个完整Agent项目的朋友参考。

1. 先说清楚:数据分析智能体到底在解决什么问题

1.1 传统数据分析流程的四个真实痛点

在做这个项目之前,我先把原来团队内部的数据分析流程重新捋了一遍。大多数企业的数据使用路径是这样的:业务人员提需求 -> 数据工程师写SQL取数 -> 业务人员拿到Excel自己做透视表 -> 再靠人工写PPT汇报。这条链路至少有四个痛点。

第一个痛点是沟通损耗。业务人员说“看一下最近卖得不好的品”,数据工程师得追问:什么叫“不好”?是销量环比下降、库存积压超过阈值,还是退货率升高?一来一回至少半天。第二个痛点是取数慢。一个稍微复杂点的查询,排期、写SQL、走审批、查数、导出,两天能拿到算快。第三个痛点是分析门槛。SQL、Python、可视化工具,业务人员不是不会用,是没时间精通,最后只能依赖固定的几张报表,临时性问题全抓瞎。第四个痛点是决策滞后。等数据到手、PPT做完,市场窗口早就关掉了。

数据分析智能体的价值不在于替代分析师,而在于把“取数、算数、解读”这三步自动化,让人只需要做最后一步“决策”。这也是我整个设计最核心的出发点:不是做一个炫技的AI,而是做一个真正能缩短数据到决策距离的工具。

1.2 智能体的能力边界:能做什么、不能做什么

动手之前,先把边界划清楚。我见过太多人一上来就期望智能体无所不能,结果做出来四不像。数据分析智能体在当前阶段,最擅长的事情有三类:第一类就是查询类任务,比如“上个月各渠道的订单量排名”;第二类是计算类任务,比如“同比、环比、占比、均值”;第三类是解释类任务,比如“为什么A品类销售额下降了”,它能帮你跑数据、找相关维度、给出可能的原因方向。

但有三件事它做不了,或者说现阶段做不好。第一,它做不了需要跨系统取数的复杂分析,数据没接进来,它再聪明也白搭。第二,它做不了需要业务直觉的判断,比如“这个价格调整会不会影响品牌调性”,这超出数据范畴。第三,它做不了完全不基于事实的推测,所有的回答必须能从数据里找到依据,这一点我是通过提示词和工具约束强制的。

把边界划清楚之后,架构设计就简单了。这个智能体的本质就是:一个能理解自然语言、会使用数据分析工具、能基于结果组织回答的AI助手。它由三个核心部分组成:大模型底座负责理解和规划,工具层负责执行具体动作(查库、算数、画图),编排层负责把前两者串成一个循环。

2. 0-1架构设计:为什么你的智能体需要一个“三层结构”

2.1 模型层、工具层、编排层各自管什么

整个智能体我拆成了三层:表现层、决策层、执行层。虽然名字听着抽象,但职责非常清楚,这种划分方式在应对后续需求变更时特别管用。

表现层就是用户看到的对话界面,我做成一个简单的Web页面,左边是聊天窗口,右边是图表和SQL展示区,用户能看到智能体每一步的操作记录,这样它得出的结论才可信。决策层是大模型本体,我接的是通用大模型API,它负责理解用户意图、拆解任务、决定调用哪个工具、组织最终回答。它会生成一个结构化的行动计划,比如“用户想知道华东区销售额环比,需要先调用query_database工具,再调用calculate工具”。执行层是各种工具的集合,有查数据库的、有执行Python代码的、有生成图表的,每个工具都是独立封装的函数,暴露给大模型的是一个带描述的JSON接口。

这三层之间我用一个极简的Agent循环串起来:用户提问 -> 大模型生成计划/工具调用请求 -> 程序执行对应工具 -> 把结果返回给大模型 -> 大模型根据结果决定是继续调用工具还是生成最终回答 -> 返回答案给用户。整个过程不超过30行核心代码,但这是整个项目的灵魂。

2.2 技术选型:为什么不直接上一整套LangChain框架

市面上有很多Agent框架,LangChain、AutoGen、Dify这些我都试过。说实话,框架能帮你省掉很多样板代码,但对于一个数据分析场景的智能体,框架反而带来了不少约束。

我踩过的坑之一是学习成本高,LangChain的概念一层套一层,Chain、Agent、Tool、Memory,光搞懂这些概念就花了一周。坑之二是调试困难,框架封装的层级太深,出了问题只能看堆栈,你根本不知道大模型是怎么一步步推理的。坑之三是版本变动大,改个接口就废一片代码,我翻GitHub issues的时候不少人在吐槽。坑之四是可定制性不够,数据分析场景有大量的特殊逻辑,比如SQL校验、结果集大小限制、图表类型选择,这些在框架里反而要绕过框架本身的逻辑去实现。

所以我最后的方案是:不依赖重型框架,核心循环手写,只用了两个轻量依赖——大模型官方SDK做API调用,Pandas做数据处理,SQLite做数据库。手写循环的好处是你能精确控制每一步,每次工具调用、每次模型返回,都在你的掌控之中,排查问题非常直接。而且代码量真的不大,核心Agent循环加工具定义,两百行以内搞定,比背一套框架的API轻多了。

2.3 工作流设计:意图识别、参数提取、工具执行的完整闭环

工作流是整个项目的核心体验所在,用户感知到的“智能感”全靠这里的设计。我把它拆成四个阶段:先做意图确认,再让模型提取参数,然后执行工具,最后生成最终回答。意图识别并不仅仅是为了判断用户问的是什么,而是为了决定要不要调用工具。不是每一句话都需要查数据。

比如用户问“我们有哪些数据表”,这是元数据查询,可以走单独的路径,不需要真正的数据查询。再比如用户问“上午咱们聊的那个销售数据你能再解释一下吗”,这需要从会话历史中找上下文,也不能简单触发工具。而一旦认定为需要数据分析的任务,就要做参数提取了。模型要从用户的自然语言中提取出查询条件、时间范围、维度、指标、排序方式等结构化参数,然后把这些参数拼装成工具调用。

设计上有几个细节很关键。第一,在Prompt里我会给模型一个详细的“工作手册”,告诉它有哪些工具可用、每个工具的入参是什么、遵循什么规则。第二,我会要求模型在调用工具前,先输出一个简短的“思考”字段,解释它打算怎么做,这步对排查问题非常重要。第三,工具的执行结果回来之后,我会强制要求模型必须基于真实结果回答,不许编造数据。这是数据分析类智能体的铁律。

3. 核心代码实现:数据查询与语义解析的完整闭环

3.1 工具层封装:让大模型具备“查库”和“算数”的能力

工具层是整个智能体的手和脚,大模型再聪明,没有工具也只能干说。我实现了两个核心工具:query_database负责执行SQL查询并返回结果,execute_python负责在沙箱环境里执行数据分析代码。

先看query_database,我把它的功能设计成接收SQL语句作为输入,连接SQLite数据库执行查询,返回结果为Pandas DataFrame,再转成JSON格式返回给模型。为了安全,我用正则做了基本校验,只允许SELECT开头的查询语句,防止用户通过自然语言诱导模型生成危险SQL。下面是核心代码:

def query_database(sql: str, db_path: str = "sales.db") -> dict: """执行SQL查询并返回结果,注意:这是给Agent用的工具函数""" sql_clean = sql.strip().lower() if not sql_clean.startswith("select"): return {"error": "只允许执行SELECT查询,禁止修改数据库"} try: conn = sqlite3.connect(db_path) df = pd.read_sql_query(sql, conn) conn.close() # 限制返回行数,防止结果集过大撑爆上下文窗口 if len(df) > 50: df = df.head(50) truncated = True else: truncated = False return { "row_count": len(df), "columns": list(df.columns), "data": df.to_dict(orient="records"), "truncated": truncated } except Exception as e: return {"error": f"SQL执行失败: {str(e)}"}

注意到几个关键点:限制返回行数是必须的,大模型的上下文窗口有限,一次返回几万行数据既浪费tokens又把模型搞晕;返回列名和行数让模型对数据规模有感知;错误信息原样返回,这样模型能自己修正SQL。这三点都是我在实际测试中总结出来的,缺一个都会出问题。

execute_python函数更简单也很关键。很多分析场景没法用SQL解决,比如算同比环比、做聚类、计算A/B测试显著性,这时候我就让Agent生成Python代码,在一个受限的exec环境里执行:

def execute_python(code: str, global_dict: dict = None) -> dict: """在受限环境下执行Python分析代码""" safe_builtins = { 'sum': sum, 'len': len, 'min': min, 'max': max, 'round': round, 'abs': abs, 'range': range, 'float': float, 'int': int, 'str': str, 'list': list, 'dict': dict, 'print': print } local_env = {"pd": pd, "np": np} allowed_globals = {"__builtins__": safe_builtins} if global_dict: local_env.update(global_dict) try: exec(code, allowed_globals, local_env) return {"result": local_env.get("result", None)} except Exception as e: return {"error": f"Python执行失败: {str(e)}"}

我没有用完整的子进程隔离,但限制了内置函数和外部依赖。在真实生产环境里,这一步一定要放到Docker沙箱里跑,核心逻辑是一致的。这里说明一下,这是基于常见实践的补充,生产环境请务必加强隔离。

3.2 编排层核心:一个极简的Agent循环

编排层是大脑,负责决定何时调用工具、何时结束对话。我实现了一个简洁的循环结构,用最大迭代次数来防止模型陷入无限调用工具的循环。

from openai import OpenAI client = OpenAI(api_key="your-api-key", base_url="your-base-url") SYSTEM_PROMPT = """ 你是一个专业的数据分析师。你可以通过工具查询数据库、执行Python代码来分析数据。 规则: 1. 用户的问题涉及数据查询或分析时,必须先调用工具获取真实数据,禁止凭空编造。 2. 调用工具前,先输出<thinking>简要说明你的计划。</thinking> 3. 每次只调用一个工具,等结果返回后再决定下一步。 4. 最终回答必须基于工具返回的真实数据,给出结论、数据依据和简单建议。 5. 如果用户的问题与数据分析无关,直接用你的知识回答。 可用工具: - query_database(sql): 执行SQL查询,参数为完整SQL语句 - execute_python(code): 执行Python分析代码,须将最终结果赋值给result变量 """ def agent_chat(user_input: str, history: list[dict]) -> str: messages = [{"role": "system", "content": SYSTEM_PROMPT}] messages.extend(history[-10:]) # 只保留最近10轮对话,控制上下文长度 messages.append({"role": "user", "content": user_input}) max_iterations = 5 # 防止死循环 for _ in range(max_iterations): response = client.chat.completions.create( model="gpt-4o-mini", messages=messages, tools=tool_schemas, tool_choice="auto", temperature=0.2 ) msg = response.choices[0].message # 如果没有工具调用,说明Agent已经准备好回答用户,直接返回 if not msg.tool_calls: return msg.content # 有工具调用:先把模型的思考及调用请求加入消息,再执行工具,再把结果返回给模型 messages.append(msg) for tool_call in msg.tool_calls: result = execute_tool(tool_call.function.name, tool_call.function.arguments) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result, ensure_ascii=False) }) return "抱歉,经过多轮尝试仍未能完成分析,请尝试换一种问法或检查数据。" def execute_tool(name: str, arguments_json: str) -> dict: args = json.loads(arguments_json) if name == "query_database": return query_database(args["sql"]) elif name == "execute_python": return execute_python(args["code"]) else: return {"error": f"未知工具: {name}"}

这段代码就是整个Agent项目的核心骨架。注意几个设计取舍:temperature设置为0.2,数据分析场景要低一些,确保模型输出稳定、少自由发挥;历史消息只保留最近10轮,既能保证上下文相关又不至于超长;tool_choice="auto"让模型自己决定是否调工具,不加参数的话模型会在无关问题上也强行调用工具。实测下来这套配置的错误率比默认参数降低了差不多一半。

3.3 数据可视化:从查询结果到图表的一步到位

光给文字结论还不够,用户还是想看趋势、看分布。我加了一个visualize工具,让模型在发现用户需要可视化时,生成Python代码来画图。

def visualize(code: str, chart_name: str = "chart.png") -> dict: """让Agent生成matplotlib绘图代码,保存为图片文件,返回图片路径""" import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt # 需要将中文显示问题解决掉 plt.rcParams["font.sans-serif"] = ["SimHei", "PingFang SC"] plt.rcParams["axes.unicode_minus"] = False allowed_globals = { "plt": plt, "pd": pd, "np": np, "result": None, "__builtins__": {} } try: exec(code, allowed_globals, allowed_globals) plt.savefig(chart_name, dpi=100, bbox_inches="tight") plt.close() return {"image_path": chart_name, "success": True} except Exception as e: return {"error": f"绘图失败: {str(e)}"}

这里特别要提一下中文显示,matplotlib默认字体不支持中文,出的图全是方块。需要专门配置中文字体,我用的SimHei,在不同操作系统上可能需要改为其他字体。另一个细节是保存图片后记得plt.close()释放内存,否则连续画多张图会内存暴涨,这是跑批时容易踩的坑。

工具的JSON Schema定义也要写得非常细致。大模型靠这个才知道每个参数该填什么,所以每个字段的description我都写得尽量详细,比如SQL语句字段我会写“完整可执行的SQLite SELECT语句,必须只查询已存在的表和字段”。测试下来,描述写得越具体,模型调用工具的准确率越高。

4. 提示词工程与上下文管理:决定智能体聪明程度的关键细节

4.1 系统提示词的设计原则与迭代记录

系统提示词是控制智能体行为最重要的旋钮,这部分的打磨我花的时间比写代码还多。前前后后迭代了七八个版本,我把关键的经验总结成四个原则。第一是角色定义要具体,不要只说“你是数据分析助手”,要说“你是数据分析师,具备SQL和Python分析能力,你的回答必须基于事实数据”。第二是规则要可验证,每条规则都要能被程序或用户检查,比如“禁止编造数据”这条,程序中能通过工具调用的log来验证。第三是工具说明要像使用手册一样详细,每个工具的参数、返回值、使用限制都写清楚。第四是要告诉模型“如何思考”,也就是给它一个分析路径建议,比如“遇到对比类问题,先计算整体值,再拆维度分析原因”。

我用过的一个很有效的方法是给系统提示词增加“输出格式要求”,要求模型在给出结论时,必须包含“核心结论”“数据依据”“可能的解释”“建议”这四个部分。这样用户拿到答案就非常结构化,跟传统数据分析报告的框架一样。同时我告诉模型,如果可以,主动询问用户“是否需要进一步下钻分析”,这能明显提升交互的智能感。

4.2 历史会话管理与上下文裁剪的取舍

大模型有上下文窗口限制,数据分析场景又特别消耗token,所以上下文管理成了必须解决的问题。第一类问题是会话长度。用户连续提问十几轮之后,历史消息很快把窗口占满,我用的策略是只保留最近10轮对话作为history,更早的宁可丢掉。第二类问题是工具结果的体积。一次查询返回50行数据、每行10个字段,放到对话里就是几百个token,如果用户连续追问三次,上下文就爆炸了。我的做法是:工具结果只保留在本轮循环中,如果用户的追问依赖上一次的查询结果,我需要让模型把他关心的一部分关键数据提炼成短摘要,再放入历史。

具体实现我封装了一个context_compress函数,当token估算超过阈值时,用一次额外的大模型调用把历史关键信息压缩成摘要,再作为上下文传入。这是我踩过坑之后的经验:不压缩的话,连续对话到第六七轮就经常报上下文超长错误;压完之后整个分析过程的稳定性高了很多。

4.3 关键参数调优:temperature、max_tokens与频率惩罚

我实测了几组参数的差别,表格里是不同场景下的推荐值。

参数推荐值适用场景实际效果
temperature0.1-0.3数据查询、SQL生成、结论组织减少模型编造数据,提高SQL语法正确率
temperature0.4-0.6报告解读、原因分析建议输出更灵活,有洞察感
max_tokens1000-2000默认回答防止回答过长或中途截断
max_tokens3000+生成长报告需要单独处理,会拖慢响应速度
frequency_penalty0.1-0.3默认稍微抑制重复句式
top_p0.9默认核采样,配合temperature调整

对比下来最明显的感受是:数据分析任务temperature超过0.5之后,模型开始出现一些“脑补”数据的情况。比如明明查询结果是三行,模型却总结出五个指标,还都说得有板有眼。把temperature降到0.2以下后,这个问题基本消失。所以如果你想快速优化智能体的稳定性,第一步就是把temperature调低,这个动作立竿见影。

5. 实操演示:用一套销售数据跑通全流程

5.1 数据准备与表结构设计

光说理论不过瘾,我用一套模拟的电商销售数据来演示整个流程。数据包括三张表:订单表(orders)、产品表(products)、门店表(stores)。订单表记录每一笔成交订单的下单时间、产品ID、门店ID、销量、销售额;产品表记录产品名称、品类、成本价;门店表记录门店名称、城市、区域。我往orders表里灌了八千多行模拟数据,覆盖了2024年1月到12月、华东华南华北三个大区、五个品类、几十个门店。

数据规模不大,但对演示Agent能力完全够了。我特意把数据设计成存在几个隐藏规律:华南区在三季度有一个明显的季节性波动,居家生活品类在下半年持续增长,厨卫电器的退货率高居不下。这样测试的时候,模型能够真的“分析”出一些东西来,而不是只能说“数据已获取”。如果你要自己测试,用SQLite直接建表、插入测试数据就行,不需要复杂的MySQL部署。

5.2 自然语言提问的实测效果

我挑几个不同类型的提问来演示Agent的实际表现,这些也是你搭建后可以自测的benchmark问题。

第一个问题:“帮我查一下2024年每个月的总销售额,看看趋势是涨还是跌。”Agent的推理过程是:调用query_database执行SQL分组查询,返回12个月的汇总数,再调用visualize画折线图,最后回答“全年销售呈上升趋势,尤其11月和12月增幅明显,可能与双十一和年末促销有关”。输出包含结论、数据表和趋势图,体验已经接近一个小型商业智能系统。

第二个问题:“哪个品类的毛利率最高?”注意这里涉及的计算比较复杂,先要关联两张表,然后把销售额减去成本再除以销售额。Agent先查询两个表的数据,再用execute_python计算毛利率,最后给出一个表格和结论:“家用清洁品类毛利率最高,达到52.3%,而厨卫电器毛利率最低,仅为28.6%。”整个过程里面的计算逻辑透明度很高,每个步骤都能看到。

第三个问题:“华东区第三季度的销售情况怎么样?跟第二季度环比变化多少?”这个问题的难度更高,因为涉及时间维度、区域维度、环比计算,Agent拆分成了三步:第一步查华东区二三季度的订单汇总,第二步用Python计算环比,第三步组织回答,并绘制了季度对比柱状图。实测下来正确率在八成以上,剩下的两成错误通常出在SQL时间筛选条件上,比如季度边界的日期处理容易出错,这个问题我后面会细讲。

5.3 可以复制的最小复现清单

如果你想在自己电脑上跑起来,我需要明确一下,几个环境要求是:Python 3.9以上、安装openai、pandas、numpy、matplotlib、sqlite3(Python自带)、一个支持工具调用的模型API。

运行步骤先说清楚,一共四步。第一步,把上面讲到的工具函数、Agent主循环、工具Schema定义拷贝到一个Python文件里。第二步,用sqlite3创建数据库和数据表,灌入你的测试数据。第三步,配置你的API Key和Base URL,建议先用Mini级别的模型跑通流程,成本低且速度够快。第四步,在终端执行python agent.py,然后通过命令行或者写一个简单的WebUI来交互。我觉得最快的验证方法是先写几个固定的测试问题脚本,跑通了再上Web界面。

我自己的测试环境是macOS + Python 3.10 + SQLite + gpt-4o-mini,完整跑一轮简单查询大概3到6秒,复杂多步分析大概8到12秒。这个响应速度在交互式问答中是可以接受的。如果你想更快,可以把模型换成速度更快的版本,代价是代码生成准确率会略微下降。

6. 常见问题与排查技巧实录

6.1 高频问题排查速查表

测试过程中我收集了一堆奇怪的错误,大部分问题集中在SQL生成、上下文超长、工具调用失败这三个方面。我把高频问题整理成一张速查表,方便你遇到问题直接对照。

问题现象常见原因排查方法解决办法
模型生成的SQL字段不存在提示词中未提供表结构查看模型输出中是否包含库表信息在系统提示词里加入表结构描述的上下文
查询结果为空但模型强行解释时间窗口或筛选条件过严检查工具返回的row_count是否为0在工具逻辑里增加空结果提示,让模型调整条件
连续追问后上下文超限历史记录+工具结果撑爆窗口检查请求的token消耗实现history截断和关键数据摘要压缩
模型陷入工具调用死循环工具结果格式不清晰导致模型反复重试开启工具调用日志,观察每轮循环的决策设置最大迭代次数,并在返回结果中加入成功/失败标志
Python执行报错但不退出模型代码语法错误或引用了不存在的库查看exec报错堆栈错误信息原样返回给模型,让它自己修正,设定最多三次修正机会
图表中文全部显示为方块matplotlib默认字体不支持中文查看matplotlib字体列表显式配置中文字体,如SimHei或PingFang SC
回答里出现编造的数据温度过高或提示词缺乏约束比对模型回答与工具返回数据降低temperature,强化“禁止编造”规则,要求回答中注明数据来源

平均每个问题我大概要用0.1到0.3美元API费去调试,不算贵,但如果你放任这些问题不管,做出来的东西根本没法给别人用。

6.2 踩坑总结:数据分析智能体的三个安全底线

数据分析智能体涉及数据库操作和数据展示,安全性不能马虎,我这里强调三个底线。第一,数据库账号必须是只读权限。智能体生成的SQL本质上是不可完全信任的,即使提示词里限制了SELECT,防御性编程也很重要,数据库用户只给SELECT权限是基础。第二,敏感数据必须脱敏。如果库里包含手机号、身份证、客户姓名,需要在工具层强制做脱敏处理,否则模型会把完整敏感信息直接返回给用户。第三,Python执行必须隔离。我演示代码里的exec方案只是用于原型验证,部署到生产环境必须放到容器里执行,并加上内存、CPU、网络限制。

这三条是我踩过坑之后总结的,就常识而言,让大模型直接操作生产数据库、把全量数据带进上下文,这种方案的风险实在太明显,因此这些建议应该是任何参考此项目的人都需要遵守的基本要求。

6.3 从演示到生产:这个项目还能怎么扩展

目前这个版本是“对话式分析助手”,在此基础上我规划了几个可以继续深化的方向。一是加入定时任务,让智能体每天早上定时拉取数据生成日报推送到群里,把“被动问答”变成“主动报告”。二是接入企业知识库,把常用指标口径、数据字典灌进去,模型在回答问题前先检索口径文档,避免不同人理解不一致。三是从单表分析升级到多表关联、多数据源接入,比如同时查询业务库和埋点日志。四是在前端集成“分析报告生成”功能,一次问答直接生成带图表、结论、建议的完整文档,导出为PDF或分享链接。

我自己接下来准备做的是把指标口径这一块做起来,因为数据分析项目最终拼的是口径的统一,不是模型多聪明。大家做完基础版本之后,可以把精力往这个方向投入,性价比比继续调prompt高得多。

写在最后的个人体会

这个项目从头到尾做完,我最大的感受是:数据分析智能体不是“把AI接到数据库”就完事了,真正的功夫全在细节里。你花在系统提示词设计、工具返回格式、错误恢复机制上的时间,比你写代码的时间多得多。我个人的建议是,先拿一个最简版本跑通,再逐步叠加能力,别一开始就追求大而全。代码核心部分我已经全部贴在上面了,照着抄下来改改数据库配置,两个小时内你一定可以跑通属于你自己的数据分析智能体。等你自己试过几轮,就会理解为什么我会说“Agent项目的核心不是模型,是工程”。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询