From ebb5d410ad2070edef9b190f4487e5d4bc921041 Mon Sep 17 00:00:00 2001 From: yuqei Date: Wed, 26 Aug 2026 17:02:55 +0800 Subject: [PATCH] feat: add controlled pilot capacity quality gates --- ...°-受控试点容量与分层质量评测报告-2026-08-26.md | 146 +++++ .../packages/ai-openapi/openapi.yaml | 129 ++++- .../services/ai-platform-api/.env.example | 7 + .../services/ai-platform-api/README.md | 16 + ...826_0028_add_stratified_quality_reviews.py | 110 ++++ .../yuqei_ai_platform_api/api/model_jobs.py | 130 ++++- .../src/yuqei_ai_platform_api/config.py | 5 + .../src/yuqei_ai_platform_api/main.py | 1 + .../src/yuqei_ai_platform_api/model_jobs.py | 540 ++++++++++++++++-- .../src/yuqei_ai_platform_api/models.py | 35 ++ .../src/yuqei_ai_platform_api/schema_check.py | 22 + .../ai-platform-api/tests/test_health.py | 4 +- .../ai-platform-api/tests/test_model_jobs.py | 248 +++++++- yuqei-ai-sdk-python/src/yuqei_sdk/__init__.py | 8 + .../src/yuqei_sdk/ai_platform.py | 120 +++- yuqei-ai-sdk-python/tests/test_ai_platform.py | 43 +- .../src/app/admin/model-jobs/page.tsx | 136 ++++- .../[jobId]/quality-reviews/route.ts | 24 + .../admin/model-jobs/review-queue/route.ts | 6 + .../src/services/contracts-client.ts | 43 +- .../contract-web/src/services/contracts.ts | 52 ++ .../packages/contract-openapi/openapi.yaml | 125 +++- .../yuqei_contract_api/api/contracts_core.py | 70 ++- .../contract-api/tests/test_legal_qa_proxy.py | 47 +- yuqei-infra/README.md | 17 + ...run-controlled-pilot-stratified-quality.py | 439 ++++++++++++++ ...est_controlled_pilot_stratified_quality.py | 81 +++ 27 files changed, 2541 insertions(+), 63 deletions(-) create mode 100644 project-docs/AI合同平台-受控试点容量与分层质量评测报告-2026-08-26.md create mode 100644 yuqei-ai-platform/services/ai-platform-api/migrations/versions/20260826_0028_add_stratified_quality_reviews.py create mode 100644 yuqei-contract-platform/apps/contract-web/src/app/api/contract-platform/admin/model-jobs/[jobId]/quality-reviews/route.ts create mode 100644 yuqei-contract-platform/apps/contract-web/src/app/api/contract-platform/admin/model-jobs/review-queue/route.ts create mode 100644 yuqei-infra/scripts/run-controlled-pilot-stratified-quality.py create mode 100644 yuqei-infra/tests/test_controlled_pilot_stratified_quality.py diff --git a/project-docs/AI合同平台-受控试点容量与分层质量评测报告-2026-08-26.md b/project-docs/AI合同平台-受控试点容量与分层质量评测报告-2026-08-26.md new file mode 100644 index 00000000..e19ca538 --- /dev/null +++ b/project-docs/AI合同平台-受控试点容量与分层质量评测报告-2026-08-26.md @@ -0,0 +1,146 @@ +# AI 合同平台受控试点容量与分层质量评测报告 + +- 日期:2026-08-26 +- 运行 ID:`stratified-v1-20260826` +- 阶段:核心产品上线收口——受控试点容量与分层质量评测 +- 自动评测结论:**通过** +- 法律内容人工验收:**未完成,不得视为已通过** +- 发布建议:可继续内部和具名受控试点;不得据此开放公共注册或宣称法律意见、司法存证或正式生产模型能力已验收 + +## 一、管理结论 + +本阶段已经建立租户级模型任务准入、100 条分层真实样本、1/3/5/10 并发阶梯、自动输出门禁、确定性人工抽样、高风险必审、不可变复核反馈和运营后台下钻能力。 + +本次 100 条样本全部由本地开发环境中的 `codex-cli-development / gpt-5.6-sol` 真实执行,未把 fallback、模拟响应或未运行质量门禁的任务计入基线。自动结构质量和既定容量门禁全部通过,但该门禁只证明输出结构、基本完整性和运行稳定性,不证明法律事实、法条时效性或具体案件结论正确。 + +| 结论项 | 结果 | 判定 | +| --- | ---: | --- | +| 真实 Provider 样本 | 100 / 100 | 通过 | +| 合同类型覆盖 | 10 类 × 每类 10 条 | 通过 | +| 自动结构质量通过 | 100 / 100 | 通过 | +| 10 并发准入拒绝 | 0 | 通过 | +| 模型执行 P95 | 179,040 ms | 通过,接近 180 秒门限 | +| fallback / 失败 / 受控重试 | 0 / 0 / 0 | 通过 | +| 待具名人工复核 | 43 | 待完成 | +| 已完成人工复核 | 0 | 未通过法律内容验收 | + +## 二、已落地能力 + +### 2.1 租户与用户准入 + +- 默认租户活动任务上限 10、单用户活动任务上限 3、租户队列上限 25。 +- 幂等键优先于容量判断;同一请求重试返回原任务,不重复占用名额。 +- 超限返回 HTTP 429 和 `Retry-After`,并提供租户活动/队列饱和度与运营告警。 +- 单 API 进程内使用同一临界区完成幂等查询、容量计数和写入,竞态测试中 20 个同时提交严格放行 10 个。 +- 当前限制是受控试点单 API 进程保证;扩展到多个 API 副本前,需要 Redis/Lua、数据库咨询锁或独立配额服务提供跨进程原子准入。 + +### 2.2 分层质量数据 + +每个任务可以携带运行 ID、样本 ID、合同类型、风险等级和行业标签。运营快照可以按评测运行过滤,并分别按合同类型、风险等级和行业计算样本量、自动通过率及人工复核进度。 + +本次固定样本覆盖:采购、销售、劳动、租赁、保密、软件服务、建设工程、数据处理、股权交易和运输合同;每类 10 条,共 100 条。风险分层为高 40、中 40、低 20;行业覆盖制造、零售、通用、科技、建筑、金融和物流。 + +### 2.3 人工复核闭环 + +- 自动质量异常、高风险样本、确定性抽样和人工接管均进入统一复核队列。 +- 高风险样本 100% 必审;其他评测样本按默认 10% 确定性抽样,重复运行同一任务不会改变抽样决定。 +- 复核人必须具名,并记录事实一致性、引用可核验性、高风险遗漏、处置结论和说明。 +- 复核事件追加写入独立表;模型原输出不被静默覆盖。 +- 数据库写入在同一事务内同时保存复核事件并更新任务状态,避免出现“任务已完成但证据未落库”或相反状态。 +- 管理页面提供模型输出与引用预览、试点租户/运行 ID 下钻和显式复核确认框。 + +## 三、真实容量曲线 + +| 并发 | 提交 | 准入重试 | 排队 P95 | 执行 P95 | 波次耗时 | +| ---: | ---: | ---: | ---: | ---: | ---: | +| 1 | 1 | 0 | 2,266 ms | 170,774 ms | 176,577 ms | +| 3 | 3 | 0 | 1,311 ms | 175,488 ms | 180,078 ms | +| 5 | 5 | 0 | 1,101 ms | 163,726 ms | 164,504 ms | +| 10 | 10 | 0 | 3,668 ms | 174,452 ms | 176,455 ms | + +全体 100 条样本: + +- 排队 P50 / P95:1,642 / 4,505 ms;最大 6,567 ms。 +- 模型执行 P50 / P95:156,712 / 179,040 ms;最大 221,713 ms。 +- 总耗时 P50 / P95:158,612 / 179,298 ms;最大 223,004 ms。 +- 3 条执行耗时超过 180 秒,其中 1 条超过 200 秒。P95 门禁通过,但长尾仍需治理。 +- 57 条任务以 `succeeded` 完成;43 条在模型真实成功和自动门禁通过后按策略进入 `awaiting_human`,不属于模型失败。 +- 每条输出至少包含 1 条引用;回答字符数最少 1,289,中位数约 1,977。这里只验证结构存在,不代表引用法律有效性已经人工核验。 + +## 四、分层结果 + +10 类合同均为 10/10 自动门禁通过。人工复核要求分布如下: + +| 合同类型 | 样本 | 自动通过 | 待人工复核 | +| --- | ---: | ---: | ---: | +| 采购 | 10 | 10 | 5 | +| 销售 | 10 | 10 | 4 | +| 劳动 | 10 | 10 | 4 | +| 租赁 | 10 | 10 | 4 | +| 保密 | 10 | 10 | 4 | +| 软件服务 | 10 | 10 | 4 | +| 建设工程 | 10 | 10 | 5 | +| 数据处理 | 10 | 10 | 4 | +| 股权交易 | 10 | 10 | 4 | +| 运输 | 10 | 10 | 5 | + +风险分层中,高风险 40/40 全部进入人工复核;中风险 40 条抽中 2 条,低风险 20 条抽中 1 条,共 43 条。当前没有自动质量异常触发的复核,人工复核完成数为 0。 + +## 五、门禁判定 + +| 门禁 | 阈值 | 实测 | 判定 | +| --- | --- | ---: | --- | +| 真实样本数 | ≥ 100 | 100 | 通过 | +| 合同类型覆盖 | 10 类且每类 ≥ 10 | 10 × 10 | 通过 | +| 总体自动门禁通过率 | ≥ 80% | 100% | 通过 | +| 各合同类型自动通过率 | ≥ 70% | 全部 100% | 通过 | +| 模型执行 P95 | ≤ 180 秒 | 179.04 秒 | 通过,黄色边界 | +| 10 并发准入拒绝 | 0 | 0 | 通过 | +| 高风险人工复核完成 | 100% | 0 / 40 | 未完成 | +| 全部抽检完成 | 100% | 0 / 43 | 未完成 | + +因此,本阶段只能签署“自动容量和结构质量门禁通过”,不能签署“法律内容质量通过”。 + +## 六、运行与工程证据 + +- PostgreSQL Alembic 已从 `20260826_0027` 升级到 `20260826_0028 (head)`;全新 SQLite 也已从零升级到同一 head。 +- 运行时 schema check 中数据库、版本、核心表、租户列和 30 个模型任务运营列均为 `ok`。 +- AI API、合同 API 和前端分别在 8101、8100、3000 启动;10 个独立 worker 参与容量运行。 +- AI API 全量测试、worker、Python SDK、合同代理、基础设施测试、Web 测试与 TypeScript 类型检查均通过;提交前会再次执行最终回归。 +- 两套 OpenAPI 3.1 文档均可被 YAML 解析,并已描述评测元数据、容量快照、复核队列、反馈对象和跨租户平台管理员下钻。 +- 源码密钥扫描的仓库全局结果仍因 38 个既有/未跟踪本地环境候选而失败;将扫描结果与本次 27 个变更文件交叉核对后,本次变更新增候选为 0。未修改或掩盖既有基线问题。 +- Chrome 只读 UAT 使用现有“UAT 法务”会话访问平台运营页,系统正确返回平台管理员权限不足,控制台无异常;未使用非管理员账号提交复核。平台管理员正向流程由自动化代理测试覆盖,仍建议由真实超级管理员完成一次最终界面确认。 +- 原始评测 JSON/Markdown 位于 `.runtime/controlled-pilot-stratified-quality/`,默认不进入 Git。证据只保存样本/任务/审计 ID、分层、状态、字符数、引用数和耗时,不保存问题正文或模型回答正文。 + +## 七、已知限制与上线边界 + +1. 100% 是自动结构门禁通过率,不是法律正确率。43 条具名复核未完成前,不应对外宣称内容质量已验收。 +2. 执行 P95 距 180 秒只剩 960 ms,且最大值 221.7 秒。受控试点应继续显示后台进度、允许取消和人工接管,不应恢复同步等待。 +3. 当前模型来源是开发期 Codex CLI 与个人订阅会话,不是可供真实客户使用的生产 Provider 合同、API 配额或服务等级协议。 +4. 本地运行时 anti-abuse 为 disabled;真实用户注册仍应保持关闭。公开试点前至少切换 observe/enforce,配置哈希密钥并验证每日调用/数据出口配额。 +5. PostgreSQL 当前缺少 `vector` 扩展和 `knowledge_chunks.embedding`,不阻断本次模型队列容量测试,但会限制真实知识库向量检索效果。 +6. 多 API 副本时,当前进程锁不能作为全局容量硬保证;必须先接入共享原子准入。 +7. 本阶段没有执行双云 KMS、外部不可变见证或司法效力验收,相关能力仍保持未执行状态。 + +## 八、下一步修改方案 + +### P0:受控试点放行前 + +1. 指定至少两名具名法律复核人员,完成 40 个高风险样本和 3 个抽样样本;每条核对事实、现行法源、引用定位和重大遗漏。 +2. 导出 43 条结构化复核反馈,按合同类型统计 corrected/rejected、不可核验引用和高风险遗漏;任一类型出现重大遗漏时重新评测该类型 10 条。 +3. 对 3 条超过 180 秒的任务分析 Provider 耗时和提示词长度;将 P95 目标留出至少 10% 余量后再扩大租户并发。 +4. 用真实超级管理员登录运营页,下钻 `controlled-pilot-v1 / stratified-v1-20260826`,验证筛选、输出预览、显式确认和复核落库,但不得用测试人员身份替代法务结论。 + +### P1:扩大到多租户前 + +1. 将准入计数迁移到 Redis/Lua、数据库咨询锁或独立配额服务,并用两个 API 副本重复 20/50 并发竞态测试。 +2. 增加双人独立复核/仲裁数据模型,避免首位复核人完成后第二位无法提交独立结论。 +3. 建立按租户和合同类型的错误预算:fallback、失败、P95、重大遗漏、引用不可核验率分别告警。 +4. 接入具备企业 API 合同、配额、数据处理条款和国内可访问性的正式模型 Provider;个人订阅仅保留开发期使用。 +5. 补齐 pgvector、anti-abuse、正式合规材料和受控注册门禁后,再评估扩大试点。 + +## 九、最终发布决定 + +本阶段建议状态为:**自动容量与结构质量门禁 PASS;法律内容验收 PENDING;公共发布 BLOCKED**。 + +允许继续的范围是内部开发和具名受控试点,租户活动任务不超过 10、单用户不超过 3,并保留后台任务、重试、fallback 和人工接管。完成 43 条人工复核及上述 P0 条件前,不开放真实用户自由注册,不将本地评测包装成第三方存证、司法效力或正式生产服务等级证明。 diff --git a/yuqei-ai-platform/packages/ai-openapi/openapi.yaml b/yuqei-ai-platform/packages/ai-openapi/openapi.yaml index a8f605e7..cdbf55fa 100644 --- a/yuqei-ai-platform/packages/ai-openapi/openapi.yaml +++ b/yuqei-ai-platform/packages/ai-openapi/openapi.yaml @@ -187,6 +187,11 @@ paths: in: header required: false schema: { type: string, maxLength: 128 } + - { name: X-Evaluation-Run-Id, in: header, required: false, schema: { type: string, maxLength: 128 } } + - { name: X-Evaluation-Case-Id, in: header, required: false, schema: { type: string, maxLength: 128 } } + - { name: X-Evaluation-Contract-Type, in: header, required: false, schema: { type: string, maxLength: 64 } } + - { name: X-Evaluation-Risk-Level, in: header, required: false, schema: { type: string, enum: [low, medium, high] } } + - { name: X-Evaluation-Industry, in: header, required: false, schema: { type: string, maxLength: 64 } } requestBody: required: true content: @@ -200,6 +205,8 @@ paths: application/json: schema: $ref: "#/components/schemas/ModelInferenceJob" + "429": + description: Tenant or user admission limit reached; retry after the response Retry-After interval /api/v1/legal/contract-review: post: operationId: runContractReview @@ -232,6 +239,11 @@ paths: in: header required: false schema: { type: string, maxLength: 128 } + - { name: X-Evaluation-Run-Id, in: header, required: false, schema: { type: string, maxLength: 128 } } + - { name: X-Evaluation-Case-Id, in: header, required: false, schema: { type: string, maxLength: 128 } } + - { name: X-Evaluation-Contract-Type, in: header, required: false, schema: { type: string, maxLength: 64 } } + - { name: X-Evaluation-Risk-Level, in: header, required: false, schema: { type: string, enum: [low, medium, high] } } + - { name: X-Evaluation-Industry, in: header, required: false, schema: { type: string, maxLength: 64 } } requestBody: required: true content: @@ -245,6 +257,8 @@ paths: application/json: schema: $ref: "#/components/schemas/ModelInferenceJob" + "429": + description: Tenant or user admission limit reached; retry after the response Retry-After interval /api/v1/model-jobs/history: get: operationId: listModelInferenceJobHistory @@ -283,6 +297,9 @@ paths: - name: target_sample_count in: query schema: { type: integer, minimum: 1, maximum: 1000, default: 30 } + - name: evaluation_run_id + in: query + schema: { type: string, maxLength: 128 } responses: "200": description: Model job operational snapshot @@ -290,6 +307,22 @@ paths: application/json: schema: $ref: "#/components/schemas/ModelJobOperationsSnapshot" + /api/v1/model-jobs/review-queue: + get: + operationId: listModelJobReviewQueue + summary: List tenant-scoped outputs requiring structured human quality review + parameters: + - name: limit + in: query + schema: { type: integer, minimum: 1, maximum: 100, default: 50 } + responses: + "200": + description: Human quality review queue + content: + application/json: + schema: + type: array + items: { $ref: "#/components/schemas/ModelInferenceJob" } /api/v1/model-jobs/{job_id}: get: operationId: getModelInferenceJob @@ -373,6 +406,36 @@ paths: application/json: schema: $ref: "#/components/schemas/ModelInferenceJob" + /api/v1/model-jobs/{job_id}/quality-reviews: + get: + operationId: listModelJobQualityReviews + summary: List immutable structured quality feedback for one model job + parameters: + - { name: job_id, in: path, required: true, schema: { type: string } } + responses: + "200": + description: Quality review feedback dataset + content: + application/json: + schema: + type: array + items: { $ref: "#/components/schemas/ModelJobQualityReview" } + post: + operationId: createModelJobQualityReview + summary: Complete human review and write structured feedback for rule learning + parameters: + - { name: job_id, in: path, required: true, schema: { type: string } } + requestBody: + required: true + content: + application/json: + schema: { $ref: "#/components/schemas/ModelJobQualityReviewRequest" } + responses: + "200": + description: Immutable quality review record + content: + application/json: + schema: { $ref: "#/components/schemas/ModelJobQualityReview" } /api/v1/legal/review: post: operationId: runLegalReview @@ -1723,6 +1786,16 @@ components: $ref: "#/components/schemas/ModelOutputQualityLabel" quality_summary: { type: string, nullable: true } quality_gate_version: { type: string } + evaluation_run_id: { type: string, nullable: true } + evaluation_case_id: { type: string, nullable: true } + evaluation_contract_type: { type: string, nullable: true } + evaluation_risk_level: { type: string, enum: [low, medium, high], nullable: true } + evaluation_industry: { type: string, nullable: true } + review_status: { type: string, enum: [not_required, required, completed] } + review_trigger: { type: string, enum: [quality_gate, high_risk, random_sample, manual], nullable: true } + reviewed_at: { type: string, format: date-time, nullable: true } + reviewed_by: { type: string, nullable: true } + review_disposition: { type: string, enum: [accepted, corrected, rejected], nullable: true } queue_latency_ms: { type: integer } execution_latency_ms: { type: integer } total_latency_ms: { type: integer } @@ -1784,7 +1857,7 @@ components: threshold: { type: number } ModelJobOperationsSnapshot: type: object - required: [metrics, alerts, completed_sample_count, target_sample_count, remaining_sample_count, baseline_ready, quality_passed, quality_needs_review, quality_failed, quality_pass_rate] + required: [metrics, alerts, completed_sample_count, target_sample_count, remaining_sample_count, baseline_ready, quality_passed, quality_needs_review, quality_failed, quality_pass_rate, review_required, review_completed, admission, quality_segments] properties: metrics: $ref: "#/components/schemas/ModelInferenceJobMetrics" @@ -1800,6 +1873,60 @@ components: quality_needs_review: { type: integer } quality_failed: { type: integer } quality_pass_rate: { type: number } + review_required: { type: integer } + review_completed: { type: integer } + evaluation_run_id: { type: string, nullable: true } + admission: { $ref: "#/components/schemas/ModelJobAdmissionSnapshot" } + quality_segments: + type: array + items: { $ref: "#/components/schemas/ModelJobQualitySegment" } + ModelJobAdmissionSnapshot: + type: object + required: [running, queued, tenant_active_limit, user_active_limit, tenant_queue_limit, active_saturation_rate, queue_saturation_rate, accepting_submissions] + properties: + running: { type: integer } + queued: { type: integer } + tenant_active_limit: { type: integer } + user_active_limit: { type: integer } + tenant_queue_limit: { type: integer } + active_saturation_rate: { type: number } + queue_saturation_rate: { type: number } + accepting_submissions: { type: boolean } + ModelJobQualitySegment: + type: object + required: [dimension, value, sample_count, quality_passed, quality_needs_review, quality_failed, review_required, review_completed, quality_pass_rate] + properties: + dimension: { type: string, enum: [contract_type, risk_level, industry] } + value: { type: string } + sample_count: { type: integer } + quality_passed: { type: integer } + quality_needs_review: { type: integer } + quality_failed: { type: integer } + review_required: { type: integer } + review_completed: { type: integer } + quality_pass_rate: { type: number } + ModelJobQualityReviewRequest: + type: object + required: [legal_fact_consistency, citation_verifiability, high_risk_omission, disposition] + properties: + legal_fact_consistency: { type: string, enum: [pass, fail, uncertain, not_applicable] } + citation_verifiability: { type: string, enum: [pass, fail, uncertain, not_applicable] } + high_risk_omission: { type: string, enum: [not_detected, detected, uncertain, not_applicable] } + disposition: { type: string, enum: [accepted, corrected, rejected] } + feedback_text: { type: string, nullable: true, maxLength: 8000 } + ModelJobQualityReview: + allOf: + - { $ref: "#/components/schemas/ModelJobQualityReviewRequest" } + - type: object + required: [id, tenant_id, job_id, review_trigger, reviewer_id, quality_gate_version, created_at] + properties: + id: { type: string } + tenant_id: { type: string } + job_id: { type: string } + review_trigger: { type: string, enum: [quality_gate, high_risk, random_sample, manual] } + reviewer_id: { type: string } + quality_gate_version: { type: string } + created_at: { type: string, format: date-time } ContractReviewRequest: type: object required: diff --git a/yuqei-ai-platform/services/ai-platform-api/.env.example b/yuqei-ai-platform/services/ai-platform-api/.env.example index 9bb80b4b..b66f17e3 100644 --- a/yuqei-ai-platform/services/ai-platform-api/.env.example +++ b/yuqei-ai-platform/services/ai-platform-api/.env.example @@ -27,6 +27,13 @@ AI_PLATFORM_CODEX_CLI_TIMEOUT_SECONDS=240 AI_PLATFORM_CODEX_CLI_WORKING_DIR=.runtime/codex-model-provider AI_PLATFORM_MODEL_JOB_DISPATCH_MODE=background_tasks AI_PLATFORM_MODEL_JOB_MAX_ATTEMPTS=3 +# Controlled-pilot admission and human-sampling policy. Idempotent replays are +# resolved before these limits, so client retries do not create duplicate work. +AI_PLATFORM_MODEL_JOB_TENANT_ACTIVE_LIMIT=10 +AI_PLATFORM_MODEL_JOB_USER_ACTIVE_LIMIT=3 +AI_PLATFORM_MODEL_JOB_TENANT_QUEUE_LIMIT=25 +AI_PLATFORM_MODEL_JOB_MANUAL_SAMPLE_RATE_PERCENT=10 +AI_PLATFORM_MODEL_JOB_HIGH_RISK_REVIEW_REQUIRED=true # Local-only convenience; protected deployments resolve api_key_secret_ref from a mounted file. AI_PLATFORM_PROVIDER_API_KEY= AI_PLATFORM_PROVIDER_BASE_URL= diff --git a/yuqei-ai-platform/services/ai-platform-api/README.md b/yuqei-ai-platform/services/ai-platform-api/README.md index c57285b6..9313324d 100644 --- a/yuqei-ai-platform/services/ai-platform-api/README.md +++ b/yuqei-ai-platform/services/ai-platform-api/README.md @@ -180,11 +180,13 @@ Core MVP endpoints: - `POST /api/v1/legal/contract-review/jobs` - `GET /api/v1/model-jobs/history` - `GET /api/v1/model-jobs/operations` +- `GET /api/v1/model-jobs/review-queue` - `GET /api/v1/model-jobs/{job_id}` - `POST /api/v1/model-jobs/{job_id}/cancel` - `POST /api/v1/model-jobs/{job_id}/retry` - `POST /api/v1/model-jobs/{job_id}/handoff` - `POST /api/v1/model-jobs/{job_id}/resolve` +- `GET|POST /api/v1/model-jobs/{job_id}/quality-reviews` - `GET /api/v1/model-jobs/metrics` - `GET /api/v1/audit/ai-runs` @@ -205,6 +207,20 @@ or user identity is rejected. Completed outputs pass a deterministic structural quality gate before being marked successful, and cancellation preserves the job record and prior attempts for operations review. +Controlled pilots can attach an evaluation run, case, contract type, risk +level, and industry to each submitted job. Admission control applies tenant +active/queue limits and a per-user active limit, returning HTTP 429 plus +`Retry-After` when saturated; an idempotent replay is always resolved first. +High-risk evaluation cases, quality-gate exceptions, and a deterministic sample +of other evaluation cases enter the structured review queue. Human reviewers +record fact consistency, citation verifiability, high-risk omissions, and a +final disposition as immutable feedback instead of overwriting model output. +`GET /api/v1/model-jobs/operations?evaluation_run_id=...` separates queue, +execution, and total latency and reports quality by contract type, risk level, +and industry. The current admission check is designed for the controlled-pilot +single API process; multiple API replicas require a shared atomic admission +counter before the same limit can be treated as a hard global guarantee. + Use `GET /api/v1/knowledge-chunks/search/explain` to inspect normalized query terms, keyword/vector/hybrid scores, inclusion rank, and filter reasons for the candidate chunks considered by retrieval. diff --git a/yuqei-ai-platform/services/ai-platform-api/migrations/versions/20260826_0028_add_stratified_quality_reviews.py b/yuqei-ai-platform/services/ai-platform-api/migrations/versions/20260826_0028_add_stratified_quality_reviews.py new file mode 100644 index 00000000..da827220 --- /dev/null +++ b/yuqei-ai-platform/services/ai-platform-api/migrations/versions/20260826_0028_add_stratified_quality_reviews.py @@ -0,0 +1,110 @@ +"""add stratified model quality reviews + +Revision ID: 20260826_0028 +Revises: 20260826_0027 +Create Date: 2026-08-26 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "20260826_0028" +down_revision = "20260826_0027" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column("model_inference_jobs", sa.Column("evaluation_run_id", sa.String(length=128), nullable=True)) + op.add_column("model_inference_jobs", sa.Column("evaluation_case_id", sa.String(length=128), nullable=True)) + op.add_column("model_inference_jobs", sa.Column("evaluation_contract_type", sa.String(length=64), nullable=True)) + op.add_column("model_inference_jobs", sa.Column("evaluation_risk_level", sa.String(length=16), nullable=True)) + op.add_column("model_inference_jobs", sa.Column("evaluation_industry", sa.String(length=64), nullable=True)) + op.add_column( + "model_inference_jobs", + sa.Column("review_status", sa.String(length=32), nullable=False, server_default="not_required"), + ) + op.add_column("model_inference_jobs", sa.Column("review_trigger", sa.String(length=32), nullable=True)) + op.add_column("model_inference_jobs", sa.Column("reviewed_at", sa.DateTime(timezone=True), nullable=True)) + op.add_column("model_inference_jobs", sa.Column("reviewed_by", sa.String(length=128), nullable=True)) + op.add_column("model_inference_jobs", sa.Column("review_disposition", sa.String(length=32), nullable=True)) + op.create_index(op.f("ix_model_inference_jobs_evaluation_run_id"), "model_inference_jobs", ["evaluation_run_id"]) + op.create_index( + op.f("ix_model_inference_jobs_evaluation_contract_type"), + "model_inference_jobs", + ["evaluation_contract_type"], + ) + op.create_index( + op.f("ix_model_inference_jobs_evaluation_risk_level"), + "model_inference_jobs", + ["evaluation_risk_level"], + ) + op.create_index(op.f("ix_model_inference_jobs_review_status"), "model_inference_jobs", ["review_status"]) + op.create_index( + "ix_model_inference_jobs_tenant_evaluation_run_created", + "model_inference_jobs", + ["tenant_id", "evaluation_run_id", "created_at"], + ) + op.create_index( + "ix_model_inference_jobs_tenant_review_status_created", + "model_inference_jobs", + ["tenant_id", "review_status", "created_at"], + ) + + op.create_table( + "model_job_quality_reviews", + sa.Column("id", sa.String(length=64), nullable=False), + sa.Column("tenant_id", sa.String(length=64), nullable=False, server_default="default"), + sa.Column("job_id", sa.String(length=64), nullable=False), + sa.Column("review_trigger", sa.String(length=32), nullable=False), + sa.Column("reviewer_id", sa.String(length=128), nullable=False), + sa.Column("legal_fact_consistency", sa.String(length=32), nullable=False), + sa.Column("citation_verifiability", sa.String(length=32), nullable=False), + sa.Column("high_risk_omission", sa.String(length=32), nullable=False), + sa.Column("disposition", sa.String(length=32), nullable=False), + sa.Column("feedback_text", sa.Text(), nullable=True), + sa.Column("quality_gate_version", sa.String(length=32), nullable=False, server_default="model-job-v1"), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index(op.f("ix_model_job_quality_reviews_tenant_id"), "model_job_quality_reviews", ["tenant_id"]) + op.create_index(op.f("ix_model_job_quality_reviews_job_id"), "model_job_quality_reviews", ["job_id"]) + op.create_index(op.f("ix_model_job_quality_reviews_created_at"), "model_job_quality_reviews", ["created_at"]) + op.create_index( + "ix_model_job_quality_reviews_tenant_job_created", + "model_job_quality_reviews", + ["tenant_id", "job_id", "created_at"], + ) + op.create_index( + "ix_model_job_quality_reviews_tenant_trigger_created", + "model_job_quality_reviews", + ["tenant_id", "review_trigger", "created_at"], + ) + + +def downgrade() -> None: + op.drop_index("ix_model_job_quality_reviews_tenant_trigger_created", table_name="model_job_quality_reviews") + op.drop_index("ix_model_job_quality_reviews_tenant_job_created", table_name="model_job_quality_reviews") + op.drop_index(op.f("ix_model_job_quality_reviews_created_at"), table_name="model_job_quality_reviews") + op.drop_index(op.f("ix_model_job_quality_reviews_job_id"), table_name="model_job_quality_reviews") + op.drop_index(op.f("ix_model_job_quality_reviews_tenant_id"), table_name="model_job_quality_reviews") + op.drop_table("model_job_quality_reviews") + op.drop_index("ix_model_inference_jobs_tenant_review_status_created", table_name="model_inference_jobs") + op.drop_index("ix_model_inference_jobs_tenant_evaluation_run_created", table_name="model_inference_jobs") + op.drop_index(op.f("ix_model_inference_jobs_review_status"), table_name="model_inference_jobs") + op.drop_index(op.f("ix_model_inference_jobs_evaluation_risk_level"), table_name="model_inference_jobs") + op.drop_index(op.f("ix_model_inference_jobs_evaluation_contract_type"), table_name="model_inference_jobs") + op.drop_index(op.f("ix_model_inference_jobs_evaluation_run_id"), table_name="model_inference_jobs") + op.drop_column("model_inference_jobs", "review_disposition") + op.drop_column("model_inference_jobs", "reviewed_by") + op.drop_column("model_inference_jobs", "reviewed_at") + op.drop_column("model_inference_jobs", "review_trigger") + op.drop_column("model_inference_jobs", "review_status") + op.drop_column("model_inference_jobs", "evaluation_industry") + op.drop_column("model_inference_jobs", "evaluation_risk_level") + op.drop_column("model_inference_jobs", "evaluation_contract_type") + op.drop_column("model_inference_jobs", "evaluation_case_id") + op.drop_column("model_inference_jobs", "evaluation_run_id") diff --git a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/api/model_jobs.py b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/api/model_jobs.py index 3f1aa0ff..8900805f 100644 --- a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/api/model_jobs.py +++ b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/api/model_jobs.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import hashlib from collections.abc import Callable from typing import Any @@ -11,6 +12,7 @@ from yuqei_ai_platform_api.contract_review import ContractReviewRequest from yuqei_ai_platform_api.legal_qa import LegalQaRequest from yuqei_ai_platform_api.model_jobs import ( ModelInferenceJob, + ModelJobAdmissionRejected, ModelInferenceJobCancelRequest, ModelInferenceJobHandoffRequest, ModelInferenceJobHistoryPage, @@ -20,6 +22,8 @@ from yuqei_ai_platform_api.model_jobs import ( ModelInferenceJobRetryRequest, ModelInferenceJobStore, ModelInferenceJobRecord, + ModelJobQualityReview, + ModelJobQualityReviewRequest, ) from yuqei_ai_platform_api.model_output_quality import evaluate_model_output_quality from yuqei_ai_platform_api.repository import AiPlatformRepository @@ -48,6 +52,11 @@ def register_model_job_routes( background_tasks: BackgroundTasks, http_request: Request, idempotency_key: str | None = Header(default=None, alias="Idempotency-Key", max_length=128), + evaluation_run_id: str | None = Header(default=None, alias="X-Evaluation-Run-Id", max_length=128), + evaluation_case_id: str | None = Header(default=None, alias="X-Evaluation-Case-Id", max_length=128), + evaluation_contract_type: str | None = Header(default=None, alias="X-Evaluation-Contract-Type", max_length=64), + evaluation_risk_level: str | None = Header(default=None, alias="X-Evaluation-Risk-Level", max_length=16), + evaluation_industry: str | None = Header(default=None, alias="X-Evaluation-Industry", max_length=64), ) -> ModelInferenceJob: tenant_id = current_tenant_id() try: @@ -58,7 +67,23 @@ def register_model_job_routes( max_attempts=settings.model_job_max_attempts, created_by=http_request.headers.get("X-User-Id"), idempotency_key=idempotency_key, + evaluation_metadata={ + "run_id": evaluation_run_id, + "case_id": evaluation_case_id, + "contract_type": evaluation_contract_type, + "risk_level": evaluation_risk_level, + "industry": evaluation_industry, + }, + tenant_active_limit=settings.model_job_tenant_active_limit, + user_active_limit=settings.model_job_user_active_limit, + tenant_queue_limit=settings.model_job_tenant_queue_limit, ) + except ModelJobAdmissionRejected as error: + raise HTTPException( + status_code=429, + detail={"code": error.code, "message": str(error)}, + headers={"Retry-After": str(error.retry_after_seconds)}, + ) from error except ValueError as error: raise HTTPException(status_code=409, detail=str(error)) from error if created: @@ -76,6 +101,11 @@ def register_model_job_routes( background_tasks: BackgroundTasks, http_request: Request, idempotency_key: str | None = Header(default=None, alias="Idempotency-Key", max_length=128), + evaluation_run_id: str | None = Header(default=None, alias="X-Evaluation-Run-Id", max_length=128), + evaluation_case_id: str | None = Header(default=None, alias="X-Evaluation-Case-Id", max_length=128), + evaluation_contract_type: str | None = Header(default=None, alias="X-Evaluation-Contract-Type", max_length=64), + evaluation_risk_level: str | None = Header(default=None, alias="X-Evaluation-Risk-Level", max_length=16), + evaluation_industry: str | None = Header(default=None, alias="X-Evaluation-Industry", max_length=64), ) -> ModelInferenceJob: tenant_id = current_tenant_id() try: @@ -86,7 +116,23 @@ def register_model_job_routes( max_attempts=settings.model_job_max_attempts, created_by=http_request.headers.get("X-User-Id"), idempotency_key=idempotency_key, + evaluation_metadata={ + "run_id": evaluation_run_id, + "case_id": evaluation_case_id, + "contract_type": evaluation_contract_type, + "risk_level": evaluation_risk_level, + "industry": evaluation_industry, + }, + tenant_active_limit=settings.model_job_tenant_active_limit, + user_active_limit=settings.model_job_user_active_limit, + tenant_queue_limit=settings.model_job_tenant_queue_limit, ) + except ModelJobAdmissionRejected as error: + raise HTTPException( + status_code=429, + detail={"code": error.code, "message": str(error)}, + headers={"Retry-After": str(error.retry_after_seconds)}, + ) from error except ValueError as error: raise HTTPException(status_code=409, detail=str(error)) from error if created: @@ -152,13 +198,26 @@ def register_model_job_routes( def get_model_job_operations( window_hours: int = Query(default=24 * 7, ge=1, le=24 * 90), target_sample_count: int = Query(default=30, ge=1, le=1000), + evaluation_run_id: str | None = Query(default=None, max_length=128), ) -> ModelJobOperationsSnapshot: return job_store.operations_snapshot( tenant_id=current_tenant_id(), window_hours=window_hours, target_sample_count=target_sample_count, + evaluation_run_id=evaluation_run_id, + tenant_active_limit=settings.model_job_tenant_active_limit, + user_active_limit=settings.model_job_user_active_limit, + tenant_queue_limit=settings.model_job_tenant_queue_limit, ) + @app.get( + f"{settings.api_prefix}/model-jobs/review-queue", + response_model=list[ModelInferenceJob], + tags=["model-jobs", "operations"], + ) + def get_model_job_review_queue(limit: int = Query(default=50, ge=1, le=100)) -> list[ModelInferenceJob]: + return job_store.review_queue(tenant_id=current_tenant_id(), limit=limit) + @app.get( f"{settings.api_prefix}/model-jobs/{{job_id}}", response_model=ModelInferenceJob, @@ -254,6 +313,40 @@ def register_model_job_routes( except ValueError as error: raise HTTPException(status_code=409, detail=str(error)) from error + @app.post( + f"{settings.api_prefix}/model-jobs/{{job_id}}/quality-reviews", + response_model=ModelJobQualityReview, + tags=["model-jobs", "operations"], + ) + def create_model_job_quality_review( + job_id: str, + payload: ModelJobQualityReviewRequest, + request: Request, + ) -> ModelJobQualityReview: + try: + _job, review = job_store.record_quality_review( + job_id, + tenant_id=current_tenant_id(), + reviewer_id=request.headers.get("X-User-Id") or "", + request=payload, + ) + return review + except KeyError as error: + raise HTTPException(status_code=404, detail="Model job not found.") from error + except ValueError as error: + raise HTTPException(status_code=409, detail=str(error)) from error + + @app.get( + f"{settings.api_prefix}/model-jobs/{{job_id}}/quality-reviews", + response_model=list[ModelJobQualityReview], + tags=["model-jobs", "operations"], + ) + def list_model_job_quality_reviews(job_id: str) -> list[ModelJobQualityReview]: + try: + return job_store.quality_reviews(job_id, tenant_id=current_tenant_id()) + except KeyError as error: + raise HTTPException(status_code=404, detail="Model job not found.") from error + def execute_model_job( *, @@ -319,16 +412,18 @@ def execute_model_job( error_message=audit.provider_error if audit is not None else "The provider run failed.", **quality_fields, ) + review_trigger = _automatic_review_trigger(claimed, quality.status, app.state.ai_platform_settings) if quality.status == "failed": return job_store.finish( job_id, tenant_id=tenant_id, result=result, - status="failed", + status="awaiting_human", provider_status=provider_status, audit_run_id=audit_run_id, error_code="output_quality_gate_failed", error_message=quality.summary, + review_trigger="quality_gate", **quality_fields, ) if audit_status == "fallback": @@ -353,6 +448,20 @@ def execute_model_job( audit_run_id=audit_run_id, error_code="output_quality_review_required", error_message=quality.summary, + review_trigger="quality_gate", + **quality_fields, + ) + if review_trigger is not None: + return job_store.finish( + job_id, + tenant_id=tenant_id, + result=result, + status="awaiting_human", + provider_status=provider_status, + audit_run_id=audit_run_id, + error_code="high_risk_review_required" if review_trigger == "high_risk" else "manual_sample_review_required", + error_message="受控试点输出需要人工抽检。", + review_trigger=review_trigger, **quality_fields, ) return job_store.finish( @@ -373,6 +482,25 @@ def execute_model_job( reset_current_tenant_id(tenant_token) +def _automatic_review_trigger( + job: ModelInferenceJobRecord, + quality_status: str, + settings: AiPlatformSettings, +) -> str | None: + if quality_status in {"failed", "needs_review"}: + return "quality_gate" + if settings.model_job_high_risk_review_required and job.evaluation_risk_level == "high": + return "high_risk" + sample_rate = max(0, min(settings.model_job_manual_sample_rate_percent, 100)) + if ( + job.evaluation_run_id + and sample_rate + and int(hashlib.sha256(job.id.encode("utf-8")).hexdigest()[:8], 16) % 100 < sample_rate + ): + return "random_sample" + return None + + def _dispatch_if_enabled( app: FastAPI, settings: AiPlatformSettings, diff --git a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/config.py b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/config.py index f18ce94d..e9c9c3ea 100644 --- a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/config.py +++ b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/config.py @@ -25,6 +25,11 @@ class AiPlatformSettings(BaseSettings): provider_fallback_enabled: bool = True model_job_dispatch_mode: Literal["background_tasks", "worker"] = "background_tasks" model_job_max_attempts: int = 3 + model_job_tenant_active_limit: int = 10 + model_job_user_active_limit: int = 3 + model_job_tenant_queue_limit: int = 25 + model_job_manual_sample_rate_percent: int = 10 + model_job_high_risk_review_required: bool = True codex_cli_path: str | None = None codex_cli_model: Literal["gpt-5.6-sol"] = "gpt-5.6-sol" codex_cli_timeout_seconds: float = 240.0 diff --git a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/main.py b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/main.py index fc3f4e1d..fd7494cb 100644 --- a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/main.py +++ b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/main.py @@ -71,6 +71,7 @@ def create_app( redis_url=resolved_settings.redis_url, ) app.state.anti_abuse_guard = anti_abuse_guard + app.state.ai_platform_settings = resolved_settings egress_budget_limits = EgressBudgetLimits( s3_records_per_day=resolved_settings.anti_abuse_s3_records_daily_limit, s4_records_per_day=resolved_settings.anti_abuse_s4_records_daily_limit, diff --git a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/model_jobs.py b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/model_jobs.py index ab77dd3c..196e8cb3 100644 --- a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/model_jobs.py +++ b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/model_jobs.py @@ -5,7 +5,7 @@ import binascii import hashlib import json from datetime import UTC, datetime, timedelta -from threading import Lock +from threading import RLock from typing import Any, Literal from uuid import uuid4 @@ -15,7 +15,7 @@ from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session, sessionmaker from yuqei_ai_platform_api.model_output_quality import ModelOutputQualityLabel -from yuqei_ai_platform_api.models import ModelInferenceJobModel +from yuqei_ai_platform_api.models import ModelInferenceJobModel, ModelJobQualityReviewModel ModelJobOperation = Literal["legal_qa", "contract_review"] @@ -32,6 +32,17 @@ ModelJobStatus = Literal[ TERMINAL_MODEL_JOB_STATUSES = {"succeeded", "fallback", "failed", "awaiting_human", "resolved", "cancelled"} RETRYABLE_MODEL_JOB_STATUSES = {"fallback", "failed", "awaiting_human"} +ReviewTrigger = Literal["quality_gate", "high_risk", "random_sample", "manual"] +ReviewDisposition = Literal["accepted", "corrected", "rejected"] +ReviewAssessment = Literal["pass", "fail", "uncertain", "not_applicable"] +HighRiskOmissionAssessment = Literal["not_detected", "detected", "uncertain", "not_applicable"] + + +class ModelJobAdmissionRejected(ValueError): + def __init__(self, code: str, detail: str, *, retry_after_seconds: int = 10) -> None: + super().__init__(detail) + self.code = code + self.retry_after_seconds = retry_after_seconds class ModelInferenceJob(BaseModel): @@ -65,6 +76,16 @@ class ModelInferenceJob(BaseModel): quality_labels: list[ModelOutputQualityLabel] = Field(default_factory=list) quality_summary: str | None = None quality_gate_version: str = "model-job-v1" + evaluation_run_id: str | None = None + evaluation_case_id: str | None = None + evaluation_contract_type: str | None = None + evaluation_risk_level: Literal["low", "medium", "high"] | None = None + evaluation_industry: str | None = None + review_status: Literal["not_required", "required", "completed"] = "not_required" + review_trigger: ReviewTrigger | None = None + reviewed_at: datetime | None = None + reviewed_by: str | None = None + review_disposition: ReviewDisposition | None = None queue_latency_ms: int = 0 execution_latency_ms: int = 0 total_latency_ms: int = 0 @@ -100,6 +121,29 @@ class ModelInferenceJobCancelRequest(BaseModel): reason: str | None = Field(default=None, max_length=1000) +class ModelJobQualityReviewRequest(BaseModel): + legal_fact_consistency: ReviewAssessment + citation_verifiability: ReviewAssessment + high_risk_omission: HighRiskOmissionAssessment + disposition: ReviewDisposition + feedback_text: str | None = Field(default=None, max_length=8000) + + +class ModelJobQualityReview(BaseModel): + id: str + tenant_id: str + job_id: str + review_trigger: ReviewTrigger + reviewer_id: str + legal_fact_consistency: ReviewAssessment + citation_verifiability: ReviewAssessment + high_risk_omission: HighRiskOmissionAssessment + disposition: ReviewDisposition + feedback_text: str | None = None + quality_gate_version: str + created_at: datetime + + class ModelInferenceJobHistoryPage(BaseModel): items: list[ModelInferenceJob] next_cursor: str | None = None @@ -138,6 +182,29 @@ class ModelJobOperationalAlert(BaseModel): threshold: float +class ModelJobAdmissionSnapshot(BaseModel): + running: int + queued: int + tenant_active_limit: int + user_active_limit: int + tenant_queue_limit: int + active_saturation_rate: float + queue_saturation_rate: float + accepting_submissions: bool + + +class ModelJobQualitySegment(BaseModel): + dimension: Literal["contract_type", "risk_level", "industry"] + value: str + sample_count: int + quality_passed: int + quality_needs_review: int + quality_failed: int + review_required: int + review_completed: int + quality_pass_rate: float + + class ModelJobOperationsSnapshot(BaseModel): metrics: ModelInferenceJobMetrics alerts: list[ModelJobOperationalAlert] @@ -149,6 +216,11 @@ class ModelJobOperationsSnapshot(BaseModel): quality_needs_review: int quality_failed: int quality_pass_rate: float + review_required: int = 0 + review_completed: int = 0 + evaluation_run_id: str | None = None + admission: ModelJobAdmissionSnapshot + quality_segments: list[ModelJobQualitySegment] = Field(default_factory=list) class ModelInferenceJobStore: @@ -158,9 +230,14 @@ class ModelInferenceJobStore: self._engine = engine self._session_factory = sessionmaker(bind=engine, expire_on_commit=False) if engine is not None else None self._records: dict[str, ModelInferenceJobRecord] = {} - self._lock = Lock() + self._reviews: dict[str, ModelJobQualityReview] = {} + # A re-entrant process lock makes idempotency lookup, admission counting, + # and insertion one critical section for the controlled-pilot API process. + # Database uniqueness remains the final idempotency guard. + self._lock = RLock() if engine is not None and auto_create: ModelInferenceJobModel.__table__.create(engine, checkfirst=True) + ModelJobQualityReviewModel.__table__.create(engine, checkfirst=True) def create( self, @@ -171,6 +248,10 @@ class ModelInferenceJobStore: max_attempts: int = 3, created_by: str | None = None, idempotency_key: str | None = None, + evaluation_metadata: dict[str, str | None] | None = None, + tenant_active_limit: int = 10, + user_active_limit: int = 3, + tenant_queue_limit: int = 25, ) -> ModelInferenceJob: job, _created = self.create_or_get( tenant_id=tenant_id, @@ -179,6 +260,10 @@ class ModelInferenceJobStore: max_attempts=max_attempts, created_by=created_by, idempotency_key=idempotency_key, + evaluation_metadata=evaluation_metadata, + tenant_active_limit=tenant_active_limit, + user_active_limit=user_active_limit, + tenant_queue_limit=tenant_queue_limit, ) return job @@ -191,11 +276,46 @@ class ModelInferenceJobStore: max_attempts: int = 3, created_by: str | None = None, idempotency_key: str | None = None, + evaluation_metadata: dict[str, str | None] | None = None, + tenant_active_limit: int = 10, + user_active_limit: int = 3, + tenant_queue_limit: int = 25, + ) -> tuple[ModelInferenceJob, bool]: + with self._lock: + return self._create_or_get_locked( + tenant_id=tenant_id, + operation=operation, + request_payload=request_payload, + max_attempts=max_attempts, + created_by=created_by, + idempotency_key=idempotency_key, + evaluation_metadata=evaluation_metadata, + tenant_active_limit=tenant_active_limit, + user_active_limit=user_active_limit, + tenant_queue_limit=tenant_queue_limit, + ) + + def _create_or_get_locked( + self, + *, + tenant_id: str, + operation: ModelJobOperation, + request_payload: dict[str, Any], + max_attempts: int = 3, + created_by: str | None = None, + idempotency_key: str | None = None, + evaluation_metadata: dict[str, str | None] | None = None, + tenant_active_limit: int = 10, + user_active_limit: int = 3, + tenant_queue_limit: int = 25, ) -> tuple[ModelInferenceJob, bool]: now = datetime.now(UTC) normalized_creator = (created_by or "").strip() or None normalized_key = _normalize_idempotency_key(idempotency_key) - fingerprint = _request_fingerprint({"created_by": normalized_creator, "request_payload": request_payload}) + metadata = _normalize_evaluation_metadata(evaluation_metadata) + fingerprint = _request_fingerprint( + {"created_by": normalized_creator, "request_payload": request_payload, "evaluation_metadata": metadata} + ) if normalized_key: existing = self._find_by_idempotency( tenant_id=tenant_id, @@ -205,6 +325,13 @@ class ModelInferenceJobStore: if existing is not None: _ensure_same_fingerprint(existing, fingerprint) return _view(existing), False + self.enforce_admission( + tenant_id=tenant_id, + created_by=normalized_creator, + tenant_active_limit=tenant_active_limit, + user_active_limit=user_active_limit, + tenant_queue_limit=tenant_queue_limit, + ) record = ModelInferenceJobRecord( id=f"model-job-{uuid4().hex[:16]}", tenant_id=tenant_id, @@ -213,6 +340,11 @@ class ModelInferenceJobStore: display_title=_display_title(operation, request_payload), idempotency_key=normalized_key, request_fingerprint=fingerprint, + evaluation_run_id=metadata.get("run_id"), + evaluation_case_id=metadata.get("case_id"), + evaluation_contract_type=metadata.get("contract_type"), + evaluation_risk_level=metadata.get("risk_level"), + evaluation_industry=metadata.get("industry"), status="queued", stage="queued", progress_percent=0, @@ -261,6 +393,70 @@ class ModelInferenceJobStore: return _view(existing), False return _view(record), True + def admission_snapshot( + self, + *, + tenant_id: str, + tenant_active_limit: int, + user_active_limit: int, + tenant_queue_limit: int, + ) -> ModelJobAdmissionSnapshot: + records = self._tenant_records(tenant_id=tenant_id) + running = sum(item.status == "running" for item in records) + queued = sum(item.status == "queued" for item in records) + active_limit = max(1, tenant_active_limit) + queue_limit = max(1, tenant_queue_limit) + return ModelJobAdmissionSnapshot( + running=running, + queued=queued, + tenant_active_limit=active_limit, + user_active_limit=max(1, user_active_limit), + tenant_queue_limit=queue_limit, + active_saturation_rate=round((running + queued) / active_limit, 4), + queue_saturation_rate=round(queued / queue_limit, 4), + accepting_submissions=(running + queued) < active_limit and queued < queue_limit, + ) + + def enforce_admission( + self, + *, + tenant_id: str, + created_by: str | None, + tenant_active_limit: int, + user_active_limit: int, + tenant_queue_limit: int, + ) -> None: + records = self._tenant_records(tenant_id=tenant_id) + active = [item for item in records if item.status in {"queued", "running"}] + active_limit = max(1, tenant_active_limit) + queue_limit = max(1, tenant_queue_limit) + user_limit = max(1, user_active_limit) + if len(active) >= active_limit: + raise ModelJobAdmissionRejected( + "tenant_active_limit_reached", + f"Tenant has reached the controlled-pilot limit of {active_limit} active model jobs.", + ) + if sum(item.status == "queued" for item in active) >= queue_limit: + raise ModelJobAdmissionRejected( + "tenant_queue_limit_reached", + f"Tenant queue has reached the controlled-pilot limit of {queue_limit} jobs.", + ) + if created_by and sum(item.created_by == created_by for item in active) >= user_limit: + raise ModelJobAdmissionRejected( + "user_active_limit_reached", + f"User has reached the controlled-pilot limit of {user_limit} active model jobs.", + ) + + def _tenant_records(self, *, tenant_id: str) -> list[ModelInferenceJobRecord]: + if self._session_factory is None: + with self._lock: + return [item for item in self._records.values() if item.tenant_id == tenant_id] + with self._session_factory() as session: + rows = session.scalars( + select(ModelInferenceJobModel).where(ModelInferenceJobModel.tenant_id == tenant_id) + ).all() + return [_row_to_record(row) for row in rows] + def _find_by_idempotency( self, *, @@ -518,6 +714,7 @@ class ModelInferenceJobStore: quality_labels: list[ModelOutputQualityLabel] | None = None, quality_summary: str | None = None, quality_gate_version: str = "model-job-v1", + review_trigger: ReviewTrigger | None = None, ) -> ModelInferenceJob: now = datetime.now(UTC) existing = self.get_record(job_id, tenant_id=tenant_id) @@ -550,7 +747,12 @@ class ModelInferenceJobStore: quality_labels=quality_labels or [], quality_summary=quality_summary, quality_gate_version=quality_gate_version, - handoff_reason="output_quality_review_required" if status == "awaiting_human" else existing.handoff_reason, + review_status="required" if status == "awaiting_human" else "not_required", + review_trigger=review_trigger if status == "awaiting_human" else None, + reviewed_at=None, + reviewed_by=None, + review_disposition=None, + handoff_reason=(review_trigger or "manual_review_required") if status == "awaiting_human" else existing.handoff_reason, handoff_note=quality_summary if status == "awaiting_human" else existing.handoff_note, execution_latency_ms=_elapsed_ms(started_at, now), total_latency_ms=_elapsed_ms(existing.created_at, now), @@ -606,6 +808,11 @@ class ModelInferenceJobStore: quality_score=0, quality_labels=[], quality_summary=None, + review_status="not_required", + review_trigger=None, + reviewed_at=None, + reviewed_by=None, + review_disposition=None, cancel_requested_at=None, cancelled_by=None, cancel_reason=None, @@ -670,6 +877,8 @@ class ModelInferenceJobStore: handoff_reason=reason, handoff_note=note, handoff_assignee=assignee, + review_status="required", + review_trigger="manual", ) def resolve(self, job_id: str, *, tenant_id: str, resolution_note: str) -> ModelInferenceJob: @@ -682,8 +891,109 @@ class ModelInferenceJobStore: progress_percent=100, retryable=False, resolution_note=resolution_note, + review_status="completed", + reviewed_at=datetime.now(UTC), + review_disposition="accepted", ) + def review_queue(self, *, tenant_id: str, limit: int = 50) -> list[ModelInferenceJob]: + records = [ + item for item in self._tenant_records(tenant_id=tenant_id) + if item.status == "awaiting_human" and item.review_status == "required" + ] + records.sort(key=lambda item: (_aware(item.updated_at), item.id)) + return [_view(item) for item in records[: max(1, min(limit, 100))]] + + def record_quality_review( + self, + job_id: str, + *, + tenant_id: str, + reviewer_id: str, + request: ModelJobQualityReviewRequest, + ) -> tuple[ModelInferenceJob, ModelJobQualityReview]: + normalized_reviewer = reviewer_id.strip() + if not normalized_reviewer: + raise ValueError("Reviewer identity is required.") + now = datetime.now(UTC) + summary = ( + f"事实一致性={request.legal_fact_consistency}; 引用可核验={request.citation_verifiability}; " + f"高风险遗漏={request.high_risk_omission}; 处置={request.disposition}" + ) + with self._lock: + if self._session_factory is None: + existing = self._records.get(job_id) + if existing is None or existing.tenant_id != tenant_id: + raise KeyError(job_id) + _ensure_reviewable(existing) + review = _new_quality_review( + existing, + reviewer_id=normalized_reviewer, + request=request, + created_at=now, + ) + self._reviews[review.id] = review + updated = existing.model_copy( + update={ + "status": "resolved", + "stage": "quality_review_completed", + "retryable": False, + "review_status": "completed", + "reviewed_at": now, + "reviewed_by": normalized_reviewer, + "review_disposition": request.disposition, + "resolution_note": summary, + "updated_at": now, + } + ) + self._records[job_id] = updated + return _view(updated), review + + with self._session_factory.begin() as session: + row = _locked_row(session, job_id) + if row is None or row.tenant_id != tenant_id: + raise KeyError(job_id) + existing = _row_to_record(row) + _ensure_reviewable(existing) + review = _new_quality_review( + existing, + reviewer_id=normalized_reviewer, + request=request, + created_at=now, + ) + session.add(_review_to_row(review)) + row.status = "resolved" + row.stage = "quality_review_completed" + row.retryable = False + row.review_status = "completed" + row.reviewed_at = now + row.reviewed_by = normalized_reviewer + row.review_disposition = request.disposition + row.resolution_note = summary + row.updated_at = now + session.flush() + return _view(_row_to_record(row)), review + + def quality_reviews(self, job_id: str, *, tenant_id: str) -> list[ModelJobQualityReview]: + if self.get_record(job_id, tenant_id=tenant_id) is None: + raise KeyError(job_id) + if self._session_factory is None: + with self._lock: + records = [ + item for item in self._reviews.values() + if item.tenant_id == tenant_id and item.job_id == job_id + ] + else: + with self._session_factory() as session: + rows = session.scalars( + select(ModelJobQualityReviewModel).where( + ModelJobQualityReviewModel.tenant_id == tenant_id, + ModelJobQualityReviewModel.job_id == job_id, + ).order_by(ModelJobQualityReviewModel.created_at.asc()) + ).all() + records = [_review_from_row(row) for row in rows] + return sorted(records, key=lambda item: (_aware(item.created_at), item.id)) + def metrics(self, *, tenant_id: str, window_hours: int) -> ModelInferenceJobMetrics: cutoff = datetime.now(UTC) - timedelta(hours=max(1, min(window_hours, 24 * 90))) if self._session_factory is None: @@ -701,38 +1011,7 @@ class ModelInferenceJobStore: ) ).all() records = [_row_to_record(row) for row in rows] - total = len(records) - completed = [ - record - for record in records - if record.status in TERMINAL_MODEL_JOB_STATUSES and record.status != "cancelled" - ] - denominator = max(len(completed), 1) - return ModelInferenceJobMetrics( - window_hours=window_hours, - total=total, - queued=sum(record.status == "queued" for record in records), - running=sum(record.status == "running" for record in records), - succeeded=sum(record.status == "succeeded" for record in records), - fallback=sum(record.fallback_used for record in records), - failed=sum(record.status == "failed" for record in records), - awaiting_human=sum(record.status == "awaiting_human" for record in records), - resolved=sum(record.status == "resolved" for record in records), - cancelled=sum(record.status == "cancelled" for record in records), - fallback_rate=round(sum(record.fallback_used for record in records) / denominator, 4), - failure_rate=round(sum(record.status == "failed" for record in records) / denominator, 4), - manual_handoff_rate=round( - sum(record.status in {"awaiting_human", "resolved"} for record in records) / denominator, - 4, - ), - retry_rate=round(sum(record.attempt_count > 1 for record in records) / max(total, 1), 4), - queue_latency_p50_ms=_percentile([record.queue_latency_ms for record in records], 0.50), - queue_latency_p95_ms=_percentile([record.queue_latency_ms for record in records], 0.95), - execution_latency_p50_ms=_percentile([record.execution_latency_ms for record in completed], 0.50), - execution_latency_p95_ms=_percentile([record.execution_latency_ms for record in completed], 0.95), - total_latency_p50_ms=_percentile([record.total_latency_ms for record in completed], 0.50), - total_latency_p95_ms=_percentile([record.total_latency_ms for record in completed], 0.95), - ) + return _metrics_from_records(records, window_hours=window_hours) def operations_snapshot( self, @@ -740,8 +1019,11 @@ class ModelInferenceJobStore: tenant_id: str, window_hours: int = 24 * 7, target_sample_count: int = 30, + evaluation_run_id: str | None = None, + tenant_active_limit: int = 10, + user_active_limit: int = 3, + tenant_queue_limit: int = 25, ) -> ModelJobOperationsSnapshot: - metrics = self.metrics(tenant_id=tenant_id, window_hours=window_hours) cutoff = datetime.now(UTC) - timedelta(hours=max(1, min(window_hours, 24 * 90))) if self._session_factory is None: with self._lock: @@ -758,7 +1040,12 @@ class ModelInferenceJobStore: ) ).all() records = [_row_to_record(row) for row in rows] - completed = [item for item in records if item.status in TERMINAL_MODEL_JOB_STATUSES] + evaluation_records = [ + item for item in records + if evaluation_run_id is None or item.evaluation_run_id == evaluation_run_id + ] + metrics = _metrics_from_records(evaluation_records, window_hours=window_hours) + completed = [item for item in evaluation_records if item.status in TERMINAL_MODEL_JOB_STATUSES] evaluated = [ item for item in completed @@ -768,6 +1055,14 @@ class ModelInferenceJobStore: quality_passed = sum(item.quality_status == "passed" for item in evaluated) quality_needs_review = sum(item.quality_status == "needs_review" for item in evaluated) quality_failed = sum(item.quality_status == "failed" for item in evaluated) + review_required = sum(item.review_status in {"required", "completed"} for item in evaluated) + review_completed = sum(item.review_status == "completed" for item in evaluated) + admission = self.admission_snapshot( + tenant_id=tenant_id, + tenant_active_limit=tenant_active_limit, + user_active_limit=user_active_limit, + tenant_queue_limit=tenant_queue_limit, + ) stale_cutoff = datetime.now(UTC) - timedelta(minutes=10) stale_active = sum( item.status in {"queued", "running"} and _aware(item.updated_at) < stale_cutoff @@ -798,6 +1093,28 @@ class ModelInferenceJobStore: alerts.append(_alert("manual_queue_backlog", "warning", "人工复核队列积压", "等待人工复核的任务达到 5 个。", metrics.awaiting_human, 5)) if stale_active: alerts.append(_alert("stale_active_jobs", "error", "存在停滞任务", "有排队或运行任务超过 10 分钟未更新。", stale_active, 0)) + if admission.active_saturation_rate >= 1: + alerts.append( + _alert( + "tenant_model_capacity_saturated", + "warning", + "租户模型容量已饱和", + "排队和运行中的模型任务已达到受控试点上限。", + admission.running + admission.queued, + admission.tenant_active_limit, + ) + ) + if admission.queue_saturation_rate >= 0.8: + alerts.append( + _alert( + "tenant_model_queue_near_limit", + "warning", + "租户模型队列接近上限", + "排队任务已达到受控试点队列上限的 80%。", + admission.queued, + admission.tenant_queue_limit * 0.8, + ) + ) quality_pass_rate = round(quality_passed / max(len(evaluated), 1), 4) return ModelJobOperationsSnapshot( metrics=metrics, @@ -810,6 +1127,11 @@ class ModelInferenceJobStore: quality_needs_review=quality_needs_review, quality_failed=quality_failed, quality_pass_rate=quality_pass_rate, + review_required=review_required, + review_completed=review_completed, + evaluation_run_id=evaluation_run_id, + admission=admission, + quality_segments=_quality_segments(evaluated), ) def _update( @@ -919,6 +1241,16 @@ def _row_to_record(row: ModelInferenceJobModel) -> ModelInferenceJobRecord: quality_labels=[ModelOutputQualityLabel.model_validate(item) for item in json.loads(row.quality_labels_json or "[]")], quality_summary=row.quality_summary, quality_gate_version=row.quality_gate_version, + evaluation_run_id=row.evaluation_run_id, + evaluation_case_id=row.evaluation_case_id, + evaluation_contract_type=row.evaluation_contract_type, + evaluation_risk_level=row.evaluation_risk_level, + evaluation_industry=row.evaluation_industry, + review_status=row.review_status, + review_trigger=row.review_trigger, + reviewed_at=_aware(row.reviewed_at) if row.reviewed_at else None, + reviewed_by=row.reviewed_by, + review_disposition=row.review_disposition, queue_latency_ms=row.queue_latency_ms, execution_latency_ms=row.execution_latency_ms, total_latency_ms=row.total_latency_ms, @@ -945,6 +1277,45 @@ def _elapsed_ms(start: datetime, end: datetime) -> int: return max(0, int((_aware(end) - _aware(start)).total_seconds() * 1000)) +def _metrics_from_records( + records: list[ModelInferenceJobRecord], + *, + window_hours: int, +) -> ModelInferenceJobMetrics: + total = len(records) + completed = [ + record + for record in records + if record.status in TERMINAL_MODEL_JOB_STATUSES and record.status != "cancelled" + ] + denominator = max(len(completed), 1) + return ModelInferenceJobMetrics( + window_hours=window_hours, + total=total, + queued=sum(record.status == "queued" for record in records), + running=sum(record.status == "running" for record in records), + succeeded=sum(record.status == "succeeded" for record in records), + fallback=sum(record.fallback_used for record in records), + failed=sum(record.status == "failed" for record in records), + awaiting_human=sum(record.status == "awaiting_human" for record in records), + resolved=sum(record.status == "resolved" for record in records), + cancelled=sum(record.status == "cancelled" for record in records), + fallback_rate=round(sum(record.fallback_used for record in records) / denominator, 4), + failure_rate=round(sum(record.status == "failed" for record in records) / denominator, 4), + manual_handoff_rate=round( + sum(record.status in {"awaiting_human", "resolved"} for record in records) / denominator, + 4, + ), + retry_rate=round(sum(record.attempt_count > 1 for record in records) / max(total, 1), 4), + queue_latency_p50_ms=_percentile([record.queue_latency_ms for record in records], 0.50), + queue_latency_p95_ms=_percentile([record.queue_latency_ms for record in records], 0.95), + execution_latency_p50_ms=_percentile([record.execution_latency_ms for record in completed], 0.50), + execution_latency_p95_ms=_percentile([record.execution_latency_ms for record in completed], 0.95), + total_latency_p50_ms=_percentile([record.total_latency_ms for record in completed], 0.50), + total_latency_p95_ms=_percentile([record.total_latency_ms for record in completed], 0.95), + ) + + def _percentile(values: list[int], percentile: float) -> int: normalized = sorted(value for value in values if value >= 0) if not normalized: @@ -967,6 +1338,48 @@ def _request_fingerprint(payload: dict[str, Any]) -> str: return hashlib.sha256(canonical.encode("utf-8")).hexdigest() +def _normalize_evaluation_metadata(value: dict[str, str | None] | None) -> dict[str, str | None]: + raw = value or {} + result: dict[str, str | None] = {} + limits = {"run_id": 128, "case_id": 128, "contract_type": 64, "risk_level": 16, "industry": 64} + for key, limit in limits.items(): + normalized = (raw.get(key) or "").strip() or None + if normalized and len(normalized) > limit: + raise ValueError(f"Evaluation metadata field {key} exceeds {limit} characters.") + result[key] = normalized + if result["risk_level"] not in {None, "low", "medium", "high"}: + raise ValueError("Evaluation risk level must be low, medium, or high.") + return result + + +def _quality_segments(records: list[ModelInferenceJobRecord]) -> list[ModelJobQualitySegment]: + dimensions = { + "contract_type": "evaluation_contract_type", + "risk_level": "evaluation_risk_level", + "industry": "evaluation_industry", + } + result: list[ModelJobQualitySegment] = [] + for dimension, field_name in dimensions.items(): + values = sorted({str(getattr(item, field_name)) for item in records if getattr(item, field_name)}) + for value in values: + selected = [item for item in records if getattr(item, field_name) == value] + passed = sum(item.quality_status == "passed" for item in selected) + result.append( + ModelJobQualitySegment( + dimension=dimension, # type: ignore[arg-type] + value=value, + sample_count=len(selected), + quality_passed=passed, + quality_needs_review=sum(item.quality_status == "needs_review" for item in selected), + quality_failed=sum(item.quality_status == "failed" for item in selected), + review_required=sum(item.review_status in {"required", "completed"} for item in selected), + review_completed=sum(item.review_status == "completed" for item in selected), + quality_pass_rate=round(passed / max(len(selected), 1), 4), + ) + ) + return result + + def _ensure_same_fingerprint(record: ModelInferenceJobRecord, fingerprint: str) -> None: if record.request_fingerprint and record.request_fingerprint != fingerprint: raise ValueError("This idempotency key was already used with a different request payload.") @@ -981,6 +1394,55 @@ def _display_title(operation: ModelJobOperation, payload: dict[str, Any]) -> str return value[:256] +def _review_to_row(review: ModelJobQualityReview) -> ModelJobQualityReviewModel: + return ModelJobQualityReviewModel(**review.model_dump()) + + +def _ensure_reviewable(record: ModelInferenceJobRecord) -> None: + if record.status != "awaiting_human" or record.review_status != "required": + raise ValueError("Only jobs in the required human-review queue can receive quality feedback.") + + +def _new_quality_review( + record: ModelInferenceJobRecord, + *, + reviewer_id: str, + request: ModelJobQualityReviewRequest, + created_at: datetime, +) -> ModelJobQualityReview: + return ModelJobQualityReview( + id=f"model-review-{uuid4().hex[:16]}", + tenant_id=record.tenant_id, + job_id=record.id, + review_trigger=record.review_trigger or "manual", + reviewer_id=reviewer_id, + legal_fact_consistency=request.legal_fact_consistency, + citation_verifiability=request.citation_verifiability, + high_risk_omission=request.high_risk_omission, + disposition=request.disposition, + feedback_text=(request.feedback_text or "").strip() or None, + quality_gate_version=record.quality_gate_version, + created_at=created_at, + ) + + +def _review_from_row(row: ModelJobQualityReviewModel) -> ModelJobQualityReview: + return ModelJobQualityReview( + id=row.id, + tenant_id=row.tenant_id, + job_id=row.job_id, + review_trigger=row.review_trigger, + reviewer_id=row.reviewer_id, + legal_fact_consistency=row.legal_fact_consistency, + citation_verifiability=row.citation_verifiability, + high_risk_omission=row.high_risk_omission, + disposition=row.disposition, + feedback_text=row.feedback_text, + quality_gate_version=row.quality_gate_version, + created_at=_aware(row.created_at), + ) + + def _history_key(record: ModelInferenceJobRecord) -> tuple[datetime, str]: return _aware(record.created_at), record.id diff --git a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/models.py b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/models.py index f473e6b2..c1b31ddf 100644 --- a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/models.py +++ b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/models.py @@ -357,6 +357,16 @@ class ModelInferenceJobModel(Base): quality_labels_json: Mapped[str] = mapped_column(Text, default="[]") quality_summary: Mapped[str | None] = mapped_column(Text, nullable=True) quality_gate_version: Mapped[str] = mapped_column(String(32), default="model-job-v1") + evaluation_run_id: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True) + evaluation_case_id: Mapped[str | None] = mapped_column(String(128), nullable=True) + evaluation_contract_type: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) + evaluation_risk_level: Mapped[str | None] = mapped_column(String(16), nullable=True, index=True) + evaluation_industry: Mapped[str | None] = mapped_column(String(64), nullable=True) + review_status: Mapped[str] = mapped_column(String(32), default="not_required", index=True) + review_trigger: Mapped[str | None] = mapped_column(String(32), nullable=True) + reviewed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + reviewed_by: Mapped[str | None] = mapped_column(String(128), nullable=True) + review_disposition: Mapped[str | None] = mapped_column(String(32), nullable=True) queue_latency_ms: Mapped[int] = mapped_column(Integer, default=0) execution_latency_ms: Mapped[int] = mapped_column(Integer, default=0) total_latency_ms: Mapped[int] = mapped_column(Integer, default=0) @@ -374,6 +384,31 @@ class ModelInferenceJobModel(Base): completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) +class ModelJobQualityReviewModel(Base): + __tablename__ = "model_job_quality_reviews" + __table_args__ = ( + Index("ix_model_job_quality_reviews_tenant_job_created", "tenant_id", "job_id", "created_at"), + Index("ix_model_job_quality_reviews_tenant_trigger_created", "tenant_id", "review_trigger", "created_at"), + ) + + id: Mapped[str] = mapped_column(String(64), primary_key=True) + tenant_id: Mapped[str] = mapped_column(String(64), default=DEFAULT_TENANT_ID, index=True) + job_id: Mapped[str] = mapped_column(String(64), index=True) + review_trigger: Mapped[str] = mapped_column(String(32)) + reviewer_id: Mapped[str] = mapped_column(String(128)) + legal_fact_consistency: Mapped[str] = mapped_column(String(32)) + citation_verifiability: Mapped[str] = mapped_column(String(32)) + high_risk_omission: Mapped[str] = mapped_column(String(32)) + disposition: Mapped[str] = mapped_column(String(32)) + feedback_text: Mapped[str | None] = mapped_column(Text, nullable=True) + quality_gate_version: Mapped[str] = mapped_column(String(32), default="model-job-v1") + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + default=lambda: datetime.now(UTC), + index=True, + ) + + class AiRunAuditModel(Base): __tablename__ = "ai_run_audits" diff --git a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/schema_check.py b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/schema_check.py index c0ae1585..37c49915 100644 --- a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/schema_check.py +++ b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/schema_check.py @@ -36,6 +36,7 @@ CORE_TABLES = ( "knowledge_reindex_executions", "knowledge_ingestion_jobs", "model_inference_jobs", + "model_job_quality_reviews", "ai_run_audits", "provider_acceptance_runs", "ai_quality_improvement_actions", @@ -47,6 +48,7 @@ TENANT_COLUMNS_BY_TABLE = { "knowledge_reindex_executions": ("tenant_id",), "knowledge_ingestion_jobs": ("tenant_id",), "model_inference_jobs": ("tenant_id",), + "model_job_quality_reviews": ("tenant_id",), "ai_run_audits": ("tenant_id",), "provider_acceptance_runs": ("tenant_id",), "ai_quality_improvement_actions": ("tenant_id",), @@ -89,6 +91,26 @@ MODEL_JOB_OPERATION_COLUMNS_BY_TABLE = { "quality_labels_json", "quality_summary", "quality_gate_version", + "evaluation_run_id", + "evaluation_case_id", + "evaluation_contract_type", + "evaluation_risk_level", + "evaluation_industry", + "review_status", + "review_trigger", + "reviewed_at", + "reviewed_by", + "review_disposition", + ), + "model_job_quality_reviews": ( + "job_id", + "review_trigger", + "reviewer_id", + "legal_fact_consistency", + "citation_verifiability", + "high_risk_omission", + "disposition", + "quality_gate_version", ), } PGVECTOR_REQUIRED_COLUMNS_BY_TABLE = { diff --git a/yuqei-ai-platform/services/ai-platform-api/tests/test_health.py b/yuqei-ai-platform/services/ai-platform-api/tests/test_health.py index a17619a0..ba7b5323 100644 --- a/yuqei-ai-platform/services/ai-platform-api/tests/test_health.py +++ b/yuqei-ai-platform/services/ai-platform-api/tests/test_health.py @@ -95,13 +95,13 @@ def test_runtime_schema_check_warns_for_local_sqlite_auto_created_schema(tmp_pat assert checks["alembic_version"]["status"] == "warning" assert checks["core_tables"]["status"] == "ok" assert checks["tenant_columns"]["status"] == "ok" - assert checks["tenant_columns"]["details"]["checked_tables"] == 12 + assert checks["tenant_columns"]["details"]["checked_tables"] == 13 assert checks["trace_governance_columns"]["status"] == "ok" assert checks["trace_governance_columns"]["details"]["checked_columns"] == 11 assert checks["secret_ref_columns"]["status"] == "ok" assert checks["secret_ref_columns"]["details"]["checked_columns"] == 2 assert checks["model_job_operation_columns"]["status"] == "ok" - assert checks["model_job_operation_columns"]["details"]["checked_columns"] == 12 + assert checks["model_job_operation_columns"]["details"]["checked_columns"] == 30 assert checks["pgvector_schema"]["status"] == "warning" diff --git a/yuqei-ai-platform/services/ai-platform-api/tests/test_model_jobs.py b/yuqei-ai-platform/services/ai-platform-api/tests/test_model_jobs.py index 1bf91272..bbb56d0b 100644 --- a/yuqei-ai-platform/services/ai-platform-api/tests/test_model_jobs.py +++ b/yuqei-ai-platform/services/ai-platform-api/tests/test_model_jobs.py @@ -1,12 +1,21 @@ -from fastapi.testclient import TestClient +from concurrent.futures import ThreadPoolExecutor -from yuqei_ai_platform_api.api.model_jobs import execute_model_job +from fastapi.testclient import TestClient +from sqlalchemy import create_engine +from sqlalchemy.pool import StaticPool + +from yuqei_ai_platform_api.api.model_jobs import _automatic_review_trigger, execute_model_job from yuqei_ai_platform_api.config import AiPlatformSettings from yuqei_ai_platform_api.main import create_app +from yuqei_ai_platform_api.model_jobs import ( + ModelInferenceJobStore, + ModelJobAdmissionRejected, + ModelJobQualityReviewRequest, +) from yuqei_ai_platform_api.repository import InMemoryAiPlatformRepository -def _app_and_client(): +def _app_and_client(**setting_overrides): repository = InMemoryAiPlatformRepository() repository.seed_defaults() app = create_app( @@ -16,12 +25,245 @@ def _app_and_client(): provider_call_mode="disabled", provider_fallback_enabled=True, model_job_dispatch_mode="worker", + **setting_overrides, ), repository=repository, ) return app, repository, TestClient(app) +def test_controlled_pilot_admission_limits_are_idempotency_safe() -> None: + _app, _repository, client = _app_and_client( + model_job_tenant_active_limit=3, + model_job_user_active_limit=2, + model_job_tenant_queue_limit=3, + ) + headers = {"X-User-Id": "pilot-user-a", "Idempotency-Key": "pilot-a-1"} + first = client.post("/api/v1/legal/qa/jobs", headers=headers, json={"question": "问题一如何处理?"}) + assert first.status_code == 202 + repeated = client.post("/api/v1/legal/qa/jobs", headers=headers, json={"question": "问题一如何处理?"}) + assert repeated.status_code == 202 + assert repeated.json()["id"] == first.json()["id"] + assert client.post( + "/api/v1/legal/qa/jobs", + headers={"X-User-Id": "pilot-user-a", "Idempotency-Key": "pilot-a-2"}, + json={"question": "问题二如何处理?"}, + ).status_code == 202 + user_limited = client.post( + "/api/v1/legal/qa/jobs", + headers={"X-User-Id": "pilot-user-a", "Idempotency-Key": "pilot-a-3"}, + json={"question": "问题三如何处理?"}, + ) + assert user_limited.status_code == 429 + assert user_limited.headers["Retry-After"] == "10" + assert user_limited.json()["detail"]["code"] == "user_active_limit_reached" + assert client.post( + "/api/v1/legal/qa/jobs", + headers={"X-User-Id": "pilot-user-b", "Idempotency-Key": "pilot-b-1"}, + json={"question": "问题四如何处理?"}, + ).status_code == 202 + tenant_limited = client.post( + "/api/v1/legal/qa/jobs", + headers={"X-User-Id": "pilot-user-c", "Idempotency-Key": "pilot-c-1"}, + json={"question": "问题五如何处理?"}, + ) + assert tenant_limited.status_code == 429 + assert tenant_limited.json()["detail"]["code"] == "tenant_active_limit_reached" + + +def test_concurrent_admission_never_exceeds_single_process_tenant_limit() -> None: + app, _repository, _client = _app_and_client() + store = app.state.model_job_store + + def submit(index: int) -> str: + try: + job, _created = store.create_or_get( + tenant_id="capacity-race-test", + operation="legal_qa", + request_payload={"question": f"并发容量问题 {index}"}, + created_by=f"capacity-user-{index}", + idempotency_key=f"capacity-race-{index}", + tenant_active_limit=10, + user_active_limit=3, + tenant_queue_limit=25, + ) + return job.id + except ModelJobAdmissionRejected: + return "rejected" + + with ThreadPoolExecutor(max_workers=20) as executor: + results = list(executor.map(submit, range(20))) + + assert sum(item != "rejected" for item in results) == 10 + assert sum(item == "rejected" for item in results) == 10 + + +def test_stratified_quality_review_queue_and_feedback_dataset() -> None: + app, _repository, client = _app_and_client() + created = client.post( + "/api/v1/legal/qa/jobs", + headers={ + "X-User-Id": "pilot-runner-01", + "Idempotency-Key": "stratified-purchase-high-01", + "X-Evaluation-Run-Id": "stratified-v1", + "X-Evaluation-Case-Id": "purchase-high-01", + "X-Evaluation-Contract-Type": "purchase", + "X-Evaluation-Risk-Level": "high", + "X-Evaluation-Industry": "manufacturing", + }, + json={"question": "高额预付款缺少履约保障时如何控制风险?"}, + ).json() + store = app.state.model_job_store + store.claim(created["id"], tenant_id="default", worker_name="pilot-worker") + stored = store.finish( + created["id"], + tenant_id="default", + result={"answer": "可审阅回答", "citations": [{"title": "示例依据"}]}, + status="awaiting_human", + provider_status="succeeded", + audit_run_id="audit-stratified-01", + quality_status="passed", + quality_score=100, + quality_summary="自动质量门禁通过,高风险样本需人工复核。", + review_trigger="high_risk", + ) + assert stored.review_status == "required" + queue = client.get("/api/v1/model-jobs/review-queue").json() + assert [item["id"] for item in queue] == [created["id"]] + assert client.post( + "/api/v1/legal/qa/jobs", + headers={"Idempotency-Key": "outside-stratified-run"}, + json={"question": "不应计入该评测运行的任务是什么?"}, + ).status_code == 202 + snapshot = client.get( + "/api/v1/model-jobs/operations?target_sample_count=1&evaluation_run_id=stratified-v1" + ).json() + assert snapshot["baseline_ready"] is True + assert snapshot["metrics"]["total"] == 1 + assert snapshot["admission"]["queued"] == 1 + assert snapshot["review_required"] == 1 + assert snapshot["quality_segments"] == [ + { + "dimension": "contract_type", + "value": "purchase", + "sample_count": 1, + "quality_passed": 1, + "quality_needs_review": 0, + "quality_failed": 0, + "review_required": 1, + "review_completed": 0, + "quality_pass_rate": 1.0, + }, + { + "dimension": "risk_level", + "value": "high", + "sample_count": 1, + "quality_passed": 1, + "quality_needs_review": 0, + "quality_failed": 0, + "review_required": 1, + "review_completed": 0, + "quality_pass_rate": 1.0, + }, + { + "dimension": "industry", + "value": "manufacturing", + "sample_count": 1, + "quality_passed": 1, + "quality_needs_review": 0, + "quality_failed": 0, + "review_required": 1, + "review_completed": 0, + "quality_pass_rate": 1.0, + }, + ] + review = client.post( + f"/api/v1/model-jobs/{created['id']}/quality-reviews", + headers={"X-User-Id": "legal-reviewer-a"}, + json={ + "legal_fact_consistency": "pass", + "citation_verifiability": "uncertain", + "high_risk_omission": "not_detected", + "disposition": "accepted", + "feedback_text": "事实一致,引用仍需在正式法源库复核。", + }, + ) + assert review.status_code == 200 + assert review.json()["reviewer_id"] == "legal-reviewer-a" + assert client.get(f"/api/v1/model-jobs/{created['id']}").json()["status"] == "resolved" + assert len(client.get(f"/api/v1/model-jobs/{created['id']}/quality-reviews").json()) == 1 + + +def test_automatic_review_policy_covers_quality_high_risk_and_sampling() -> None: + app, _repository, client = _app_and_client( + model_job_manual_sample_rate_percent=100, + model_job_high_risk_review_required=True, + ) + created = client.post( + "/api/v1/legal/qa/jobs", + headers={ + "X-Evaluation-Run-Id": "policy-test-v1", + "X-Evaluation-Case-Id": "case-001", + "X-Evaluation-Risk-Level": "high", + }, + json={"question": "高风险条款如何审查?"}, + ).json() + record = app.state.model_job_store.get_record(created["id"], tenant_id="default") + assert record is not None + assert _automatic_review_trigger(record, "failed", app.state.ai_platform_settings) == "quality_gate" + assert _automatic_review_trigger(record, "passed", app.state.ai_platform_settings) == "high_risk" + sampled = record.model_copy(update={"evaluation_risk_level": "medium"}) + assert _automatic_review_trigger(sampled, "passed", app.state.ai_platform_settings) == "random_sample" + settings = app.state.ai_platform_settings.model_copy( + update={"model_job_manual_sample_rate_percent": 0} + ) + assert _automatic_review_trigger(sampled, "passed", settings) is None + + +def test_sqlalchemy_quality_review_is_persisted_with_resolved_job() -> None: + engine = create_engine( + "sqlite+pysqlite:///:memory:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + store = ModelInferenceJobStore(engine, auto_create=True) + job = store.create( + tenant_id="review-db-test", + operation="legal_qa", + request_payload={"question": "数据库复核事务如何验证?"}, + ) + store.claim(job.id, tenant_id="review-db-test", worker_name="review-db-worker") + store.finish( + job.id, + tenant_id="review-db-test", + result={"answer": "待具名人员复核", "citations": []}, + status="awaiting_human", + provider_status="succeeded", + audit_run_id="review-db-audit", + quality_status="passed", + quality_score=100, + quality_summary="高风险抽检。", + review_trigger="high_risk", + ) + resolved, event = store.record_quality_review( + job.id, + tenant_id="review-db-test", + reviewer_id="reviewer-db-test", + request=ModelJobQualityReviewRequest( + legal_fact_consistency="uncertain", + citation_verifiability="uncertain", + high_risk_omission="uncertain", + disposition="rejected", + feedback_text="测试数据库事务,不代表法务结论。", + ), + ) + + assert resolved.status == "resolved" + assert resolved.reviewed_by == "reviewer-db-test" + assert store.quality_reviews(job.id, tenant_id="review-db-test") == [event] + engine.dispose() + + def test_model_job_is_queued_polled_and_completed_with_explicit_fallback() -> None: app, repository, client = _app_and_client() diff --git a/yuqei-ai-sdk-python/src/yuqei_sdk/__init__.py b/yuqei-ai-sdk-python/src/yuqei_sdk/__init__.py index cdf5d55a..71e16c13 100644 --- a/yuqei-ai-sdk-python/src/yuqei_sdk/__init__.py +++ b/yuqei-ai-sdk-python/src/yuqei_sdk/__init__.py @@ -41,8 +41,12 @@ from yuqei_sdk.ai_platform import ( ModelInferenceJob, ModelInferenceJobHistoryPage, ModelInferenceJobMetrics, + ModelJobAdmissionSnapshot, ModelJobOperationalAlert, ModelJobOperationsSnapshot, + ModelJobQualityReview, + ModelJobQualityReviewRequest, + ModelJobQualitySegment, ModelOutputQualityLabel, PromptTemplate, PromptTemplateChange, @@ -104,8 +108,12 @@ __all__ = [ "ModelInferenceJob", "ModelInferenceJobHistoryPage", "ModelInferenceJobMetrics", + "ModelJobAdmissionSnapshot", "ModelJobOperationalAlert", "ModelJobOperationsSnapshot", + "ModelJobQualityReview", + "ModelJobQualityReviewRequest", + "ModelJobQualitySegment", "ModelOutputQualityLabel", "PromptTemplate", "PromptTemplateChange", diff --git a/yuqei-ai-sdk-python/src/yuqei_sdk/ai_platform.py b/yuqei-ai-sdk-python/src/yuqei_sdk/ai_platform.py index 4a98f5dd..04f1a521 100644 --- a/yuqei-ai-sdk-python/src/yuqei_sdk/ai_platform.py +++ b/yuqei-ai-sdk-python/src/yuqei_sdk/ai_platform.py @@ -1,5 +1,5 @@ from datetime import datetime -from typing import Any +from typing import Any, Literal import httpx from pydantic import BaseModel, Field @@ -77,6 +77,16 @@ class ModelInferenceJob(BaseModel): quality_labels: list[ModelOutputQualityLabel] = Field(default_factory=list) quality_summary: str | None = None quality_gate_version: str = "model-job-v1" + evaluation_run_id: str | None = None + evaluation_case_id: str | None = None + evaluation_contract_type: str | None = None + evaluation_risk_level: Literal["low", "medium", "high"] | None = None + evaluation_industry: str | None = None + review_status: Literal["not_required", "required", "completed"] = "not_required" + review_trigger: Literal["quality_gate", "high_risk", "random_sample", "manual"] | None = None + reviewed_at: datetime | None = None + reviewed_by: str | None = None + review_disposition: Literal["accepted", "corrected", "rejected"] | None = None queue_latency_ms: int = 0 execution_latency_ms: int = 0 total_latency_ms: int = 0 @@ -125,6 +135,47 @@ class ModelJobOperationalAlert(BaseModel): threshold: float +class ModelJobAdmissionSnapshot(BaseModel): + running: int + queued: int + tenant_active_limit: int + user_active_limit: int + tenant_queue_limit: int + active_saturation_rate: float + queue_saturation_rate: float + accepting_submissions: bool + + +class ModelJobQualitySegment(BaseModel): + dimension: Literal["contract_type", "risk_level", "industry"] + value: str + sample_count: int + quality_passed: int + quality_needs_review: int + quality_failed: int + review_required: int + review_completed: int + quality_pass_rate: float + + +class ModelJobQualityReviewRequest(BaseModel): + legal_fact_consistency: Literal["pass", "fail", "uncertain", "not_applicable"] + citation_verifiability: Literal["pass", "fail", "uncertain", "not_applicable"] + high_risk_omission: Literal["not_detected", "detected", "uncertain", "not_applicable"] + disposition: Literal["accepted", "corrected", "rejected"] + feedback_text: str | None = None + + +class ModelJobQualityReview(ModelJobQualityReviewRequest): + id: str + tenant_id: str + job_id: str + review_trigger: Literal["quality_gate", "high_risk", "random_sample", "manual"] + reviewer_id: str + quality_gate_version: str + created_at: datetime + + class ModelJobOperationsSnapshot(BaseModel): metrics: ModelInferenceJobMetrics alerts: list[ModelJobOperationalAlert] = Field(default_factory=list) @@ -136,6 +187,11 @@ class ModelJobOperationsSnapshot(BaseModel): quality_needs_review: int quality_failed: int quality_pass_rate: float + review_required: int = 0 + review_completed: int = 0 + evaluation_run_id: str | None = None + admission: ModelJobAdmissionSnapshot + quality_segments: list[ModelJobQualitySegment] = Field(default_factory=list) class ContractReviewRequest(BaseModel): @@ -790,11 +846,24 @@ class AiPlatformClient: request: LegalQaRequest, *, idempotency_key: str | None = None, + evaluation_run_id: str | None = None, + evaluation_case_id: str | None = None, + evaluation_contract_type: str | None = None, + evaluation_risk_level: str | None = None, + evaluation_industry: str | None = None, context: RequestContext | None = None, ) -> ModelInferenceJob: headers = (context or RequestContext()).to_headers() if idempotency_key: headers["Idempotency-Key"] = idempotency_key + evaluation_headers = { + "X-Evaluation-Run-Id": evaluation_run_id, + "X-Evaluation-Case-Id": evaluation_case_id, + "X-Evaluation-Contract-Type": evaluation_contract_type, + "X-Evaluation-Risk-Level": evaluation_risk_level, + "X-Evaluation-Industry": evaluation_industry, + } + headers.update({key: value for key, value in evaluation_headers.items() if value}) response = self._client.post( f"{self._api_prefix}/legal/qa/jobs", json=request.model_dump(exclude_none=True), @@ -880,16 +949,63 @@ class AiPlatformClient: *, window_hours: int = 24 * 7, target_sample_count: int = 30, + evaluation_run_id: str | None = None, context: RequestContext | None = None, ) -> ModelJobOperationsSnapshot: response = self._client.get( f"{self._api_prefix}/model-jobs/operations", - params={"window_hours": window_hours, "target_sample_count": target_sample_count}, + params={ + "window_hours": window_hours, + "target_sample_count": target_sample_count, + **({"evaluation_run_id": evaluation_run_id} if evaluation_run_id else {}), + }, headers=(context or RequestContext()).to_headers(), ) response.raise_for_status() return ModelJobOperationsSnapshot.model_validate(response.json()) + def list_model_job_review_queue( + self, + *, + limit: int = 50, + context: RequestContext | None = None, + ) -> list[ModelInferenceJob]: + response = self._client.get( + f"{self._api_prefix}/model-jobs/review-queue", + params={"limit": limit}, + headers=(context or RequestContext()).to_headers(), + ) + response.raise_for_status() + return [ModelInferenceJob.model_validate(item) for item in response.json()] + + def create_model_job_quality_review( + self, + job_id: str, + request: ModelJobQualityReviewRequest, + *, + context: RequestContext | None = None, + ) -> ModelJobQualityReview: + response = self._client.post( + f"{self._api_prefix}/model-jobs/{job_id}/quality-reviews", + json=request.model_dump(exclude_none=True), + headers=(context or RequestContext()).to_headers(), + ) + response.raise_for_status() + return ModelJobQualityReview.model_validate(response.json()) + + def list_model_job_quality_reviews( + self, + job_id: str, + *, + context: RequestContext | None = None, + ) -> list[ModelJobQualityReview]: + response = self._client.get( + f"{self._api_prefix}/model-jobs/{job_id}/quality-reviews", + headers=(context or RequestContext()).to_headers(), + ) + response.raise_for_status() + return [ModelJobQualityReview.model_validate(item) for item in response.json()] + def handoff_model_job( self, job_id: str, diff --git a/yuqei-ai-sdk-python/tests/test_ai_platform.py b/yuqei-ai-sdk-python/tests/test_ai_platform.py index 313df3df..b30e1bcd 100644 --- a/yuqei-ai-sdk-python/tests/test_ai_platform.py +++ b/yuqei-ai-sdk-python/tests/test_ai_platform.py @@ -15,6 +15,7 @@ from yuqei_sdk import ( KnowledgeImportBatchCreate, KnowledgeReindexRequest, LegalQaRequest, + ModelJobQualityReviewRequest, PromptTemplateCreate, PromptTemplatePublishRequest, PromptTemplateRollbackRequest, @@ -143,6 +144,19 @@ def test_ai_platform_client_supports_model_job_operations() -> None: return httpx.Response(202, json=job_payload()) if request.url.path.endswith("/model-jobs/history"): return httpx.Response(200, json={"items": [job_payload()], "next_cursor": None, "has_more": False}) + if request.url.path.endswith("/model-jobs/review-queue"): + return httpx.Response(200, json=[job_payload("awaiting_human")]) + if request.url.path.endswith("/quality-reviews") and request.method == "POST": + return httpx.Response(200, json={ + "id": "model-review-1", "tenant_id": "tenant-1", "job_id": "model-job-1", + "review_trigger": "high_risk", "reviewer_id": "user-1", + "legal_fact_consistency": "pass", "citation_verifiability": "uncertain", + "high_risk_omission": "not_detected", "disposition": "accepted", + "feedback_text": "引用待正式法源库复核。", "quality_gate_version": "model-job-v1", + "created_at": "2026-08-26T08:10:00Z", + }) + if request.url.path.endswith("/quality-reviews") and request.method == "GET": + return httpx.Response(200, json=[]) if request.url.path.endswith("/cancel"): return httpx.Response(200, json=job_payload("cancelled")) if request.url.path.endswith("/model-jobs/operations"): @@ -159,6 +173,13 @@ def test_ai_platform_client_supports_model_job_operations() -> None: "remaining_sample_count": 30, "baseline_ready": False, "quality_passed": 0, "quality_needs_review": 0, "quality_failed": 0, "quality_pass_rate": 0, + "review_required": 0, "review_completed": 0, "evaluation_run_id": "stratified-v1", + "admission": { + "running": 0, "queued": 1, "tenant_active_limit": 10, "user_active_limit": 3, + "tenant_queue_limit": 25, "active_saturation_rate": 0.1, + "queue_saturation_rate": 0.04, "accepting_submissions": True, + }, + "quality_segments": [], }) return httpx.Response(404) @@ -167,17 +188,37 @@ def test_ai_platform_client_supports_model_job_operations() -> None: created = client.create_legal_qa_job( LegalQaRequest(question="付款期限如何约定?"), idempotency_key="submit-1", + evaluation_run_id="stratified-v1", + evaluation_case_id="purchase-01", + evaluation_contract_type="purchase", + evaluation_risk_level="high", + evaluation_industry="manufacturing", context=context, ) history = client.list_model_jobs(operation="legal_qa", context=context) cancelled = client.cancel_model_job(created.id, reason="不再需要", context=context) - operations = client.get_model_job_operations(context=context) + operations = client.get_model_job_operations(evaluation_run_id="stratified-v1", context=context) + review_queue = client.list_model_job_review_queue(context=context) + review = client.create_model_job_quality_review( + created.id, + ModelJobQualityReviewRequest( + legal_fact_consistency="pass", + citation_verifiability="uncertain", + high_risk_omission="not_detected", + disposition="accepted", + feedback_text="引用待正式法源库复核。", + ), + context=context, + ) assert created.id == "model-job-1" assert history.items[0].display_title == "付款期限如何约定?" assert cancelled.status == "cancelled" assert operations.remaining_sample_count == 30 + assert review_queue[0].status == "awaiting_human" + assert review.reviewer_id == "user-1" assert seen_requests[0].headers["Idempotency-Key"] == "submit-1" + assert seen_requests[0].headers["X-Evaluation-Contract-Type"] == "purchase" history_query = parse_qs(seen_requests[1].url.query.decode()) assert history_query["operation"] == ["legal_qa"] assert "status" not in history_query diff --git a/yuqei-contract-platform/apps/contract-web/src/app/admin/model-jobs/page.tsx b/yuqei-contract-platform/apps/contract-web/src/app/admin/model-jobs/page.tsx index 53b67a79..654236ad 100644 --- a/yuqei-contract-platform/apps/contract-web/src/app/admin/model-jobs/page.tsx +++ b/yuqei-contract-platform/apps/contract-web/src/app/admin/model-jobs/page.tsx @@ -2,11 +2,16 @@ import { Activity, AlertTriangle, CheckCircle2, Clock3, RefreshCw, ShieldAlert } from "lucide-react"; import { useCallback, useEffect, useState } from "react"; -import type { ModelJobOperationsSnapshot } from "@/services/contracts"; -import { getModelJobOperations } from "@/services/contracts-client"; +import type { ModelInferenceJob, ModelJobOperationsSnapshot, ModelJobQualityReviewRequest } from "@/services/contracts"; +import { getModelJobOperations, getModelJobReviewQueue, submitModelJobQualityReview } from "@/services/contracts-client"; export default function ModelJobOperationsPage() { const [snapshot, setSnapshot] = useState(null); + const [reviewQueue, setReviewQueue] = useState([]); + const [runIdInput, setRunIdInput] = useState(""); + const [evaluationRunId, setEvaluationRunId] = useState(""); + const [tenantIdInput, setTenantIdInput] = useState(""); + const [targetTenantId, setTargetTenantId] = useState(""); const [loading, setLoading] = useState(true); const [error, setError] = useState(null); @@ -14,13 +19,18 @@ export default function ModelJobOperationsPage() { setLoading(true); setError(null); try { - setSnapshot(await getModelJobOperations()); + const [nextSnapshot, nextReviewQueue] = await Promise.all([ + getModelJobOperations(evaluationRunId, targetTenantId), + getModelJobReviewQueue(50, targetTenantId), + ]); + setSnapshot(nextSnapshot); + setReviewQueue(nextReviewQueue); } catch (requestError) { setError(requestError instanceof Error ? requestError.message : "模型任务运营指标加载失败。"); } finally { setLoading(false); } - }, []); + }, [evaluationRunId, targetTenantId]); useEffect(() => { void loadSnapshot(); }, [loadSnapshot]); @@ -41,17 +51,25 @@ export default function ModelJobOperationsPage() { {snapshot && ( <> +
+

