DeepSeek 驱动的数据分析 Agent 实战

用大模型让数据"开口说话"——从架构设计到代码落地的完整指南

一、为什么要做数据分析 Agent

1.1 传统数据分析的痛点

做过数据分析的同学都知道，日常工作中最折磨人的不是算法本身，而是从"业务问题"到"拿到结果"这条路上的各种摩擦：

痛点	典型场景	真实代价
SQL 门槛高	运营同学想知道"上个月华东区 GMV 多少"，得找数据分析师写 SQL	排队等半天，分析师还得反复确认口径
代码成本大	画个趋势图要写一堆 matplotlib 样板代码	80% 时间在调样式，20% 在思考业务
沟通损耗重	"我要看环比" -> 分析师理解成"同比" -> 返工	需求理解偏差是最大的隐性成本
重复劳动多	每周固定出几张报表，逻辑一模一样	机械重复，毫无技术含量

核心矛盾很清晰：业务人员懂业务但不会写代码，技术人员会写代码但不一定懂业务。中间这条鸿沟，正是大模型 Agent 可以填平的地方。

1.2 大模型能解决什么

大语言模型（LLM）在数据分析场景中，核心能力可以拆成三层：

自然语言理解：把"帮我看看上个月各区域销售情况"解析成结构化意图——时间范围（上月）、分组维度（区域）、指标（销售额）。这一层纯靠 LLM 就能搞定。
代码生成：根据解析出的意图，自动生成 SQL 查询或 Python 分析代码。这也是 LLM 的强项，尤其是 DeepSeek-Coder 系列在代码生成上的表现相当能打。
推理与规划：面对复杂分析需求（比如"找出销量下滑的原因"），需要 Agent 自主拆解任务、规划执行步骤。这一层需要 LLM + 工具调用 + 业务上下文共同协作。

一句话总结：LLM 负责"理解"和"生成"，Agent 框架负责"调度"和"执行"。

二、系统架构总览

2.1 整体架构

整个系统由两个核心 Agent 协同工作，各司其职：

▲ 系统架构：用户提问 → 编排器路由 → NL2SQL Agent 查数据 / 可视化 Agent 画图表

2.2 两个 Agent 的分工

维度	NL2SQL 查询 Agent	数据可视化 Agent
职责	把自然语言翻译成 SQL，查询数据库	把查询结果翻译成 Python 代码，生成图表
核心工具	LangChain SQLDatabaseToolkit	自定义 Python 代码执行工具
输入	用户的自然语言问题	查询返回的结构化数据
输出	查询结果（表格数据）	统计分析 + 可视化图表
典型问题	"技术部有多少人？"	"画一个各部门平均薪资的柱状图"

2.3 技术栈一览

组件	技术选型	选型理由
大模型	DeepSeek-Chat / DeepSeek-Coder	国产头部，代码能力强，性价比高
Agent 框架	LangChain 1.3.1	生态成熟，create_agent API 简洁可靠
数据库	MySQL 8.0+	通用性强，适合演示
数据处理	Pandas + NumPy	Python 数据分析标准库
可视化	Matplotlib + Seaborn	灵活可控，适合 Agent 生成代码
沙箱执行	自定义 exec 沙箱	轻量级，可控性强

三、技术选型与环境准备

3.1 依赖安装

# 核心框架（锁定 langchain 1.3.1，API 稳定可靠）
pip install langchain==1.3.1 langchain-openai langchain-community

# 数据库驱动（MySQL）
pip install mysql-connector-python pymysql

# 数据处理与可视化
pip install pandas numpy matplotlib seaborn

3.2 初始化大模型客户端

import os
from langchain_openai import ChatOpenAI

# 通过环境变量读取 API Key，避免硬编码泄露风险
llm = ChatOpenAI(
    model="deepseek-v4-flash",                          # DeepSeek 的对话模型
    api_key=os.getenv("DEEPSEEK_API_KEY"),          # 从环境变量加载
    base_url="https://api.deepseek.com",            # DeepSeek API 地址
    temperature=0,                                   # 设为 0 保证 SQL 生成的确定性
)

print("✅ DeepSeek 模型初始化完成")

**为什么 temperature=0？** 生成 SQL 和分析代码属于确定性任务，不需要随机性。温度设为 0 可以让模型输出更稳定、可复现。

四、数据层搭建：建表与灌数据

4.1 表结构设计

我们用一个电商订单系统作为演示场景，涉及三张核心表：

4.2 建表与数据初始化

import os
import mysql.connector

