在这里插入图片描述

上一次我们聊了专栏的开篇。我们把最基础的API调用框架给搭起来了。你如果跟着敲了代码,应该已经能拿到大模型返回的那句“你好”了。

但是呢,那只是个开始。我们用TimechoAI,总不能真的只拿它来闲聊吧。我们是要处理时序数据的。所以今天这篇,我们要往前走一大步。我们要把一段真正的时间序列数据,塞进我们的代码里。然后让大模型帮我们找出这堆数据里的异常点。

这个的话,其实就是时序大模型最核心的一个用法。你把这套逻辑搞明白了,后面那些花里胡哨的功能,都是在这一套基础上加东西。

一、 为什么直接发数字给大模型是不行的

1.1 大模型其实是个“文盲”

在动手写代码之前,我们得先搞清楚一个概念。很多新手会有一个误解。他们觉得,既然叫时序大模型,那我直接把一组数字丢给它不就行了?

比如你直接在问题里写:“[10, 12, 11, 80, 13, 12],帮我找异常。”

你这么干,往往仅仅只会得到一个很笼统的回答。它可能会说“第4个数字明显偏大”。但是它不知道这代表什么。为什么看不懂呢?原因在于大模型的底层的机制。它本质上是在预测下一个词的概率。它对自然语言的理解能力很强,但是对纯数字序列的感知其实很弱。

这就好比你给一个人看一串密电码。如果没有密码本,他看半天也看不出个所以然来。大模型也是一样,你需要给它一个“密码本”。这个密码本是什么呢?就是上下文和格式。

1.2 TimechoAI推荐的输入格式长啥样

那怎么给它建上下文呢?你去看它的开发文档:https://ai.timecho.com/docs/ ,里面其实有提到怎么构造时序数据的输入。

通常来说,最稳妥的办法是把数字变成带有明确语义的文本。你不能只给数字,你得带上时间戳,还得带上单位,甚至带上指标的名字。

比如,你不要写“80”。你要写成“2023-10-24 10:05:00,CPU使用率是80%”。

你把数据变成这种一句一句的人话之后,大模型瞬间就能看懂了。它知道这是在说时间,这是在说指标,这是在说数值。它推理起来就准多了。

这其实也是写提示词的一个核心技巧。你不要指望模型去猜你的意图。你把格式规整得越像人话,它给出的结果就越靠谱。

二、 准备一份像样的测试数据

2.1 别去网上乱找,自己造最靠谱

既然要测,我们手里得有数据。你去网上找真实的数据集,往往还要下载、解压、清洗,太麻烦了。对于我们现在这个阶段,其实自己用代码造一批假数据是最省事的。

我们造什么样的数据呢?我们就造一个服务器的CPU监控数据。假设每分钟记录一次。我们造大概100个数据点。在这100个点里,前几十个都是正常的,在20%到30%之间波动。然后中间突然来一个飙升,飙到90%。接着又掉回正常水平。

这种数据最典型了。我们就用它来测试大模型能不能把那个飙升的点给揪出来。

2.2 用Python生成一段带刺的数据

我们打开上一次建的那个Python文件,或者新建一个也行。我们先写一段生成假数据的代码。

import random

def generate_fake_cpu_data():
    data_points = []
    base_time = "2023-10-24 10:00:"
    
    for i in range(100):
        # 拼接时间,假设从0秒开始,每次加60秒
        time_str = base_time + str(i * 60).zfill(2)
        
        # 造异常数据:在第50个点附近搞个突刺
        if i == 50:
            value = random.randint(85, 95) # 突然飙到90左右
        else:
            value = random.randint(20, 30) # 正常值在20到30之间
            
        data_points.append(f"{time_str}, CPU使用率, {value}%")
        
    return data_points

你看这段代码,逻辑其实很直白。我用了一个循环跑了100次。每次循环我都算出一个时间。然后我加了个判断。如果循环到了第50次,也就是 i == 50 的时候,我给它赋一个85到95之间的随机数。其他时候呢,我就给20到30之间的随机数。

