从 main.py 出发，亲手走完一个企业级 Agent 后端

整条故事线

flowchart LR
  A["浏览器/Swagger"] --> B["FastAPI 路由"]
  B --> C["领域 Service"]
  C --> D["Repository"]
  C --> E["Dispatcher"]
  E --> F["Worker"]
  F --> G["Agent Runtime"]
  G --> H["Skill + Tool + Policy"]
  H --> I["Evidence / Approval"]
  F --> D
  D --> J["Trace / SSE / Result"]
            

图加载失败也不影响阅读：它表达的就是“请求进入 API，经服务和仓库存下任务，由 Worker 执行 Agent，最后把可审计结果写回仓库”。

三个容易混淆的词

为什么使用 src/ 布局？

源码不直接躺在 backend/ 根目录，而在 backend/src/devinfra_agent。这样测试必须像真实安装后的用户一样导入包，可避免“只因当前目录碰巧在搜索路径里所以能运行”的假成功。

HTTP 请求在后端眼里是什么？

这些命令分别做什么

cd C:\Users\Jack1\Desktop\offer\devinfra-agent-platform\backend
uv sync --all-groups
$env:DEVINFRA_MODE="offline"
$env:DEVINFRA_OFFLINE_EAGER="1"
uv run uvicorn devinfra_agent.main:app --reload

cd

把当前工作目录切到 backend，后面的 pyproject.toml 才能被找到。

uv sync

uv 读取项目清单和锁文件，创建/更新 .venv 并安装运行、开发依赖。

DEVINFRA_MODE

环境变量；offline 让组合根选择内存仓库、确定性模型和 fixture 工具。

DEVINFRA_OFFLINE_EAGER

设为 1 后，内存 dispatcher 收到 Run 就立即安排 Worker 执行，适合本地演示。

uv run

在项目虚拟环境里执行后面的命令，不必手动激活 .venv。

uvicorn

ASGI 服务器；它导入 devinfra_agent.main 模块中的 app 对象。

--reload

源码保存后自动重启开发服务器；只用于开发，不用于生产。

第一次手动验证

1. 打开 http://127.0.0.1:8000/docs，这是 FastAPI 自动生成的 Swagger UI。
2. 调用 POST /api/auth/login，用户名 maintainer，密码 maintainer-password。
3. 复制 access_token，点 Swagger 右上角 Authorize，填 Bearer 空格 token。
4. 调用 POST /api/runs，body 使用 {"workflow":"ci_triage","prompt":"分析退款测试失败","metadata":{}}。
5. 拿响应里的 Run ID 调用 GET /api/runs/{run_id}/trace，观察 step、model_call、tool_call、evidence。

建议的第一组断点

"""FastAPI application entry point."""

from .composition import create_app_from_environment

app = create_app_from_environment()

"""..."""

模块文档字符串，告诉读者这个文件的职责。

from .composition

开头的点表示“从当前 devinfra_agent 包内导入”。

create_app_from_environment

工厂函数：读环境变量、组装服务、创建 FastAPI 对象。

app

模块级变量。命令里的 devinfra_agent.main:app，冒号后就是它。

为什么不在入口里直接连接数据库？

入口只声明“我要一个 app”。composition.py 才负责根据环境选择内存实现或生产实现。这叫组合根（composition root）：所有具体依赖集中装配，业务类只接收自己需要的对象。

AppServices 是什么？

它是应用的“服务箱”：repository、dispatcher、events、runs、approvals、worker、auth、skills、tools、knowledge、evaluations、health 都放在同一个数据类里。app.state.services 保存这只箱子，测试也能换入自定义假实现。

tools = runtime.tools if runtime is not None else ToolRegistry.fixture()
skills = runtime.skills if runtime is not None else SkillRegistry.default()
actual_runtime = runtime or AgentRuntime(
    gateway=DeterministicModelGateway(),
    tools=tools,
    skills=skills,
    graph=LangGraphAdapter(checkpointer=InMemorySaver()),
)
dispatcher = InMemoryDispatcher()
services = cls.compose(
    repository=InMemoryRepository(),
    dispatcher=dispatcher,
    runtime=actual_runtime,
    jwt_secret="offline-development-secret",
    user_store=UserStore.seeded(),
)

