用 Pydantic 定义诊断 schema,结合 OpenAI SDK 和 Oxlo 平台,将堆栈跟踪和错误日志转化为结构化诊断结果和置信度评分。
我正在构建一个结构化的故障排查 Agent,它会摄入堆栈跟踪、错误日志和源代码,然后返回排序后的诊断结果和修复建议。它帮助后端工程师缩短复现生产环境疑难 Bug 的时间。我们会将其接入 Oxlo.ai,这样即使是大体积的日志负载也只需按请求数付固定费率。
pip install openai pydantic
需要一个来自 https://portal.oxlo.ai 的 Oxlo.ai API 密钥。
Oxlo.ai 客户端使用标准的 OpenAI SDK,因此不需要额外的适配器。
为了保持 Agent 的可复现性,在向 API 发送任何内容之前,我用 Pydantic 强制执行 JSON schema。这保证了每条诊断都包含相同的字段。
import json
from pydantic import BaseModel, Field
from typing import List
class Diagnosis(BaseModel):
summary: str = Field(description="One-line description of the bug")
root_cause: str = Field(description="Detailed explanation of why it happened")
affected_files: List[str] = Field(description="List of file names or paths involved")
suggested_fix: str = Field(description="Concrete code or configuration change")
confidence: int = Field(description="Integer 1-10, 10 being certain")
系统提示词扮演一位拒绝猜测的高级工程师。它要求逐步推理,且只输出 JSON。
SYSTEM_PROMPT = """You are a senior site-reliability engineer diagnosing production issues.
You will receive an error log, a code snippet, and a list of dependencies.
Follow these rules:
1. Reason silently about the most likely root cause before proposing a fix.
2. Consider race conditions, dependency mismatches, and environment assumptions.
3. Respond with valid JSON only, no markdown fences, no commentary outside the JSON.
4. Use this exact schema:
{
"summary": "...",
"root_cause": "...",
"affected_files": ["..."],
"suggested_fix": "...",
"confidence": 0
}
"""
原始日志充满噪音,所以我去掉尾部空白,并用清晰的标签包裹每个部分,以减少模型的歧义。
def build_user_message(error_log: str, code_snippet: str, dependencies: str) -> str:
return (
"<error_log>\n"
f"{error_log.strip()}\n"
"</error_log>\n\n"
"<code_snippet>\n"
f"{code_snippet.strip()}\n"
"</code_snippet>\n\n"
"<dependencies>\n"
f"{dependencies.strip()}\n"
"</dependencies>\n\n"
"Diagnose the issue and return the JSON object described in your instructions."
)
我将组装好的负载发送给 Oxlo.ai。我使用 Llama 3.3 70B 以获得可靠的结构化推理,但你也可以换用 DeepSeek R1 671B 或 Qwen 3 32B 来获得更强的逻辑能力。由于 Oxlo.ai 按请求数而非按 token 数计费,我可以包含完整的堆栈跟踪而无需担心长度。详细信息见 https://oxlo.ai/pricing。
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
def run_diagnosis(user_message: str) -> str:
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
)
return response.choices[0].message.content
通过 Pydantic 解析 JSON 可以尽早捕获幻觉字段,并生成一份干净的报告以便粘贴到工单中。
def print_diagnosis(raw_json: str) -> None:
d = Diagnosis.model_validate_json(raw_json)
print(f"Summary : {d.summary}")
print(f"Confidence : {d.confidence}/10")
print(f"Root Cause : {d.root_cause}")
print(f"Affected : {', '.join(d.affected_files)}")
print(f"Suggested Fix:\n{d.suggested_fix}")
以下是一个端到端的真实测试,包含一个 Python 堆栈跟踪和一个有 Bug 的函数。
if __name__ == "__main__":
ERROR_LOG = """Traceback (most recent call last):
File "/app/worker.py", line 42, in process_event
user = payload["user"]["profile"]
KeyError: 'profile'"""
CODE = """import json
def process_event(raw: str):
payload = json.loads(raw)
user = payload["user"]["profile"]
return user["email"]"""
DEPS = "python 3.11, pydantic 2.5, redis 5.0"
msg = build_user_message(ERROR_LOG, CODE, DEPS)
raw = run_diagnosis(msg)
print_diagnosis(raw)
输出结果:
Summary : Missing profile key in user payload causes KeyError in process_event
Confidence : 9/10
Root Cause : The code assumes the nested key "profile" always exists under "user", but the upstream event schema does not guarantee it.
Affected : worker.py
Suggested Fix:
Use .get() or validate the schema before accessing nested keys.
Example: user = payload.get("user", {}).get("profile")
Alternatively, define a Pydantic model for the payload and validate it on entry.
将 run_diagnosis 函数接入 Slack Bot,这样值班工程师可以粘贴日志并在几秒内获得分类线程。你还可以加入第二轮,使用 Oxlo.ai 上的 DeepSeek R1 671B 来审查任何置信度低于 7 的诊断,并提出替代假设。