# ============================================================
# 第一步：建立数据库连接
# ============================================================
# 使用环境变量管理敏感信息，这是生产级代码的基本规范
conn = mysql.connector.connect(
    host=os.getenv("DB_HOST", "127.0.0.1"),
    port=int(os.getenv("DB_PORT", 3306)),
    user=os.getenv("DB_USER", "root"),
    password=os.getenv("DB_PASSWORD", ""),
    database=os.getenv("DB_NAME", "analytics_demo"),
)
cursor = conn.cursor()

# ============================================================
# 第二步：创建表结构
# ============================================================

# 员工表：存储公司内部人员信息
# 用于分析：部门人数分布、薪资结构、入职趋势等
cursor.execute("""
CREATE TABLE IF NOT EXISTS employees (
    id          INT PRIMARY KEY AUTO_INCREMENT,  -- 自增主键
    name        VARCHAR(50)  NOT NULL,           -- 姓名
    department  VARCHAR(50)  NOT NULL,           -- 所属部门
    salary      DECIMAL(10,2) NOT NULL,          -- 月薪（精确到分）
    hire_date   DATE                             -- 入职日期
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
""")

# 产品表：存储在售商品信息
# 用于分析：品类销售、价格分布、库存预警等
cursor.execute("""
CREATE TABLE IF NOT EXISTS products (
    id           INT PRIMARY KEY AUTO_INCREMENT,  -- 自增主键
    product_name VARCHAR(100) NOT NULL,           -- 商品名称
    category     VARCHAR(50)  NOT NULL,           -- 商品分类
    price        DECIMAL(10,2) NOT NULL,          -- 单价
    stock        INT DEFAULT 0                    -- 当前库存量
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
""")

# 订单表：记录每笔交易
# 用于分析：销售趋势、员工绩效、产品热度等
cursor.execute("""
CREATE TABLE IF NOT EXISTS orders (
    id           INT PRIMARY KEY AUTO_INCREMENT,  -- 自增主键
    employee_id  INT NOT NULL,                    -- 下单员工（外键）
    product_id   INT NOT NULL,                    -- 购买商品（外键）
    quantity     INT NOT NULL,                    -- 购买数量
    order_date   DATE NOT NULL,                   -- 下单日期
    FOREIGN KEY (employee_id) REFERENCES employees(id),   -- 关联员工表
    FOREIGN KEY (product_id)  REFERENCES products(id)     -- 关联产品表
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
""")

# ============================================================
# 第三步：插入演示数据
# ============================================================

# 员工数据：5 个员工，分属 3 个部门
employees_data = [
    (1, "张三", "技术部",   20000.00, "2023-01-15"),
    (2, "李四", "销售部",   11000.00, "2023-02-20"),
    (3, "王五", "技术部",   16000.00, "2022-11-10"),
    (4, "赵六", "人力资源", 5000.00, "2023-03-01"),
    (5, "钱七", "销售部",   17000.00, "2022-12-05"),
]

# 产品数据：4 款商品，覆盖 2 个品类
products_data = [
    (1, "笔记本电脑", "电子产品", 6999.00, 500),
    (2, "机械键盘",   "电子产品", 399.00,  1000),
    (3, "办公椅",     "办公用品", 499.00,  300),
    (4, "显示器",     "电子产品", 1200.00, 400),
]

# 订单数据：5 笔订单，模拟真实购买行为
orders_data = [
    (1, 1, 1, 2, "2024-01-15"),  # 张三买了 2 台笔记本
    (2, 2, 2, 15, "2024-01-16"),  # 李四买了 15 个键盘
    (3, 3, 1, 10, "2024-01-17"),  # 王五买了 10 台笔记本
    (4, 5, 3, 6, "2024-01-18"),  # 钱七买了 6 把办公椅
    (5, 2, 4, 5, "2024-01-19"),  # 李四买了 5 台显示器
]

# executemany 批量插入，比逐条 insert 高效得多
cursor.executemany("INSERT IGNORE INTO employees VALUES (%s,%s,%s,%s,%s)", employees_data)
cursor.executemany("INSERT IGNORE INTO products  VALUES (%s,%s,%s,%s,%s)", products_data)
cursor.executemany("INSERT IGNORE INTO orders     VALUES (%s,%s,%s,%s,%s)", orders_data)

conn.commit()
conn.close()

print("✅ 数据库初始化完成")
print("   - employees 表：5 条员工记录")
print("   - products  表：4 条产品记录")
print("   - orders    表：5 条订单记录")

五、NL2SQL 自然语言查询 Agent

5.1 工作原理

