LLM辅助数据分析实战：从SQL到归因-IT搬运工

LLM辅助数据分析实战：从SQL到归因

发布日期: 2026/08/09 阅读总量: 0

凌晨两点，订单量跌了37%

周三凌晨2点，公司大促刚开场15分钟，监控群炸了：订单量从每分钟320单跌到200单，跌幅37%。我是值班DBA，第一反应是连上MySQL，跑了条SQL：

SELECT 
    DATE_FORMAT(create_time, '%H:%i') AS minute,
    COUNT(*) AS order_cnt
FROM orders
WHERE create_time >= NOW() - INTERVAL 15 MINUTE
GROUP BY minute;

5秒出结果：02:00是315单，02:05剩210单，02:10只有198单。数据是准的，但为什么跌？SQL只告诉我结果，没说原因。

我又查了支付回调表、库存表、用户行为日志……花了40分钟，才发现是支付渠道回调超时，回调耗时P99从800ms飙到12s。但这是我在看了8张表、2000多行日志之后猜出来的。真正止损用了差不多2小时。

这次事故之后我就在想：能不能让LLM来做这个排查？不是让它生成SQL给我，而是让它像资深DBA一样，自己查表、自己看日志、自己给结论。

到2024年12月，我把这套流程跑通了。本文完整记录方案对比、代码实现和效果数据。环境：Python 3.11.7 + LlamaIndex 0.10.43 + MySQL 8.0.35 + 通义千问qwen-plus（2024-11-28版）。

方案对比：规则引擎 vs LLM+RAG vs 微调

方案A：规则引擎（传统方案）

大多数人面对「订单量突跌」会怎么做？写一堆监控规则：支付失败率超过阈值就告警，库存不足就告警，接口超时就告警。

我之前的监控系统就是这套。规则长这样：

// PHP 8.3 | 规则引擎核心判断逻辑
class OrderDropRule {
    private array $rules = [
        'payment_timeout' => [
            'query' => 'SELECT AVG(callback_cost_ms) AS avg_ms, COUNT(*) AS cnt FROM payment_callback WHERE create_time > NOW() - INTERVAL 5 MINUTE',
            'threshold' => 2000, // 平均回调耗时超过2000ms触发
            'level' => 'critical'
        ],
        'inventory_shortage' => [
            'query' => 'SELECT COUNT(*) AS zero_stock FROM inventory WHERE stock = 0 AND updated_at > NOW() - INTERVAL 5 MINUTE',
            'threshold' => 5,
            'level' => 'warning'
        ],
        'refund_surge' => [
            'query' => 'SELECT COUNT(*) AS c FROM refund_requests WHERE created_at > NOW() - INTERVAL 5 MINUTE',
            'threshold' => 50,
            'level' => 'warning'
        ]
    ];

    public function analyze(): array
    {
        $results = [];
        foreach ($this->rules as $name => $rule) {
            $rows = DB::query($rule['query']);
            $value = (float)$rows[0][array_key_first($rows[0])];
            if ($value >= $rule['threshold']) {
                $results[] = ['rule' => $name, 'value' => $value, 'level' => $rule['level']];
            }
        }
        return $results;
    }
}

这套方案的致命问题：规则是人想出来的，而故障往往不在人的预期里。12月那次事故，支付回调P99超时但均值正常（大量请求成功拉低了均值），规则没有覆盖P99指标，没触发告警。

真实数据对比（2024年11月-2025年1月，76个告警）：

指标	规则引擎	LLM+RAG
告警总数	76	76
误报次数	27（35.5%）	5（6.6%）
平均定位耗时	47分钟	12分钟
能覆盖的事故类型	预定义的12种	不限（LLM自主拆解）

方案B：LLM + RAG + 工具调用（本文方案）

核心思路：不预先穷举规则，而是给LLM几个工具（查SQL、查日志、看系统指标），让它自己决定查什么、怎么查。

架构图：

# docker-compose.yml | 基础依赖
version: "3.9"
services:
  mysql:
    image: mysql:8.0.35
    environment:
      MYSQL_ROOT_PASSWORD: root
      MYSQL_DATABASE: shop
    ports:
      - "3306:3306"
    volumes:
      - ./sql/init.sql:/docker-entrypoint-initdb.d/init.sql
  
  ollama:
    image: ollama/ollama:0.3.6
    ports:
      - "11434:11434"
    # 实际生产我们用 qwen-plus API，Ollama 仅用于本地开发调试