runtime is not None

调用方传了自定义 Runtime 就复用它，便于测试；否则使用默认离线实现。

actual_runtime

“实际采用的 runtime”。变量改名是为了区别可选参数 runtime。

gateway

模型网关；确定性版本不访问网络，返回可预测结果。

checkpointer

图状态存档器；离线存在内存里，生产存在 PostgreSQL。

cls.compose

cls 是 AppServices 类；compose 继续把底层对象组装成领域服务。

登录的真实调用顺序

@app.post("/api/auth/login")
def login(request: LoginRequest) -> dict[str, str]:
    return {"access_token": container.auth.login(request), "token_type": "bearer"}

def allow(*roles: Role):
    def dependency(user: User = Depends(current_user)) -> User:
        if user.role not in roles:
            raise APIProblem(403, "forbidden", "Permission denied")
        return user
    return dependency

approver = allow(Role.MAINTAINER, Role.ADMIN)

@app.post

装饰器，把下面的函数注册为 POST 路由；函数名可读，URL 才是外部接口。

request: LoginRequest

FastAPI 根据类型标注读 JSON，并让 Pydantic 校验字段。

container

当前 AppServices 对象；它在 create_app 中生成。

*roles

可变位置参数；调用 allow(A, B) 后，函数里得到角色元组。

Depends

FastAPI 依赖注入：先运行 current_user 解析 Authorization，再把 User 传进来。

approver

已经配置好的依赖，只允许 Maintainer 或 Admin；审批路由直接复用。

auth.py 的职责拆分

密码不是明文保存。hash_password 使用 PBKDF2-HMAC-SHA256、随机 salt 和 120000 次迭代；verify_password 重算并用常量时间方式比较。JWT 签名用 HMAC；它是签名令牌，不是加密保险箱，payload 不应放秘密。

统一错误为什么重要？

errors.py 把业务错误、Pydantic 校验错、FastAPI HTTP 错、数据库完整性错和未知异常都转成统一结构：code / message / request_id / details / retryable。未知异常不回显堆栈或密钥。

operations.py 在每个请求外面做什么？

• 生成或沿用 X-Request-ID，写入响应，方便串联日志。
• 对登录和普通 API 分别限流；离线内存计数，生产 Redis 计数。
• 记录 Prometheus 请求总量和耗时；/metrics 暴露文本。
• configure_telemetry 可安装 OpenTelemetry FastAPI instrumentor。

class ContractModel(BaseModel):
    model_config = ConfigDict(extra="forbid")

class RunRequest(ContractModel):
    run_id: UUID = Field(default_factory=uuid4)
    workflow: WorkflowName
    prompt: str = Field(min_length=1)
    requested_by: Role
    steps: list[PlanStep] = Field(default_factory=list)
    tools: list[ToolSpec] = Field(default_factory=list)
    metadata: dict[str, Any] = Field(default_factory=dict)

BaseModel

Pydantic 基类；负责解析、校验、序列化和生成 JSON Schema。

extra="forbid"

传入模型未声明的字段就报错，避免拼错字段被静默忽略。

default_factory

每次创建对象时调用工厂。UUID 用 uuid4 生成；list/dict 用新容器，避免多个对象共享可变默认值。

WorkflowName

StrEnum 枚举；只允许定义过的工作流字符串。

metadata

扩展信息，例如 service、action_tool、action_arguments；灵活但仍受到后续策略校验。

文件之间怎样转换？

flowchart LR
  JSON["HTTP JSON"] --> P["Pydantic API model"]
  P --> S["Service 参数"]
  S --> R["Record dataclass"]
  R --> O["SQLAlchemy Model"]
  O --> DB[("PostgreSQL")]
  R --> Z["serializer / _run_json"]
  Z --> OUT["响应 JSON"]
          

