揭秘「零故障」运维：Prophet 时序预测与 AI 模型如何联手驯服服务器风险？

Prophet 预测原理简述

1. 加性模型（Additive Model）

Prophet 假设时间序列可以分解为：

y(t) = g(t) + s(t) + h(t) + ε_t

g(t)：趋势（Trend）分量，描述长期增长或下降趋势

s(t)：季节性（Seasonality）分量，描述周期性变化（如日、周、年周期）

h(t)：节假日（Holiday）分量，描述特殊事件对数据的影响

ε_t：误差项（Noise）

2. 趋势建模

Prophet 支持线性趋势和饱和增长（logistic growth）两种趋势模型。

可以自动检测并拟合趋势中的“转折点”（changepoints），即趋势发生变化的时间点。

3. 季节性建模

使用傅里叶级数（Fourier series）建模季节性成分，能灵活拟合各种周期性波动。

支持多种季节性（如日、周、年等），可自定义。

4. 节假日建模

用户可指定节假日或特殊事件，Prophet 会自动建模这些事件对数据的影响。

5. 鲁棒性与自动化

Prophet 对异常值和缺失值有较强的鲁棒性。

只需少量参数即可自动建模，适合非专业数据科学用户。

核心代码

1、zabbix API 获取历史数据def get_item_history(auth, itemid, time_from, time_till):    payload = {    "jsonrpc": "2.0",    "method": "history.get",    "params": {        "output": "extend",        "history": 0, # 0: float        "itemids": itemid,        "sortfield": "clock",        "sortorder": "ASC",        "time_from": time_from,        "time_till": time_till,        "limit": 1000            },      "auth": auth,      "id": 2    }    r = requests.post(ZABBIX_URL, json=payload)    result = r.json()['result']    print(f"[DEBUG] get_item_history for itemid={itemid}: {result[:3]} ... total={len(result)}")    return result

2、数据清洗def preprocess(data):    df = pd.DataFrame(data)    df['clock'] = pd.to_numeric(df['clock'], errors='coerce')    df['value'] = pd.to_numeric(df['value'], errors='coerce')    df = df.dropna(subset=['clock', 'value'])    if df.empty:        return pd.DataFrame(columns=['ds', 'y'])    df['clock'] = pd.to_datetime(df['clock'], unit='s')    df = df.set_index('clock').sort_index()    # 只保留 value 列    df = df[['value']]    # 强制 value 为 float64    df['value'] = df['value'].astype('float64')    # 重采样    df = df.resample('5min').mean().interpolate()    df = df.reset_index()    df = df.rename(columns={'clock': 'ds', 'value': 'y'})    return df

3、时序模型做本地趋势预测def forecast_trend(df, periods=24):    model = Prophet()    model.fit(df)    future = model.make_future_dataframe(periods=periods, freq='5min')    forecast = model.predict(future)    return forecast[['ds', 'yhat']].tail(periods)

4. 构造AI大模型解释Prompt(提示词工程)def build_prompt(df, forecast, metric_name="磁盘使用率", threshold=THRESHOLD, hostname="主机"):    history = df['y'].tolist()[-24:]    future = forecast['yhat'].tolist()    prompt = (    f"主机：{hostname}\n"    f"以下是该主机最近24小时的{metric_name}历史数据（每5分钟采样，单位%）：\n"    f"{history}\n"    f"Prophet模型预测未来2小时的{metric_name}为：\n{future}\n"    f"当前阈值为{threshold}%。这是磁盘空闲率指标，值越小说明可用存储空间越小，存储越紧张，需要尽快清理空间或者磁盘扩容，请判断该指标是否有异常降低的趋势，未来2小时内是否有可能持续低于阈值，低于阈值的时候就需要提醒该清理空间或扩容磁盘了，并给出预警建议。"    )    return prompt

5.调用OpenAI大模型def ask_ai(prompt):    client = OpenAI(api_key="sk-xxxxxx", base_url="https://api.deepseek.com")    messages = [{"role": "user", "content": prompt}]    response = client.chat.completions.create(    model="deepseek-reasoner",    messages=messages    )    reasoning_content = response.choices[0].message.reasoning_content    content = response.choices[0].message.content    return content

效果展示

以磁盘预警为例，当我设置磁盘空闲率为20%阈值时，磁盘初始化值使用率为80%左右，当我定时批量写入数据，将磁盘空闲率降低到22%左右，这时还没触发告警阈值，此时运行预警程序，则会提示，磁盘使用率将要达到阈值，即将触发报警

而当我开始清理磁盘时，空间逐步释放，这时候再次运行程序，则提示磁盘空间逐步释放，无法清理

总结

这是一次数据预测模型与AI模型相结合的尝试，经过前面几次尝试，发现，直接将源数据吐给大模型，并不能实现准确的分析与评估，需要数学模型结合，对数据进行清洗，然后由AI给出总结报告，这样得出的结论还是比较准备。