# 路线 08：运行监控与评测证据

## 学习场景

部署后的服务需要持续回答三个问题：依赖是否可用，业务任务是否继续推进，已有回答与知识检索是否满足评测条件。本路线沿应用启动、工作进程心跳、指标采集和两个评测脚本，追踪这些结论所使用的代码与数据。

图中后端路径相对 `src/zhifu/`；`frontend/`、`scripts/`、`ops/` 等路径相对仓库根目录。连续箭头表示函数直接调用；间断箭头注明框架调度、HTTP、跨进程或者持久数据传递。配置、依赖框架和数据节点单独标明。返回[路线目录](README.md)。

## 图 1：服务启动、健康检查与任务监控

```mermaid
flowchart TD
    BOOT["Dockerfile / compose.runtime.yaml<br/>配置：Uvicorn 与工作进程入口"]
    APP["app.py<br/>create_app()<br/>注册路由、生命周期和指标"]
    LIFE["app.py<br/>lifespan()<br/>创建客户端并检查依赖"]
    READY["app.py<br/>ready() / database_probe()<br/>检查 MySQL 与两套 Redis"]
    HTTP_METRICS["core/telemetry.py<br/>instrument()<br/>注册 HTTP 计数与耗时采集"]
    HTTP_HANDLER["依赖 Instrumentator<br/>注册的 /metrics 处理函数<br/>提供当前进程 HTTP 指标"]
    RUN["operations/worker.py<br/>run(role) → run_monitored()<br/>运行指定工作进程"]
    HEART["core/telemetry.py<br/>heartbeat(role) / loop()<br/>定期更新本进程心跳"]
    MAIN["modules/agent_runtime/worker.py<br/>main()<br/>以 Agent 工作进程为例"]
    REDIS[("Redis 数据<br/>monitor:worker:{role}:*<br/>模型并发额度")]
    DB[("MySQL 数据<br/>AgentRun、MemoryJob<br/>BusinessEvent、ModelAttempt")]
    MONITOR["monitoring/api.py<br/>lifespan()<br/>创建 exporter 查询客户端"]
    METRICS["monitoring/api.py<br/>metrics()<br/>查询任务、用量、心跳与额度"]
    PROM["ops/prometheus/prometheus.yml<br/>配置：Prometheus 定时采集"]
    ALERT["ops/prometheus/alerts.yml<br/>配置：告警表达式与持续时间"]
    BOOT -. "Uvicorn factory 调用" .-> APP
    APP -. "FastAPI 生命周期调用" .-> LIFE
    APP --> HTTP_METRICS
    HTTP_METRICS -. "注册依赖提供的路由" .-> HTTP_HANDLER
    BOOT -. "healthcheck HTTP GET /health/ready" .-> READY
    BOOT -. "python -m 启动独立进程" .-> RUN
    RUN -->|进入上下文| HEART
    RUN -. "role=agent：asyncio.create_task 调度 main" .-> MAIN
    HEART -. "每 5 秒写入，TTL 为 20 秒" .-> REDIS
    BOOT -. "exporter 的 FastAPI 生命周期" .-> MONITOR
    PROM -. "HTTP GET /metrics" .-> METRICS
    PROM -. "HTTP GET /metrics" .-> HTTP_HANDLER
    DB -. "采集时查询任务与 token" .-> METRICS
    REDIS -. "采集时查询心跳与有效额度" .-> METRICS
    PROM -. "读取规则并计算告警条件" .-> ALERT
```

API 的 `create_app()` 创建 FastAPI 应用，并调用 `instrument()` 注册 HTTP 指标。Uvicorn 启动应用时，由 FastAPI 进入 `lifespan()`，初始化 MySQL、两套 Redis 与知识检索客户端。`/health/live` 返回进程存活结果，`/health/ready` 查询 MySQL 和两套 Redis，失败返回 `503`。

统一工作进程入口 `run(role)` 通过 `MODULES` 选择模块 `main()`，`run_monitored()` 进入 `heartbeat()`，并通过 `asyncio.create_task()` 调度业务函数与停止等待。停止信号触发后取消业务任务，等待清理；心跳上下文退出时删除本进程的 Redis key。

`monitoring/api.py` 是独立 exporter。`metrics()` 在 8 秒期限内查询待处理任务、最早等待时间、最近 15 分钟记录、token、工作进程心跳与有效模型额度。Prometheus 的 `operations` 采集超时配置为 4 秒。告警规则进一步检查心跳缺失、等待时间、HTTP 错误比例及 Agent 失败数量。

