自动化数据分析报告解放分析师双手

时间线锚点：截至目前，企业数据分析师每周仍在重复一项低价值工作——手动从数据库拉取数据、用 Excel 画图、粘贴到周报模板中。这套流程通常耗时 3 到 4 个小时。而本章将在 60 分钟内 带你搭建一个“数据查询-图表生成-报告分发”的全自动 Skill，让你从此告别这套手工流。

当运营同学在周五下午 5 点急急忙忙来找你，说要看本周的用户留存数据时，你不需要再打开 SQL 客户端、导出 CSV、调整图表颜色、再截图粘贴到邮件里——因为你的 Skill 已经在 Slack 里等他提问了。

Skill 的真正力量不在于“多复杂的代码”，而在于把别人半天的查找压缩到 30 秒的对话里。当数据库查询、matplotlib 绘图、邮件发送这些工具链被编排成一个标准化的能力包时，你会发现：AI 落地的最大障碍从来不是技术，而是“他们不知道可以这么简单”。

你需要什么

在开始之前，请确认你已具备以下环境：

⚠️ 踩坑预警：不要把你的主库连接权限直接写进 Skill。请创建一个只读数据库用户，并将凭据放在环境变量或 .env 文件中，Skill 只通过变量名引用它们。

最终成果

你将在本章结束时拥有一个名为 data-insights-reporter 的 Skill。当用户在对话中说：

“帮我生成本周的用户活跃趋势报告，发到 #data-weekly 频道。”

Agent 会依次完成以下操作：

1. 解析意图：提取时间范围、指标名称、分发渠道
2. 连接数据库执行只读查询：安全地运行 SQL，返回格式化数据
3. 调用 matplotlib 生成图表：将 DataFrame 渲染为 PNG 图像
4. 组装报告并分发：将图表嵌入 Markdown 报告，通过 Slack Webhook 发送

整个过程无需分析师手动介入。这份“把半天压缩到 30 秒”的体验，正是工具链编排的核心价值。

步骤一：连接数据库与执行查询

目标

编写一个“数据库查询”工具函数，它接受一个 SQL 查询字符串，返回结构化的查询结果。这个函数将被 Skill 的描述文件引用，成为 Agent 可调用的工具。

动作

在你的项目目录下创建 tools/db_query.py：

# tools/db_query.py
import os
import psycopg2
import pandas as pd
from dotenv import load_dotenv

load_dotenv()

def run_query(sql: str, params: tuple = None) -> pd.DataFrame:
    """
    执行只读查询，返回 pandas DataFrame。
    Skill 会通过函数签名自动理解参数含义。
    """
    conn = psycopg2.connect(
        host=os.getenv("DB_HOST"),
        port=os.getenv("DB_PORT", 5432),
        dbname=os.getenv("DB_NAME"),
        user=os.getenv("DB_READONLY_USER"),  # 注意：使用只读账号
        password=os.getenv("DB_READONLY_PASSWORD")
    )
    try:
        # 执行查询并直接加载到 DataFrame
        df = pd.read_sql_query(sql, conn, params=params)
        return df
    finally:
        conn.close()

预期结果

在终端中测试：

python -c "from tools.db_query import run_query; print(run_query('SELECT 1 AS test').to_dict())"

输出应为 {'test': {0: 1}}，说明数据库连接成功。

将函数注册到 Skill

在 SKILL.md（或你所用框架的技能描述文件）中添加工具声明：

## Tools

### run_query
- **描述**：对只读数据库副本执行 SQL 查询，返回表格数据。
- **参数**：
  - `sql`（必填）：要执行的 SELECT 查询语句
  - `params`（可选）：SQL 参数化查询的元组，防止注入
- **返回**：包含列名和行数据的结构化结果

⚠️ 踩坑经验：务必使用参数化查询（params 参数），不要用字符串拼接构造 SQL。即使你的 Skill 面向内部用户，习惯性的安全写法能避免未来被外部请求触发时的注入风险。

步骤二：集成 Python 绘图库生成图表

