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 的核心区别在于——它不只是一个翻译官它是一个有工作流的执行者。我们设计的架构包含三层第一层意图识别与路由。拿到一句话后先判断用户是要查一个数还是做一段分析还是监控一件事。不同意图走不同的处理链路。第二层任务拆解。如果是分析类需求比如分析上个月用户留存下降的原因智能体会自动拆成多个子任务拉取上月留存数据按渠道、用户分层钻取对比前几个月趋势找到下降最严重的分群关联同期产品变更记录第三层工具调用与编排。每个子任务可能调用不同的工具——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_factorylist) # 依赖的前置任务 params: Dict[str, Any] field(default_factorydict) 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人机协作边界智能体不是全自动的。有些环节必须让人把关环节自动做人工确认意图理解自动识别展示理解摘要用户确认SQL 生成自动生成复杂查询展示 SQL用户可修改数据分析自动执行自动执行结论生成自动生成用户审核修改数据发送不出智能体用户手动发送我们的底线是智能体不能自动发送数据给任何人。生成的分析结果要先给用户看用户点发送才出去。这是数据安全的基本要求。决策3知识积累智能体每一次执行成功后会把需求描述 → 子任务拆解方案 → 执行结果记录下来。下次遇到类似需求时LLM 可以直接复用已有的拆解方案准确率和速度都大幅提升。这就相当于团队积累了一份常见分析需求的最佳执行计划知识库——新人来了也能直接上手。四、实际效果从 40 分钟到 5 分钟我们的数据智能体上线 3 个月后的数据指标之前人工之后智能体简单查询耗时10-15 分钟30 秒分析类需求耗时30-50 分钟3-5 分钟图表生成额外 10-15 分钟自动包含结论文档编写额外 20-30 分钟自动生成初稿人均日处理需求数5-8 个15-20 个省下来的时间不是用来摸鱼的而是用来看智能体生成的报告做更深入的思考和判断。人的价值从怎么算变成了算完之后该怎么看。五、总结数据智能体的本质不是替代分析师而是给每个分析师配一个不知疲倦的搭子。这个搭子帮你做所有机械性的工作——取数、清洗、算指标、画图、写初稿——让你把精力集中在这个数据说明了什么和下一步该做什么上。如果你也想搭一个建议顺序是先做简单查询智能体自然语言 → SQL → 执行验证可行性再加多步骤分析能力拆解 → 执行 → 汇总覆盖复杂需求最后加自动监控定时巡检 异常告警从被动到主动记住一个原则智能体的目标不是 100% 自动化而是让分析师从做数据变成看数据。我是朱大喜一个给自己写了一个 AI 搭子的数据分析师。你想好给你的工作流程配一个什么样的 AI 搭档了吗评论区聊聊你的 idea~