LLM需要三类工具：

query_mysql(sql)：执行只读SQL，返回结果集（截断到50行）
search_logs(keyword, time_range)：搜索应用日志关键词
get_metric(metric_name)：拉取监控系统指标（如RT、错误率）

方案C：微调专用模型（不推荐）

我试过用Qwen2.5-7B在16张A100上做LoRA微调，用历史事故报告作为训练集，目标是让模型直接「看出」问题原因。结论是：样本量不够就是浪费时间。我们只有几百份事故报告，微调后的模型在测试集上准确率只有61%，还经常一本正经地胡说。微调适合场景极度固定、样本超万份的场景，不适合数据分析这种长尾极长的工作。

代码实现：完整可跑

1. 初始化和建表

先准备MySQL测试数据，包含订单表、支付回调表、库存表：

-- init.sql | MySQL 8.0.35
CREATE DATABASE IF NOT EXISTS shop DEFAULT CHARACTER SET utf8mb4;

USE shop;

CREATE TABLE orders (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    order_no VARCHAR(32) NOT NULL,
    user_id BIGINT NOT NULL,
    amount DECIMAL(10,2) NOT NULL,
    status TINYINT NOT NULL DEFAULT 0 COMMENT '0-待支付 1-已支付 2-已取消',
    create_time DATETIME NOT NULL,
    INDEX idx_create_time (create_time)
) ENGINE=InnoDB;

CREATE TABLE payment_callback (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    order_id BIGINT NOT NULL,
    channel VARCHAR(16) NOT NULL COMMENT 'alipay/wechat',
    callback_cost_ms INT NOT NULL COMMENT '回调耗时(ms)',
    success TINYINT NOT NULL DEFAULT 1,
    create_time DATETIME NOT NULL,
    INDEX idx_create_time (create_time)
) ENGINE=InnoDB;

-- 插入模拟数据：正常时段（前10分钟）和异常时段（后5分钟）
INSERT INTO orders (order_no, user_id, amount, status, create_time)
SELECT 
    CONCAT('SN', LPAD(n, 8, '0')),
    FLOOR(RAND() * 10000),
    ROUND(RAND() * 500 + 50, 2),
    1,
    DATE_SUB(NOW(), INTERVAL (15 - n % 15) MINUTE)
FROM (
    SELECT 1 AS n UNION ALL SELECT 2 UNION ALL SELECT 3 UNION ALL SELECT 4 UNION ALL SELECT 5
    UNION ALL SELECT 6 UNION ALL SELECT 7 UNION ALL SELECT 8 UNION ALL SELECT 9 UNION ALL SELECT 10
) numbers;

-- 模拟支付回调异常：后5分钟回调耗时飙升
INSERT INTO payment_callback (order_id, channel, callback_cost_ms, success, create_time)
SELECT 
    id,
    IF(RAND() > 0.5, 'alipay', 'wechat'),
    CASE WHEN create_time > DATE_SUB(NOW(), INTERVAL 5 MINUTE) 
         THEN FLOOR(RAND() * 8000 + 4000)  -- 异常时段：4s-12s
         ELSE FLOOR(RAND() * 500 + 100)    -- 正常时段：100ms-600ms
    END,
    1,
    create_time
FROM orders;

2. LLM工具函数

这是核心。每个工具都是有边界的函数，不是直接暴露给LLM任意执行——用参数白名单和只读事务保证安全。

先是数据库查询工具：

# tools.py | Python 3.11
import pymysql
import json
import re
from typing import Any

DB_CONFIG = {
    "host": "127.0.0.1",
    "port": 3306,
    "user": "root",
    "password": "root",
    "database": "shop",
    "charset": "utf8mb4",
    "autocommit": True,
}

# 只允许SELECT开头的SQL，防止LLM写坏数据
_FORBIDDEN_KEYWORDS = re.compile(r"\b(INSERT|UPDATE|DELETE|DROP|ALTER|TRUNCATE|GRANT)\b", re.IGNORECASE)

