12.3 数据分析 Agent

上一节我们构建了代码生成 Agent——它能规划需求、写代码、跑测试、自我修复。如果把视角从"生成代码"转到"查询数据"，就来到了数据分析 Agent 的领域。代码生成 Agent 的产物是代码，数据分析 Agent 的产物是洞察。前者关心"代码能不能跑通"，后者关心"数据说了什么"。

数据分析 Agent 的核心能力是 Text-to-SQL——让不懂 SQL 的业务人员也能用自然语言查询数据库。但"能查"只是起点，还要"查得安全""查得准""看得懂"。本节我们就来拆解这套从自然语言到数据洞察的完整流水线。下一节我们将进入多 Agent 协作系统，看看当单个 Agent 不够用时，如何让多个 Agent 分工合作。

生活类比：图书馆参考咨询台

想象你走进一座大型图书馆，面对几百万册图书不知所措。你走到参考咨询台，对馆员说："我想找去年畅销的科幻小说。"馆员不会让你自己去翻卡片目录，而是先问你几个澄清问题——"中文还是英文？""科幻还是奇幻？"——然后帮你把查询翻译成图书馆的分类编码，去书库里取出书，最后再给你做个简单的读书笔记。

数据分析 Agent 就是这座图书馆的"数字馆员"。它做四件事：把你的问题翻译成数据库能懂的查询语言（Text-to-SQL）、执行查询取数据（安全执行）、分析数据给你洞察（数据分析）、画张图帮你直观理解（数据可视化）。区别在于，数据库比图书馆大得多，查询语言比分类编码复杂得多，而且安全风险也高得多——一不小心就可能泄露用户隐私或被 SQL 注入攻击。

一、Text-to-SQL：从自然语言到数据查询

Text-to-SQL 是数据分析 Agent 最核心的能力。它的本质是把自然语言（"上个月销售额最高的 5 个产品"）翻译成数据库能执行的 SQL（SELECT product_name, SUM(amount) FROM orders ... LIMIT 5）。

这个翻译过程需要三样东西：表结构描述（让 LLM 知道有哪些表、哪些列）、查询规则约束（只准 SELECT、不准删数据）、安全验证（生成后检查有没有危险操作）。下面的 TextToSQL 类把这三件事封装在了一起，每一行都加了注释：

from typing import List, Dict, Optional          # 类型注解
from dataclasses import dataclass                # 数据类装饰器
import sqlite3                                   # SQLite数据库接口
import re                                        # 正则表达式

@dataclass
class TableSchema:                               # 表结构定义
    """表结构定义"""                              #   描述一张表的元信息
    name: str                                     #   表名，如"orders"
    columns: List[Dict[str, str]]                 #   列信息列表
    description: str                             #   表的用途说明
    sample_data: List[Dict] = None               #   示例数据，帮助LLM理解