NL2SQL Agent 的核心思路是：让 LLM 借助工具链自主完成"看表结构 → 写 SQL → 检查语法 → 执行查询 → 总结结果"的全流程。

▲ NL2SQL Agent 工作流：侦察表结构 → 生成 SQL → 检查语法 → 执行查询 → 总结结果

关键点：Agent 不是一次性生成 SQL 就完事了，它会先侦察（看表结构）、再行动（写 SQL）、后验证（检查语法），这套工作流比人肉写 SQL 还严谨。

5.2 搭建 SQL Agent

import os
from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent  # langchain 1.3.1 的新 API

# ============================================================
# 第一步：连接数据库
# ============================================================
# SQLDatabase.from_uri 接受标准的数据库连接 URI
# LangChain 内部会用 SQLAlchemy 管理连接池
db_uri = (
    f"mysql+pymysql://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}"
    f"@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}"
)
db = SQLDatabase.from_uri(db_uri)

# 验证连接：打印可用的表名
print(f"数据库连接成功")
print(f"   可用表：{db.get_usable_table_names()}")

# ============================================================
# 第二步：初始化大模型
# ============================================================
llm = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    temperature=0,  # SQL 生成需要确定性输出
)

# ============================================================
# 第三步：创建 SQL 工具包
# ============================================================
# SQLDatabaseToolkit 会自动注册 4 个工具：
#   1. sql_db_list_tables    — 列出数据库中所有表
#   2. sql_db_schema         — 获取指定表的 DDL 结构
#   3. sql_db_query_checker  — 检查 SQL 语法是否正确
#   4. sql_db_query          — 执行 SQL 并返回结果
toolkit = SQLDatabaseToolkit(db=db, llm=llm)
tools = toolkit.get_tools()

print(f"\nSQL 工具包已加载（{len(tools)} 个工具）：")
for t in tools:
    print(f"   - {t.name}")

# ============================================================
# 第四步：定义 System Prompt（Agent 的"岗位说明书"）
# ============================================================
SQL_AGENT_PROMPT = """你是一名专业的 SQL 数据分析师。

## 工作流程
1. 先用 sql_db_list_tables 查看数据库中有哪些表
2. 用 sql_db_schema 获取相关表的字段结构和类型
3. 生成 SQL 之前，用 sql_db_query_checker 检查语法
4. 确认无误后，用 sql_db_query 执行查询
5. 用中文总结查询结果，给出简洁的业务洞察

## 约束
- 只使用数据库中实际存在的表和字段，不要凭空编造
- 单次查询结果限制在 50 条以内
- 如果查询出错，分析错误原因后重新生成 SQL
- 回答要简洁专业，不要啰嗦
"""

# ============================================================
# 第五步：组装 Agent（langchain 1.3.1 一行搞定）
# ============================================================
# create_agent 返回一个可直接调用的 Runnable，无需再套 AgentExecutor
# 它内部自动处理工具调用、循环迭代、错误重试等逻辑
sql_agent = create_agent(
    model=llm,
    tools=tools,
    system_prompt=SQL_AGENT_PROMPT,
)

print("\nNL2SQL Agent 创建完成，可以开始提问了！")

5.3 查询示例

示例 1：简单计数查询

# 简单问题：Agent 只需一条 COUNT SQL 就能搞定
# create_agent 的 invoke 接口使用 messages 格式
response = sql_agent.invoke({
    "messages": [{"role": "user", "content": "公司一共有多少名员工？"}]
})

# 最终回答在最后一条消息里
final_msg = response["messages"][-1]
print(f"回答：{final_msg.content}")

Agent 内部执行过程（简化版）：

调用 sql_db_list_tables -> 发现 employees, products, orders
调用 sql_db_schema("employees") -> 获取字段：id, name, department, salary, hire_date
生成 SQL: SELECT COUNT(*) FROM employees
返回结果: 5
总结: "公司共有 5 名员工"

示例 2：带条件的聚合查询

# 中等难度：需要 JOIN + WHERE + GROUP BY
response = sql_agent.invoke({
    "messages": [{"role": "user", "content": "列出技术部所有员工的姓名和薪资，按薪资从高到低排序"}]
})
final_msg = response["messages"][-1]
print(f"回答：{final_msg.content}")

示例 3：跨表关联查询

# 高难度：需要 JOIN 三张表 + 聚合计算
response = sql_agent.invoke({
    "messages": [{"role": "user", "content": "销售部的员工总共下了多少订单？订单总金额是多少？"}]
})
final_msg = response["messages"][-1]
print(f"回答：{final_msg.content}")

