企业官网建设流程全解析

1. 为什么我要自己搭一个 CI 故障处理 Agent

CI 挂了这件事，但凡在团队里负责过流水线的同学都懂那种感受。凌晨两点，手机叮的一声，GitHub Actions 的失败通知弹出来，你揉着眼睛打开日志，翻了三百行才发现是某个依赖版本对不上，或者某个环境变量在 PR 里被误删了。修完之后你躺在床上想：这种活儿，为什么不能交给一个 Agent 去干？

我最初做这个项目的动机特别朴素——我们团队每周大概有 15 到 20 次 CI 失败，其中真正需要人脑判断的不到三分之一，剩下的全是重复性的模式匹配：依赖冲突、缓存失效、环境变量缺失、测试超时、lint 报错。这些故障有明确的日志特征，有固定的修复套路，完全可以交给一个 AI Agent 去自动捕获、分析、甚至直接提 PR 修复。

所以这个项目的核心目标就三个词：捕获、分析、修复。捕获是指从 GitHub Actions 的 webhook 里实时拿到失败事件；分析是指用 LLM 去读日志、定位根因；修复是指 Agent 根据分析结果生成补丁，走 PR 流程提交。整个系统用 Python 写，Web 框架选 FastAPI，LLM 调用走 Ollama 本地部署（也可以用云端 API），部署方式是一台轻量服务器加一个 webhook 端点。

这篇文章适合谁看？如果你已经会用 GitHub Actions，写过一点 Python，对 FastAPI 有基本了解，想动手做一个真正能跑起来的 AI Agent 而不是停留在 demo 阶段，那这篇就是写给你的。我不会只给你讲架构图，我会把目录结构、关键代码、踩过的坑、参数怎么调，全部摊开讲。

2. 整体架构设计与技术选型思路

2.1 系统分层的核心逻辑

这个 Agent 的架构我改过三版，第一版是单体脚本，所有逻辑塞在一个main.py里，跑了两天就发现根本没法维护。第二版拆成了三个模块但耦合严重，webhook 接收和日志分析混在一起，一个环节超时整个请求就挂了。第三版才定型为现在的四层结构：

• 接入层：FastAPI 接收 GitHub webhook，做签名验证和事件过滤
• 任务层：把失败事件丢进异步任务队列，避免阻塞 webhook 响应
• 分析层：调用 LLM 分析日志，输出结构化的根因和修复建议
• 执行层：根据分析结果生成补丁，通过 GitHub API 创建分支和 PR

为什么要做异步？因为 GitHub webhook 有超时限制，默认 10 秒内必须返回 2xx，否则 GitHub 会认为投递失败并重试。而 LLM 分析日志动辄十几秒甚至几十秒，同步处理必然超时。我试过同步方案，结果就是 GitHub 疯狂重试，同一个失败事件被处理了七八次，PR 都提重了。

提示：webhook 端点必须快速返回，任何耗时操作都要异步化。这是做 CI Agent 的第一条铁律。

2.2 为什么选 FastAPI 而不是 Flask

热词里有人问 Flask 和 FastAPI 的比较，我在这里直接给结论：做 AI Agent 的后端，FastAPI 几乎是默认选择。原因有三个。

第一是原生异步支持。FastAPI 基于 Starlette，async def是真异步，处理 webhook 的同时可以并发跑 LLM 调用。Flask 是同步框架，虽然也能用线程池凑合，但写起来别扭，高并发下性能差距明显。

第二是Pydantic 数据校验。GitHub webhook 的 payload 结构复杂，嵌套好几层，用 Pydantic 定义模型之后，字段校验、类型转换、文档生成全部自动搞定。Flask 里你得手动request.json.get()一层层取，容易出错。

第三是自动生成 API 文档。FastAPI 自带 Swagger UI，调试 webhook 的时候直接在浏览器里测，省了 Postman 的功夫。

# 用 Pydantic 定义 webhook payload 的核心字段 from pydantic import BaseModel from typing import Optional class WorkflowRun(BaseModel): id: int name: str conclusion: Optional[str] = None head_branch: str head_sha: str logs_url: str class WebhookPayload(BaseModel): action: str workflow_run: Optional[WorkflowRun] = None repository: dict