class TextToSQL:                                  # Text-to-SQL转换器
    """Text-to-SQL转换器"""                       #   把自然语言转成安全SQL

    SCHEMA_PROMPT = """你是一个SQL专家。请根据以下数据库表结构，将自然语言问题转换为SQL查询。

## 数据库表结构
{schema_text}

## 查询规则
1. 只生成SELECT查询，禁止INSERT/UPDATE/DELETE/DDL
2. 使用参数化查询的占位符格式
3. 如果问题无法用SQL表达，返回：UNSUPPORTED
4. 只输出SQL语句，不要额外解释

## 自然语言问题
{question}

SQL查询："""

    def __init__(self, llm_client):               # 初始化转换器
        self.llm = llm_client                     #   保存LLM客户端
        self.schemas: Dict[str, TableSchema] = {}  # 表结构注册表

    def register_schema(self, schema: TableSchema):  # 注册表结构
        """注册表结构"""                           #   把表结构存起来供后续使用
        self.schemas[schema.name] = schema         #   以表名为键存储

    def _build_schema_text(self, relevant_tables: List[str] = None) -> str:
        """构建Schema描述文本"""                   #   把表结构格式化成文本
        tables = relevant_tables or list(self.schemas.keys())  # 指定表或全部
        schema_parts = []                         #   收集各表的描述
        for table_name in tables:                 #   遍历每张表
            schema = self.schemas.get(table_name)  #   取出表结构
            if not schema:                        #   没注册过就跳过
                continue
            columns_text = "\n".join([            #   把每列格式化成一行
                f"  - {col['name']} ({col['type']}): {col.get('description', '')}"
                for col in schema.columns         #     遍历所有列
            ])
            schema_parts.append(f"### 表：{table_name}\n描述：{schema.description}\n列：\n{columns_text}")
        return "\n\n".join(schema_parts)          #   用空行连接各表描述

    async def generate_sql(self, question: str, relevant_tables: List[str] = None) -> Dict:
        """生成SQL查询"""                         #   主入口：自然语言->SQL
        schema_text = self._build_schema_text(relevant_tables)  # 构建表结构文本
        prompt = self.SCHEMA_PROMPT.format(schema_text=schema_text, question=question)  # 填模板
        response = await self.llm.chat(prompt)     #   调LLM生成SQL
        sql = self._extract_sql(response)         #   从回复中提取SQL
        return {"sql": sql, "question": question, "is_valid": self._validate_sql(sql)}

    def _validate_sql(self, sql: str) -> tuple[bool, str]:
        """SQL安全检查"""                         #   生成后必须验证
        if sql == "UNSUPPORTED":                  #   LLM明确表示无法转换
            return True, ""                        #     这是合法的
        sql_upper = sql.upper().strip()            #   转大写便于匹配
        forbidden = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "CREATE", "TRUNCATE"]
        for keyword in forbidden:                 #   逐个检查危险关键字
            if sql_upper.startswith(keyword):     #   以危险关键字开头
                return False, f"禁止执行 {keyword} 操作"
        dangerous_functions = ["SLEEP(", "BENCHMARK(", "LOAD_FILE(", "INTO OUTFILE"]
        for func in dangerous_functions:           #   检查危险函数
            if func.upper() in sql_upper:         #   SQL中包含危险函数
                return False, f"禁止使用危险函数 {func}"
        return True, ""                            #   通过所有检查

    def _extract_sql(self, text: str) -> str:
        """从LLM回复中提取SQL"""                  #   LLM经常把SQL包在```里
        match = re.search(r'```(?:sql)?\s*\n(.*?)```', text, re.DOTALL)  # 先找代码块
        if match:                                 #   找到了
            return match.group(1).strip()         #     返回代码块内容
        return text.strip()                       #   没代码块就返回原文

这个转换器的关键设计在于安全优先。注意 Prompt 里明确写了"只生成 SELECT 查询，禁止 INSERT/UPDATE/DELETE"——这是第一道防线。但 LLM 不一定听话，所以 _validate_sql 方法做了第二道防线：检查生成的 SQL 是否以危险关键字开头、是否包含危险函数。两道防线都过了，才会真正执行。

二、数据分析 Agent 架构

有了 Text-to-SQL 转换器，下一步是把它组装成一个完整的数据分析 Agent。这个 Agent 的工作流分四步：生成 SQL -> 执行查询 -> 分析结果 -> 建议可视化。就像图书馆馆员先帮你查目录编码、再进书库取书、然后给你讲讲这本书讲了什么、最后画个思维导图帮你理清脉络。

import asyncio                                      # 异步IO
import pandas as pd                                 # 数据处理库
from typing import Any, Callable, List, Dict, Optional  # 类型注解