5.4 调试：查看 Agent 的思考过程

在实际开发中，你经常需要看 Agent 到底干了什么——生成了什么 SQL、中间结果是什么。create_agent 返回的 messages 列表包含完整的推理链：

# 遍历所有消息，还原 Agent 的完整推理链
response = sql_agent.invoke({
    "messages": [{"role": "user", "content": "每个部门各有多少人？"}]
})

for i, msg in enumerate(response["messages"]):
    msg_type = msg.__class__.__name__

    # AIMessage 中如果有 tool_calls，说明 Agent 调用了工具
    if hasattr(msg, "tool_calls") and msg.tool_calls:
        print(f"\n步骤 {i+1} [{msg_type}]：")
        for tc in msg.tool_calls:
            print(f"  调用工具：{tc['name']}")
            print(f"  参数：{tc.get('args', {})}")

    # ToolMessage 是工具返回的结果
    elif msg_type == "ToolMessage":
        print(f"  工具返回：{msg.content[:200]}")

    # AIMessage 的最终文本回答
    elif msg.content:
        print(f"\n最终回答：{msg.content}")

六、数据可视化 Agent 与代码沙箱

6.1 设计思路

NL2SQL Agent 解决了"查数据"的问题，但用户往往还想"看图表"。我们需要一个能自主编写 Python 代码、执行数据分析、生成可视化图表的 Agent。

核心挑战在于：如何让 Agent 安全地执行动态生成的代码？

▲ 可视化 Agent 工作流：Agent 生成代码 → 沙箱隔离执行 → 结果回传 → 解读后返回用户

6.2 加载数据到内存

import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np

# ============================================================
# 配置 matplotlib 中文显示
# ============================================================
# Windows 用 SimHei（黑体），macOS 用 PingFang SC
# 如果还是乱码，试试安装 fonts-noto-cjk 并清除缓存
plt.rcParams["font.sans-serif"] = ["SimHei", "PingFang SC", "DejaVu Sans"]
plt.rcParams["axes.unicode_minus"] = False  # 解决负号显示为方块的问题

# ============================================================
# 从数据库加载数据到 Pandas DataFrame
# ============================================================
# 用 SQLAlchemy engine 复用连接，避免重复创建连接池
from sqlalchemy import create_engine

engine = create_engine(db_uri)

employees_df = pd.read_sql("SELECT * FROM employees", engine)
products_df  = pd.read_sql("SELECT * FROM products", engine)
orders_df    = pd.read_sql("SELECT * FROM orders", engine)

print("✅ 数据加载完成")
print(f"   员工表：{len(employees_df)} 行 × {len(employees_df.columns)} 列")
print(f"   产品表：{len(products_df)} 行 × {len(products_df.columns)} 列")
print(f"   订单表：{len(orders_df)} 行 × {len(orders_df.columns)} 列")
print(f"\n📋 员工表示例：")
print(employees_df.head(3).to_string(index=False))

6.3 构建 Python 代码执行沙箱

这是整个可视化 Agent 的核心组件——一个受控的代码执行环境。

import traceback
from io import StringIO
from contextlib import redirect_stdout
from langchain.tools import tool

# ============================================================
# 定义沙箱的"白名单"——只有这些库和数据可以被代码访问
# ============================================================
# 这是一种简单的安全策略：不在白名单里的东西，代码碰不到
SANDBOX_GLOBALS = {
    # 数据：Agent 可以分析这三张表
    "employees_df": employees_df,
    "products_df":  products_df,
    "orders_df":    orders_df,
    # 工具库：Agent 可以用这些库做分析和画图
    "pd":  pd,       # pandas — 数据处理
    "plt": plt,      # matplotlib — 基础绑图
    "sns": sns,      # seaborn — 统计可视化
    "np":  np,       # numpy — 数值计算
}