serializers.py 的 document_json、eval_case_json 等只做 Record → JSON 字典。Run/Event/Approval 的序列化目前在 app.py 的私有函数里。若你给数据库加字段，通常要同时检查 ORM Model、Record、迁移、Repository 转换、serializer 和 API schema。

@app.post("/api/runs", status_code=status.HTTP_202_ACCEPTED)
async def create_run(
    request: RunCreate,
    user: User = Depends(run_mutator),
    idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key")] = None,
) -> dict[str, Any]:
    run = await container.runs.create(
        workflow=request.workflow,
        prompt=request.prompt,
        metadata=request.metadata,
        requested_by=user.role,
        requested_by_user=user.username,
        idempotency_key=idempotency_key,
    )
    return _run_json(run)

status.HTTP_202_ACCEPTED

返回 202 表示请求已接收、后台工作未必完成；API 不等待 Agent 跑完。

run_mutator

上一章创建的角色依赖，viewer 会在进入函数前被拒绝。

Annotated

把“值类型”和 FastAPI 元数据放一起；这里表示可选字符串来自名为 Idempotency-Key 的请求头。

run

RunService.create 返回已持久化的 RunRecord，不一定是刚创建的。

_run_json

私有序列化函数；前导下划线表示模块内部使用约定。

为什么同一个请求不能创建两次？

网络重试、用户双击、代理超时都可能让客户端重复 POST。RunService.create 对影响请求含义的字段做稳定 JSON 序列化，再计算 SHA-256 fingerprint。调用方没给 key 时，使用 derived:{fingerprint}；给了 key 时，数据库把 key 与 fingerprint 绑定。

fingerprint = hashlib.sha256(
    json.dumps(
        {
            "workflow": workflow.value,
            "prompt": prompt,
            "metadata": metadata,
            "requested_by_user": requested_by_user,
            "rerun_of": str(rerun_of) if rerun_of else None,
        },
        sort_keys=True,
        separators=(",", ":"),
    ).encode()
).hexdigest()
key = idempotency_key or f"derived:{fingerprint}"
persisted, created = await self.repository.create_run_request(
    run, key, fingerprint, self.clock()
)
if created:
    await self.dispatcher.dispatch(persisted.id)

hashlib.sha256

把任意长度字节变成固定 64 位十六进制摘要；内容相同则摘要相同。

sort_keys

字典键顺序不影响 fingerprint。

separators

去掉 JSON 非必要空格，保证序列化稳定。

.encode()

哈希函数接收 bytes；把 Python 字符串编码成 UTF-8 字节。

persisted

仓库实际保存/查到的 Run；重复请求时它是第一次请求的记录。

created

布尔值：这次是否首次插入。它决定是否需要新分发。

Outbox 解决哪扇“时间窗口”？

如果“写数据库成功”后进程崩溃，来不及给 Celery 发消息，Run 会永久 pending。Repository 在同一事务里写 Run 和 dispatch_outbox。立即分发失败也没关系，Celery Beat 的 relay 会重新 claim pending outbox，再投递。

sequenceDiagram
  participant API
  participant DB as Repository/DB
  participant Broker as Dispatcher/Broker
  API->>DB: create_run_request(run, key, fingerprint)
  DB-->>API: 同一事务写 run + outbox
  API->>Broker: dispatch(run_id)
  alt 成功
    API->>DB: mark_dispatch_succeeded
  else 失败
    API->>DB: mark_dispatch_failed
    Note over DB,Broker: relay 稍后补投递
  end
          

为什么 repositories.py 有 2081 行？

它提供两套相同行为：InMemoryRepository 让单测和离线模式快速确定；SqlAlchemyRepository 用真实事务、唯一约束和行锁保证跨进程正确。ThreadedRepository 再把同步 SQLAlchemy 调用放到线程，避免阻塞异步事件循环。

ORM Model 与 Alembic

models.py 用 SQLAlchemy Declarative 定义 21 类表模型：users、runs、run_events、run_steps、model_calls、tool_calls、evidence、artifacts、approvals、audit_logs、skills、tools、documents、chunks、eval_cases、eval_runs、eval_results、eval_scores、feedback、dispatch_outbox、action_claims。

