告别造假数据，直接连数据库查真实时序数据喂给TimechoAI大模型

最近我在写一些关于数据分析的文章，发现自己一直在本地用代码生成假数据。我们造了CPU的数据、内存的数据，把这些拼凑在一起，交给TimechoAI去分析。其实这样做有它的好处，代码运行得很顺畅，而且不需要担心外部环境的问题。但在真实的工作场景下，数据可不都是那么简单生成的呀。

我觉得，数据通常是乖乖地躺在数据库里的。如果我们不能把真正的数据提取出来，填给大模型，那之前的练习简直就是在玩过家家，无法应用到真实的业务中去。因此，今天我们来做件正儿八经的事：用Python直接连接数据库，把真实数据查询出来，喂给模型。这可是一条完整的数据处理链路哦。

为什么在本地造数据没有实际意义？

1. 造假数据没法反映真实的脏数据问题

想象一下，你用 random.randint(20, 30) 生成的数据，简直完美无瑕。没有缺失值，格式也一丝不苟，时间戳顺序更是一点差错都没有。但现实中的时序数据往往是“脏”的，某个传感器在某一时刻断网，数据就缺失了；或者设备故障，传来的数据变成了负数，甚至是空字符串。

如果模型遇到这样的脏数据，没提前处理好，就可能会报错或者给你一些令人烦恼的结果。用真实数据来测试，才能发现哪一步需要清洗数据。这是造假数据无法测试出来的。

2. 真实的业务流是和数据库直接打交道的

再说说老板的需求吧，实时分析或定时分析是必须的。他可不想每天早上手动导出CSV文件，然后再跑一遍Python。在真实的业务流程中，应该是每隔一小时自动执行一次任务，连接数据库，捞取最近一小时的数据，然后将结果自动发到工作群。实现这一流程的第一步，就是得有代码能连接到数据库。没有这一步，接下来谈的自动化都是空话。

如何选择数据库来练手？

1. 为什么用SQLite演示

那我们用哪个数据库呢？MySQL？PostgreSQL？还是时序数据库IoTDB？其实对于本教程来说，用哪种数据库底层都差不多。因为到Python这层，出来的数据结构差不多。为了不让大家浪费时间在安装数据库上，我决定用SQLite。它太省事了，Python自带这个库，完全不需要下载任何安装包，也不需要配置用户名和密码，简单得很。

只需指定一个文件名，如果文件不存在，它会自动创建。这在演示时特别方便。如果是用MySQL，你得先装好MySQL Server，还得建库建表授权，新手容易卡在这一步。今天我们不在这上面浪费时间，直接将精力放在如何提取数据上。

2. 创建表并插入真实数据

我们先来创建一个SQLite数据库文件，并建立一张表。这次我们不造随机数据，而是插入一些模拟真实故障的数据。

import sqlite3