@tool
def execute_python_code(code: str) -> str:
    """
    执行 Python 代码进行数据分析和可视化。

    可用变量：
      - employees_df: 员工表 DataFrame（字段：id, name, department, salary, hire_date）
      - products_df:  产品表 DataFrame（字段：id, product_name, category, price, stock）
      - orders_df:    订单表 DataFrame（字段：id, employee_id, product_id, quantity, order_date）
      - pd:   pandas 库
      - plt:  matplotlib.pyplot
      - sns:  seaborn
      - np:   numpy

    使用示例：
      # 统计各部门平均薪资
      result = employees_df.groupby('department')['salary'].mean()
      print(result)

      # 画柱状图
      plt.figure(figsize=(10, 6))
      employees_df.groupby('department')['salary'].mean().plot(kind='bar')
      plt.title('各部门平均薪资')
      plt.show()
    """
    # 准备隔离的执行环境
    # globals_dict 提供白名单变量，locals_dict 收集执行过程中产生的新变量
    exec_globals = dict(SANDBOX_GLOBALS)
    exec_locals = {}

    # 用 StringIO 捕获 print() 的输出
    # 这样 Agent 生成的代码里所有的 print 语句都会被收集
    output_buffer = StringIO()

    try:
        # redirect_stdout 会把标准输出重定向到我们的 buffer
        with redirect_stdout(output_buffer):
            exec(code, exec_globals, exec_locals)

        result = output_buffer.getvalue()

        # 如果代码没有 print 任何东西，给个默认提示
        if not result.strip():
            result = "✅ 代码执行成功（无文本输出，可能已生成图表）"

        return f"执行成功：\n{result}"

    except Exception as e:
        # 出错时返回完整的错误堆栈，方便 Agent 自我修正
        error_detail = traceback.format_exc()
        return f"❌ 执行出错：{e}\n\n{error_detail}"


print("✅ Python 代码执行沙箱创建成功")

安全提示：上面的 exec 沙箱是最简实现，适合 Demo 和内部工具。生产环境建议用 Docker 容器沙箱 或 WebAssembly 隔离，彻底杜绝恶意代码风险。详见第七章。

6.4 搭建可视化 Agent

from langchain.agents import create_agent

# ============================================================
# 定义可视化 Agent 的 System Prompt
# ============================================================
VISUALIZATION_PROMPT = """你是一名资深数据分析师，精通 Python、Pandas 和 Matplotlib 数据可视化。

## 可用数据
1. employees_df — 员工表（字段：id, name, department, salary, hire_date）
2. products_df  — 产品表（字段：id, product_name, category, price, stock）
3. orders_df    — 订单表（字段：id, employee_id, product_id, quantity, order_date）

## 工作流程
1. 理解用户的分析需求
2. 用 execute_python_code 工具编写并执行 Python 代码
3. 先做数据探索（head、describe、info），再做深入分析
4. 用中文解释分析结果，给出业务洞察

## 代码规范
- 绑图前设置中文字体：plt.rcParams['font.sans-serif'] = ['SimHei', 'PingFang SC', 'DejaVu Sans']
- 设置 plt.rcParams['axes.unicode_minus'] = False
- 图表尺寸统一用 plt.figure(figsize=(10, 6))
- 必须添加标题、坐标轴标签，让图表自解释
- 用 print() 输出关键统计量，不要只画图不说话
- 图表标题用英文（避免渲染问题），但用中文向用户解释结果

## 注意事项
- 每次只执行一段完整的代码，不要拆成多段
- 先探索数据结构，再做分析——不要上来就画图
- 结果要有业务洞察，不只是"最大值是 XXX"
"""

# ============================================================
# 组装可视化 Agent（langchain 1.3.1 写法）
# ============================================================
viz_tools = [execute_python_code]

# 同样用 create_agent 一行搞定，和 SQL Agent 的创建方式完全一致
visualization_agent = create_agent(
    model=llm,
    tools=viz_tools,
    system_prompt=VISUALIZATION_PROMPT,
)

print("数据可视化 Agent 创建完成")

6.5 可视化示例

示例 1：基础统计分析

import warnings
warnings.filterwarnings("ignore")  # 抑制 matplotlib 的字体警告

response = visualization_agent.invoke({
    "messages": [{"role": "user", "content": "分析一下员工薪资的分布情况，包括均值、中位数、最大最小值等统计量"}]
})
final_msg = response["messages"][-1]
print(f"\n分析结果：\n{final_msg.content}")

Agent 内部生成的代码（大致）：

# 数据探索
print("=== 薪资统计概览 ===")
print(employees_df['salary'].describe())

# 计算额外统计量
median_salary = employees_df['salary'].median()
print(f"\n中位数薪资：{median_salary:,.0f} 元")
print(f"薪资范围：{employees_df['salary'].min():,.0f} ~ {employees_df['salary'].max():,.0f} 元")
print(f"薪资标准差：{employees_df['salary'].std():,.0f} 元")

示例 2：分组对比图表

response = visualization_agent.invoke({
    "messages": [{"role": "user", "content": "画一个柱状图，展示各部门的平均薪资对比"}]
})
final_msg = response["messages"][-1]
print(f"\n分析结果：\n{final_msg.content}")