Alembic 的三次迁移按时间演进：首次建完整 schema → 增加 outbox/action claim 并修复旧审批数据 → 增加 durable eval 执行字段和约束。改 Model 不会自动改已有数据库，必须新增 migration。

SSE 为什么从数据库重放？

async def replay():
    cursor = after
    while not await raw_request.is_disconnected():
        events = await container.repository.events_after(run_id, cursor)
        for event in events:
            cursor = event.sequence
            data = json.dumps(_event_json(event), separators=(",", ":"))
            yield f"id: {event.sequence}\nevent: {event.event}\ndata: {data}\n\n"
        await container.events.publisher.wait(run_id, 15)
        yield ": heartbeat\n\n"

replay

异步生成器；每次 yield 一小段文本给客户端，而不是一次性返回。

cursor / after

最后看过的事件序号；来自 Last-Event-ID，断线重连从它后面继续。

is_disconnected

浏览器关掉连接后停止循环，避免服务器继续白做事。

event.sequence

数据库事实序号，同时作为 SSE id。

heartbeat

SSE 注释行，防止代理因长时间无数据关闭连接。

离线与生产的 Dispatcher

celery_entrypoint.py 只允许 production，复用 AppSettings 和 build_services，并配置每 5 秒 relay Run outbox、每 10 秒 relay eval queue。dispatch.py 注册具体 Celery task，设置 late acknowledgement、失败重试和 delivery identity。

async def execute(self, run_id: UUID | str, *, delivery_id: str) -> bool:
    key = UUID(str(run_id))
    acquired = await self.repository.acquire_run_lease(
        key, delivery_id, self.clock(), timedelta(minutes=2)
    )
    if not acquired:
        raise RunLeaseBusy(...)
    try:
        run = await self.repository.get_run(key)
        await self.repository.update_run(key, status=RunStatus.RUNNING, now=self.clock())
        outcome = await self.runtime.run(
            RunRequest(...),
            resource_scopes=scopes_for_role(run.requested_by),
        )
        await self._project_runtime(key, outcome)
        ...
    finally:
        await self.repository.release_run_lease(key, delivery_id)

delivery_id

本次消息投递的稳定身份，用作 lease owner；不是 Run ID。

key

无论调用者传 UUID 还是字符串，都归一化成 UUID 对象。

acquired

是否拿到 2 分钟 lease。False 说明另一个 Worker 正在负责或任务已结束。

outcome

Runtime 的完整结果：status、summary、trace、模型/工具记录、证据、checkpoint 和待审批动作。

finally

成功、失败、取消都释放 lease；否则任务要等过期才能恢复。

_project_runtime 为什么存在？

Runtime 只关心 Agent 逻辑，返回普通内存对象；Worker 把这些事实投影成数据库记录。每个工具原始结果先写 object_store.put，数据库只保存 URI、checksum、size，避免大日志塞进表和模型上下文。

nodes: Sequence[GraphNode] = (
    ("validate", self._validate),
    ("route", self._route),
    ("specialist", self._specialist),
    ("verify", self._verify),
    ("action_gate", self._action_gate),
    ("finalize", self._finalize),
)

nodes

有顺序的元组；每项是“节点名称 + 异步处理函数”。

Sequence[GraphNode]

类型接口；只要求可按顺序读取，不要求一定是 list。

self._validate

绑定方法对象，此处不加括号，因为是把函数交给图稍后调用。

specialist

根据 workflow 选出的 CI/incident/issue/knowledge 专家子流程。

action_gate

所有读证据完成并验证后，才允许处理请求中的 action_tool。

stateDiagram-v2
  [*] --> validate
  validate --> route
  route --> specialist
  specialist --> evidence_plan
  evidence_plan --> tools
  tools --> normalize
  normalize --> critique
  critique --> evidence_plan: needs_replan 且有预算
  critique --> verify: 足够
  verify --> action_gate: verification_passed
  action_gate --> finalize: 无动作/R0/R1/R3
  action_gate --> waiting: R2 interrupt
  waiting --> action_gate: resume approved/rejected
  finalize --> [*]
          