最后我把它们拼成一个字符串。格式就是“时间, 指标名, 数值”。这种格式人眼一看就很舒服,大模型看着也舒服。

这里我用了 zfill(2) 这个小函数。这是干嘛的呢?因为秒数如果是个位数,比如“0秒”、“60秒”,直接拼上去就变成了“10:00:0”,不太好看。用这个函数补个零,就变成了“10:00:00”,规规矩矩的。这种小细节在处理时序数据的时候很重要。格式乱七八糟的话,模型很容易解析错。

三、 把数据揉进Prompt里面

3.1 纯文本拼接的方式

数据造好了,我们现在要把它塞给大模型。但是我们不能直接把那个列表丢进去。我们得用自然语言把它包起来。这就是构造Prompt的过程。

我们可以写一个专门的函数来干这件事。

def build_prompt(data_list):
    # 把列表里的每一行用换行符连起来,变成一个大字符串
    data_str = "\n".join(data_list)
    
    prompt = f"""
    下面是一段服务器监控的时序数据。格式是:时间, 指标名称, 数值。
    请你仔细分析这段数据,找出其中是否存在异常波动的点。
    如果有,请告诉我异常发生的时间点,以及当时的数值。
    最后简单分析一下可能的原因。
    
    数据如下:
    {data_str}
    """
    return prompt

你看,我先把那个列表用 "\n".join() 拼成了一个长字符串。每一行数据中间就会换一行。这样排布得很清晰。

然后我用了一个f-string。把刚才拼好的数据字符串,直接塞到了一段话的后面。这段话就是我们的指令。我告诉它格式是什么,我告诉它要找什么,我还告诉它最后要给个原因分析。

这种写法的话,其实就是在手把手教大模型怎么干活。你指令给得越细,它跑偏的概率就越低。

3.2 思考一下数据量的问题

这里我要插一句。你注意看,我们才造了100条数据。每条数据大概几十个字符。拼起来也就两千多字符。

这其实是个很聪明的做法。为什么这么说呢?因为大模型的输入是有长度限制的。这个限制叫上下文窗口。如果你一次丢一万条数据进去,它要么直接报错,要么就把前面的数据给忘了。

所以在真实的业务里,你不可能把一年的数据全丢给它。你往往只能截取其中一段。比如截取异常发生前后的半个小时数据。这就叫截取上下文窗口。这个概念我们后面会反复提到。今天你只要知道,别一次性塞太多数据就行了。

四、 改造上一讲的代码骨架

4.1 动态传入时序字符串

上一次我们写了一个 ask_timecho_ai 函数。那时候我们传进去的问题是写死的“什么是时序数据库”。现在我们要把它改一改,让它能接收我们刚才拼好的长字符串。

其实改动非常小。你根本不用动核心的请求逻辑。你只需要在调用的时候,把参数换掉就行。

我们把完整的代码整理一下。你看看现在长什么样了。

import requests
import random

def generate_fake_cpu_data():
    data_points = []
    base_time = "2023-10-24 10:00:"
    
    for i in range(100):
        time_str = base_time + str(i * 60).zfill(2)
        
        if i == 50:
            value = random.randint(85, 95)
        else:
            value = random.randint(20, 30)
            
        data_points.append(f"{time_str}, CPU使用率, {value}%")
        
    return data_points

def build_prompt(data_list):
    data_str = "\n".join(data_list)
    prompt = f"""
    下面是一段服务器监控的时序数据。格式是:时间, 指标名称, 数值。
    请你仔细分析这段数据,找出其中是否存在异常波动的点。
    如果有,请告诉我异常发生的时间点,以及当时的数值。
    最后简单分析一下可能的原因。
    
    数据如下:
    {data_str}
    """
    return prompt