def init_db():
    conn = sqlite3.connect('monitor.db')
    cursor = conn.cursor()
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS metrics (
            time TEXT NOT NULL,
            metric_name TEXT NOT NULL,
            value REAL
        )
    ''')
    
    cursor.execute('DELETE FROM metrics')
    
    fake_data = [
        ('2023-10-25 10:00:00', 'CPU', 25.5),
        ('2023-10-25 10:01:00', 'CPU', 26.1),
        ('2023-10-25 10:02:00', 'CPU', 88.9), # 异常点
        ('2023-10-25 10:03:00', 'CPU', 90.2), # 异常点
        ('2023-10-25 10:04:00', 'CPU', 27.0),
        ('2023-10-25 10:00:00', 'Memory', 45.0),
        ('2023-10-25 10:01:00', 'Memory', 46.2),
        ('2023-10-25 10:02:00', 'Memory', 85.5), # 跟着高
        ('2023-10-25 10:03:00', 'Memory', 87.1), # 跟着高
        ('2023-10-25 10:04:00', 'Memory', 46.8),
    ]
    
    cursor.executemany('INSERT INTO metrics (time, metric_name, value) VALUES (?, ?, ?)', fake_data)
    conn.commit()
    conn.close()
    print("数据库初始化完成，数据已写入 monitor.db")

看这段代码，逻辑很简单。我们建了一张叫 metrics 的表，有三个字段：时间、指标名和数值。时间用 TEXT 类型存储，这样后面取出来能直接拼接给大模型，省去转换格式的麻烦。

如何用Python连数据库

1. 直接用内置库

SQLite是Python自带的，所以我们只需一句话就能连接。

import sqlite3

如果是连MySQL，你还得去终端里安装一些额外的库，搞得麻烦。用自带的库，直接省去这些麻烦。

2. 写一个查询数据的函数

接下来，我们写个函数从刚才那张表里查数据。为了灵活起见，要允许传入查询的时间范围。毕竟，不可能每次都把全表扫出来。

def query_data_from_db(start_time, end_time):
    conn = sqlite3.connect('monitor.db')
    cursor = conn.cursor()
    
    sql = """
        SELECT time, metric_name, value 
        FROM metrics 
        WHERE time >= ? AND time

这里有几个要点。首先，用 ? 作为占位符，防止SQL注入，这是个好习惯。其次，finally 中的 conn.close() 是必须的，确保连接能被正常关闭。

把数据洗成模型能用的格式

1. 数据库里拿出来的是啥结构

我们上面的 cursor.fetchall() 返回的 results 是一个列表，里面是元组，形似：

[
    ('2023-10-25 10:00:00', 'CPU', 25.5),
    ('2023-10-25 10:00:00', 'Memory', 45.0),
    ('2023-10-25 10:01:00', 'CPU', 26.1),
    ...
]

这种结构人类看着没问题，但大模型看不懂。我们需要一个“洗数据”的函数，把这个元组列表转换成字符串。

2. 写个洗数据函数

转换逻辑其实很简单，遍历这个列表，把每个元素拼起来。

def format_db_results(results):
    if not results:
        return ""
    
    text_lines = []
    for row in results:
        time_str = row[0]
        metric = row[1]
        value = row[2]
        
        line = f"{time_str}, {metric}, {value}%"
        text_lines.append(line)
        
    return "\n".join(text_lines)

经过这样的处理，结果就变成了一段规整的文本，与之前手动生成的数据格式完全一致。这样你以后换数据库，只要保证输出格式不变，后面的代码就不需要改动。

串联整个自动化的流程

1. 画个图理清链路

现在我们所有的组件都已搭建完毕，该建库的建库，该查数据的查了，该洗数据的洗了。最后一步就是把它们串联起来。

在串联之前，我觉得画个图把整个链路理清楚是个好主意。你会发现中间需要一个“查到数据了吗？”的判断。这是非常关键的一步，因为如果查到的结果是空列表，直接扔给模型可能会导致错误。

2. 主函数怎么写

接下来把主流程写出来。这里我直接用之前的 ask_timecho_ai 函数，逻辑不变。

import requests

if __name__ == "__main__":
    my_key = "sk-你的真实KEY粘贴在这里" 
    
    init_db()
    print("\n")
    
    start = "2023-10-25 10:00:00"
    end = "2023-10-25 10:04:00"
    print(f"正在从数据库查询 {start} 到 {end} 的数据...")
    
    raw_results = query_data_from_db(start, end)
    
    if not raw_results:
        print("警告：数据库里这段时间没数据，停止调用大模型。")
    else:
        print(f"查到了 {len(raw_results)} 条数据，开始转换格式...\n")
        
        clean_text = format_db_results(raw_results)
        
        prompt = f"""
        你是一个资深的运维专家。下面是从数据库查出来的真实监控数据。
        格式是：时间, 指标名称, 数值。
        请对比CPU和Memory这两个指标在时间线上的变化，找出异常波动点并分析它们之间的关联性。
        
        数据如下：
        {clean_text}
        """
        
        print("正在请求TimechoAI进行真实数据分析...")
        final_answer = ask_timecho_ai(prompt, my_key)
        
        print("\n=== 真实数据交叉分析报告 ===")
        print(final_answer)

这段代码逻辑清晰，跑一下你就能在控制台看到大模型对那些真实数据库记录的分析。它会告诉你10:02和10:03这两个时间点，CPU和内存同时出现了异常飙升。

真实场景中的几个坑

1. 时间范围没卡准

刚刚我们只查了5分钟的数据，只有10条。但在真实生产环境中，监控指标每秒钟一条，一天就86400条。如果你不小心把时间范围设成了一整年， cursor.fetchall() 会瞬间把几千万条数据拉到内存中。这时候，程序会卡死或者直接报错，原因就是你一次性全捞出来了。

2. 解决办法

解决这个问题有两个办法：一个是SQL加限制，例如加入 LIMIT 500 只取最近的500条数据；另一个是对时间范围进行合理的限制，确保不会一次拉取太多数据。

总之，通过这次的实践，你能看到如何将数据库中的真实数据提取出来，转化为有意义的分析报告。这不仅是一个技术活，更是让冰冷数据变得有温度的过程。希望你能在实际工作中运用这些知识，提升数据处理的能力！