完整发布检查由 `operations/release_check.py:check()` 单独执行：核对两套数据库迁移版本、Redis、RabbitMQ、Milvus 向量维度、一次 embeddings 请求、API 和 OMS 健康接口，以及生成模型配置完整性。

### 图 1 源码导航

1. [create_app()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/app.py:36)、[lifespan()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/app.py:40)、[live() 与 ready()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/app.py:145)：区分应用创建、运行期初始化与健康请求。
2. [run() 与 run_monitored()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/operations/worker.py:49)、[Agent main()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/worker.py:210)、[heartbeat()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/core/telemetry.py:31)：跟踪工作进程入口、心跳和停止处理。
3. [instrument()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/core/telemetry.py:12)、[exporter lifespan()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/monitoring/api.py:21)、[metrics()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/monitoring/api.py:43)：区分 HTTP 指标和任务指标的数据来源。
4. [Prometheus 采集配置](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/ops/prometheus/prometheus.yml:9)、[告警规则](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/ops/prometheus/alerts.yml:1)、[release_check.check()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/operations/release_check.py:36)：核对采集周期、告警条件与发布检查覆盖范围。

## 图 2：Agent 评测清单、执行额度与结果审核

```mermaid
flowchart TD
    CLI["scripts/evaluate_agent_runs.py<br/>__main__ 模块入口<br/>解析参数并选择校验或评测"]
    INPUT["scripts/evaluate_agent_runs.py<br/>load_inputs()<br/>校验资料、源码与审核证据"]
    VALIDATE["evaluation.py<br/>Dataset / AgentManifest<br/>Pydantic 校验；Review.verify()"]
    EVAL["scripts/evaluate_agent_runs.py<br/>evaluate()<br/>锁定清单并逐项处理"]
    INIT["scripts/evaluate_agent_runs.py<br/>initialize_report()<br/>绑定输出文件、额度及期限"]
    OBSERVE["scripts/evaluate_agent_runs.py<br/>observe()<br/>读取运行、模型尝试与外部操作"]
    DB[("MySQL 持久数据<br/>AgentRun、ModelAttempt<br/>ExternalOperation")]
    EXECUTE["modules/agent_runtime/worker.py<br/>execute_one()<br/>限定 run_ids 领取并执行"]
    GRAPH["modules/agent_runtime/runtime.py<br/>build_graph() / model()<br/>构建运行图及模型请求"]
    CALLBACK["modules/agent_runtime/budget_callback.py<br/>BudgetCallback.on_chat_model_start()<br/>模型外呼前执行额度检查"]
    TAKE["scripts/evaluate_agent_runs.py<br/>PersistentAllowance.take()<br/>原子预留次数并保存报告"]
    REDIS[("Redis 持久数据<br/>evaluation-budget:{manifest_hash}<br/>calls、deadline、output、mode")]
    JUDGE["evaluation.py<br/>Judgment.model_validate() / verify()<br/>校验结论与证据摘要"]
    WRITE["evaluation.py<br/>write_report()<br/>写入完整 JSON 报告"]
    CLI --> INPUT
    INPUT --> VALIDATE
    CLI -. "未指定 validate-only：asyncio.run" .-> EVAL
    EVAL --> INIT
    INIT -. "读取或登记持久评测状态" .-> REDIS
    EVAL -->|执行前及适用的执行后| OBSERVE
    DB -. "读取并核对清单绑定的业务记录" .-> OBSERVE
    EVAL -->|execute 且运行状态为 QUEUED| EXECUTE
    EXECUTE -->|调用 build_graph| GRAPH
    GRAPH -. "LangGraph 调度 model；LangChain 触发模型开始回调" .-> CALLBACK
    CALLBACK --> TAKE
    TAKE -. "Redis Lua 检查期限与次数并增加 calls" .-> REDIS
    TAKE --> WRITE
    EVAL -->|提供 judgments 或复用已有审核| JUDGE
    EVAL -->|汇总类别、尝试次数与用量| WRITE
```

`load_inputs()` 使用 Pydantic 读取数据集和运行清单，核对数据文件摘要、当前后端源码摘要、数据集合及审核证据。`--validate-only` 在完成这些校验后结束。普通评测进入 `evaluate()`，读取已有业务记录，同时维护 Redis 评测记录和本地报告。