Runtime 的重要对象

图适配器为什么有三层？

• DeterministicGraphAdapter：普通 for 循环依次执行节点，最容易测试。
• LangGraphAdapter：构建 StateGraph；有 checkpointer 时支持 interrupt 与 Command(resume=...)。
• AsyncPostgresLangGraphAdapter：真正执行时才打开 PostgreSQL saver，避免应用启动/组合阶段就联网。

模型网关的安全边界

DeterministicModelGateway 可按 stage 注入脚本响应，没有脚本就返回稳定默认值。OpenAICompatibleGateway 校验 base URL，把允许工具放 system message，把外部证据包装为“不可信用户数据”，要求 JSON Schema 结构化输出，再把 native tool calls 和 JSON proposals 都解析成 ToolProposal。

超时与取消

_await_operation 同时等待实际 operation、CancellationToken 和 durable poll。谁先完成决定结果：取消则 cancel operation 并抛 RunCancelled；超时则清理所有 task 并抛 TimeoutError；正常则返回 operation result。

四级风险不是“提示颜色”

def decide(self, tool: ToolDefinition, context: PolicyContext) -> PolicyDecision:
    if not context.input_valid:
        return PolicyDecision(ActionDecision.BLOCKED, "invalid_input")
    if not context.budget_remaining:
        return PolicyDecision(ActionDecision.BLOCKED, "budget_exhausted")
    if tool.name not in context.skill_allowed_tools:
        return PolicyDecision(ActionDecision.BLOCKED, "tool_not_allowed_by_skill")
    if context.role not in tool.allowed_roles:
        return PolicyDecision(ActionDecision.BLOCKED, "role_denied")
    if tool.required_scope and tool.required_scope not in context.resource_scopes:
        return PolicyDecision(ActionDecision.BLOCKED, "resource_scope_denied")
    return decisions[tool.risk]

fail closed

任何前置条件不明确就 BLOCKED，不尝试“智能猜测”。

input_valid

Run 自身有效且工具 arguments 满足 input_schema；不是模型说“我检查过”。

skill_allowed_tools

当前 workflow 能看见的工具集合，来自 SkillRegistry。

allowed_roles

Tool 自己声明哪些角色可能使用。

required_scope

更细的资源权限，例如 deployments:write。

decisions

风险到动作的固定映射，模型无法修改。

ToolRegistry 做了四件事

1. 注册工具且拒绝重名；输出排序稳定的 definitions/specs。
2. _argument_error 做项目所需的最小 JSON Schema 校验：required、额外字段、string 类型和最小长度。
3. 执行 handler，记录 ToolAuditRecord；并发相同 idempotency key 共享同一 task。
4. R2 强制使用 idempotent_handler，Fixture store 模拟下游幂等数据库。

R2 审批的完整恢复链

sequenceDiagram
  participant RT as Runtime
  participant LG as LangGraph Checkpoint
  participant DB as Repository
  participant U as Approver
  participant W as Worker
  participant T as R2 Tool
  RT->>LG: interrupt(pending_action)
  RT-->>W: AWAITING_APPROVAL + checkpoint
  W->>DB: create approval
  U->>DB: claim decision once
  DB-->>W: dispatch same run
  W->>DB: lease + begin execution + claim action key
  W->>LG: resume("approved")
  LG->>T: idempotent_handler(arguments, action_key)
  W->>DB: receipt + audit + terminal status
          

ApprovalService.validate_action 在批准时重新读当前 Tool、Skill 和 Policy，防止等待期间策略变化。RunWorker._resume_approval 用 {run_id}:action:{tool} claim 动作，再恢复 checkpoint；即使批准后崩溃重投递，下游也按相同 key 返回旧 receipt。

工具结果怎样变成证据？

EvidenceNormalizer.normalize(tool_name, result, sequence) 生成 citation（如 E1）、清洗 source URI、计算内容 hash，并标注外部内容为不可信。NormalizedEvidence.model_text() 给模型的是明确包裹的数据；to_contract() 才转成最终 EvidenceItem。