def ask_timecho_ai(question, api_key):
    url = "https://ai.timecho.com/v1/chat/completions"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {api_key}"
    }
    payload = {
        "model": "timecho-model", 
        "messages": [
            {
                "role": "user",
                "content": question
            }
        ]
    }
    
    try:
        response = requests.post(url, headers=headers, json=payload, timeout=30)
        response.raise_for_status() 
        result_dict = response.json()
        answer = result_dict['choices'][0]['message']['content']
        return answer
    except requests.exceptions.HTTPError as err:
        return f"请求报错了,状态码是:{err}"
    except Exception as e:
        return f"发生了不知道什么错:{e}"

if __name__ == "__main__":
    # 这里填你自己的KEY
    my_key = "sk-你的真实KEY粘贴在这里" 
    
    # 1. 生成数据
    print("正在生成测试数据...")
    raw_data = generate_fake_cpu_data()
    
    # 2. 构造提示词
    print("正在构造提示词...")
    my_prompt = build_prompt(raw_data)
    
    # 3. 发送给大模型
    print("正在请求TimechoAI接口,请稍候...")
    final_answer = ask_timecho_ai(my_prompt, my_key)
    
    # 4. 打印结果
    print("\n=== 大模型的分析结果 ===")
    print(final_answer)

你对比一下,是不是很简单?核心的 ask_timecho_ai 函数我一行都没改。我只是在外面套了一层生成数据的逻辑,和一层拼提示词的逻辑。

另外你注意看,我把 timeout 从上次的10秒改成了30秒。为什么呢?因为这次我们发过去的数据量变大了。大模型读这么多数据,还要思考,是需要花点时间的。如果你还设10秒,很可能它还没算完,你这边就强行断开了,那就会报超时的错。

4.2 跑一下整个流程的框架图

在点运行之前,我们再用图把这次的流程梳理一遍。这次比上一次多了一步数据处理的过程。

Python生成假数据

拼接成指定格式的文本

加上指令封装成Prompt

通过API发给TimechoAI

大模型解析文本寻找突刺

返回异常时间点和原因分析

你看,前面三步A、B、C都是在我们自己电脑上完成的。这就叫数据预处理。你预处理做得越好,后面大模型干起活来就越轻松。

五、 跑一下看看,大模型能看出啥

5.1 观察它抓异常的准确度

现在你可以去跑这段代码了。如果你的网络没问题,KEY也没问题,等个十几秒,你应该能在控制台看到大模型的回答。

我跑出来的结果大概是这样的(每次跑可能稍微有点不一样,但大差不差):

“经过分析,这段数据在大部分时间里,CPU使用率稳定在20%到30%之间,属于正常波动范围。
但是,在 2023-10-24 10:50:00 这个时间点,CPU使用率突然飙升至89%(具体数值根据随机数变化),出现了一个非常明显的异常突刺。
除此之外,前后数据均恢复正常。”

你看,它精准地把那个第50个点给抓出来了。它不仅抓出来了,还把时间点给你翻译成了 readable 的格式。这其实就是大模型最厉害的地方。如果你用传统SQL去找,你得写个 WHERE value > 80 的语句。但大模型不需要你告诉它阈值是多少,它自己看一眼就知道那个点不对劲。

5.2 它给出的原因分析靠不靠谱

除了找异常点,我们在Prompt里还让它分析了原因。它通常会怎么回呢?它可能会编一些常见的原因。比如:

“可能的原因分析:

  1. 该时间点可能触发了某个定时任务,比如日志切割、数据备份,导致CPU瞬间被拉满。
  2. 可能是外界发起了突发的流量请求,或者有某个死循环的bug被触发了。
  3. 也有可能是垃圾回收(GC)机制被触发,导致短暂的CPU占用过高。”

你仔细看这段分析。它其实并不知道你的服务器上到底跑的什么程序。它是在“瞎猜”。但是它猜得很有逻辑。它把IT运维里常见的导致CPU突刺的原因给你列出来了。

这在实际工作里非常有用。老板看到报告,至少知道该往哪个方向去排查了。是去查定时任务,还是去查流量日志。这就是大模型提供的附加价值。它不仅仅给你数据,还给你提供排查思路。