试点租户与评测运行筛选

平台管理员可下钻到指定试点租户和运行;留空查看当前租户窗口全量。

+
+
setTenantIdInput(event.target.value)} placeholder="试点租户 ID,例如 controlled-pilot-v1" /> setRunIdInput(event.target.value)} placeholder="运行 ID,例如 stratified-v1-20260826" />{(evaluationRunId || targetTenantId) && }
+ {(targetTenantId || snapshot.evaluation_run_id) &&

当前范围:租户 {targetTenantId || "当前租户"} · 运行 {snapshot.evaluation_run_id || "全部"}

} +
+
+
- +
-

30 条真实样本基线

只计入已运行质量门禁的完成任务。

+

分层真实样本基线

只计入真实 Provider 且已运行质量门禁的完成任务。

{snapshot.baseline_ready ? "基线已就绪" : `还差 ${snapshot.remaining_sample_count} 条`} @@ -76,12 +94,39 @@ export default function ModelJobOperationsPage() { 执行 P50 / P95{formatDuration(snapshot.metrics.execution_latency_p50_ms)} / {formatDuration(snapshot.metrics.execution_latency_p95_ms)} fallback / 失败率{formatRate(snapshot.metrics.fallback_rate)} / {formatRate(snapshot.metrics.failure_rate)} 重试率{formatRate(snapshot.metrics.retry_rate)} + 抽检要求 / 已完成{snapshot.review_required} / {snapshot.review_completed}
+
+
+

受控试点容量

{snapshot.admission.accepting_submissions ? "可接收" : "已限流"}
+
+ + + + + + +
运行 / 排队{snapshot.admission.running} / {snapshot.admission.queued}
租户活动上限{snapshot.admission.tenant_active_limit}
单用户活动上限{snapshot.admission.user_active_limit}
队列上限{snapshot.admission.tenant_queue_limit}
活动 / 队列饱和度{formatRate(snapshot.admission.active_saturation_rate)} / {formatRate(snapshot.admission.queue_saturation_rate)}
+
+
+ +
+

分层质量结果

{snapshot.quality_segments.length} 个分层
+
+ {snapshot.quality_segments.length === 0 ?
当前窗口尚无带分层标签的样本。
: ( + + {snapshot.quality_segments.map((segment) => )} +
维度分层样本通过率抽检
{dimensionLabel(segment.dimension)}{segment.value}{segment.sample_count}{formatRate(segment.quality_pass_rate)}{segment.review_completed}/{segment.review_required}
+ )} +
+
+
+

运营告警

{snapshot.alerts.length} 条
@@ -102,12 +147,91 @@ export default function ModelJobOperationsPage() { )}
+ +
+