目标

编写一个“图表生成”工具函数，它接受 DataFrame 和图表配置，输出一张 PNG 图像的文件路径。Agent 将把这张图嵌入到最终的报告中。

动作

创建 tools/chart_maker.py：

# tools/chart_maker.py
import matplotlib
matplotlib.use('Agg')  # 非 GUI 后端，服务器环境必须设置
import matplotlib.pyplot as plt
import pandas as pd
from datetime import datetime
import os

# 中文字体配置（踩坑重灾区）
plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']  # 根据系统调整
plt.rcParams['axes.unicode_minus'] = False  # 解决负号显示异常

def generate_line_chart(
    df: pd.DataFrame, 
    x_col: str, 
    y_col: str, 
    title: str,
    output_dir: str = "reports/images"
) -> str:
    """
    根据 DataFrame 生成折线图，返回图像文件路径。
    """
    os.makedirs(output_dir, exist_ok=True)

    fig, ax = plt.subplots(figsize=(10, 5))
    ax.plot(df[x_col], df[y_col], marker='o', linewidth=2, markersize=4)
    ax.set_title(title, fontsize=14, fontweight='bold')
    ax.set_xlabel(x_col)
    ax.set_ylabel(y_col)
    ax.grid(True, alpha=0.3)

    # 生成唯一文件名
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    filepath = f"{output_dir}/chart_{timestamp}.png"
    fig.savefig(filepath, dpi=150, bbox_inches='tight')
    plt.close(fig)  # 释放内存

    return filepath

预期结果

准备一份测试数据并运行：

import pandas as pd
from tools.chart_maker import generate_line_chart

df = pd.DataFrame({
    "日期": ["周一", "周二", "周三", "周四", "周五"],
    "活跃用户": [1200, 1350, 1100, 1480, 1600]
})
path = generate_line_chart(df, "日期", "活跃用户", "本周活跃用户趋势")
print(f"图表已保存至：{path}")

你应该在 reports/images/ 目录下看到一张清晰的折线图。

⚠️ 踩坑经验：

1. 中文字体乱码：这是最高频的报障问题。服务器通常不包含 Windows 字体，你需要在 plt.rcParams['font.sans-serif'] 中指定一个系统中存在的支持中文的字体（如 WenQuanYi Zen Hei），或者统一使用英文标签。
2. matplotlib.use('Agg') 必须放在 import pyplot 之前，否则在无图形界面的服务器上会直接报错。
3. plt.close() 别忘了。在高频调用场景下，不关闭 figure 会在几分钟内耗尽服务器内存。

在 Skill 中声明

### generate_line_chart
- **描述**：从表格数据生成折线图，用于展示趋势类指标。
- **参数**：
  - `df`：pandas DataFrame 对象
  - `x_col`：X 轴列名（如日期维度）
  - `y_col`：Y 轴列名（如指标值）
  - `title`：图表标题
- **返回**：生成的 PNG 图像文件路径

步骤三：调度与分发

目标

编写一个“报告组装与分发”编排函数，它将前两步的输出串联起来：调用 run_query 获取数据，调用 generate_line_chart 生成图像，然后将图像嵌入 Markdown 报告，通过 Slack Webhook 发送。

动作

创建 tools/report_dispatcher.py：

# tools/report_dispatcher.py
import os
import requests
from datetime import datetime, timedelta
from tools.db_query import run_query
from tools.chart_maker import generate_line_chart
from dotenv import load_dotenv

load_dotenv()