当然,如果你觉得它分析得不够深入,你可以在Prompt里加一点前置条件。比如你告诉它“这是一台Java Web服务器”。那它分析的原因就会偏向于Full GC、线程死锁这些方向。这个我们后面的文章会细讲怎么调教它。

六、 数据太长被截断怎么办

6.1 Token限制是个大麻烦

前面我们只造了100条数据。那如果我造1000条呢?1万条呢?

你试着把循环里的 range(100) 改成 range(2000)。然后再跑一次。

这时候你大概率会遇到一个新的报错。或者即使没报错,它返回的结果也会变得很敷衍,比如只回你一句“数据太长无法分析”。为什么会出现这个情况呢?原因在于Token。

大模型处理文本不是按字数算的,是按Token算的。一个词可能是一个Token,一个数字也可能是一个Token。我们那个请求体,加上数据,加上返回的结果,总Token数不能超过模型的上限。

如果你塞进去的数据已经把输入的额度占满了,它就没有余力去生成输出的回答了。这就好比你让一个人看一本书,看完立刻让他写读后感。书太厚的话,他连看都看不完,哪还有精力写东西。

6.2 简单的降采样处理

那遇到这种长数据怎么办呢?通常来说,最简单的办法就是降采样。

什么意思呢?比如你本来有1万条每一秒的数据。你可以每隔10秒取一条。这样你就把数据压缩到了1000条。虽然损失了一些细节,但是整体的趋势和大的异常波峰是保留下来了。

我们改一下生成数据的代码,模拟一下降采样的逻辑。

def generate_downsampled_data():
    data_points = []
    base_time = "2023-10-24 10:00:"
    
    # 原本是每秒一条,现在我们每10秒记录一次
    for i in range(1000):
        real_seconds = i * 10 
        time_str = base_time + str(real_seconds).zfill(2)
        
        # 模拟一个持续了一小段时间的异常,而不是一个点
        if 500 <= i <= 510:
            value = random.randint(85, 95)
        else:
            value = random.randint(20, 30)
            
        data_points.append(f"{time_str}, CPU使用率, {value}%")
        
    return data_points

你看这里,我不仅把取值的间隔拉大了。我还把异常的情况改了一下。原来是一个点突刺,现在变成了连续11个点的高位运行。这种数据在真实世界里更常见。比如一个任务跑了十几秒才结束。

你把这段数据丢给模型,它依然能给你找出来。这就是降采样的好处。用少量的数据点,换取模型能够正常输出的空间。

当然了,降采样是一门学问。怎么采才能不把异常给采丢掉,这里面有很多算法。我们在这个专栏的中间部分,会专门花几篇文章讲各种降采样算法的Python实现。今天你就知道有这么个思路就行了。

七、 什么时候该用接口而不是网页

7.1 网页端的局限性

写到这儿,可能有朋友会问。既然你在 https://ai.timecho.com/realtime 这个应用示例页面也能输入数据让它分析,那我干嘛还要费劲写Python代码呢?

这个问题问得好。如果你只是偶尔看一眼数据,比如今天老板就让你查一次。那你直接去网页端,把数据粘贴进去,确实更快。

但是网页端有几个很致命的缺点。第一,你不能粘贴太多字。网页端的输入框往往有字数限制,你粘个几千行它就直接卡死了。第二,它没有记忆性。你查完CPU,想接着查内存,你得重新把上下文再描述一遍。

7.2 代码批处理的优势

用代码调接口就不一样了。一旦你把流程写成了脚本,你就可以玩批处理了。

什么意思呢?比如你有100台服务器。每台服务器都有一个监控数据文件。你在代码里写个for循环,遍历这100个文件。然后挨个去调API。最后把100台服务器的分析结果汇总成一个Excel表格。

这种活儿,你在网页端用手点,点断手都点不完。但是用代码跑,也就是喝杯咖啡的功夫。

