告别造假数据，直接连数据库查真实时序数据喂给TimechoAI大模型
前面三篇文章，我们其实都在做一件事情。那就是在本地用代码造假数据。我们造了CPU的数据，造了内存的数据，然后把这些假数据拼成一个长长的字符串，扔给TimechoAI去分析。
这么干是有好处的。好处就是你不用担心环境问题，代码跑起来就很顺滑。但是呢，它掩盖了一个很现实的问题。我们在真实的工作里，数据是不可能靠 random.randint 去生成的。数据都是乖乖地躺在数据库里的。如果你不能把数据库里的真实数据掏出来喂给大模型，那前面学的那些东西基本上就停留在玩具阶段，没法用到真正的业务里头去。
所以今天这篇，我们要干一件正儿八经的事情。我们要用Python连上数据库，把数据查出来，然后直接丢给大模型。我们要把这个链路彻底打通。
一、 为什么总在本地造数据不是个事儿
1.1 造假数据掩盖了真实的脏数据问题
你用 random.randint(20, 30) 造出来的数据，那叫一个干净。没有任何缺失值，格式完美，时间戳一个接一个一点都不差。
但是现实里的时序数据是什么样的呢？往往很脏。比如某个传感器在某个时刻断网了，那一条数据就没传上来。你的时间序列里就断了一截。或者设备出了点故障，传上来的数值是个负数，或者干脆是个空字符串。
大模型遇到这种脏数据会怎么样呢？如果你不提前处理，它可能就会报错，或者给你胡说八道一通。所以我们必须得用真实数据来测，看看在这个流程里，哪一步需要对脏数据进行清洗。这是造假数据永远测不出来的情况。
1.2 真实的业务流一定是对着数据库干的
另外还有一个原因。老板要的是实时分析，或者至少是定时分析。他不可能接受你每天早上手动导出一个CSV文件，然后再跑一遍Python脚本。
真实的业务流应该是这样的：有一个定时任务，每隔一小时触发一次。这个任务连上数据库，执行一句SQL，把最近一小时的数据捞出来。然后转手塞给大模型的API。拿到结果后，自动发到工作群里。
你要实现这个，第一步就是得把“连数据库查数据”这事儿在代码里搞定。没有这一步，后面那些自动化的东西全都是空谈。
二、 挑个数据库来练手
2.1 为什么拿SQLite来做演示
连数据库，我们连哪个呢？MySQL？还是PostgreSQL？还是时序数据库比如IoTDB？
其实对于咱们这个教程来说，用什么数据库底层都一样。因为到了Python这一层，查出来的数据结构都是差不多的。
为了不让大家在装数据库上浪费半天时间，我决定用SQLite来演示。为什么选它呢？因为它太省事了。Python自带了这个库，你不需要去下载任何安装包，不需要配置用户名密码，不需要启动什么系统服务。它就是一个文件。
你只要指定一个文件名，如果文件不存在，它就自动给你创建一个。这在教程演示里特别方便。如果是MySQL，你得先装MySQL Server，还得建库建表授权，新手很容易卡在这一步。我们今天不把精力花在那上面，我们聚焦在怎么把数据掏出来这个动作上。
2.2 建表和塞点真数据进去
既然用SQLite，我们先写一段代码，在当前目录下建一个数据库文件，并且造一张表。这次我们不造随机数据了，我们造一些模拟真实故障的固定数据。
import sqlite3
def init_db():
# 连接到sqlite数据库，如果文件不存在会自动创建
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 类型来存。为什么不存时间戳格式呢？其实说白了，存成 2023-10-25 10:00:00 这种字符串，后面取出来直接拼给大模型最方便，省去了类型转换的麻烦。
然后我塞了十条数据。有CPU的，有内存的。我故意让10:02和10:03这俩时间点，两个指标都高。这就模拟了一个典型的联动异常。
三、 Python连数据库的那点破事
3.1 不用装任何额外库，直接import
前面说了，SQLite是Python自带的。所以我们连库的时候，一句话就搞定。
import sqlite3
如果你是连MySQL，你还得去终端里敲 pip install pymysql 或者 pip install mysql-connector-python。如果网络不好，安装失败了，你还得去搞镜像源。这些都是坑。我们用自带的库，直接跳过这些坑。
3.2 写一个查数据的函数，别把逻辑混在一起
我们写一个专门的函数，用来从刚才那张表里查数据。为了灵活，我们得允许传入查询的时间范围。你不能每次都把全表扫出来，那样数据量大了肯定撑不住。
def query_data_from_db(start_time, end_time):
# 连接数据库
conn = sqlite3.connect('monitor.db')
cursor = conn.cursor()
# 写SQL语句。用问号占位符防止SQL注入，这是个好习惯
sql = """
SELECT time, metric_name, value
FROM metrics
WHERE time >= ? AND time <= ?
ORDER BY time ASC, metric_name ASC
"""
try:
# 把时间参数传进去执行
cursor.execute(sql, (start_time, end_time))
# 查出所有结果
results = cursor.fetchall()
return results
except Exception as e:
print(f"查数据库报错了：{e}")
return None
finally:
# 无论报不报错，最后一定要关掉连接
conn.close()
这里面有几个细节我要唠叨一下。
第一个是那个 ? 号。你写SQL的时候，千万不要直接用字符串拼接的方式把时间塞进去。比如 f"WHERE time >= '{start_time}'"。这么写如果遇到别有用心的人传进来一个带分号的字符串，就容易引发SQL注入攻击。用 ? 占位符，数据库驱动会自动帮你处理转义，安全得多。
第二个是 finally 里面的 conn.close()。这个特别重要。数据库的连接是一种资源。你打开了不关，资源就被占着。如果你写了个循环一直查却不关连接，很快数据库就会报“连接数过多”的错误。写在 finally 里，就能保证就算中间出错了，连接也能被正常关闭。
四、 把查出来的数据洗成大模型喜欢的样子
4.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),
...
]
这种结构，人眼看着还行，但是大模型看不懂。大模型需要的是上节课我们讲的那种带有明显标识的文本字符串。所以我们需要写一个“洗数据”的函数，把这个元组列表给转成字符串。
4.2 写个洗数据函数，转成字符串
转换的逻辑很简单，就是遍历这个列表，然后把每个元素拼起来。
def format_db_results(results):
if not results:
return ""
text_lines = []
for row in results:
# row[0]是时间，row[1]是指标名，row[2]是数值
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)
你看，经过这么一洗，出来的结果就是一段规规矩矩的文本了。跟咱们之前手造假数据的格式一模一样。这也是为什么我前面建表的时候，字段顺序要按照时间、指标、数值来排。因为这样取出来的元组顺序就是对的，洗起来最方便。
这其实就是在做数据适配层的工作。你以后要是换了MySQL，或者是换了时序数据库，只要最后你能把数据拼成这种格式的字符串，后面的调用大模型的代码就可以一行不改。
五、 串联整个自动化的流水线
5.1 画个图看看完整链路
现在我们把各个零件都做出来了。该建库的建库了，该查数据的查了，该洗数据的洗了。最后一步就是把它们串起来。
在串之前，我们先用图把整个链路梳理清楚。
你看这个图，我中间加了一个判断“查到数据了吗？”。这是一个很关键的步骤。很多人写代码不讲究这个。如果数据库里碰巧这段时间没数据，查出来是个空列表。你如果不判断，直接就把空字符串丢给大模型。大模型懵了，可能给你回一句“你没有提供数据”。你这接口额度不就白白浪费了吗？
5.2 主函数怎么写，哪里容易报错
我们把主流程写出来。这里我们把之前写的 ask_timecho_ai 函数直接拿过来用，核心逻辑不变。
import requests
# 这里假设 ask_timecho_ai 函数已经定义好了，跟上一篇一样
# def ask_timecho_ai(question, api_key): ...
if __name__ == "__main__":
my_key = "sk-你的真实KEY粘贴在这里"
# 1. 初始化数据库（如果是第一次跑，这步会建库建表；如果已经跑过了，其实可以注释掉）
init_db()
print("\n")
# 2. 设定我们要查的时间范围
start = "2023-10-25 10:00:00"
end = "2023-10-25 10:04:00"
print(f"正在从数据库查询 {start} 到 {end} 的数据...")
# 3. 查数据
raw_results = query_data_from_db(start, end)
# 4. 关键判断：如果查不到数据，直接打住
if not raw_results:
print("警告：数据库里这段时间没数据，停止调用大模型。")
else:
print(f"查到了 {len(raw_results)} 条数据，开始转换格式...\n")
# 5. 洗数据
clean_text = format_db_results(raw_results)
# 6. 拼提示词（复用上节课的交叉分析逻辑）
prompt = f"""
你是一个资深的运维专家。下面是从数据库查出来的真实监控数据。
格式是：时间, 指标名称, 数值。
请对比CPU和Memory这两个指标在时间线上的变化，找出异常波动点并分析它们之间的关联性。
数据如下：
{clean_text}
"""
# 7. 调接口
print("正在请求TimechoAI进行真实数据分析...")
final_answer = ask_timecho_ai(prompt, my_key)
print("\n=== 真实数据交叉分析报告 ===")
print(final_answer)
这段主逻辑非常清晰。你对照着上面的流程图看，一步都没落下。
你跑一下这段代码。如果一切正常，你会在控制台看到大模型针对那几条真实数据库记录给出的分析。它会告诉你10:02和10:03这两个时间点，CPU和内存同时出现了异常飙升。
到这一步为止，其实你已经具备了一个初级AI运维工程师的代码能力了。你已经能把冷冰冰的数据库记录，变成一份有温度的分析报告了。
六、 真实场景下必踩的几个坑
6.1 时间范围没卡准，查出几十万条直接爆炸
刚才我们在演示的时候，只查了5分钟的数据，一共就10条。这很轻松。
但是你换到真实的生产环境里去。一个监控指标，一秒钟采一条，一天就是86400条。如果你手一抖，把 start_time 和 end_time 的跨度设成了一整年。你的 cursor.fetchall() 会瞬间把几千万条数据拉到你的Python进程的内存里。
接着你的程序就会卡死，或者直接报一个 MemoryError 然后崩溃退出。为什么会出现这个问题呢？原因在于你把所有数据一次性全捞出来了。
怎么解决呢？通常来说有两个办法。第一个办法是在SQL里加限制。比如加上 LIMIT 500，只取最近的500条。第二个办法是不用 fetchall()，改用 fetchmany(size=100) 分批去取。不过对于我们喂给大模型这个场景来说，因为大模型本身就有Token限制，你一次也不可能塞几千万条数据进去。所以最简单的办法就是在SQL里严格卡死时间范围，比如只查最近半小时，并且加上 LIMIT 兜底。
6.2 查出来是空列表，大模型会懵逼
这个坑我刚才在流程图里已经提过了，这里再强调一下它的严重性。
如果你的监控系统有故障，或者你的SQL写错了条件。导致查出来的 raw_results 是个空列表 []。你如果不做判断，直接走到 format_db_results 函数里，它会返回一个空字符串 ""。
然后这个空字符串被塞进了Prompt里发给大模型。大模型看到的就是“数据如下：\n（后面啥也没有）”。它不知道是你忘了放数据，它可能就会胡乱给你编一段分析，或者直接报错说输入格式不对。
而且这白白浪费了一次API请求的Token和等待时间。所以在主流程里加那个 if not raw_results: 的判断，是极其必要的防御性编程。
6.3 数据库连接没关导致锁表
刚才在写 query_data_from_db 函数的时候，我特意把 conn.close() 写在了 finally 块里。
在SQLite里，如果你打开了一个连接正在写数据，另一个连接是进不去的，它会被锁住。虽然在我们这个简单的单线程脚本里不太会遇到这种并发的锁表问题。但是如果你以后把这段查数据的代码放到了一个Web服务里，比如Flask或者Django里。多个用户同时访问，你如果不关连接，数据库马上就锁死了，谁都查不了。
所以，用完就关，这是一个必须刻在脑子里的习惯。千万别指望Python的垃圾回收机制去帮你关。它什么时候回收是不确定的，但是数据库的连接数是实打实被占着的。
七、 再唠叨一句API KEY的管理
7.1 别再硬编码了
你看我们上面写的代码，my_key = "sk-你的真实KEY粘贴在这里"。这种写法叫硬编码。
你在自己电脑上跑一跑就算了。如果你要把这个脚本传到公司的代码仓库里，你把真实的KEY写在代码里，这是非常危险的事情。只要有人能看这个仓库，你的KEY就泄露了。别人就可以拿你的KEY去刷接口，最后扣的是你的钱或者你们公司的钱。
你去后台 https://ai.timecho.com/settings/keys 看一眼，如果发现额度不对劲，第一件事就是去把旧的KEY删掉，重新生成一个。
正确的做法是把它设成系统环境变量。在你的操作系统里设一个叫 TIMECHO_API_KEY 的变量，把那串字符存进去。然后代码里这样写：
import os
# 从环境变量里把KEY读出来，如果没读到就给个空字符串
my_key = os.environ.get("TIMECHO_API_KEY", "")
if not my_key:
print("错误：找不到环境变量 TIMECHO_API_KEY，程序退出。")
exit()
这样你的代码里就干干净净，没有任何敏感信息了。把这个习惯养起来，对你未来的职业发展有好处。
7.2 去开发文档里确认最新的调用规范
虽然我们这套调用的代码现在跑得挺好。但是你要知道，API这些东西是官方说了算的。他们哪天要是升级了接口，改了参数名，你的代码可能就跑不通了。
所以遇到奇怪报错的时候，除了百度，一定要去这里看看：https://ai.timecho.com/docs/
看看有没有发什么更新公告，或者接口变更说明。别死盯着自己那几行代码找问题，有时候可能真不是你的锅，是上游变了。
八、 总结与下期预告
8.1 这一步打通意味着什么
今天这篇我们干了一件挺实在的事。我们把数据源从本地的假列表，切换到了真正的数据库。
我们学了怎么用最简单的SQLite建表插数据。学了怎么用游标去查数据。更重要的是，我们学了怎么把数据库里冷冰冰的元组，洗成大模型能看懂的自然语言文本。
这一步打通意味着什么呢？意味着这套代码已经脱离了玩具的范畴。你只要把那个 sqlite3 换成 pymysql，把SQL语句稍微改改，它就能直接连上你们公司企业层级里的正式MySQL数据库去干活了。它已经具备了落地的基础。
8.2 下期搞点刺激的
不过呢，现在我们调接口的方式还是有点傻。我们是发一次请求，然后干等着。代码卡在 requests.post 那里一动不动，直到大模型把几千字全想完了，才一口气返回来。
如果大模型思考的时间很长，比如要分析一个很复杂的数据，用户等个一两分钟，体验是很差的。正常的聊天软件不都是一个个字往外蹦的吗？
所以下一篇文章，我们要玩点刺激的。我们要去研究怎么调流式输出的接口。也就是Server-Sent Events (SSE)。我们要让大模型想出一个字，我们就立刻在控制台打印一个字，像打字机一样。
那个的话，代码的写法就跟现在完全不一样了。底层不能用简单的 requests.post，得用 stream=True 配合迭代去写。如果你觉得现在的等待过程太无聊了，那就一定要看下一期。我们下篇见。
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/weixin_43151418/article/details/164164009
