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

告别造假数据直接连数据库查真实时序数据喂给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我们先写一段代码在当前目录下建一个数据库文件并且造一张表。这次我们不造随机数据了我们造一些模拟真实故障的固定数据。importsqlite3definit_db():# 连接到sqlite数据库如果文件不存在会自动创建connsqlite3.connect(monitor.db)cursorconn.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自带的。所以我们连库的时候一句话就搞定。importsqlite3如果你是连MySQL你还得去终端里敲pip install pymysql或者pip install mysql-connector-python。如果网络不好安装失败了你还得去搞镜像源。这些都是坑。我们用自带的库直接跳过这些坑。3.2 写一个查数据的函数别把逻辑混在一起我们写一个专门的函数用来从刚才那张表里查数据。为了灵活我们得允许传入查询的时间范围。你不能每次都把全表扫出来那样数据量大了肯定撑不住。defquery_data_from_db(start_time,end_time):# 连接数据库connsqlite3.connect(monitor.db)cursorconn.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))# 查出所有结果resultscursor.fetchall()returnresultsexceptExceptionase:print(f查数据库报错了{e})returnNonefinally:# 无论报不报错最后一定要关掉连接conn.close()这里面有几个细节我要唠叨一下。第一个是那个?号。你写SQL的时候千万不要直接用字符串拼接的方式把时间塞进去。比如fWHERE 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 写个洗数据函数转成字符串转换的逻辑很简单就是遍历这个列表然后把每个元素拼起来。defformat_db_results(results):ifnotresults:returntext_lines[]forrowinresults:# row[0]是时间row[1]是指标名row[2]是数值time_strrow[0]metricrow[1]valuerow[2]# 拼成我们熟悉的格式linef{time_str},{metric},{value}%text_lines.append(line)# 用换行符连起来return\n.join(text_lines)你看经过这么一洗出来的结果就是一段规规矩矩的文本了。跟咱们之前手造假数据的格式一模一样。这也是为什么我前面建表的时候字段顺序要按照时间、指标、数值来排。因为这样取出来的元组顺序就是对的洗起来最方便。这其实就是在做数据适配层的工作。你以后要是换了MySQL或者是换了时序数据库只要最后你能把数据拼成这种格式的字符串后面的调用大模型的代码就可以一行不改。五、 串联整个自动化的流水线5.1 画个图看看完整链路现在我们把各个零件都做出来了。该建库的建库了该查数据的查了该洗数据的洗了。最后一步就是把它们串起来。在串之前我们先用图把整个链路梳理清楚。没查到查到了定时任务触发/手动执行初始化数据库环境执行SQL查询指定时间段数据查到数据了吗?直接退出,不调接口省钱将元组列表格式化为文本拼接交叉分析提示词调用TimechoAI API打印大模型分析报告你看这个图我中间加了一个判断“查到数据了吗”。这是一个很关键的步骤。很多人写代码不讲究这个。如果数据库里碰巧这段时间没数据查出来是个空列表。你如果不判断直接就把空字符串丢给大模型。大模型懵了可能给你回一句“你没有提供数据”。你这接口额度不就白白浪费了吗5.2 主函数怎么写哪里容易报错我们把主流程写出来。这里我们把之前写的ask_timecho_ai函数直接拿过来用核心逻辑不变。importrequests# 这里假设 ask_timecho_ai 函数已经定义好了跟上一篇一样# def ask_timecho_ai(question, api_key): ...if__name____main__:my_keysk-你的真实KEY粘贴在这里# 1. 初始化数据库如果是第一次跑这步会建库建表如果已经跑过了其实可以注释掉init_db()print(\n)# 2. 设定我们要查的时间范围start2023-10-25 10:00:00end2023-10-25 10:04:00print(f正在从数据库查询{start}到{end}的数据...)# 3. 查数据raw_resultsquery_data_from_db(start,end)# 4. 关键判断如果查不到数据直接打住ifnotraw_results:print(警告数据库里这段时间没数据停止调用大模型。)else:print(f查到了{len(raw_results)}条数据开始转换格式...\n)# 5. 洗数据clean_textformat_db_results(raw_results)# 6. 拼提示词复用上节课的交叉分析逻辑promptf 你是一个资深的运维专家。下面是从数据库查出来的真实监控数据。 格式是时间, 指标名称, 数值。 请对比CPU和Memory这两个指标在时间线上的变化找出异常波动点并分析它们之间的关联性。 数据如下{clean_text}# 7. 调接口print(正在请求TimechoAI进行真实数据分析...)final_answerask_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(size100)分批去取。不过对于我们喂给大模型这个场景来说因为大模型本身就有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的变量把那串字符存进去。然后代码里这样写importos# 从环境变量里把KEY读出来如果没读到就给个空字符串my_keyos.environ.get(TIMECHO_API_KEY,)ifnotmy_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得用streamTrue配合迭代去写。如果你觉得现在的等待过程太无聊了那就一定要看下一期。我们下篇见。