class DataAnalysisAgent:                            # 数据分析Agent
    """数据分析Agent"""                              #   端到端的分析流水线

    def __init__(self, llm_client, db_connection, text_to_sql: TextToSQL):  # 依赖注入
        self.llm = llm_client                       #   LLM客户端
        self.db = db_connection                     #   数据库连接
        self.tts = text_to_sql                     #   Text-to-SQL转换器
        self.analysis_history: List[Dict] = []      #   分析历史记录

    async def analyze(self, question: str) -> Dict:  # 主入口
        """执行数据分析"""                           #   自然语言->SQL->数据->洞察
        result = {                                   #   初始化结果字典
            "question": question,                   #     原始问题
            "sql": None,                            #     生成的SQL
            "data": None,                           #     查询结果
            "analysis": None,                       #     分析洞察
            "visualization_suggestion": None,       #     可视化建议
            "error": None                            #     错误信息
        }
        try:
            # Step 1: Text-to-SQL 生成查询
            sql_result = await self.tts.generate_sql(question)  # 调转换器
            if not sql_result["is_valid"][0]:       #   SQL验证失败
                result["error"] = f"SQL验证失败：{sql_result['is_valid'][1]}"
                return result                        #   直接返回错误
            result["sql"] = sql_result["sql"]       #   存SQL
            if sql_result["sql"] == "UNSUPPORTED":  #   LLM说无法转换
                result["analysis"] = "该问题无法通过SQL查询回答，请尝试其他方式"
                return result                        #   返回说明

            # Step 2: 安全执行查询
            data = await self._execute_query(sql_result["sql"])  # 只读模式执行
            result["data"] = data                   #   存数据
            if data is None or len(data) == 0:      #   没查到数据
                result["analysis"] = "查询未返回任何数据"
                return result                        #   返回空结果提示

            # Step 3: 数据分析与洞察生成
            analysis = await self._analyze_data(question, data, sql_result["sql"])  # 调LLM分析
            result["analysis"] = analysis           #   存分析结果

            # Step 4: 可视化方案建议
            viz = await self._suggest_visualization(data, question)  # 调LLM建议图表
            result["visualization_suggestion"] = viz  # 存可视化建议
        except Exception as e:                     #   捕获异常
            result["error"] = str(e)                #   存错误信息
        self.analysis_history.append(result)        #   记录到历史
        return result                                #   返回完整结果

    async def _execute_query(self, sql: str) -> Optional[pd.DataFrame]:
        """安全执行SQL查询"""                       #   只读模式，防止误操作
        try:
            if hasattr(self.db, 'execute'):         #   SQLite数据库
                self.db.execute("PRAGMA query_only = ON")  # 开启只读模式
                df = pd.read_sql_query(sql, self.db)  # 执行查询，返回DataFrame
                self.db.execute("PRAGMA query_only = OFF")  # 关闭只读模式
            else:                                   #   其他数据库
                df = pd.read_sql_query(sql, self.db)  # 直接执行
            return df                               #   返回DataFrame
        except Exception as e:                     #   执行出错
            raise RuntimeError(f"SQL执行错误：{str(e)}")  # 包装错误抛出

注意 _execute_query 中的 PRAGMA query_only = ON——这是 SQLite 的只读模式开关，开启后即使 SQL 里有 DELETE 或 DROP 也会被拒绝执行。这是第三道安全防线，防止前两道防线都被绕过时造成不可逆的破坏。

_analyze_data 方法把查询结果喂给 LLM，让它生成人类可读的分析报告。关键技巧是先做数据摘要（行数、列名、数值统计、前 5 行样本），避免把全部数据塞进 Prompt 导致 Token 爆炸。

    async def _analyze_data(self, question: str, data: pd.DataFrame, sql: str) -> str:
        """分析查询结果"""                          #   把数据摘要+问题喂给LLM
        summary = {                                 #   生成数据摘要
            "shape": data.shape,                    #     (行数, 列数)
            "columns": data.columns.tolist(),       #     列名列表
            "dtypes": data.dtypes.to_dict(),        #     各列数据类型
            "numeric_summary": data.describe().to_dict() if len(data.select_dtypes('number').columns) > 0 else {},  # 数值统计
            "sample": data.head(5).to_dict(orient='records')  # 前5行样本
        }
        prompt = f"""请分析以下数据查询结果，回答用户问题。  # 分析Prompt

## 用户问题
{question}

## 执行的SQL
```sql
{sql}

数据摘要

• 行数：{summary['shape'][0]}，列数：
• 列名：
• 数值统计：
• 前5行数据：

要求

1. 用中文回答，简洁明了

2. 突出关键发现和数据洞察

3. 如果数据不足以回答问题，请明确指出

4. 建议下一步可以做的分析 """ return await self.llm.chat(prompt) # 调LLM生成分析

   async def _suggest_visualization(self, data: pd.DataFrame, question: str) -> Dict: """建议可视化方案""" # 让LLM推荐图表类型 prompt = f"""根据以下数据特征和用户问题，建议合适的可视化方案。

数据列：{data.columns.tolist()} 数据类型：{data.dtypes.to_dict()} 数据行数：{len(data)} 用户问题：

请以JSON格式返回建议：