def query_mysql(sql: str) -> dict[str, Any]:
    """执行只读SQL查询，返回结构化结果"""
    sql = sql.strip()
    if not sql.startswith("SELECT"):
        return {"error": "只允许SELECT查询"}
    if _FORBIDDEN_KEYWORDS.search(sql):
        return {"error": "SQL包含危险关键字"}
    
    conn = pymysql.connect(**DB_CONFIG)
    try:
        with conn.cursor() as cur:
            cur.execute(sql)
            # 限制最多取20行，防止超长结果集塞爆上下文
            rows = cur.fetchmany(20)
            columns = [desc[0] for desc in cur.description] if cur.description else []
        return {"columns": columns, "rows": rows}
    except Exception as e:
        return {"error": str(e)}
    finally:
        conn.close()

然后是日志搜索和监控指标工具。日志我们用ES的API，实际生产可以把这段替换成ClickHouse或Loki：

# tools.py 续
import requests
from datetime import datetime, timedelta

def search_logs(keyword: str, minutes: int = 10) -> list[dict]:
    """从日志系统搜索包含keyword的日志（生产环境替换为ES/Loki API）"""
    # 这里用简化实现，生产环境按需调整
    log_url = "http://127.0.0.1:9200/logstash-*/_search"
    query = {
        "query": {
            "bool": {
                "must": [
                    {"match": {"message": keyword}},
                    {"range": {"@timestamp": {
                        "gte": f"now-{minutes}m",
                        "lt": "now"
                    }}}
                ]
            }
        },
        "size": 20,
        "sort": [{"@timestamp": "desc"}]
    }
    resp = requests.post(log_url, json=query, timeout=10)
    if resp.status_code != 200:
        return {"error": f"日志查询失败: {resp.status_code}"}
    hits = resp.json().get("hits", {}).get("hits", [])
    return [
        {
            "time": h["_source"].get("@timestamp"),
            "message": h["_source"].get("message", "")[:500]
        }
        for h in hits
    ]

def get_metric(metric_name: str, minutes: int = 10) -> list[dict]:
    """获取监控指标：支持 p99_rt, error_rate, qps, pay_success_rate"""
    # 同样简化实现，实际对接Prometheus API
    # 示例返回：查询Prometheus
    prom_url = "http://127.0.0.1:9090/api/v1/query_range"
    query_map = {
        "p99_rt": 'histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket[1m])) by (le))',
        "error_rate": 'sum(rate(http_errors_total[1m])) / sum(rate(http_requests_total[1m]))',
        "pay_success_rate": 'sum(rate(pay_success_total[1m])) / sum(rate(pay_attempts_total[1m]))',
    }
    if metric_name not in query_map:
        return {"error": f"未知指标: {metric_name}, 可选: {list(query_map.keys())}"}
    
    now = datetime.now()
    start = now - timedelta(minutes=minutes)
    params = {
        "query": query_map[metric_name],
        "start": start.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "end": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "step": "60"
    }
    resp = requests.get(prom_url, params=params, timeout=10)
    if resp.status_code != 200:
        return {"error": f"Prometheus查询失败: {resp.status_code}"}
    
    result = resp.json().get("data", {}).get("result", [])
    values = []
    for item in result:
        for ts, val in item.get("values", []):
            values.append({"time": ts, "value": float(val)})
    return values

3. LLM主流程：Agent式分析

实际上我用LlamaIndex的Function Calling定义来注册这些工具，然后用ReAct Agent让LLM自主决定调用顺序。工作流：

喂给LLM一个初始描述：「订单量突降，请分析原因」
LLM自己决定先查什么：先看时间分布 → 查支付回调 → 查库存 → 看日志关键词
每得到一个结果，LLM判断是否继续深入（比如发现回调耗时高，继续按channel分组）
最终输出归因结论，用json格式统一解析

# agent.py | LlamaIndex 0.10.43
from llama_index.core.agent import FunctionCallingAgentWorker
from llama_index.core.tools import FunctionTool
from llama_index.llms.dashscope import DashScope
from tools import query_mysql, search_logs, get_metric

# 用qwen-plus，temperature=0.1保证稳定性（这个参数很关键，后面避坑讲）
llm = DashScope(
    model="qwen-plus",
    api_key="sk-xxx",
    temperature=0.1,
    max_tokens=2048,
)

# 注册三个工具
sql_tool = FunctionTool.from_defaults(
    query_mysql,
    name="query_mysql",
    description="执行MySQL只读SQL，输入必须是SELECT语句。适用于查订单量、支付耗时、库存等业务数据。"
)
search_tool = FunctionTool.from_defaults(
    search_logs,
    name="search_logs",
    description="按关键词搜索应用日志，输入是日志关键词和时间范围。适用于查看异常报错、超时日志。"
)
metric_tool = FunctionTool.from_defaults(
    get_metric,
    name="get_metric",
    description="查询监控指标，支持p99_rt、error_rate、pay_success_rate。适用于确认系统整体健康状况。"
)