人工抽检与接管队列

记录事实一致性、引用可核验性和高风险遗漏,形成规则迭代数据。

{reviewQueue.length} 条
+
+ {reviewQueue.length === 0 ?
当前没有待人工抽检任务。
: ( +
+ {reviewQueue.map((job) => )} +
+ )} +
+
)} ); } +function QualityReviewCard({ job, onReviewed }: { job: ModelInferenceJob; onReviewed: () => Promise }) { + const result = job.result && typeof job.result === "object" ? job.result as Record : null; + const answer = typeof result?.answer === "string" ? result.answer : typeof result?.summary === "string" ? result.summary : ""; + const citations = Array.isArray(result?.citations) ? result.citations.filter((item): item is Record => Boolean(item) && typeof item === "object") : []; + const [payload, setPayload] = useState({ + legal_fact_consistency: "uncertain", + citation_verifiability: "uncertain", + high_risk_omission: "uncertain", + disposition: "accepted", + feedback_text: "", + }); + const [submitting, setSubmitting] = useState(false); + const [confirmed, setConfirmed] = useState(false); + const [error, setError] = useState(null); + + async function submit() { + setSubmitting(true); + setError(null); + try { + await submitModelJobQualityReview(job.id, payload, job.tenant_id); + await onReviewed(); + } catch (requestError) { + setError(requestError instanceof Error ? requestError.message : "抽检结论提交失败。"); + } finally { + setSubmitting(false); + } + } + + return
+
+ {job.display_title} +

{job.evaluation_contract_type ?? "未分层"} · 风险 {job.evaluation_risk_level ?? "未知"} · 触发 {reviewTriggerLabel(job.review_trigger)}

+
查看模型输出与引用({citations.length} 条)
{answer || "当前任务没有可审阅的文本输出。"}{citations.length > 0 && <>

引用

    {citations.map((citation, index) =>
  1. {citationLabel(citation)}
  2. )}
}
+ {job.quality_summary &&

自动门禁:{job.quality_summary}

} +
+ setPayload((current) => ({ ...current, legal_fact_consistency: value }))} /> + setPayload((current) => ({ ...current, citation_verifiability: value }))} /> + setPayload((current) => ({ ...current, high_risk_omission: value }))} /> + setPayload((current) => ({ ...current, disposition: value }))} /> +
+