让 LLM Agent 替你跑三个平台：告警排查自动化的落地思路

import json
import openai

# --- 工具定义 ---
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_logs",
            "description": "在日志平台搜索指定服务在指定时间窗口的关键词日志",
            "parameters": {
                "type": "object",
                "properties": {
                    "service": {"type": "string", "description": "服务名，如 order-service"},
                    "keyword": {"type": "string", "description": "搜索关键词，如 timeout、error"},
                    "start_time": {"type": "string", "description": "开始时间，ISO 8601 格式"},
                    "end_time": {"type": "string", "description": "结束时间，ISO 8601 格式"},
                },
                "required": ["service", "keyword", "start_time", "end_time"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "query_metrics",
            "description": "查询 APM 平台指定服务的指标曲线",
            "parameters": {
                "type": "object",
                "properties": {
                    "service": {"type": "string", "description": "服务名"},
                    "metric": {"type": "string", "description": "指标名，如 p99_latency、error_rate"},
                    "start_time": {"type": "string", "description": "开始时间"},
                    "end_time": {"type": "string", "description": "结束时间"},
                },
                "required": ["service", "metric", "start_time", "end_time"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_trace_detail",
            "description": "根据 trace ID 获取链路追踪详情",
            "parameters": {
                "type": "object",
                "properties": {
                    "trace_id": {"type": "string", "description": "链路追踪 ID"},
                },
                "required": ["trace_id"],
            },
        },
    },
]

# --- 工具执行（模拟，实际替换为真实 API 调用） ---
def execute_tool(name: str, args: dict) -> str:
    if name == "search_logs":
        # 实际: 调用 Elasticsearch / Loki API
        return json.dumps({
            "total": 12,
            "hits": [
                {"timestamp": "2024-01-15T01:58:03Z", "level": "ERROR",
                 "message": "upstream timeout: service-B /api/data took 3200ms"},
                {"timestamp": "2024-01-15T01:58:04Z", "level": "WARN",
                 "message": "retrying call to service-B, attempt 2/3"},
            ],
            "trace_ids": ["trace-abc123", "trace-def456"],
        })

    if name == "query_metrics":
        # 实际: 调用 Prometheus / Grafana API
        return json.dumps({
            "service": args["service"],
            "metric": args["metric"],
            "datapoints": [
                {"time": "01:57", "value": 120},
                {"time": "01:58", "value": 3200},  # 延迟飙升
                {"time": "01:59", "value": 150},
                {"time": "02:00", "value": 110},    # 恢复
            ],
        })

    if name == "get_trace_detail":
        # 实际: 调用 Jaeger / SkyWalking API
        return json.dumps({
            "trace_id": args["trace_id"],
            "root_span": {"service": "service-B", "operation": "/api/data",
                          "duration_ms": 3200, "status": "ERROR"},
            "spans": [
                {"service": "service-B", "operation": "GC pause",
                 "duration_ms": 2800, "tags": {"gc.type": "Full"}},
            ],
        })

    return json.dumps({"error": f"unknown tool: {name}"})

# --- Agent 主循环 ---
client = openai.OpenAI()  # 需设置 OPENAI_API_KEY

def run_alert_agent(alert_event: str, max_rounds: int = 5) -> str:
    """接收告警描述，自动排查并返回结论"""

    messages = [
        {
            "role": "system",
            "content": (
                "你是一个告警排查助手。根据告警信息，依次调用日志搜索、指标查询、"
                "链路追踪工具来收集数据，然后综合分析给出根因结论。"
                "结论格式：根因、影响范围、持续时间、是否自愈、建议动作。"
            ),
        },
        {"role": "user", "content": alert_event},
    ]

    for _ in range(max_rounds):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto",
        )

        msg = response.choices[0].message

        # 如果 LLM 不再调用工具，说明分析完成
        if not msg.tool_calls:
            return msg.content or "排查完成，但未生成结论。"

        # 执行所有工具调用（可并行，这里简化为顺序执行）
        messages.append(msg)
        for tc in msg.tool_calls:
            result = execute_tool(tc.function.name, json.loads(tc.function.arguments))
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": result,
            })

    return "排查轮次超限，请人工介入。"

# --- 运行 ---
alert = """
告警事件：
- 服务: order-service
- 时间: 2024-01-15 01:57 ~ 02:00
- 异常: P99 延迟从 120ms 飙升至 3200ms，错误率同步上升
- trace 样本: trace-abc123
请排查根因。
"""

report = run_alert_agent(alert)
print(report)

根因：上游服务 service-B 在 01:58 发生 Full GC 停顿（耗时 2800ms），
导致下游 order-service 调用 /api/data 超时。

影响范围：order-service 的 /api/data 接口，P99 延迟峰值 3200ms。
持续时间：约 1 分钟（01:58 ~ 01:59）。
是否自愈：已自愈，02:00 指标恢复正常。
建议动作：无需紧急干预；建议优化 service-B 的 GC 配置，
考虑增大堆内存或切换到 G1/ZGC 收集器以减少 Full GC 频率。