AI 数据智能体：让每个分析师都有一个数据分析搭子

一、分析师的一天

早上9:30，三个需求同时进来了：

• 运营："大喜，昨天活动的转化漏斗帮我拉一下，半小时后开会用"
• 产品经理："上周上线的改版 AB 实验结论有了吗？"
• 老板："这个季度的用户留存趋势图，加上同比环比，10 点前给我"

换成一年前，我的流程是这样的：打开三个 SQL 编辑器 → 分别写三段查询 → 切回 Excel/Python 画图 → 粘贴到文档 → 发飞书。一套流程走下来 40 分钟起。

后来我们自己搭了一个 AI 数据智能体，把上面这套流程从"手动拼接"变成了"一句话驱动"。

数据智能体不是 ChatBI。ChatBI 只管"把自然语言变成 SQL"，智能体管的是"从需求到结论"的完整链路。

flowchart LR A[用户自然语言需求] --> B{智能体路由} B -->|查询类| C[语义层翻译] C --> D[SQL 生成 + 执行] B -->|分析类| E[分析计划生成] E --> F[步骤1: 取数] F --> G[步骤2: 清洗] G --> H[步骤3: 计算指标] H --> I[步骤4: 可视化] I --> J[步骤5: 生成结论] B -->|监控类| K[定时调度] K --> L[自动巡检 + 异常告警] D --> M[格式化输出] J --> M L --> M M -->|数据| N[图表 + 表格] M -->|文字| O[分析结论 + 建议] style B fill:#7B68EE,color:#fff style M fill:#27AE60,color:#fff

二、智能体的核心架构

数据智能体和 ChatBI 的核心区别在于——它不只是一个"翻译官"，它是一个有工作流的执行者。

我们设计的架构包含三层：

第一层：意图识别与路由。拿到一句话后，先判断用户是要"查一个数"，还是"做一段分析"，还是"监控一件事"。不同意图走不同的处理链路。

第二层：任务拆解。如果是分析类需求（比如"分析上个月用户留存下降的原因"），智能体会自动拆成多个子任务：

1. 拉取上月留存数据
2. 按渠道、用户分层钻取
3. 对比前几个月趋势
4. 找到下降最严重的分群
5. 关联同期产品变更记录

第三层：工具调用与编排。每个子任务可能调用不同的工具——SQL 引擎、Python 脚本、可视化库、文档生成。智能体要做的是按依赖关系串起来。