flowchart LR
  T["ToolResult: content + source_uri + metadata"] --> N["EvidenceNormalizer"]
  N --> E["NormalizedEvidence: citation + hash + trust"]
  E --> M["有限模型上下文"]
  T --> O["Object Store 原文"]
  E --> P["Repository evidence 索引"]
  P --> F["FinalResult 引用 E1/E2"]
          

RAG 从文档到命中

1. KnowledgeService.create 生成 DocumentRecord，再把 KnowledgeDocument 交给 retriever.ingest。
2. DocumentChunker 按字符上限切块并保留 overlap；每块有稳定 ID、hash、locator、metadata。
3. Embedding provider 生成 dense vector；同时 tokenize/sparse_vector 形成稀疏表示。
4. Qdrant 请求使用 named dense/sparse prefetch 与 RRF fusion；内存版本自己计算 cosine、BM25 和 rank fusion。
5. Qdrant 不可用时 HybridKnowledgeService 退到 lexical fallback，并返回 degraded=True 与 limitations，绝不伪造 citation。

三种外部适配器

MCP 是协议，不是授权。MCP Server 宣称有某工具后，Python 端仍把它映射成带 risk、role、scope、schema、timeout 的 ToolDefinition，最终仍走 Policy。

两条相关但不同的评测路径

harness/ 每个模块做什么

三种 replay

cd backend
uv run devinfra-eval --name pr-smoke --mode exact --subset smoke
uv run devinfra-eval --name robustness --mode full --subset robustness --repetitions 3

devinfra-eval 来自 pyproject.toml [project.scripts]，入口是 devinfra_agent.harness.cli:main。

失败标注

FailureService.list 找失败/取消 Run 并附已有 feedback；label 校验 Run 存在且确实失败，再保存 reviewer、label、comment。它让人工失败分类反哺评测集和修复优先级，但当前代码不会自动训练模型。

三个健康接口不是重复

ComponentCheck.required 区分关键和可降级依赖：Qdrant 可选，因为 RAG 能 lexical fallback；Redis/MCP 生产组合中是 required。HTTPHealthProbe 同样执行 scheme/host allowlist，避免健康检查变成 SSRF 工具。

离线 → 生产替换表

Docker Compose 启动顺序

Postgres、Redis、Qdrant、MinIO、MCP 先各自健康 → API 运行 alembic upgrade head 后启动 Uvicorn → Worker 等 API 健康再启动 Celery worker + beat → Web 等 API。可观测 profile 额外启动 OTel Collector、Prometheus 和 Grafana。

docker compose --profile observability up --build

观测链

operations.py 结构化 JSON 日志带 request_id；Prometheus client 统计 route/status/latency；OpenTelemetry instrumentation 产生 HTTP trace 并经 OTLP HTTP exporter 发到 Collector；Prometheus 拉 /metrics；Grafana 展示。它们解决的是“发生了什么、慢在哪、请求如何串起来”，不是业务事实存储。

最实用的测试命令

# 只跑一个行为，失败时显示完整信息
uv run --project backend pytest backend/tests/test_run_api.py -vv

# 只跑一个测试函数
uv run --project backend pytest backend/tests/test_runtime.py::test_ci_runtime_normalizes_evidence_and_finalizes_with_citations -vv

# 全部后端测试与覆盖率门禁
uv run --project backend pytest backend/tests --cov=devinfra_agent --cov-fail-under=85

# 静态检查：导入、未使用变量、异步误用、风格
uv run --project backend ruff check backend/src backend/tests

按现象找第一断点

六个安全修改配方

读测试的正确顺序

1. 先读测试名，它是一句行为规范。
2. 看 Arrange：构造了哪些服务、假时间、脚本模型和工具。
3. 看 Act：真正调用哪个公开入口。
4. 看 Assert：系统承诺什么，而不只是实现细节。
5. 用测试名反搜源码，再在入口打断点单步。

运行与开发依赖：每个包为什么存在

实际使用的标准库模块

全部源码模块与公开/关键私有入口

脚本与迁移索引

31 个 API 路由逐条反查