用LLMOps构建生产级AI告警分诊Agent

我们将在 Oxlo.ai 上构建一个值班事故分级 Agent，它读取合成服务遥测数据并返回结构化的严重程度评估和修复步骤。它帮助平台团队自动化事故调查的前五分钟，并减少告警疲劳。

在开始之前，确保准备好以下内容：

一个来自 https://portal.oxlo.ai 的 Oxlo.ai API Key

OpenAI SDK：pip install openai

步骤 1：设置 Oxlo.ai 客户端

我在模块级别一次性配置客户端，并在每个请求中复用它。Oxlo.ai 暴露了一个完全兼容 OpenAI 的端点，因此标准 SDK 无需任何适配器代码即可工作。

from openai import OpenAI
import json

client = OpenAI(
    base_url="https://api.oxlo.ai/v1",
    api_key="YOUR_OXLO_API_KEY"
)

步骤 2：定义系统提示词

系统提示词是 Agent 的职位描述。我将其输出格式明确化，以便后续可以程序化解析响应。

SYSTEM_PROMPT = """You are an expert Site Reliability Engineer. Your job is to triage incoming incidents based on the provided telemetry.

Analyze the service metrics, error logs, and deployment history. Respond with a single JSON object containing exactly these keys:
- severity: one of "critical", "high", "medium", "low"
- summary: a one-sentence description of the issue
- root_cause: your best guess at the underlying cause
- remediation_steps: an ordered list of actionable strings
- affected_services: a list of service names that might be impacted

Be concise. Do not include markdown formatting or explanation outside the JSON."""

步骤 3：模拟遥测数据摄入

在生产环境中，这个辅助函数会查询 Prometheus、Datadog 或 Splunk。对于本教程，我编写了一个小型函数，为虚构的微服务返回真实的遥测数据。

def fetch_telemetry(service: str):
    # Simulated data source. Replace with real API calls to your observability stack.
    return {
        "service": service,
        "timestamp": "2025-01-15T14:32:00Z",
        "cpu_percent": 94.2,
        "memory_percent": 87.5,
        "error_rate_5m": 12.4,
        "p99_latency_ms": 2300,
        "recent_deploy": "payment-service:v2.3.1 deployed 14 minutes ago",
        "error_logs": [
            "Connection timeout to inventory-db after 3000ms",
            "Retry exhaustion on /api/v1/charge"
        ]
    }

步骤 4：构建用户消息

在将遥测数据交给模型之前，我将其格式化为纯文本报告。保持消息结构化但人类可读可以提高解析可靠性。

def build_user_message(telemetry: dict) -> str:
    lines = [
        f"Service: {telemetry['service']}",
        f"Time: {telemetry['timestamp']}",
        f"CPU: {telemetry['cpu_percent']}%",
        f"Memory: {telemetry['memory_percent']}%",
        f"5m Error Rate: {telemetry['error_rate_5m']}%",
        f"P99 Latency: {telemetry['p99_latency_ms']}ms",
        f"Recent Deploy: {telemetry['recent_deploy']}",
        "Recent Error Logs:",
    ]
    for log in telemetry["error_logs"]:
        lines.append(f"  - {log}")
    return "\n".join(lines)

步骤 5：运行分级 Agent

现在我将所有内容连接起来。我将格式化的遥测数据发送到 Oxlo.ai 上的 Llama 3.3 70B，并将结果解析为 JSON。因为系统提示词锁定了输出格式，所以我可以将响应当作 API 契约来处理。

def triage(service: str):
    telemetry = fetch_telemetry(service)
    user_message = build_user_message(telemetry)

    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_message},
        ],
    )

    content = response.choices[0].message.content.strip()
    return json.loads(content)

if __name__ == "__main__":
    result = triage("payment-api")
    print(json.dumps(result, indent=2))

针对模拟的 payment-api 事故执行脚本，会立即生成一份结构化报告。通过 Oxlo.ai 调用 Llama 3.3 70B 没有冷启动问题，因此第一次请求和第十次请求的返回速度一样快。

{
  "severity": "critical",
  "summary": "Payment API is experiencing connection timeouts and retry exhaustion following a recent deployment.",
  "root_cause": "The v2.3.1 deploy likely introduced a regression in database connection pooling or timeout configuration.",
  "remediation_steps": [
    "Check connection pool settings in payment-service:v2.3.1",
    "Verify inventory-db health and network latency",
    "Consider rolling back to payment-service:v2.3.0",
    "Scale payment-api replicas horizontally if db connections are exhausted"
  ],
  "affected_services": [
    "payment-api",
    "inventory-db"
  ]
}

这就是 LLMOps 事故分级管线的核心。下一步是将这个函数包装在 FastAPI 处理器中，并连接到 PagerDuty Webhook，这样每个高优先级告警都会触发自动预分析。你还可以将过去的事故报告存储在向量数据库中，并将三个最相似的历史案例预置到系统提示词中，为 Oxlo.ai 提供 few-shot 上下文。