这段模型定义看起来简单，但它帮我挡掉了大量脏数据。GitHub 会发各种类型的事件过来，不是每个都跟 CI 失败有关，Pydantic 校验不过的直接返回 422，干净利落。

2.3 LLM 选型：本地 Ollama 还是云端 API

这是很多人纠结的点。我的建议是：开发阶段用 Ollama 本地跑，生产环境看预算和隐私要求决定。

Ollama 的好处是免费、数据不出本地、调试方便。我用qwen2.5-coder:7b这个模型做日志分析，效果够用，7B 的模型在一张 12G 显存的卡上跑得很流畅。坏处是分析质量不如 GPT-4 级别的大模型，复杂故障的根因定位偶尔会跑偏。

云端 API 的好处是分析质量高，坏处是要花钱，而且 CI 日志里可能包含敏感信息（比如内部服务地址、部分配置），发到外部 API 有合规风险。

我的折中方案是：日志脱敏后再送分析。用一个正则规则把 IP、域名、token 之类的敏感串替换掉，然后再发给 LLM。这样即使用云端 API 也相对安全。

import re SENSITIVE_PATTERNS = [ (r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b', '[IP]'), (r'ghp_[A-Za-z0-9]{36}', '[TOKEN]'), (r'https?://[^\s]+\.internal[^\s]*', '[INTERNAL_URL]'), ] def sanitize_log(text: str) -> str: for pattern, replacement in SENSITIVE_PATTERNS: text = re.sub(pattern, replacement, text) return text

2.4 目录结构设计

FastAPI 项目的目录结构直接影响后期维护成本。我见过太多人把所有路由塞在main.py里，写到五百行就崩溃了。下面是我实际在用的结构：

ci-agent/ ├── app/ │ ├── __init__.py │ ├── main.py # 应用入口，注册路由和中间件 │ ├── config.py # 配置管理，读环境变量 │ ├── models/ │ │ ├── webhook.py # webhook payload 模型 │ │ └── analysis.py # 分析结果模型 │ ├── routers/ │ │ ├── webhook.py # webhook 接收路由 │ │ └── health.py # 健康检查 │ ├── services/ │ │ ├── github.py # GitHub API 封装 │ │ ├── llm.py # LLM 调用封装 │ │ └── analyzer.py # 日志分析核心逻辑 │ ├── tasks/ │ │ └── queue.py # 异步任务队列 │ └── utils/ │ ├── sanitize.py # 日志脱敏 │ └── logger.py # 日志配置 ├── tests/ ├── requirements.txt └── .env.example

这个结构的关键在于按职责分层，routers只管收请求，services管业务逻辑，models管数据结构，utils管工具函数。改一个功能只需要动对应的目录，不会牵一发动全身。

3. 核心模块拆解与关键实现细节

3.1 Webhook 接收与签名验证

GitHub webhook 的安全机制是靠X-Hub-Signature-256头，用你配置的 secret 对 payload 做 HMAC-SHA256 签名。不验证签名的话，任何人都能伪造请求触发你的 Agent，轻则浪费 LLM 额度，重则被恶意利用提垃圾 PR。

import hmac import hashlib from fastapi import Request, HTTPException async def verify_signature(request: Request, secret: str) -> bytes: signature = request.headers.get("X-Hub-Signature-256") if not signature: raise HTTPException(status_code=401, detail="Missing signature") body = await request.body() expected = "sha256=" + hmac.new( secret.encode(), body, hashlib.sha256 ).hexdigest() if not hmac.compare_digest(signature, expected): raise HTTPException(status_code=401, detail="Invalid signature") return body

这里有个坑我踩过：必须用await request.body()拿原始字节，不能用request.json()。因为签名是对原始字节算的，一旦经过 JSON 解析再序列化，字节顺序和空格可能变了，签名就对不上。而且hmac.compare_digest要用它做恒定时间比较，别用==，防止时序攻击。

3.2 日志拉取与预处理

GitHub Actions 的日志不是直接给你的，得通过 API 拿。workflow_run事件里有个logs_url，但那个 URL 需要认证才能下载，返回的是一个 zip 包。

import httpx import zipfile import io async def fetch_logs(logs_url: str, token: str) -> str: headers = {"Authorization": f"Bearer {token}"} async with httpx.AsyncClient() as client: resp = await client.get(logs_url, headers=headers, follow_redirects=True) resp.raise_for_status() with zipfile.ZipFile(io.BytesIO(resp.content)) as zf: # 日志按 job 分文件，合并所有内容 all_logs = [] for name in zf.namelist(): with zf.open(name) as f: all_logs.append(f.read().decode("utf-8", errors="ignore")) return "\n".join(all_logs)

日志拉下来之后不能直接丢给 LLM，因为完整日志可能几万行，超出上下文窗口不说，还浪费 token。我的做法是先做失败步骤定位：GitHub 日志里有明确的##[error]标记和步骤分隔符，先提取出失败的那个 step 的日志片段，通常只有几百行，再送分析。

def extract_failure_section(full_log: str) -> str: lines = full_log.split("\n") # 找到最后一个 error 标记的位置 error_indices = [i for i, l in enumerate(lines) if "##[error]" in l] if not error_indices: return full_log[-3000:] # 没找到就取尾部 start = max(0, error_indices[-1] - 200) end = min(len(lines), error_indices[-1] + 50) return "\n".join(lines[start:end])

这个-200和+50是我调出来的经验值。往前 200 行基本能覆盖到失败步骤的完整上下文，往后 50 行能包含错误堆栈的尾部。取太多浪费 token，取太少丢关键信息。

3.3 LLM 分析提示词设计

提示词设计是这个项目的灵魂。我前后改了十几版，核心思路是让 LLM 输出结构化 JSON，而不是自由文本。自由文本你没法程序化处理，结构化 JSON 才能驱动后续的自动修复。

ANALYSIS_PROMPT = """你是一个 CI 故障分析专家。分析下面的 CI 日志，输出 JSON 格式的结果。 日志内容： {log_content} 请严格按以下 JSON 格式输出，不要输出任何其他内容： {{ "root_cause": "一句话描述根因", "category": "dependency|env|test|timeout|lint|infra|unknown", "confidence": 0.0 到 1.0 之间的浮点数, "fix_suggestion": "具体的修复建议", "fix_type": "auto|manual", "files_to_change": ["需要修改的文件路径"] }} 判断规则： - 如果是依赖版本问题，category 填 dependency - 如果是环境变量缺失，category 填 env - 如果测试断言失败，category 填 test - 如果超时，category 填 timeout - 如果 lint 报错，category 填 lint - 如果无法确定，category 填 unknown，confidence 低于 0.5 """

这里的关键设计是fix_type字段。不是所有故障都能自动修，比如测试断言失败，可能是代码逻辑真的有问题，Agent 不该乱改。只有依赖问题、环境变量缺失、lint 这类有明确修复模式的，才标记为auto，允许 Agent 自动提 PR。

confidence字段也很重要。低于 0.6 的分析结果我直接丢弃，只发通知让人来处理。宁可漏修，不可错修。

3.4 调用 Ollama 的封装

FastAPI 调 Ollama 就是发个 HTTP 请求，但有几个细节要注意。

import httpx import json async def analyze_with_ollama(log_content: str, model: str = "qwen2.5-coder:7b") -> dict: prompt = ANALYSIS_PROMPT.format(log_content=log_content) async with httpx.AsyncClient(timeout=120.0) as client: resp = await client.post( "http://localhost:11434/api/generate", json={ "model": model, "prompt": prompt, "stream": False, "format": "json", # 关键：强制 JSON 输出 "options": { "temperature": 0.1, # 低温度保证稳定性 "num_predict": 500 } } ) resp.raise_for_status() result = resp.json() return json.loads(result["response"])

format: "json"这个参数是 Ollama 的特性，能强制模型输出合法 JSON，省去了自己解析和容错的麻烦。temperature设成 0.1，因为分析任务要的是稳定和准确，不需要创造性。num_predict限制输出长度，防止模型啰嗦。

超时设 120 秒，因为 7B 模型分析长日志确实慢，设短了会频繁超时。如果你用更小的模型或者更短的日志，可以适当调低。

3.5 自动修复与 PR 创建

分析出结果之后，fix_type为auto的走自动修复流程。以依赖问题为例，Agent 需要：拉取仓库代码、修改依赖文件、提交、创建 PR。

from github import Github def create_fix_pr(repo_name: str, analysis: dict, token: str): g = Github(token) repo = g.get_repo(repo_name) # 从默认分支创建修复分支 base = repo.default_branch branch_name = f"ci-agent/fix-{int(time.time())}" ref = repo.get_git_ref(f"heads/{base}") repo.create_git_ref(f"refs/heads/{branch_name}", ref.object.sha) # 修改文件 for file_path in analysis["files_to_change"]: file = repo.get_contents(file_path, ref=branch_name) content = file.decoded_content.decode() new_content = apply_fix(content, analysis) repo.update_file( file_path, f"fix: {analysis['root_cause']}", new_content, file.sha, branch=branch_name ) # 创建 PR pr = repo.create_pull( title=f"[CI Agent] {analysis['root_cause']}", body=format_pr_body(analysis), head=branch_name, base=base ) return pr

PR 的 body 里我会把分析结果、置信度、原始日志链接都带上，方便人 review。Agent 提的 PR 必须人工审核后才能合并，这是底线，不能让 Agent 直接推主干。

4. 完整实操流程与部署方案

4.1 环境准备与依赖安装

先把环境搭起来。Python 版本建议 3.10 以上，因为用到了match语法和一些新的类型标注特性。

# 创建虚拟环境 python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate # 安装依赖 pip install fastapi uvicorn httpx pydantic PyGithub python-dotenv

requirements.txt长这样：

fastapi==0.115.0 uvicorn[standard]==0.32.0 httpx==0.27.2 pydantic==2.9.2 PyGithub==2.4.0 python-dotenv==1.0.1

Ollama 单独装，去官网下安装包，装完拉模型：

ollama pull qwen2.5-coder:7b ollama serve # 启动服务，默认监听 11434

4.2 配置管理

所有敏感配置走环境变量，.env文件不进版本库。

# app/config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): github_token: str webhook_secret: str ollama_base_url: str = "http://localhost:11434" ollama_model: str = "qwen2.5-coder:7b" confidence_threshold: float = 0.6 max_log_lines: int = 500 class Config: env_file = ".env" settings = Settings()

.env.example给个模板：

GITHUB_TOKEN=ghp_xxxxxxxxxxxx WEBHOOK_SECRET=your_random_secret OLLAMA_BASE_URL=http://localhost:11434 OLLAMA_MODEL=qwen2.5-coder:7b CONFIDENCE_THRESHOLD=0.6

4.3 主应用入口

# app/main.py from fastapi import FastAPI from app.routers import webhook, health from app.utils.logger import setup_logger setup_logger() app = FastAPI(title="CI Agent", version="1.0.0") app.include_router(health.router) app.include_router(webhook.router, prefix="/webhook") @app.on_event("startup") async def startup(): # 启动异步任务队列的 worker from app.tasks.queue import start_workers await start_workers()

4.4 Webhook 路由实现

# app/routers/webhook.py from fastapi import APIRouter, Request, BackgroundTasks, HTTPException from app.config import settings from app.models.webhook import WebhookPayload from app.services.analyzer import process_failure from app.utils.sanitize import verify_signature router = APIRouter() @router.post("/github") async def github_webhook(request: Request, background_tasks: BackgroundTasks): body = await verify_signature(request, settings.webhook_secret) payload = WebhookPayload.model_validate_json(body) # 只处理 workflow_run 的 completed 事件 if payload.action != "completed": return {"status": "ignored", "reason": "not completed"} if payload.workflow_run.conclusion != "failure": return {"status": "ignored", "reason": "not failure"} # 丢到后台处理，立即返回 background_tasks.add_task( process_failure, payload.workflow_run, payload.repository["full_name"] ) return {"status": "accepted"}

用BackgroundTasks是最简单的异步方案，适合单机部署。如果你要处理大量并发，建议换成 Celery 或者 arq 这类正经的任务队列。我用 BackgroundTasks 是因为我们的失败频率不高，一天几十次，够用了。

4.5 分析流程串联

# app/services/analyzer.py from app.services.github import fetch_logs, create_fix_pr from app.services.llm import analyze_with_ollama from app.utils.sanitize import sanitize_log, extract_failure_section from app.config import settings async def process_failure(workflow_run, repo_name: str): # 1. 拉日志 full_log = await fetch_logs(workflow_run.logs_url, settings.github_token) # 2. 提取失败片段 failure_section = extract_failure_section(full_log) # 3. 脱敏 clean_log = sanitize_log(failure_section) # 4. LLM 分析 analysis = await analyze_with_ollama(clean_log, settings.ollama_model) # 5. 置信度过滤 if analysis["confidence"] < settings.confidence_threshold: await notify_human(workflow_run, analysis) return # 6. 自动修复 if analysis["fix_type"] == "auto": pr = create_fix_pr(repo_name, analysis, settings.github_token) await notify_pr_created(pr, analysis) else: await notify_human(workflow_run, analysis)

这个流程里每一步都有日志记录，方便排查。我用的loguru做日志，比标准库的 logging 好用太多。

4.6 部署到服务器

本地跑通之后就是部署。我用的是最朴素的方式：一台轻量云服务器，systemd 管理进程，Nginx 做反向代理。

# /etc/systemd/system/ci-agent.service [Unit] Description=CI Agent After=network.target [Service] User=deploy WorkingDirectory=/opt/ci-agent Environment="PATH=/opt/ci-agent/venv/bin" ExecStart=/opt/ci-agent/venv/bin/uvicorn app.main:app --host 0.0.0.0 --port 8000 Restart=always [Install] WantedBy=multi-user.target

Nginx 配置：

server { listen 443 ssl; server_name ci-agent.yourdomain.com; location /webhook/ { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; } }

然后在 GitHub 仓库的 Settings -> Webhooks 里填上你的 URL，content type 选application/json，secret 填跟.env里一致的字符串，事件选Workflow runs。

4.7 Windows 打包的注意事项

热词里有人问 FastAPI 在 Windows 上打包的问题。如果你要在 Windows 服务器上跑，uvicorn直接跑没问题，但如果想打包成 exe，用 PyInstaller 会遇到一些坑。

主要问题是 PyInstaller 不会自动收集 FastAPI 和 Pydantic 的隐式依赖。需要写个 spec 文件，手动加上hiddenimports：

# build.spec a = Analysis( ['app/main.py'], hiddenimports=[ 'uvicorn.logging', 'uvicorn.loops.auto', 'uvicorn.protocols.http.auto', 'pydantic.deprecated.decorator', ], ... )

说实话，Windows 打包 FastAPI 项目性价比不高，除非你有强制的 Windows 部署要求，否则建议直接跑 Python 脚本，用 NSSM 注册成 Windows 服务就行。

5. 常见问题排查与避坑经验

5.1 问题速查表

5.2 踩过的坑：uvicorn 日志丢失

热词里有人提到uvicorn fastapi 日志丢失问题，这个我深有体会。FastAPI 默认的日志配置会覆盖你自定义的 logging，导致你自己写的日志打不出来。

解决方案有两个。一是用loguru完全接管日志，在应用启动时移除 uvicorn 的默认 handler：

import logging from loguru import logger class InterceptHandler(logging.Handler): def emit(self, record): logger_opt = logger.opt(depth=6, exception=record.exc_info) logger_opt.log(record.levelname, record.getMessage()) logging.basicConfig(handlers=[InterceptHandler()], level=0)

二是启动 uvicorn 时指定--log-config指向你自己的配置文件。我推荐第一种，简单直接。

5.3 踩过的坑：LLM 输出不稳定

同一个日志，LLM 有时候分析得准，有时候跑偏。除了降低 temperature，我还加了两个措施。

一是在提示词里给示例。few-shot 能显著提升稳定性。我在提示词里塞了两个分析示例，一个依赖问题，一个环境变量问题，模型照着格式输出，准确率提升明显。

二是加校验层。分析结果回来之后，用规则再校验一遍。比如category是dependency，但files_to_change里没有requirements.txt或package.json，那这个结果就存疑，降级为人工处理。

def validate_analysis(analysis: dict) -> bool: if analysis["category"] == "dependency": dep_files = {"requirements.txt", "package.json", "go.mod", "Cargo.toml"} if not any(f in dep_files for f in analysis["files_to_change"]): return False if analysis["confidence"] > 0.9 and analysis["category"] == "unknown": return False # 高置信度但类别未知，矛盾 return True

5.4 踩过的坑：重复处理

GitHub webhook 会重试，网络抖动也会导致重复投递。如果不做幂等，同一个失败会被处理多次，提多个重复 PR。

我的做法是用workflow_run.id做去重键，处理前先查 Redis 或者内存缓存，处理过的直接跳过。

processed_runs = set() # 生产环境用 Redis async def process_failure(workflow_run, repo_name): run_id = workflow_run.id if run_id in processed_runs: logger.info(f"Run {run_id} already processed, skipping") return processed_runs.add(run_id) # ... 后续处理

内存缓存重启就丢了，生产环境一定要用 Redis，设个 24 小时的过期时间。

5.5 实操心得：从保守开始

我最大的心得是：Agent 的能力要逐步放开，不要一上来就全自动。

我的第一版是纯通知，只分析不修复，跑了两周，观察分析准确率。准确率稳定在 80% 以上之后，才放开依赖类问题的自动修复。又跑了一个月，才放开环境变量类。测试类和代码逻辑类的问题，我到现在都没放开自动修复，因为风险太高。

这个渐进过程很重要。你直接全自动，万一 Agent 改错了代码合进主干，造成的损失远大于省下的时间。

提示：自动修复的范围要按类别逐步放开，每放开一类，先观察两周的准确率和误报率。

5.6 成本与性能优化

如果你用云端 API，成本是个绕不开的话题。我的优化手段有三个。

第一是日志截断。前面说的extract_failure_section只取失败片段，能把 token 消耗降低 80% 以上。完整日志几万行，失败片段通常几百行。

第二是结果缓存。相同的错误模式，分析结果可以复用。我用错误信息的 hash 做 key，缓存分析结果，命中率大概 30%。

第三是模型分级。简单问题用小模型，复杂问题用大模型。先用小模型分析，如果 confidence 低于阈值，再用大模型重试一次。这样大部分请求走小模型，成本可控。

async def analyze_with_fallback(log_content: str) -> dict: # 先试小模型 result = await analyze_with_ollama(log_content, "qwen2.5-coder:7b") if result["confidence"] >= 0.6: return result # 小模型没把握，换大模型 logger.info("Low confidence, retrying with larger model") return await analyze_with_ollama(log_content, "qwen2.5-coder:32b")

这套组合拳下来，我们的月均 API 成本控制在很低的水平，而 CI 故障的平均处理时间从 25 分钟降到了 4 分钟。

5.7 后续可以扩展的方向

这个 Agent 跑稳之后，我陆续加了几个扩展。一是故障统计看板，把所有分析结果存库，用 Grafana 展示故障类型分布、修复成功率、平均处理时间。二是知识库沉淀，把每次的分析结果和修复方案存起来，下次遇到类似问题直接检索，不用再调 LLM。三是多仓库支持，现在一个 Agent 实例可以服务多个仓库，通过 webhook 的 repository 字段区分。

如果你也想动手做，我的建议是先跑通最小闭环：webhook 接收 -> 日志拉取 -> LLM 分析 -> 通知。这个闭环跑通了，再逐步加自动修复、加缓存、加看板。别一上来就追求大而全，容易烂尾。