而且你还可以把它接进你的报警系统里。比如Zabbix或者Prometheus发现某个指标超了,它不发邮件了,而是触发你的这个Python脚本。脚本直接去调TimechoAI,拿到一段人话分析,然后把这段人话发到你的企业微信或者钉钉群里。你一看消息就知道发生了什么,连图都不用点开看。

这才是时序大模型真正落地的样子。它不是一个单独的聊天框,它是你自动化流水线上的一个节点。

八、 常见报错再补充两个

8.1 报错:context_length_exceeded

我们在刚才改大循环数据量的时候,很有可能会碰到这个报错。或者类似的提示,说超出了最大上下文长度。

这个原因刚才其实已经解释过了。就是你塞的数据太长了。

怎么解决呢?第一个办法就是我上面说的降采样。把数据条数减少。第二个办法是检查你的Prompt。你是不是在Prompt里写了太多废话?把那些不必要的寒暄和解释都删掉,只留核心指令和数据,能省下不少Token。

如果还是不行,那说明这批数据真的太大了。你可能得考虑切分。比如把一天的 data 切成24份,每份一个小时,然后分24次去请求。最后再把结果拼起来。这个逻辑写起来也不复杂,就是多套一层循环的事。

8.2 报错:invalid_api_key的变体

有时候你会碰到一种很诡异的情况。你昨天跑代码还好好的,今天一跑就提示KEY无效。

你跑去后台 https://ai.timecho.com/settings/keys 看一眼,发现KEY还在啊,没被删。

这往往是为什么呢?原因在于你把它存到了一个txt文件里,然后读取文件的时候,可能不小心把结尾的换行符 \n 也读进去了。

你的KEY本来是 sk-abc123。结果读出来变成了 sk-abc123\n。你带着这个带换行符的KEY去请求,服务器当然认不出来。

怎么避免呢?在读取文件拿到KEY之后,习惯性地加一个 .strip() 方法。这个方法会把字符串首尾的空格和换行符全都删掉。

# 正确的读取KEY姿势
with open('key.txt', 'r') as f:
    my_key = f.read().strip()

这种小细节如果不注意,真的能卡你大半天。因为你总觉得逻辑没问题,根本想不到是多了个不可见字符。

九、 总结一下今天的进展

9.1 今天学到了什么核心逻辑

好了,今天这篇内容也比较多了。我们来梳理一下。

今天我们迈出了真正实战的第一步。我们知道了不能直接丢裸数字给大模型。我们要把数字包装成“时间, 指标, 数值”这样的文本格式。

我们学会了用Python自己造假数据,用来测试我们的逻辑。我们学会了怎么把数据和指令拼接成一个完整的Prompt。然后我们把它塞进了上一次写好的那个请求函数里,成功拿到了异常分析的结果。

我们还聊到了Token限制这个很现实的问题。知道了数据太长会报错,也知道了可以用降采样的办法来缓解这个问题。

这些其实都是时序大模型应用里最基础的“套路”。你把这套套路摸熟了,后面换什么模型,换什么接口,本质上的逻辑都是这一套。

9.2 下一期我们玩点更高级的

在下一篇文章,也就是第03篇里,我们不会只停留在找单一的异常点了。我们要让大模型做更复杂的任务。

我们要试着丢给它两条甚至三条不同的时序数据。比如同时给它CPU的数据、内存的数据、还有网络流量的数据。我们要让它做交叉分析。

比如我们要问它:“CPU飙升的时候,内存有没有跟着涨?网络流量有没有突增?”我们要看看它在面对多个变量关联分析的时候,表现到底怎么样。

那个的话,Prompt的写法又会不一样了。数据结构的组织也会更复杂一点。如果你觉得今天这块消化得差不多了,那就跟着我继续往下看吧。代码这个东西,光看不练是没用的。一定要自己敲一遍,把那个报错都踩一遍,你才算真正掌握了。我们下篇见。

更多推荐