# data_agent.py — 数据智能体核心引擎 import json from typing import Dict, List, Any, Callable from dataclasses import dataclass, field from enum import Enum class TaskType(Enum): """子任务类型枚举""" QUERY = "query" # SQL 查询 PYTHON = "python" # Python 分析脚本 VIZ = "visualization" # 图表生成 REPORT = "report" # 报告生成 @dataclass class SubTask: """子任务定义""" task_id: str task_type: TaskType description: str # 这个子任务做什么 depends_on: List[str] = field(default_factory=list) # 依赖的前置任务 params: Dict[str, Any] = field(default_factory=dict) result: Any = None # 执行结果 class DataAgent: """ 数据智能体：接收分析需求，拆解为子任务，串行/并行执行，汇总结果 核心设计理念： 1. 每种子任务类型有专门的执行器（单一职责） 2. 子任务之间通过 DAG（有向无环图）表达依赖关系 3. 无依赖的子任务可以并行执行 """ def __init__(self): # 注册各类型的执行器 # 设计成可插拔的：新的执行器可以随时注册 self.executors: Dict[TaskType, Callable] = { TaskType.QUERY: self._execute_query, TaskType.PYTHON: self._execute_python, TaskType.VIZ: self._execute_visualization, TaskType.REPORT: self._generate_report, } # 工具注册表：智能体可以调用的外部能力 self.tools = {} def analyze(self, user_request: str, context: Dict = None) -> Dict: """分析入口：接收自然语言，返回完整分析结果""" # 步骤1：意图识别 intent = self._classify_intent(user_request) # 步骤2：根据意图生成分析计划（子任务 DAG） plan = self._generate_plan(intent, user_request, context or {}) # 步骤3：按拓扑序执行子任务 results = self._execute_plan(plan) # 步骤4：汇总结果，生成最终输出 return self._synthesize_results(user_request, results) def _classify_intent(self, request: str) -> str: """意图分类：查询 / 分析 / 监控""" # 简化示例：实际用的是 LLM 分类 query_keywords = ['查', '多少', '是什么', '拉一下', '看一下'] analysis_keywords = ['为什么', '分析', '原因', '对比', '趋势'] monitor_keywords = ['监控', '报警', '定时', '关注'] request_lower = request.lower() for kw in analysis_keywords: if kw in request_lower: return 'analysis' for kw in monitor_keywords: if kw in request_lower: return 'monitor' return 'query' def _generate_plan(self, intent: str, request: str, context: Dict) -> List[SubTask]: """根据意图生成子任务执行计划 这是智能体最核心的能力：把一个模糊的需求拆成可执行的步骤 举例："分析上月用户留存下降原因" → [ SubTask(取上月留存数据), SubTask(取前三个月留存的基线数据, 依赖=无), SubTask(按渠道/层级 drill-down, 依赖=[取上月数据, 取基线数据]), SubTask(生成留存对比图表, 依赖=[drill-down]), SubTask(生成分析报告, 依赖=[图表]) ] """ # 实际场景用 LLM 生成计划 # 这里展示一个分析类需求的经典拆解模板 if intent == 'analysis': return [ SubTask("step_1", TaskType.QUERY, f"拉取{request}相关数据"), SubTask("step_2", TaskType.QUERY, "拉取同期基线数据用于对比", depends_on=[]), # 和 step_1 无依赖，可以并行 SubTask("step_3", TaskType.PYTHON, "多维度 drill-down 分析", depends_on=["step_1", "step_2"], params={"dimensions": ["channel", "user_level", "region"]}), SubTask("step_4", TaskType.VIZ, "生成趋势对比图", depends_on=["step_3"]), SubTask("step_5", TaskType.REPORT, "汇总分析结论和建议", depends_on=["step_4"]), ] # 查询类简单得多 return [ SubTask("step_1", TaskType.QUERY, f"查询{request}"), SubTask("step_2", TaskType.VIZ, "结果可视化", depends_on=["step_1"]), ] def _execute_plan(self, plan: List[SubTask]) -> Dict[str, Any]: """执行子任务 DAG 关键逻辑： 1. 找出所有"依赖已满足"的子任务 2. 并行执行它们 3. 等待完成，释放新的可执行任务 4. 直到所有任务完成 """ results = {} completed = set() while len(completed) < len(plan): # 找出所有依赖已满足的待执行任务 ready_tasks = [ task for task in plan if task.task_id not in completed and all(dep in completed for dep in task.depends_on) ] if not ready_tasks: # 死锁检测：如果没有可执行任务但还有未完成的， # 说明有循环依赖或依赖的任务失败了 remaining = [t.task_id for t in plan if t.task_id not in completed] raise RuntimeError(f"子任务死锁！无法执行的任务: {remaining}") # 执行所有就绪任务 for task in ready_tasks: executor = self.executors[task.task_type] # 将前序任务的结果注入当前任务上下文 # 比如 step_3 需要 step_1 和 step_2 的查询结果作为输入 task_context = { dep: results[dep] for dep in task.depends_on if dep in results } task.result = executor(task.params, task_context) results[task.task_id] = task.result completed.add(task.task_id) return results def _execute_query(self, params: Dict, context: Dict) -> Any: """执行 SQL 查询""" # 实际实现会调用语义层翻译 + 执行引擎 sql = params.get('sql', '') return {"status": "ok", "data": f"执行 {sql} 的结果"} def _execute_python(self, params: Dict, context: Dict) -> Any: """执行 Python 分析脚本""" return {"status": "ok", "data": "Python 分析完成"} def _execute_visualization(self, params: Dict, context: Dict) -> Any: """生成图表""" return {"status": "ok", "chart_url": "base64_image_data"} def _generate_report(self, params: Dict, context: Dict) -> Any: """生成分析报告""" return {"status": "ok", "report": "分析报告 Markdown 内容"} def _synthesize_results(self, request: str, results: Dict) -> Dict: """汇总子任务结果，生成面向用户的最终输出""" return { "request": request, "summary": "分析完成", "details": results, "charts": [], }

三、关键设计决策

做一个能用的数据智能体，关键设计决策就这么几个：

决策1：子任务粒度

太粗了，一个"分析"就是一个任务，智能体就成了 ChatBI。太细了，"取数"要拆成"选表→选字段→过滤→聚合"，调度开销太大。

我们的经验是：一个子任务 = 一个工具的一次调用 + 前后 2-3 步预处理。比如"按渠道 drill-down"是一个子任务，包含取渠道维表、分组聚合、排序 Top N。

决策2：人机协作边界

智能体不是全自动的。有些环节必须让人把关：

我们的底线是：智能体不能自动发送数据给任何人。生成的分析结果要先给用户看，用户点"发送"才出去。这是数据安全的基本要求。

决策3：知识积累

智能体每一次执行成功后，会把"需求描述 → 子任务拆解方案 → 执行结果"记录下来。下次遇到类似需求时，LLM 可以直接复用已有的拆解方案，准确率和速度都大幅提升。

这就相当于团队积累了一份"常见分析需求的最佳执行计划"知识库——新人来了也能直接上手。

四、实际效果：从 40 分钟到 5 分钟

我们的数据智能体上线 3 个月后的数据：

省下来的时间不是用来摸鱼的，而是用来看智能体生成的报告，做更深入的思考和判断。人的价值从"怎么算"变成了"算完之后该怎么看"。

五、总结

数据智能体的本质不是替代分析师，而是给每个分析师配一个不知疲倦的搭子。这个搭子帮你做所有机械性的工作——取数、清洗、算指标、画图、写初稿——让你把精力集中在"这个数据说明了什么"和"下一步该做什么"上。

如果你也想搭一个，建议顺序是：

1. 先做"简单查询智能体"（自然语言 → SQL → 执行），验证可行性
2. 再加"多步骤分析"能力（拆解 → 执行 → 汇总），覆盖复杂需求
3. 最后加"自动监控"（定时巡检 + 异常告警），从被动到主动

记住一个原则：智能体的目标不是 100% 自动化，而是让分析师从"做数据"变成"看数据"。

我是朱大喜，一个给自己写了一个 AI 搭子的数据分析师。你想好给你的工作流程配一个什么样的 AI 搭档了吗？评论区聊聊你的 idea~