先看代码结构

先把目录结构贴出来，朋友们有个整体感知。

monitor_agent
├── agent.py
├── llm.py
├── main.py
├── tools.py

这个结构非常小，但麻雀虽小，五脏俱全，它已经包含了一个 AIOps 场景最核心的四层：

• （1）入口层：main.py
• （2）编排层：agent.py
• （3）模型调用层：llm.py
• （4）外部数据/工具层：tools.py

如果用一句人话来解释这套结构，那就是：
用户提问题 → Agent 判断是不是线上问题 → 去告警系统和日志系统拿事实 → 把事实喂给大模型 → 大模型输出分析结论。

这就是一个非常典型的 AIOps 雏形。

程序入口：main.py

先看入口文件。

这段代码的作用，就是定义一个问题，然后把问题交给 ai_agent() 去处理，最后把结果打印出来。

from agent import ai_agent

if __name__ == "__main__":
    question = "现在线上出现了什么问题吗？"
    answer = ai_agent(question)
    print("AI 分析结果：\n")
    print(answer)

这个文件的职责非常单纯，核心只有两件事：

• （1）构造用户输入
• （2）调用 Agent，并输出结果

入口层不应该掺杂太多业务逻辑，它只负责启动流程。真正的分析、工具调用、模型推理，都应该下沉到其他模块里。

核心大脑：agent.py

这个项目里最关键的文件，就是 agent.py。

这段代码的作用，是把“意图识别、告警查询、日志查询、Prompt 构造、LLM 推理”串成一个完整流程。

from tools import (
    get_active_alerts,
    query_nginx_error_log,
    query_nginx_access_log
)
from llm import llm

def ai_agent(question: str) -> str:
    # 1. 判断用户意图
    if "线上" not in question:
        return "我只能回答线上运行状态相关问题"

    # 2. 查询告警
    alerts = get_active_alerts()
    if not alerts:
        return "当前无告警，线上运行正常"

    # 3. 查询日志
    error_log = query_nginx_error_log(alerts["service"])
    access_log = query_nginx_access_log(alerts["service"])

    # 4. 构造推理 Prompt
    prompt = f"""
你是一个经验丰富的 SRE。

当前检测到告警：
{alerts}

Nginx error.log：
{error_log}

Nginx access.log：
{access_log}

请分析：
1. 当前是否存在真实故障
2. 故障发生在哪一层（nginx / upstream / network）
3. 根因是什么
4. 给出修复建议

请基于日志，不要编造不存在的事实。
"""

    # 5. 调用 LLM
    result = llm(prompt)
    return result

1）意图识别

第一步是这句：

if "线上" not in question:
    return "我只能回答线上运行状态相关问题"

这其实是一个非常粗糙但有效的路由逻辑。
它的意思是：只有当用户的问题里包含“线上”两个字，Agent 才继续走后面的排障流程。

这个实现当然比较 demo，但它表达了一个重要设计思想：
Agent 在调用昂贵工具和大模型之前，先做任务分类。

否则的话，随手问一句“帮我写个 SQL”，它也去查告警、捞日志、调 LLM 分析线上故障，那 token 和接口费就属于原地升天。

2）查询告警

接下来是：

alerts = get_active_alerts()
if not alerts:
    return "当前无告警，线上运行正常"

这一步做了第一个事实收集。
在真实生产环境里，这里通常会接：

• Prometheus Alertmanager
• Grafana Alerting
• Zabbix
• 飞书/钉钉告警聚合接口

而这个 demo 里，返回的是一条模拟告警：

{
    "alertname": "nginx_5xx_high",
    "service": "backend-service",
    "since": "2026-03-02 09:41",
    "severity": "critical"
}

它给 Agent 提供了最基础的排障上下文，包括：

• 告警名称
• 受影响服务
• 发生时间
• 严重级别

3）查询日志

接着是日志部分：

error_log = query_nginx_error_log(alerts["service"])
access_log = query_nginx_access_log(alerts["service"])

获取nginx的日志，包括access.log与error.log，获取现场的第一手记录

4）构造 Prompt

再往后，就是这套 AIOps 方案的灵魂地带：Prompt 编排。

这里把告警、Nginx error 日志、access 日志统一拼成了一段结构化输入，并明确要求模型回答四件事：

• （1）是否是真实故障
• （2）故障位于哪一层
• （3）根因是什么
• （4）修复建议是什么

而且最后还加了一句：

请基于日志，不要编造不存在的事实。

这句非常重要。
因为排障场景和普通问答不一样，胡说八道的成本很高。
模型在创作场景里会发散，这是优点；
但在 SRE 场景里发散，就是事故二次伤害。

5）调用 LLM

最后才是模型调用：

result = llm(prompt)
return result

注意这个顺序很关键：
先拿事实，再让模型做归纳。
而不是上来就把用户问题扔给大模型，让它空口分析。

模型封装层：llm.py

接着看 llm.py。

这段代码的作用，是把大模型调用封装成一个统一函数，Agent 不用关心底层 API 细节。

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("API_BASE_URL")
)

MODEL = os.getenv("DEFAULT_MODEL")