`evaluate()` 使用清单摘要作为互斥锁标识。`initialize_report()` 将同一清单绑定到唯一输出文件，恢复时读取原有次数与期限。启用 `--execute` 时还检查 Agent 和 memory 常驻工作进程的心跳；只有本次观测状态为 `QUEUED` 的指定运行会交给 `execute_one()`。

`execute_one()` 调用 `build_graph()`，随后通过 `graph.ainvoke()` 进入 LangGraph。模型调用使用 `BudgetCallback`；LangChain 在模型请求开始前调用 `on_chat_model_start()`，后者调用本次评测传入的 `PersistentAllowance.take()`。次数及总输出上限共同限制可预留请求数，Redis 时间检查总期限，预留次数随报告保存。Agent 内部节点与业务工具继续阅读 [路线 02](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/docs/reading-routes/02-agent-runtime.md)。

`observe()` 核对租户、运行图版本、模型名、发布参数摘要、单次输出上限和初始问题，再读取模型尝试及关联外部操作。存在模型尝试的记录进入 `REVIEW_REQUIRED`；结论文件必须匹配当前 `observation_sha256`。运行仍在 `QUEUED` 或 `RUNNING` 时不能接受该审核结论。`Judgment` 禁止将存在权限越界或重复外部操作的结果登记为 `PASSED`。

### 图 2 源码导航

1. [CLI 入口](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/scripts/evaluate_agent_runs.py:224)、[load_inputs()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/scripts/evaluate_agent_runs.py:82)、[Dataset.validate_isolation()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/evaluation.py:51)、[AgentManifest](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/evaluation.py:106)：理解数据集、运行绑定和固定参数。
2. [evaluate()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/scripts/evaluate_agent_runs.py:139)、[initialize_report()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/scripts/evaluate_agent_runs.py:46)、[PersistentAllowance.take()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/scripts/evaluate_agent_runs.py:26)：追踪互斥、报告恢复和持久额度。
3. [execute_one()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/worker.py:137)、[build_graph()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/runtime.py:244)、[BudgetCallback.on_chat_model_start()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/budget_callback.py:49)：识别直接调用与框架触发的模型回调。
4. [observe()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/scripts/evaluate_agent_runs.py:106)、[Review.verify()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/evaluation.py:74)、[Judgment.validate_safety()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/evaluation.py:90)、[write_report()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/evaluation.py:24)：核对观测、审核和最终报告之间的数据关系。

## 图 3：知识检索评测与 Recall@10

```mermaid
flowchart TD
    CLI["scripts/evaluate_retrieval.py<br/>__main__ 模块入口<br/>加载清单与审核资料"]
    MANIFEST["evaluation.py<br/>RetrievalManifest.model_validate_json()<br/>读取有类型约束的检索清单"]
    LABELS["evaluation.py<br/>RetrievalManifest.validate_labels()<br/>检查版本、唯一性和标注范围"]
    REVIEW["evaluation.py<br/>Review.verify()<br/>核对适用的审核证据"]
    EVAL["scripts/evaluate_retrieval.py<br/>evaluate()<br/>逐项请求并计算召回结果"]
    API["modules/knowledge/api.py<br/>retrieve()<br/>认证与店铺读取权限检查"]
    SEARCH["modules/knowledge/vector.py<br/>RetrievalClient.search()<br/>获取混合检索候选"]
    EMBED["modules/knowledge/vector.py<br/>RetrievalClient.embeddings()<br/>请求查询向量"]
    VECTOR[("独立依赖服务<br/>BGE embeddings / Milvus 索引")]
    VERIFY["modules/knowledge/api.py<br/>verified_results()<br/>按当前发布版本过滤候选"]
    DB[("MySQL 持久数据<br/>知识文档、发布版本和访问权限")]
    WRITE["evaluation.py<br/>write_report()<br/>保存每项结果与最终状态"]
    CLI --> MANIFEST
    MANIFEST -. "Pydantic model_validator 调用" .-> LABELS
    CLI -->|清单包含审核资料时| REVIEW
    CLI -. "asyncio.run" .-> EVAL
    EVAL -. "HTTP POST knowledge/search；limit=10；FastAPI 路由" .-> API
    API --> SEARCH
    SEARCH --> EMBED
    EMBED -. "HTTP POST /embeddings" .-> VECTOR
    SEARCH -. "Milvus 客户端 hybrid_search" .-> VECTOR
    API -->|得到候选后，重新开启事务| VERIFY
    DB -. "读取当前发布版本与权限" .-> VERIFY
    VERIFY -. "retrieve 返回 HTTP items" .-> EVAL
    EVAL -->|逐项及 finally 保存| WRITE
```

