用LLM构建分布式系统诊断工具：日志分析实战

我们要构建的是一个分布式日志诊断工具——一个用 LLM 来定位跨服务故障的小型 Python Agent。当在十个容器上 tail 日志已经无法满足需求时，这个工具就能派上用场。

依赖安装：

pip install openai

还需要一个来自 https://portal.oxlo.ai 的 Oxlo.ai API Key。

步骤 1：模拟微服务集群

需要一个被监控对象，所以创建了三个假服务。payment-service 被配置为故障状态，这样就能有一个真实的分布式故障可供检测。

import random
from datetime import datetime, timedelta

class ServiceNode:
    def __init__(self, name, failure_mode=False):
        self.name = name
        self.healthy = True
        self.logs = []
        self.failure_mode = failure_mode

    def generate_logs(self, count=5):
        self.logs = []
        for i in range(count):
            ts = (datetime.utcnow() - timedelta(seconds=i * 15)).isoformat()
            if self.failure_mode:
                msg = f"{ts} ERROR {self.name} connection reset by peer, upstream timeout"
                self.healthy = False
            else:
                msg = f"{ts} INFO {self.name} health=OK req_id={random.randint(1000, 9999)}"
            self.logs.append(msg)
        return "\n".join(self.logs)

nodes = [
    ServiceNode("user-service"),
    ServiceNode("payment-service", failure_mode=True),
    ServiceNode("notification-service"),
]

步骤 2：从所有节点收集日志

协调器从每个节点收集最新日志，汇聚成一个统一的上下文块，这样模型就能看到整个集群的状态。

def collect_cluster_state(nodes):
    blocks = []
    for node in nodes:
        logs = node.generate_logs(count=6)
        blocks.append(f"--- {node.name} ---\n{logs}")
    return "\n\n".join(blocks)

cluster_state = collect_cluster_state(nodes)
print(cluster_state)

步骤 3：定义 System Prompt

需要结构化的推理，而不是散文式的输出。这个 prompt 告诉模型扮演 SRE 角色，并以我们可解析的固定格式回复。

SYSTEM_PROMPT = """You are a distributed systems SRE. You are given logs from multiple microservices.
Analyze the logs to identify the root cause. Respond in this exact format:

Root Cause: 
Affected Services: 
Recommended Action: 
Confidence: 

Be concise. If no issue is found, state "No issue detected"."""

步骤 4：用 Oxlo.ai 进行诊断

将完整的日志包发送给 Oxlo.ai 上的 Llama 3.3 70B。由于 Oxlo.ai 采用扁平化的按请求定价，从三个服务拉取六行日志并不会增加调用成本。当从三个服务扩展到三十个服务时，这就很重要了。

from openai import OpenAI

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

def diagnose(state: str) -> str:
    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": state},
        ],
    )
    return response.choices[0].message.content

diagnosis = diagnose(cluster_state)
print(diagnosis)

步骤 5：根据诊断结果执行操作

最后，解析结构化的回复结果，对模拟集群执行建议的操作。在生产环境中，这个 hook 会调用 kubectl 或你的容器编排工具。

def act(diagnosis: str, nodes):
    affected = []
    action = "none"
    for line in diagnosis.splitlines():
        if line.startswith("Affected Services:"):
            raw = line.split(":", 1)[1]
            affected = [s.strip() for s in raw.split(",")]
        if line.startswith("Recommended Action:"):
            action = line.split(":", 1)[1].strip().lower()

    for node in nodes:
        if node.name in affected and action == "restart":
            node.healthy = True
            node.failure_mode = False
            node.logs = []
            print(f"Executed restart on {node.name}")

act(diagnosis, nodes)

print("\nPost-recovery state:")
print(collect_cluster_state(nodes))

将所有代码保存到 diagnostician.py，替换 YOUR_OXLO_API_KEY，然后运行：

python diagnostician.py

输出结果：

--- user-service ---
2024-05-20T14:12:00 INFO user-service health=OK req_id=4821
2024-05-20T14:11:45 INFO user-service health=OK req_id=3912
...

--- payment-service ---
2024-05-20T14:12:00 ERROR payment-service connection reset by peer, upstream timeout
2024-05-20T14:11:45 ERROR payment-service connection reset by peer, upstream timeout
...

--- notification-service ---
2024-05-20T14:12:00 INFO notification-service health=OK req_id=9912
...

Root Cause: payment-service is failing due to upstream timeouts likely caused by connection resets.
Affected Services: payment-service
Recommended Action: restart
Confidence: high

Executed restart on payment-service

Post-recovery state:
--- payment-service ---
2024-05-20T14:12:45 INFO payment-service health=OK req_id=5543
...

下一步

接下来可以将 ServiceNode 类连接到真实的 HTTP 健康检查端点，这样 Agent 就能监控真实的容器了。如果想更进一步，可以将日志通过 Oxlo.ai 的 embedding 端点发送，存储到向量数据库中，这样 Agent 在诊断前就能检索相似的历史故障案例。