AkshayShah03/forge

# Forge **一个由 AI agent 团队驱动的事件响应系统。只需提供警报和日志路径，它就会自动进行调查，找出根本原因，并起草事后分析报告。** ## 它的实际功能 当生产环境警报触发时，通常的流程是：某人被唤醒，SSH 登录到服务器，滚动浏览日志，检查最近的部署，提出假设，验证假设，然后再花一个小时编写事后分析报告。这个过程大部分是机械化的。真正有价值的部分在于最后的专业判断。 Forge 负责处理机械化的部分。你只需描述事件（什么崩溃了、何时发生、日志在哪里、哪个服务受到影响），系统就会启动三个并行运行的 agent： * **研究员 (researcher)** 搜索已知的故障模式和针对该警报类型的操作手册指南 * **程序员 (coder)** 读取日志文件，计算延迟百分位，统计每个请求的数据库查询次数，并查询 git 历史记录，以找到与异常发生时间相关的部署 * **分析师 (analyst)** 接收这两组调查结果，按置信度对根本原因假设进行排序，并起草结构化的事后分析报告 事后分析报告会经过一个评估 agent 的处理，它会根据一套标准对其进行打分：根本原因是否具体（一个 commit hash、一个文件、一次查询）？是否引用了证据（日志中的实际数据）？如果分数低于阈值，orchestrator 会根据反馈重新规划并再次运行。 最终输出的是一份真正可用的事后分析报告，而不是一句简单写着“观察到延迟增加”的废话总结。 ## 架构 ``` Incident description (what broke, log path, repo path) | v Orchestrator — plans the investigation into 3 subtasks | +——— Researcher (web search for known causes + runbook patterns) | +——— Coder (reads logs, parses latency stats, queries git history) | v (analyst waits for both above to finish) Analyst (correlates findings, ranks hypotheses, writes postmortem) | v Critic (scores on specificity + evidence + completeness) | +——— score < 0.75: re-plan and retry (up to 3 iterations) | +——— score >= 0.75: return final postmortem ``` 每个 agent 节点在移交给下一步之前，都会将其状态写入检查点。如果调查过程中进程崩溃，它会从上次中断的地方继续执行。 ## 快速开始 你需要一个 Groq API 密钥（可在 console.groq.com 免费获取）或在本地运行的 Ollama。不需要任何其他外部服务。 **使用 Groq：** ``` git clone https://github.com/AkshayShah03/forge cd forge uv venv .venv --python 3.11 && uv pip install -r requirements-azure.txt GROQ_API_KEY=your_key PYTHONPATH=. .venv/bin/uvicorn api.azure_main:app --reload ``` 打开 `http://localhost:8000`。UI 中预填了示例事件，可直接进行试用。 **使用 Ollama（无需 API 密钥）：** ``` ollama pull qwen2.5:3b PYTHONPATH=. .venv/bin/uvicorn api.azure_main:app --reload ``` 注意：较小的模型（3B）可以处理简单的事件，但在处理多步骤工具调用时会比较吃力。使用 `llama3.1:8b` 或更大的模型会获得更好的效果。 ## 如何使用 当你提供具体细节时，系统的效果最好。与其说“我们的 API 很慢”，不如尝试这样输入： ``` P95 latency in checkout-service jumped from 45ms to 385ms at 2026-06-08T14:32Z. A deploy (v2.4.1) finished at 14:02. Logs are at /var/log/checkout/. The repo is at /home/app/checkout-service. Investigate and write a postmortem. ``` coder agent 会找到日志文件，计算延迟统计数据，并检查在那个时间点代码仓库中发生了什么变化。analyst 会将异常时间戳与部署窗口进行匹配，并尝试确定导致问题的具体 commit 或代码变更。 你也可以运行预置的 demo，它会生成模拟日志并提交一个逼真的事件： ``` python scripts/demo_incident.py ``` 这会在 `/tmp/forge-demo/logs/` 目录中创建日志文件，其中包含一个模拟的 N+1 查询性能衰退（每个请求的数据库查询次数在已知的时间戳从 2 次跃升至 23 次）以及附带的部署日志。系统无需任何额外设置即可对此进行调查。 ## 技术决策 **使用 LangGraph 而非循环。** 普通的异步循环处理正常路径（happy path）是没有问题的。LangGraph 提供了带有类型化 reducer 的已编译状态机，这意味着路由函数是纯函数，并且可以使用普通的 dict 进行测试。当评估器让某个结果不合格并将其路由回 orchestrator 时，该路径在图中是清晰明确的，而不是隐藏在循环体中的 `if/else` 语句。 **依赖追踪。** 分析师依赖于研究员和程序员。orchestrator 在路由到下一个 agent 之前，会通过检查 `subtask_results` 中包含哪些子任务来追踪这种依赖关系。在当前的实现中，研究员和程序员是顺序运行的（并行执行已列入计划），但这种依赖契约意味着它们可以在不更改分析师或评估器代码的情况下实现并行化。 **作为质量门控的 Critic。** 基于规则的检查只能告诉你输出是否为格式良好的 JSON。它无法判断“根本原因是流量过高”是否是一份有价值的事后分析。Critic 会读取原始的事件描述和完整的 agent 输出，并像资深工程师审查草稿一样对其进行打分。它生成的反馈（“根本原因太模糊，没有引用 commit hash”）能够以一种单纯的数字分数无法实现的方式驱动重试。 **使用带有会话的 Azure Service Bus。** 按 `user_id` 实现 FIFO。如果你提交了多个事件，它们将按提交顺序运行，而不会阻塞其他用户。Worker 每隔 50 秒续订一次 Service Bus 消息锁，因为 agent 的运行时间超过了默认的 60 秒锁超时时间。如果不这样做，消息将重新进入队列，而第二个 worker 会在运行中途将其拾取。 ## 各 agent 可用的工具 | Agent | 工具 | |---|---| | Orchestrator | web 搜索 | | Researcher | web 搜索, RAG 检索, 文件读取 | | Coder | Python 执行, 文件读/写, 日志文件读取器, 目录列表, git log 查询 | | Analyst | SQL 查询, Python 执行, RAG 检索, 日志文件读取器 | | Critic | 无（纯 LLM 评估） | 工具的作用域在 `agent_system/tools/registry.py` 中强制执行。添加新工具意味着添加一个函数，并在应拥有该功能的角色的 scope 中将其列出。 ## 运行测试 ``` .venv/bin/python -m pytest tests/ -v --tb=short ``` 93 个测试，全在一秒以内完成。路由函数是纯函数，因此单元测试无需 mock。集成测试使用带有预设响应的 `FakeChatModel`，因此整个图的运行无需调用任何 API。 测试套件涵盖： * 所有路由逻辑（orchestrator、子 agent、critic、重试循环） * 三种新工具（`read_log_file`、`list_log_files`、`query_git_log`），包括不存在的路径和零小时 git 时间窗口等边缘情况 * orchestrator 和 critic 输出的 JSON 解析鲁棒性（markdown 代码围栏、内嵌 JSON、格式错误的响应） * 完整的 3-agent 事件工作流端到端测试 * 依赖阻塞和解除阻塞（分析师等待研究员和程序员双方完成） * API endpoint（提交、轮询、SSE 流、历史记录） * Service Bus worker（消息解析、锁续订、达到最大投递次数后转入死信队列） ## 局限性 Web 搜索工具需要 Tavily API 密钥。如果没有，研究员将回退到模型的训练知识，这对于常见的故障模式（N+1 查询、GC 压力、锁竞争）很有效，但不适用于特定服务或最近发生的问题。 评估器每次迭代大约会增加一次额外的 LLM 调用。对于你信任首次处理结果的事件，可以设置 `max_iterations=1`。 默认情况下（未设置 `POSTGRES_URL`），系统使用内存检查点。重启后任务历史记录将会丢失。将 `POSTGRES_URL` 指向一个 PostgreSQL 实例即可在重启后保留数据。 `query_git_log` 工具读取的是本地 git 仓库。如果服务仓库位于远程服务器上或只能通过 API 访问，你需要添加一个工具，用于从 GitHub 或 GitLab 获取 commit 历史记录。 ## 生产部署 Terraform 会在 Azure 上配置所有资源： ``` cd infra/azure/terraform terraform init && terraform apply ``` 这将创建用于 API 和 worker 的 Container Apps、一个 Service Bus Premium 命名空间（会话支持所必需）、用于检查点的 PostgreSQL Flexible Server，以及用于密钥的 Key Vault。GitHub Actions 会在代码推送到 main 分支时进行部署。 Worker 支持水平扩展。每个副本处理不同的用户会话，因此并发的事件不会相互阻塞。每月的成本约为 700 美元，主要是因为 Service Bus Premium 需要一个消息单元才能启用会话。 ## 项目结构 ``` forge/ ├── agent_system/ │ ├── agents/ │ │ ├── orchestrator.py plans subtasks, tracks dependencies │ │ ├── sub_agents.py researcher, coder, analyst with role-specific prompts │ │ └── critic.py quality scoring, drives retry │ ├── graph/ │ │ └── builder.py LangGraph StateGraph, AgentGraphFactory │ ├── state/ │ │ └── schema.py AgentState TypedDict, initial_state factory │ ├── tools/ │ │ └── registry.py tool definitions, scopes per agent role │ └── llm.py Groq / Ollama factory ├── api/ │ └── azure_main.py FastAPI endpoints + browser UI ├── worker/ │ └── servicebus_worker.py Service Bus consumer, lock renewal, dead-letter ├── tests/ │ ├── unit/ routing, state schema, tools, parsing, worker │ └── integration/ API endpoints, full graph end to end ├── scripts/ │ ├── demo_incident.py generates synthetic logs and runs a full investigation │ ├── init_db.py creates the Postgres checkpoint schema │ └── smoke_test.py mocked end-to-end graph run └── infra/azure/terraform/ Container Apps, Service Bus, Postgres, Key Vault ```

标签：AI风险缓解, 人工智能, 多智能体, 测试用例, 用户模式Hook绕过, 自动化分析, 请求拦截, 跨站脚本, 运维监控, 逆向工具