`RetrievalManifest` 固定店铺、文档版本、查询文本、预期文档以及总运行时间。`validate_labels()` 检查文档版本、查询唯一性和预期文档范围；`origin=production_reviewed` 要求人工相关性审核及模型部署证据，入口再调用 `Review.verify()` 核对证据文件摘要。

`evaluate()` 对每条查询请求 `limit=10`，请求由 FastAPI 分派给 `retrieve()`。后端先验证身份和店铺权限，通过 `RetrievalClient.search()` 获取候选，再开启事务重新认证并调用 `verified_results()`，保留当前有效发布资料。向量和知识权限内部过程可结合知识路线阅读。

脚本检查返回资料是否在清单内，且版本相同，然后按 `document_id` 去重。单条 `recall_at_10` 为命中的预期文档数除以预期文档数；报告同时计算类别平均值和全部查询平均值。全部平均值达到 `0.9` 时状态为 `PASSED`，否则为 `FAILED`。执行中断时 `finally` 将尚未完成的 `RUNNING` 改为 `INCOMPLETE` 并保存已有记录。报告固定登记 `generation_attempts=0`；检索过程仍调用 embeddings 与 Milvus。

### 图 3 源码导航

1. [检索评测 CLI](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/scripts/evaluate_retrieval.py:68)、[RetrievalManifest.validate_labels()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/evaluation.py:147)、[Review.verify()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/evaluation.py:74)：核对清单标注与审核条件。
2. [evaluate()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/scripts/evaluate_retrieval.py:15)、[retrieve()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/knowledge/api.py:289)：跟踪脚本请求和后端入口。
3. [RetrievalClient.search()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/knowledge/vector.py:98)、[embeddings()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/knowledge/vector.py:66)、[verified_results()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/knowledge/api.py:243)：理解候选获取和发布资料校验。
4. [write_report()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/evaluation.py:24)：检查完整报告如何经 `.pending` 文件替换目标文件。

## 关键数据与状态

- `monitor:worker:{role}:*` 表示单个工作进程心跳；TTL 使异常退出后的记录自动失效。`zhifu_workers_alive` 在采集时统计有效记录。
- `zhifu_pending_jobs` 与 `zhifu_oldest_pending_seconds` 反映任务等待；`zhifu_jobs_15m` 按记录创建时间筛选后统计当前状态；`zhifu_model_tokens_15m` 汇总对应时间范围内记录的用量。
- `manifest_sha256` 关联评测互斥锁、持久额度与报告。`source_sha256` 固定当前后端 Python 源码，`release_sha256` 固定运行发布参数。
- `reserved_calls` 记录评测预留次数；`recorded_attempts` 统计观测到的模型尝试；`attempts_without_usage` 保留用量未知记录。这些字段分别保存不同阶段的证据。
- Agent 案例状态包含 `NOT_RUN`、`REVIEW_REQUIRED`、`PASSED`、`FAILED`。数据库运行状态和评测审核状态分别维护。
- 检索报告状态包含 `RUNNING`、`PASSED`、`FAILED`、`INCOMPLETE`；`recall_at_10` 使用去重后的文档标识计算。

## 阅读检查问题

1. `/health/ready` 正常时，怎样从心跳、待处理数量、等待时间和模型额度判断后台任务是否持续推进？发布检查还覆盖哪些依赖？
2. 同一 Agent 评测进程中断并重新运行时，哪些字段阻止扩大调用次数？哪些摘要保证审核结果对应当前源码、运行参数和观测证据？
3. 检索报告的单项召回率、类别平均值和最终状态分别怎样计算？文档版本变化、请求异常与审核资料缺失分别在哪个位置结束处理？

## 测试文件阅读入口

本路线只阅读用例及其断言；编写本文未执行测试、评测脚本或真实模型调用。

- [test_monitoring.py](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/tests/test_monitoring.py:20)：HTTP 指标标签、身份信息保护、真实数据库指标和工作进程心跳。
- [test_evaluation_validation.py](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/tests/test_evaluation_validation.py:13)：数据集合、审核摘要、结果约束、检索标注与报告替换。
- [test_evaluation_runtime.py](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/tests/test_evaluation_runtime.py:24)：持久额度恢复、总输出上限、指定运行领取和真实工具结果观测。
- [process_recovery_acceptance.py](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/tests/process_recovery_acceptance.py:23)：自然租约到期、重新领取与旧执行 token 失效，用于连接运行监控和恢复证据。