{{
    "chart_type": "bar|line|pie|scatter|heatmap|table",
    "title": "图表标题",
    "x_axis": "X轴字段",
    "y_axis": "Y轴字段",
    "reason": "选择理由"
}}
```"""
        response = await self.llm.chat(prompt)      #   调LLM
        import json                                 #   导入json
        try:
            return json.loads(self._extract_json(response))  # 解析JSON建议
        except:                                     #   解析失败
            return {"chart_type": "table", "title": "数据表格", "reason": "默认展示"}  # 兜底

    def _extract_json(self, text: str) -> str:
        """从LLM回复中提取JSON"""                   #   和提取SQL类似的逻辑
        match = re.search(r'```json\s*\n(.*?)```', text, re.DOTALL)  # 先找json代码块
        if match:                                   #   找到了
            return match.group(1).strip()           #     返回JSON文本
        match = re.search(r'\{.*\}', text, re.DOTALL)  # 再找裸JSON
        return match.group(0) if match else "{}"    # 没有就返回空对象

三、数据可视化 Agent

拿到数据和分析后，最后一步是可视化。人类对图形的感知远强于对数字表格的感知——一张柱状图传达的信息，往往胜过一张密密麻麻的 Excel 表。VisualizationAgent 负责把 LLM 建议的图表类型变成真正的图片。

它支持五种常见图表：柱状图（对比排名）、折线图（趋势变化）、饼图（占比分布）、散点图（相关性）、热力图（多维相关）。生成后转为 base64 编码返回，方便嵌入 Web 页面。

import matplotlib                                   # 绑图库
matplotlib.use('Agg')                               # 非交互式后端，适合服务端
import matplotlib.pyplot as plt                     # pyplot接口
import seaborn as sns                               # 高级统计图表
import io                                           # 内存字节流
import base64                                       # base64编码

class VisualizationAgent:                           # 数据可视化Agent
    """数据可视化Agent"""                            #   根据配置生成图表

    # 中文字体配置（避免中文乱码）
    plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']  # 优先黑体
    plt.rcParams['axes.unicode_minus'] = False       #   负号显示修复

    def __init__(self):                             # 初始化
        self.chart_generators = {                    #   图表类型->生成函数映射
            "bar": self._generate_bar,               #   柱状图
            "line": self._generate_line,             #   折线图
            "pie": self._generate_pie,               #   饼图
            "scatter": self._generate_scatter,       #   散点图
            "heatmap": self._generate_heatmap,       #   热力图
        }

    def generate_chart(self, data: pd.DataFrame, viz_config: Dict) -> str:
        """生成图表并返回base64编码"""                #   主入口
        chart_type = viz_config.get("chart_type", "bar")  # 取图表类型
        generator = self.chart_generators.get(chart_type, self._generate_table)  # 取生成函数
        fig, ax = plt.subplots(figsize=(10, 6))       #   创建10x6英寸的画布
        try:
            generator(data, viz_config, ax)          #   调用生成函数画图
            plt.title(viz_config.get("title", "数据可视化"), fontsize=14)  # 设标题
            plt.tight_layout()                       #   自动调整布局
            buf = io.BytesIO()                       #   创建内存字节流
            plt.savefig(buf, format='png', dpi=100, bbox_inches='tight')  # 存为PNG
            buf.seek(0)                              #   移到流开头
            img_base64 = base64.b64encode(buf.read()).decode('utf-8')  # 编码为base64
            return f"data:image/png;base64,{img_base64}"  # 返回data URI
        finally:
            plt.close(fig)                           #   释放画布资源

    def _generate_bar(self, data: pd.DataFrame, config: Dict, ax):
        """柱状图"""                                 #   适合对比和排名
        x_col = config.get("x_axis", data.columns[0])  # X轴字段
        y_col = config.get("y_axis", data.columns[1])  # Y轴字段
        data_sorted = data.sort_values(y_col, ascending=False).head(10)  # 取Top10
        ax.bar(data_sorted[x_col], data_sorted[y_col])  # 画柱状图
        ax.set_xlabel(x_col)                        #   X轴标签
        ax.set_ylabel(y_col)                        #   Y轴标签
        plt.xticks(rotation=45)                     #   X轴标签旋转45度

    def _generate_line(self, data: pd.DataFrame, config: Dict, ax):
        """折线图"""                                 #   适合展示趋势变化
        x_col = config.get("x_axis", data.columns[0])  # X轴字段
        y_col = config.get("y_axis", data.columns[1])  # Y轴字段
        ax.plot(data[x_col], data[y_col], marker='o')  # 画折线，带圆点标记
        ax.set_xlabel(x_col)                        #   X轴标签
        ax.set_ylabel(y_col)                        #   Y轴标签

    def _generate_pie(self, data: pd.DataFrame, config: Dict, ax):
        """饼图"""                                   #   适合展示占比分布
        label_col = config.get("x_axis", data.columns[0])  # 标签列
        value_col = config.get("y_axis", data.columns[1])  # 数值列
        ax.pie(data[value_col], labels=data[label_col], autopct='%1.1f%%')  # 画饼图

    def _generate_scatter(self, data: pd.DataFrame, config: Dict, ax):
        """散点图"""                                 #   适合展示相关性
        x_col = config.get("x_axis", data.columns[0])  # X轴字段
        y_col = config.get("y_axis", data.columns[1])  # Y轴字段
        ax.scatter(data[x_col], data[y_col], alpha=0.6)  # 画散点，半透明
        ax.set_xlabel(x_col)                        #   X轴标签
        ax.set_ylabel(y_col)                        #   Y轴标签

    def _generate_heatmap(self, data: pd.DataFrame, config: Dict, ax):
        """热力图"""                                 #   适合展示多维相关性
        numeric_data = data.select_dtypes(include='number')  # 只取数值列
        if len(numeric_data.columns) >= 2:           #   至少2列才能算相关
            sns.heatmap(numeric_data.corr(), annot=True, cmap='coolwarm', ax=ax)  # 画热力图

generate_chart 方法用 try/finally 确保画布资源一定被释放——plt.close(fig) 是必须的，否则在长时间运行的服务中会造成内存泄漏。这是数据处理服务常见的"隐形坑"。

四、数据安全与权限控制

数据分析 Agent 直连数据库，安全是重中之重。除了前面讲的三道 SQL 防线，还需要数据权限控制（不同用户看到不同数据）和SQL 注入防护（防止恶意输入绕过防线）。

数据权限分三级：只读、读写、管理员。除此之外，还要支持列级脱敏——比如普通员工看不到完整手机号，只能看到前 3 位加星号。DataAccessControl 类把这些都封装在了一起：

from enum import Enum                                # 枚举类型
from functools import wraps                          # 装饰器工具

class DataPermission(Enum):                          # 数据权限枚举
    READ_ONLY = "read_only"                          #   只读权限
    READ_WRITE = "read_write"                        #   读写权限
    ADMIN = "admin"                                 #   管理员权限

class DataAccessControl:                             # 数据访问控制
    """数据访问控制"""                               #   权限+脱敏+限流

    def __init__(self):                             # 初始化
        self.user_permissions: Dict[str, DataPermission] = {}  # 用户权限表
        self.table_permissions: Dict[str, List[str]] = {}  # 表->允许用户列表
        self.column_masks: Dict[str, Dict[str, List[str]]] = {}  # 表->用户->脱敏列
        self.rate_limits: Dict[str, Dict] = {}       #   用户速率限制

    def grant_permission(self, user_id: str, permission: DataPermission):
        """授予用户权限"""                           #   设置用户权限级别
        self.user_permissions[user_id] = permission   #   存入权限表

    def mask_sensitive_data(self, user_id: str, table: str, data: pd.DataFrame) -> pd.DataFrame:
        """脱敏敏感数据"""                           #   根据用户身份脱敏
        if table in self.column_masks:               #   该表有脱敏配置
            user_masks = self.column_masks[table].get(user_id, [])  # 取该用户的脱敏列
            for col in user_masks:                   #   遍历需要脱敏的列
                if col in data.columns:              #   列存在
                    dtype = data[col].dtype          #     取数据类型
                    if dtype == 'object':            #     字符串类型
                        data[col] = data[col].apply(  #     脱敏：前3位+***
                            lambda x: x[:3] + '***' if isinstance(x, str) and len(x) > 3 else x
                        )
                    elif 'int' in str(dtype) or 'float' in str(dtype):  # 数值类型
                        data[col] = data[col].apply(lambda x: round(x, -2))  # 精度降级到百位
        return data                                  #   返回脱敏后的数据

    def check_rate_limit(self, user_id: str) -> bool:
        """检查速率限制"""                           #   防止滥用
        import time                                  #   时间模块
        now = time.time()                            #   当前时间戳
        user_limit = self.rate_limits.get(user_id, {"count": 0, "reset_time": now + 60, "max": 60})
        if now > user_limit["reset_time"]:           #   超过重置周期
            user_limit = {"count": 0, "reset_time": now + 60, "max": user_limit["max"]}  # 重置
        if user_limit["count"] >= user_limit["max"]: #   超过最大请求数
            return False                             #     拒绝请求
        user_limit["count"] += 1                    #     计数+1
        self.rate_limits[user_id] = user_limit       #     存回
        return True                                  #   允许请求

SQLInjectionGuard 是最后一道安全组件，对用户输入和生成的 SQL 做额外检查。它用多层规则过滤：禁止多条语句、禁止 UNION、禁止 SQL 注释、禁止访问系统表、限制语句长度。

class SQLInjectionGuard:                             # SQL注入防护
    """SQL注入防护"""                               #   多层规则检查

    @staticmethod
    def sanitize_input(value: str) -> str:
        """清理用户输入"""                           #   移除危险字符
        dangerous = ["'", '"', ';', '--', '/*', '*/', 'xp_', 'sp_', 'exec', 'execute']
        cleaned = value                              #   复制一份
        for char in dangerous:                       #   遍历危险字符
            cleaned = cleaned.replace(char, '')      #   逐个移除
        return cleaned[:1000]                        #   限制最大长度

    @staticmethod
    def validate_query(sql: str) -> tuple[bool, str]:
        """验证SQL查询安全性"""                      #   多层规则检查
        sql_upper = sql.upper()                      #   转大写
        checks = [                                   #   检查规则列表
            (lambda s: s.count(';') <= 1, "不允许包含多条语句"),  # 防多语句注入
            (lambda s: 'UNION' not in s, "不允许使用UNION"),       # 防UNION注入
            (lambda s: '--' not in s, "不允许SQL注释"),             # 防注释注入
            (lambda s: 'INFORMATION_SCHEMA' not in s.upper(), "不允许访问系统表"),  # 防信息泄露
            (lambda s: len(s) < 2000, "SQL语句过长"),              # 防超长注入
        ]
        for check, error_msg in checks:              #   逐条检查
            if not check(sql_upper):                 #   某条没过
                return False, error_msg              #     返回失败
        return True, ""                              #   全部通过

五、常见误区

1. 误区一：只靠 Prompt 约束 LLM 不写危险 SQL。Prompt 里写了"只生成 SELECT"，但 LLM 可能不听话，也可能被精心构造的问题诱导。必须加上 _validate_sql 的代码级验证，做"不信任何 LLM 输出"的纵深防御。

2. 误区二：Schema 描述太简单。只给 LLM 列名和类型，不给列的语义说明，LLM 就会猜错字段含义。比如 status 字段取值 1/2/3，LLM 不知道 1 代表"已支付"还是"已取消"。每个字段都应该配 description 和 sample_data。

3. 误区三：把全部查询结果塞进 LLM 分析。如果查询返回 10 万行数据，直接全塞进 Prompt 会 Token 爆炸且成本飙升。必须先做 data.describe() 和 data.head(5) 的摘要，只把摘要喂给 LLM。

4. 误区四：脱敏只做字符串类型。很多人只对姓名、手机号做脱敏，忘了金额、薪资等数值字段。数值脱敏可以用精度降级（如 round(x, -2) 取整到百位），既保留趋势又隐藏精确值。

5. 误区五：可视化图表类型让 LLM 自由发挥。如果不给 LLM 图表类型的约束，它可能建议用饼图展示 50 个类别的占比——结果成了不可读的"彩色披萨"。应该限制图表类型的适用场景。

六、本节小结

要点	说明
Text-to-SQL	通过 Schema 注入和 Prompt 工程，让 LLM 将自然语言转换为安全 SQL
安全防护	三层纵深防御：Prompt 约束 -> 代码级 SQL 验证 -> 数据库只读模式
数据可视化	自动化图表类型选择，支持柱状图、折线图、饼图、散点图、热力图
权限控制	基于角色的访问控制 + 列级数据脱敏 + 速率限制
端到端流程	自然语言 -> SQL -> 执行 -> 分析 -> 可视化，形成完整的数据分析闭环

数据分析 Agent 是 Agent 在企业 BI 场景中最直接的价值体现。它让不会写 SQL 的运营、产品、高管也能自助查数据，把数据分析师从重复性取数需求中解放出来。掌握了这套架构后，你可以把它迁移到日志分析、监控大盘、财务报表等任何"自然语言 -> 数据 -> 洞察"型场景。下一节我们将进入多 Agent 协作系统，看看当任务太复杂、单个 Agent 搞不定时，如何让多个 Agent 分工协作。