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

学习场景

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

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

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

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 源码导航

create_app()、lifespan()、live() 与 ready()：区分应用创建、运行期初始化与健康请求。
run() 与 run_monitored()、Agent main()、heartbeat()：跟踪工作进程入口、心跳和停止处理。
instrument()、exporter lifespan()、metrics()：区分 HTTP 指标和任务指标的数据来源。
Prometheus 采集配置、告警规则、release_check.check()：核对采集周期、告警条件与发布检查覆盖范围。

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

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。

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

图 2 源码导航

CLI 入口、load_inputs()、Dataset.validate_isolation()、AgentManifest：理解数据集、运行绑定和固定参数。
evaluate()、initialize_report()、PersistentAllowance.take()：追踪互斥、报告恢复和持久额度。
execute_one()、build_graph()、BudgetCallback.on_chat_model_start()：识别直接调用与框架触发的模型回调。
observe()、Review.verify()、Judgment.validate_safety()、write_report()：核对观测、审核和最终报告之间的数据关系。

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

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 源码导航

检索评测 CLI、RetrievalManifest.validate_labels()、Review.verify()：核对清单标注与审核条件。
evaluate()、retrieve()：跟踪脚本请求和后端入口。
RetrievalClient.search()、embeddings()、verified_results()：理解候选获取和发布资料校验。
write_report()：检查完整报告如何经 .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 使用去重后的文档标识计算。

阅读检查问题

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

测试文件阅读入口

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

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

08-operations-evaluation.md 17 KB Permalink Geschiedenis Ruwe