HeitorBuscariolo/ai-incident-triage

# AI Incident 分诊 [](https://github.com/HeitorBuscariolo/ai-incident-triage/actions/workflows/ci.yml) 一个小型后端服务，用于监控应用日志流，通过滚动统计检测异常（错误率激增、延迟激增），并使用 LLM 将原始日志突发转换为结构化的根因诊断——这正是待命工程师在凌晨 3 点本需要手动进行的第一道筛查工作。 ``` log events queue worker ┌─────────┐ POST /logs ┌───────┐ ┌──────────────────┐ ┌─────────────────┐ │ services │ ───────────▶│ asyncio│─▶│ AnomalyDetector │──▶│ Groq (Llama 3.3) │ │ /simulator│ │ Queue │ │ rolling window per│ │ tool-forced JSON │ └─────────┘ └───────┘ │ service, error-rate│ └─────────────────┘ │ │ / latency baseline │ │ ▼ └──────────────────┘ ▼ SQLite: log_events │ SQLite: diagnoses ▼ SQLite: incidents ──▶ GET /incidents ``` ## 为什么开发这个项目 基于日志/指标的异常检测在*标记*问题方面已经是一个解决过的问题。最繁琐的部分是通读嘈杂的突发日志并形成关于*原因*的假设。这非常适合使用 LLM 来完成：有限的上下文、针对非结构化文本的模式匹配，以及在这种任务中，“我不确定，但这是我最好的猜测和置信度”是一个可以接受的答案。 ## 架构说明 - **Ingestion** (`app/main.py`)：`POST /logs` 会同步将事件写入 SQLite（以便稍后进行查询以获取上下文），并将其推送到进程内的 `asyncio.Queue` 中，供检测器异步消费。 这使得本演示无需依赖外部组件（不需要 Redis/Docker）——以后将队列替换为 Redis Streams 或 SQS 只是直接替换，而无需重写代码。 - **Detection** (`app/detector.py`)：基于服务的滚动时间窗口（30秒）。如果窗口内 ≥30% 的事件是错误，则触发 `error_rate_spike`；如果窗口的平均延迟超过根据前 20 个健康观测值建立的**冻结基线**的 3 倍，则触发 `latency_spike`。 基线被刻意冻结，而不是使用滑动平均值——否则缓慢的退化（如 `memory_leak` 场景）会拉高其自身的基线，从而永远无法达到阈值。冷却机制可以防止针对同一个持续发生的事件重复触发。 - **Diagnosis** (`app/diagnosis.py`)：触发时，worker 会提取事件窗口中的日志行，并将其发送到 Groq 的 Llama 3.3 70B，同时带有强制的工具调用 (`report_diagnosis`)，以确保输出始终是有效的结构化 JSON（`root_cause`、`confidence`、`suggested_action`、`evidence`）——绝不产生需要解析的自由格式文本。 - **Storage** (`app/db.py`)：通过标准库 `sqlite3` 使用纯 SQLite，包含三个表（`log_events`、`incidents`、`diagnoses`）。没有使用 ORM —— schema 足够小，直接使用原生 SQL 也具有很高的可读性。 ## 设置 ``` python -m venv .venv .venv/Scripts/activate # or source .venv/bin/activate on macOS/Linux pip install -r requirements.txt cp .env.example .env # then paste your GROQ_API_KEY into .env ``` 在 [console.groq.com](https://console.groq.com) 获取免费的 Groq API 密钥 — 无需信用卡。 ## 运行演示 启动服务器： ``` uvicorn app.main:app --port 8000 ``` 在另一个终端中，触发一个故障场景： ``` python simulator/generate_logs.py --scenario timeout_cascade --service payments-api python simulator/generate_logs.py --scenario bad_deploy --service checkout-api python simulator/generate_logs.py --scenario memory_leak --service inventory-api ``` 然后检查诊断结果 —— 可以是 JSON 格式，或者在浏览器仪表板中查看： ``` curl http://localhost:8000/incidents ``` 打开 **http://localhost:8000/** 查看一个小型的实时更新仪表板，其中包含事件及其诊断。 每个场景都模拟了不同的现实世界故障模式： | 场景 | 发生的情况 | 预期触发 | |---|---|---| | `timeout_cascade` | 下游服务开始超时，出现 504 错误 | `latency_spike` / `error_rate_spike` | | `bad_deploy` | 出现一条部署日志，随后是一连串的 500 错误 | `error_rate_spike` | | `memory_leak` | 延迟逐渐上升，没有错误 | `latency_spike` | 诊断输出示例： ``` { "service": "checkout-api", "trigger_reason": "error_rate_spike", "root_cause": "The recent deployment of version v1.4.2 introduced a bug in the CheckoutHandler.process() method, causing a NullReferenceException.", "confidence": "high", "suggested_action": "Revert the deployment to the previous version and review the changes made in v1.4.2 to identify and fix the bug." } ``` ## 运行测试 ``` pip install -r requirements-dev.txt pytest tests/ -v ``` 测试涵盖了独立运行的异常检测器（没有网络调用）：健康流量永远不会触发，错误率激增和延迟激增能够正确触发，冻结的基线不会被缓慢泄漏所拉高，冷却机制能抑制立即重复触发，并且滚动窗口之外的稀疏旧错误永远不会累积成误报。 ## 可能的扩展 - 将诊断结果发布到 Slack，而不是（或者除了）`/incidents` endpoint - 将进程内队列替换为 Redis Streams，以支持多个 worker 进程 - 添加一个人工反馈列 (`diagnoses.human_verdict`)，并随着时间的推移跟踪诊断准确率作为一种评估 - 提供一个最小化的仪表板 UI，而不是原始的 JSON

标签：后端服务, 安全规则引擎, 异常检测, 故障诊断, 脚本检测, 计算机取证, 运维监控, 逆向工具