def llm(prompt: str) -> str:
    response = client.chat.completions.create(
        model=MODEL,
        messages=[
            {
                "role": "system",
                "content": "你是一个严谨的 SRE，只能基于给定事实进行分析。"
            },
            {
                "role": "user",
                "content": prompt
            }
        ],
        temperature=0.2  # ⚠️ 排障一定要低温度
    )

    return response.choices[0].message.content

1）环境变量解耦

先看客户端初始化：

client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("API_BASE_URL")
)

这里做得比较灵活。
通过环境变量传入 api_key 和 base_url，说明它并不强绑定某一家模型平台。
只要兼容 OpenAI SDK 协议，不管后面接的是：

• OpenAI
• 阿里百炼
• DeepSeek 兼容网关
• LiteLLM
• 自建代理

都可以复用这一层。

2）System Prompt 约束

模型调用里有一段很值得注意：

"content": "你是一个严谨的 SRE，只能基于给定事实进行分析。"

这属于给模型戴“安全头盔”。虽然不能 100% 防幻觉，但至少明确了角色和边界。

对 AIOps 来说，这种约束最好再继续加强，比如补充：

• 不允许猜测未提供的数据
• 结论必须引用对应日志证据
• 证据不足时要明确说“不足以判断”
• 建议按置信度输出

3）低温度设置

最后这句，是整份代码里笔者最喜欢的一句注释：

temperature=0.2  # ⚠️ 排障一定要低温度

写文案、起标题、搞创意，温度高一点没问题；
但线上故障分析要的是稳定、克制、少脑补。

否则模型一兴奋，上一秒还是 upstream timeout，下一秒就能给你联想到数据库雪崩、交换机抖动、机房断电、宇宙射线翻转 bit 位，属实太会整活。

所以排障、审计、风控、配置审查这类任务，低温度是基本操作。

工具层：tools.py

这段代码的作用，是模拟告警平台和日志系统的查询接口。

def get_active_alerts():
    """
    模拟：从告警平台获取当前告警
    """
    return {
        "alertname": "nginx_5xx_high",
        "service": "backend-service",
        "since": "2026-03-02 09:41",
        "severity": "critical"
    }

def query_nginx_error_log(service: str):
    """
    模拟：查询 Nginx error.log
    """
    return """
    2026/03/02 09:41:12 [error] 1234#1234: *5678 upstream timed out
    (110: Connection timed out) while reading response header from upstream,
    upstream: "http://10.244.0.73:10000/test"
    """

def query_nginx_access_log(service: str):
    """
    模拟：查询 access.log 中的 5xx
    """
    return """
    10.0.0.1 - - [02/Mar/2026:09:41:12 +0000] "GET /test HTTP/1.1" 504
    10.0.0.2 - - [02/Mar/2026:09:41:13 +0000] "GET /test HTTP/1.1" 504
    """

1）get_active_alerts()

这个函数负责返回当前活动告警。
虽然现在是写死的模拟数据，但它对应的真实职责很明确：

• 从告警平台拉取当前触发中的告警
• 返回可用于后续分析的结构化数据

如果未来要接生产系统，这里可以对接 Prometheus Alertmanager API 或内部告警中心。

2）query_nginx_error_log(service)

这个函数负责拿 Nginx 错误日志。
返回的关键日志是：

upstream timed out
(110: Connection timed out) while reading response header from upstream

这个信息其实已经非常关键了。
它说明：

• Nginx 已经把请求转发给 upstream 了
• 问题不是客户端没连上 Nginx
• 问题也不是 Nginx 自己语法错误或者进程挂了
• 而是上游服务在规定时间内没有返回响应头

这类故障在线上特别常见，本质上就是：
网关层看到 504，但锅大概率不在网关本身，而在上游处理慢、阻塞、卡死、网络超时或线程池耗尽。

如果未来要接生产系统，这里可以对接 日志中心。

3）query_nginx_access_log(service)

这个函数负责返回访问日志里的 5xx 样本。
日志里连续出现两个 504：

"GET /test HTTP/1.1" 504

如果未来要接生产系统，这里可以对接 日志中心。

这套代码的执行链路，到底是怎么跑起来的？

很多朋友第一次看这种 Agent 代码，会觉得文件很少，但脑子里流程有点绕。这里笔者用人话串一遍。

第一步：用户发起问题

入口是：

question = "现在线上出现了什么问题吗？"

程序启动后，用户问题被传给 ai_agent(question)。

第二步：Agent 判断是否受理

agent.py 会先检查问题里是否包含“线上”。
如果没有，就拒绝受理。

这一步相当于一个非常简陋的 Intent Router。

第三步：拉告警

如果问题命中线上场景，Agent 就调用：

alerts = get_active_alerts()

拿到当前活动告警。

第四步：按服务维度查日志

拿到告警里 service=backend-service 之后，再继续查：

query_nginx_error_log(alerts["service"])
query_nginx_access_log(alerts["service"])

这一步把日志事实补齐。

第五步：把所有事实拼成 Prompt

Agent 会把“告警 + error.log + access.log”统一喂给大模型。

第六步：模型输出故障分析

最后由 llm.py 调用模型，产出一段排障结论。

整个链路可以用一张简化图表示：

用户问题
   ↓
main.py
   ↓
agent.py（意图识别 + 编排）
   ↓
tools.py（告警 + 日志）
   ↓
llm.py（模型推理）
   ↓
输出分析结果

这就是一个最小可运行的 AIOps Agent 闭环。

运行结果