agent_worker = FunctionCallingAgentWorker.from_tools(
    tools=[sql_tool, search_tool, metric_tool],
    llm=llm,
    verbose=True,
    max_iterations=10,
)

agent = agent_worker.as_agent()

# 分析入口
def analyze_incident(incident_desc: str) -> str:
    response = agent.chat(f"""
    你现在是资深数据分析师。以下异常需要通过SQL查询、日志检索和监控指标逐步定位原因。
    
    异常描述：{incident_desc}
    
    请按数据分析的逻辑逐步操作：
    1. 先用SQL确认异常现象的具体特征（时间范围、影响量）
    2. 用监控指标检查系统核心指标是否有异常
    3. 根据前两步的线索用日志检索缩小范围
    4. 给出最终归因结论
    
    最终输出格式为JSON：
    {{
        "conclusion": "最终归因结论",
        "evidence": ["证据1", "证据2"],
        "action": "止损建议"
    }}
    """)
    return str(response)

4. 运行与解析输出

跑起来之后，LLM会自己去调SQL和日志。我们从verbose输出里能看到它的思考链路：

# bash | 运行agent.py
$ python agent.py

> Running step 1: Calling function query_mysql with args:
  {"sql": "SELECT DATE_FORMAT(create_time, '%H:%i') AS minute, COUNT(*) AS cnt 
           FROM orders WHERE create_time > NOW() - INTERVAL 15 MINUTE GROUP BY minute"}
  Observation: {'columns': ['minute', 'cnt'], 'rows': [['03:05', '312'], ['03:10', '198'], ['03:15', '205']]}

> Running step 2: LLM思考：订单量从312降到198，降幅明显。继续查支付回调是否异常。
  Calling function query_mysql with args:
  {"sql": "SELECT channel, AVG(callback_cost_ms) AS avg_ms, ROUND(MAX(callback_cost_ms)) AS max_ms, 
           COUNT(*) AS cnt FROM payment_callback 
           WHERE create_time > NOW() - INTERVAL 5 MINUTE GROUP BY channel"}
  Observation: {'columns': ['channel', 'avg_ms', 'max_ms', 'cnt'], 
                'rows': [['alipay', '6200.5', '12000', '102'], ['wechat', '480.2', '900', '98']]}

> Running step 3: LLM思考：支付宝渠道平均耗时6.2秒，明显异常。查看支付宝相关日志。
  Calling function search_logs with args:
  {"keyword": "alipay timeout", "minutes": 10}
  Observation: [{'time': '2025-01-12T03:12:11', 'message': 'alipay callback timeout after 8000ms, order=...'}, ...]

最后LLM输出归因报告：

{
  "conclusion": "订单量突降由支付宝支付渠道回调超时导致。03:05后支付宝回调平均耗时从420ms飙升至6200ms，大量订单支付成功后回调未及时确认，订单状态停留在待支付，造成订单量统计下降。",
  "evidence": [
    "支付宝渠道回调平均耗时6200ms，远高于微信渠道480ms",
    "日志中出现alipay callback timeout after 8000ms错误",
    "订单量下降时间与回调异常时间完全吻合（03:05起）"
  ],
  "action": "1. 紧急降级：将支付宝渠道支付确认改为异步重试+补偿对账\n2. 联系支付宝开放平台确认服务端是否异常\n3. 设置支付宝回调P99耗时告警（阈值3s）"
}

这段JSON可以直接接入告警系统，自动创建工单。

效果数据：真实压测对比

我在一套模拟环境做了20轮测试，构造了4类故障：

类型A：支付回调超时（支付宝渠道）
类型B：库存扣减异常（乐观锁冲突导致超卖）
类型C：缓存雪崩（Redis集群单节点故障）
类型D：上游服务限流（订单服务依赖的会员服务）

每类故障跑5轮，记录准确率和耗时：

故障类型	规则引擎定位准确率	LLM+RAG定位准确率	LLM平均耗时
A-支付回调超时	4/5	5/5	1分52秒
B-库存扣减异常	2/5（规则没覆盖超卖场景）	4/5	2分08秒
C-缓存雪崩	0/5（无相关规则）	5/5	2分35秒
D-上游限流	1/5	4/5	1分47秒

费用成本：使用qwen-plus，一次分析平均消耗token约18500个（输入15000 + 输出3500），按qwen-plus价格（输入0.0008元/1K，输出0.002元/1K）计算，单次成本约0.019元。20轮测试总成本0.38元。对比一个初级DBA加班2小时的工时成本（约100元），差距不用我算。

避坑：这5个坑我一个一个踩过的

坑1：temperature=0.7，LLM每次给的结论都不一样

刚开始我把temperature调成0.7，结果同样的数据，第一遍分析是「支付回调超时」，第二遍变成「库存不足」——因为模型在「猜测」哪个更像答案。排查类场景必须压低随机性：temperature=0.1。0是完全确定性但有时太机械，0.1是实践下来的甜点值。

坑2：LLM把表名写错了，查了半天全是空结果

qwen-plus对MySQL表名记忆不牢，出现过把 orders 写成 order、把 payment_callback 写成 payment_log。不报错，但结果为空。LLM看到空结果就换方向，浪费了好几步。

解决：在System Prompt里注入表结构信息，让LLM照着写：

# system_prompt.py | 注入表结构
TABLE_SCHEMA = """
数据库shop包含以下表：
1. orders(id BIGINT, order_no VARCHAR(32), user_id BIGINT, amount DECIMAL(10,2), status TINYINT, create_time DATETIME)
   含义：订单表。status: 0-待支付 1-已支付 2-已取消
2. payment_callback(id BIGINT, order_id BIGINT, channel VARCHAR(16), callback_cost_ms INT, success TINYINT, create_time DATETIME)
   含义：支付回调记录表。channel: alipay/wechat。success: 1-成功 0-失败
3. inventory(id BIGINT, sku_id BIGINT, stock INT, updated_at DATETIME)
   含义：库存表
"""

坑3：把全量数据塞给LLM，上下文爆了

最初我尝试「把最近15分钟的订单明细全查出来给LLM」，结果一次查询返回8000行，直接干爆上下文窗口。后面改成：先让LLM写聚合SQL，只返回统计结果（按分钟、按渠道分组），把上下文控制在2000token以内。LLM需要看明细时，再单独查询限流后的小结果集。

坑4：日志检索用embedding匹配，效果差还慢

最开始的版本用向量检索找日志，结果匹配到的全是语义相近但没用的日志（比如「timeout」匹配到「timezone」）。后来改成关键词精确匹配 + 时间范围过滤，准确率反而更高，延迟还从ms级变少了。数据分析场景，日志检索用BM25或普通keyword就够了，向量检索适合「模糊问法」但不太适合「准确定位»。

坑5：LLM传了个负数limit给SQL，MySQL直接报错

有一次LLM生成SQL：SELECT * FROM orders LIMIT -1。MySQL报语法错误，Agent卡死。原因：Function Calling的参数没校验。后面在工具函数里加了防御：

# tools.py 防御式参数校验
def query_mysql(sql: str) -> dict[str, Any]:
    sql = sql.strip()
    if not sql.startswith("SELECT"):
        return {"error": "只允许SELECT查询"}
    if _FORBIDDEN_KEYWORDS.search(sql):
        return {"error": "SQL包含危险关键字"}
    # 过滤负数limit
    sql = re.sub(r"LIMIT\s+-\d+", "LIMIT 20", sql, flags=re.IGNORECASE)
    # 去掉末尾分号
    sql = sql.rstrip(";")
    ...

所有LLM传入的参数都当用户输入处理，不自信任任何值。

什么时候别用这套方案

LLM+RAG不是银弹。以下场景我建议你继续用规则：

故障类型固定且就几种：写规则10分钟搞定，没必要上LLM
延迟敏感：LLM推理需要几秒到几十秒，紧急止损还是靠实时告警
敏感数据：SQL查询能力暴露给LLM，DBA得评估安全风险
没预算：虽然单次不到2分钱，但公司如果对数据出域很敏感就另说

总结一句话：LLM负责把排查从「人肉看表」变成「人看结论」，但表和库还是你的，别全交给它。

上面所有代码，包括完整的agent.py、tools.py、init.sql，我都放在 GitHub 的 llm-data-analysis 仓库里，下载就能跑通。数据是模拟的，但逻辑和生产一致。如果你也遇到类似的数据分析困境，直接改表名前缀就能用。