def weekly_active_users_report():
    """
    生成本周活跃用户趋势报告，并发送到 Slack 频道。
    此函数是工具链的编排中心。
    """
    # 步骤 1：计算时间范围
    today = datetime.now()
    week_start = today - timedelta(days=today.weekday())  # 本周一

    # 步骤 2：执行数据库查询
    sql = """
        SELECT 
            DATE(login_time) AS 日期,
            COUNT(DISTINCT user_id) AS 活跃用户数
        FROM user_activity_log
        WHERE login_time >= %s AND login_time < %s
        GROUP BY DATE(login_time)
        ORDER BY 日期
    """
    df = run_query(sql, params=(week_start, today))

    if df.empty:
        return "本周无活跃数据。"

    # 步骤 3：生成图表
    chart_path = generate_line_chart(
        df, 
        x_col="日期", 
        y_col="活跃用户数",
        title=f"本周活跃用户趋势（{week_start.strftime('%m/%d')} - {today.strftime('%m/%d')}）"
    )

    # 步骤 4：组装 Markdown 报告
    peak_day = df.loc[df["活跃用户数"].idxmax()]
    report = f"""📊 *本周活跃用户周报*

> 统计周期：{week_start.strftime('%Y-%m-%d')} 至 {today.strftime('%Y-%m-%d')}

- 日均活跃用户：*{df['活跃用户数'].mean():.0f}*
- 最高活跃日：{peak_day['日期']}（{peak_day['活跃用户数']} 人）
- 趋势：图表如下👇
"""

    # 步骤 5：通过 Slack Webhook 发送
    webhook_url = os.getenv("SLACK_WEBHOOK_URL")

    # 先发送文本报告
    requests.post(webhook_url, json={"text": report})

    # 再上传图表文件
    with open(chart_path, 'rb') as f:
        requests.post(
            webhook_url,
            files={"file": f},
            data={"initial_comment": "📈 活跃趋势图"}
        )

    return "报告已发送至 Slack #data-weekly 频道。"

预期结果

在终端执行：

python -c "from tools.report_dispatcher import weekly_active_users_report; print(weekly_active_users_report())"

你的 Slack #data-weekly 频道应该立即收到两条消息：一条包含本周的活跃用户摘要数据，另一条附带折线图。

在 Skill 中声明编排能力

## Capabilities

### 周报生成
- **触发语**："本周活跃数据"、"活跃用户趋势"、"用户周报"
- **流程**：
  1. 自动计算本周一至今天的时间范围
  2. 查询 `user_activity_log` 表获取每日活跃用户数
  3. 生成折线图展示趋势
  4. 组装 Markdown 报告并发送到 Slack
- **分发渠道**：默认 `#data-weekly`，可在触发时指定其他频道

回顾

你现在已经完成了一个完整的数据分析自动化 Skill。我们来梳理一下整个过程：

总计 60 分钟，你交付了一个能替代分析师 3-4 小时手工工作的自动化 Skill。

这套“连接数据库 → 执行查询 → 生成图表 → 组装报告 → 分发到 IM”的工具链，是 Agent Skills 在数据领域最典型的落地模式。它证明了一个核心认知：

Skill 封装的不只是“回答问题的能力”，而是把多个工具编排成一条完整的业务流水线。当用户只需要一句自然语言就能触发这条流水线时，AI 就从“玩具”变成了“生产力”。

行动清单

在进入下一章之前，你可以按以下顺序巩固本节内容：

1. 检查数据库权限：确认你拥有一个只读账号，并将其凭据写入 .env 文件
2. 跑通第一步：在终端中执行 run_query("SELECT 1")，确认连接正常
3. 生成第一张图：用测试数据调用 generate_line_chart，调整字体直到中文正常显示
4. 配置分发渠道：创建 Slack Webhook（或准备 SMTP 凭据），测试消息发送
5. 触发完整链路：在 Agent 对话中说“生成本周活跃用户趋势报告”，观察端到端输出

当数据分析从“手动拉取”变成“对话触发”后，你可能会开始思考：Skill 能不能驱动更多的实时交互？比如在游戏中，让 NPC 根据玩家行为动态生成对话和决策？

下一章 “游戏 NPC 行为控制不再是脚本的天下” 将带你进入一个更具创造性的领域——用 Skills 驱动 NPC 的动态对话与决策，让 AI 从“分析后台数据”走到“与玩家实时互动”的前台。你会发现，本章学到的工具链编排能力，在那个场景下依然适用。