智能数据分析Agent

​

智能数据分析Agent——用户用自然语言提问，Agent自动查询数据、执行分析、生成图表和报告。​

​

需求分析：从自然语言到SQL与图表生成​

项目目标​

构建一个数据分析Agent，核心能力：​

​

​ 能力​ ​

​ 用户输入示例​ ​

​ Agent输出​ ​

​ 数据查询​ ​	​ "上个月的销售额是多少"​ ​	​ SQL查询 + 结果表格​ ​
​ 趋势分析​ ​	​ "销售额的增长趋势如何"​ ​	​ 趋势图 + 文字分析​ ​
​ 异常检测​ ​	​ "有没有异常的订单"​ ​	​ 异常数据列表 + 原因推测​ ​
​ 报告生成​ ​	​ "生成本月销售报告"​ ​	​ 完整分析报告（含图表）​ ​

​

架构设计​

​

代码块​

Plain Text

用户自然语言问题​

|​

[意图识别] ──> 查询类 / 分析类 / 报告类​

|​

[Schema理解] ──> 读取数据库元数据​

|​

[SQL生成] ──> 生成并执行SQL​

|​

[结果处理] ──> 格式化 / 生成图表 / 文字解读​

|​

[安全检查] ──> 确认无敏感数据泄露​

|​

返回结果​

​

技术选型​

​

​ 组件​ ​

​ 选型​ ​

​ 理由​ ​

​ LLM​ ​	​ GPT-4o​ ​	​ SQL生成和数据分析需要强推理能力​ ​
​ 数据库​ ​	​ SQLite（演示）/ PostgreSQL（生产）​ ​	​ 轻量且兼容​ ​
​ 图表库​ ​	​ matplotlib + Plotly​ ​	​ 静态图 + 交互图​ ​
​ 框架​ ​	​ LangChain​ ​	​ Agent编排 + 工具管理​ ​
​ 前端​ ​	​ Streamlit​ ​	​ 快速构建交互界面​ ​

​

数据预处理与Schema映射策略​

示例数据准备​

​

代码块​

Python

import sqlite3​

import pandas as pd​

import numpy as np​

​

def create_sample_database(db_path: str = "sales.db"):​

"""创建示例销售数据库"""​

conn = sqlite3.connect(db_path)​

​

# 订单表​

orders_data = {​

"order_id": [f"ORD{i:06d}" for i in range(1, 1001)],​

"customer_id": [f"C{np.random.randint(1, 200):04d}" for _ in range(1000)],​

"product_id": [f"P{np.random.randint(1, 50):04d}" for _ in range(1000)],​

"order_date": pd.date_range("2024-01-01", periods=1000, freq="8H"),​

"quantity": [np.random.randint(1, 20) for _ in range(1000)],​

"unit_price": [round(np.random.uniform(10, 500), 2) for _ in range(1000)],​

}​

df_orders = pd.DataFrame(orders_data)​

df_orders["total_amount"] = df_orders["quantity"] * df_orders["unit_price"]​

df_orders.to_sql("orders", conn, if_exists="replace", index=False)​

​

Schema映射：让LLM理解数据库结构​

LLM需要知道数据库的表结构才能生成正确的SQL。关键策略是精简但完整地描述Schema：​

​

代码块​

Python

def get_db_schema(db_path: str) -> str:​

​

Schema描述优化​

直接把完整Schema塞给LLM太浪费Token。优化策略：​

​

代码块​

Python

def get_condensed_schema(db_path: str, relevant_tables: list[str] = None) -> str:​

"""精简Schema描述，只包含关键信息"""​

conn = sqlite3.connect(db_path)​

cursor = conn.cursor()​

​

cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")​

all_tables = [t[0] for t in cursor.fetchall()]​

​

代码生成与执行：PandasAI的核心实现逻辑​

SQL生成与执行工具​

​

代码块​

Python

from langchain_core.tools import tool​

import sqlite3​

import time​

​

@tool​

def execute_sql(query: str, db_path: str = "sales.db") -> str:​

"""执行SQL查询并返回结果。​

​

Args:​

query: SQL查询语句（仅支持SELECT）​

db_path: 数据库文件路径​

"""​

# 安全检查​

forbidden_keywords = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "CREATE"]​

​

构建数据分析Agent​

​

代码块​

Python

​

使用示例​

​

代码块​

Python

result = agent_executor.invoke({​

"input": "各产品类别的销售额排名如何？请生成柱状图"​

})​

print(result["output"])​

​

安全围栏：防止恶意代码执行与数据泄露​

三层安全防护​

​

代码块​

YAML

Layer 1: 提示词层 - 告知Agent安全规则​

Layer 2: 工具层 - 工具内部做安全检查​

Layer 3: 输出层 - 检查返回内容是否包含敏感数据​

​

输出脱敏​

​

代码块​

Python

import re​

​

class OutputSanitizer:​

"""输出脱敏处理器"""​

​

查询审计​

​

代码块​

Python

import json​

from datetime import datetime​

​

class QueryAuditor:​

"""查询审计日志"""​

​

def __init__(self, log_path: str = "query_audit.jsonl"):​

self.log_path = log_path​

​

def audit(self, user: str, question: str, sql: str, result_summary: str):​

record = {​

"timestamp": datetime.now().isoformat(),​

"user": user,​

"question": question,​

"sql": sql,​

"result_summary": result_summary[:200],​

"risk_level": self._assess_risk(sql),​

}​

​

结果解释：将数据洞察转化为自然语言报告​

分析结果的结构化输出​

​

代码块​

Python

​

完整的端到端流程​

​

代码块​

Python

class DataAnalysisAgent:​

"""完整的智能数据分析Agent"""​

​

​

本章小结​

​

​ 模块​ ​

​ 核心实现​ ​

​ 关键要点​ ​

​ 需求分析​ ​	​ 意图识别 + 架构设计​ ​	​ 先定义能力边界，再设计架构​ ​
​ Schema映射​ ​	​ 精简描述 + 示例数据​ ​	​ LLM需要足够但不冗余的Schema信息​ ​
​ SQL生成执行​ ​	​ Function Calling + 安全检查​ ​	​ 只允许SELECT，禁止修改操作​ ​
​ 安全围栏​ ​	​ 提示词 + 工具层 + 输出层​ ​	​ 三层防护，输出脱敏​ ​
​ 结果解释​ ​	​ 结构化报告 + 具体数字​ ​	​ 数据洞察用数字说话​ ​

​

下一章，我们将构建第二个实战项目——自动化研发运维Agent。​