diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 00000000..d5b4bd85 --- /dev/null +++ b/.gitattributes @@ -0,0 +1,2 @@ +project-docs/legal-policy-candidates/**/*.md text eol=lf +project-docs/legal-policy-candidates/**/*.json text eol=lf diff --git a/project-docs/AI合同平台-模型任务运营闭环与质量门禁实施报告-2026-08-26.md b/project-docs/AI合同平台-模型任务运营闭环与质量门禁实施报告-2026-08-26.md new file mode 100644 index 00000000..cfba7eb0 --- /dev/null +++ b/project-docs/AI合同平台-模型任务运营闭环与质量门禁实施报告-2026-08-26.md @@ -0,0 +1,94 @@ +# AI 合同平台:模型任务运营闭环与质量门禁实施报告 + +日期:2026-08-26 + +阶段:核心产品上线收口 / 模型任务运营闭环与质量门禁 + +模型:`gpt-5.6-sol`(开发与测试阶段) + +## 1. 阶段结论 + +本阶段已经完成模型任务历史、幂等提交、用户取消、运营告警、输出质量自动校验和 30 个真实模型样本基线。30 个样本均由真实 Provider 成功返回,无 fallback,自动质量门禁全部通过。 + +模型自身执行耗时 P95 为 157.162 秒,低于 3 分钟目标;但 30 个任务集中突发提交时,队列等待 P95 达到 1,487.088 秒。这说明当前模型质量和单任务执行时间达到受控试点门槛,但并发容量尚不足以支撑大规模突发流量。系统已将执行耗时与排队耗时拆分监控,并对排队异常产生运营告警,不能把本轮结果解释为“端到端 3 分钟内完成”。 + +## 2. 已交付能力 + +### 2.1 用户侧任务闭环 + +- 法律问答任务历史页,支持游标分页和仅查看本人任务。 +- `Idempotency-Key` 提交幂等;同一用户、同一操作和同一请求可安全重放。 +- 相同幂等键但请求内容或用户不同会返回冲突,避免错误复用结果。 +- 排队中和执行中的任务均可取消;取消记录保留,不删除历史。 +- 取消与模型完成发生竞争时,以已经确认的取消状态为准,迟到结果不会覆盖取消结果。 +- 历史项展示 Provider 状态、质量状态、执行进度和失败/接管信息。 + +### 2.2 自动质量门禁 + +质量校验采用版本化的确定性规则 `model-job-v1`,覆盖: + +- 空输出和过短输出; +- 未解析模板占位符; +- 用户问题丢失; +- 提示词泄露迹象; +- 引用缺失或引用结构不完整; +- 合同审查风险项缺失或为空。 + +质量结论分为 `passed`、`needs_review`、`failed`,并保存分数、标签、摘要和规则版本。只有真实 Provider 成功且已经完成质量评估的结果才计入正式样本基线;fallback 不计入。 + +### 2.3 运营后台与告警 + +- 新增模型任务运营页,展示任务状态、真实样本完成度、质量通过率、fallback/失败/重试/人工接管比例及队列、执行、总耗时分位值。 +- 覆盖样本不足、排队 P95 超标、执行 P95 超标、fallback 比例过高、失败比例过高、人工接管积压和长时间未完成任务等告警。 +- 取消任务从失败率分母中排除,避免用户主动取消污染可靠性口径。 +- PostgreSQL Worker 使用短连接模式,降低多个 Worker 持续占用数据库连接造成的资源争用。 + +## 3. 真实样本质量与耗时基线 + +基线运行 ID:`fixed-v1-20260826` + +| 指标 | 结果 | 门槛/解释 | +| --- | ---: | --- | +| 提交样本 | 30 | 固定中文合同/法律问答样本 | +| 真实 Provider 合格样本 | 30 | 不含 fallback | +| 自动质量通过 | 30 | 100% | +| 需要人工复核 | 0 | 无阻断项 | +| 自动质量失败 | 0 | 无阻断项 | +| fallback | 0 | 不参与合格样本计数 | +| 执行耗时 P50 | 144.806 秒 | 模型实际处理时间 | +| 执行耗时 P95 | 157.162 秒 | 低于 180 秒,门禁通过 | +| 排队耗时 P50 | 589.677 秒 | 突发容量信号 | +| 排队耗时 P95 | 1,487.088 秒 | 超过 30 秒,运营告警 | +| 端到端耗时 P50 | 739.257 秒 | 排队 + 执行 | +| 端到端耗时 P95 | 1,636.160 秒 | 不能对外承诺 3 分钟 | + +正式基线证据保存在本机 `.runtime/model-job-quality-baseline/fixed-v1-20260826-20260826T062134Z.json` 及同名 Markdown 文件中。该目录不进入 Git,避免把运行数据当成产品源代码;本报告保存验收结论与关键指标。 + +## 4. 浏览器 UAT 结果 + +已在登录态浏览器完成用户侧真实交互验证: + +1. 历史接口最初因前端 SDK 把空参数编码为 `status=&cursor=` 而返回 422;已修复为不发送空查询参数。 +2. 提交一条不含真实客户数据的测试法律问答,页面正确显示排队状态和可取消操作。 +3. 用户点击取消后,页面显示“任务已由用户取消”,任务状态保留为“已取消”,历史记录仍可查询。 +4. 当前浏览器账号为 UAT 法务角色,访问平台管理员运营页被正确拒绝;管理员接口另以平台管理员身份完成授权和数据返回验证。 + +## 5. 发布判断 + +本阶段代码可进入受控预发布验证,但仍不建议直接开放大规模真实用户注册: + +- 质量样本数量已达到首个 30 条门槛,但样本规模仍小,后续应按合同类型、行业和风险等级持续扩充。 +- 单任务执行时间达标,突发排队容量未达标。试点期应限制并发并观察队列告警,随后再评估 Worker 横向扩容或任务调度隔离。 +- 自动质量规则用于发现明显结构问题,不等同于律师意见或法律正确性证明;高风险输出仍需人工复核。 +- 本地 PostgreSQL 缺少 `pgvector` 时仅作为开发环境警告;预发布和生产环境仍保持失败关闭。 +- 本阶段不宣称第三方存证、司法效力或双云灾备能力。 + +## 6. 下一阶段建议 + +建议进入“受控试点容量与分层质量评测”: + +1. 按试点并发量执行 1、3、5、10 并发阶梯压测,建立队列容量曲线并配置限流。 +2. 将样本扩充到至少 100 条,按采购、销售、劳动、租赁、保密等合同类型分层统计。 +3. 增加法律事实一致性、引用可核验性和高风险漏检的人工抽检口径。 +4. 对质量失败自动进入人工接管队列,并回写处理结论,形成规则迭代数据集。 +5. 在真实用户开放前完成正式模型 API 账户、费用预算、隐私与合规材料验收。 diff --git a/yuqei-ai-platform/packages/ai-openapi/openapi.yaml b/yuqei-ai-platform/packages/ai-openapi/openapi.yaml index 6f2f0d64..a8f605e7 100644 --- a/yuqei-ai-platform/packages/ai-openapi/openapi.yaml +++ b/yuqei-ai-platform/packages/ai-openapi/openapi.yaml @@ -180,6 +180,13 @@ paths: post: operationId: createLegalQaJob summary: Queue a durable legal QA model job + parameters: + - $ref: "#/components/parameters/TenantIdHeader" + - $ref: "#/components/parameters/UserIdHeader" + - name: Idempotency-Key + in: header + required: false + schema: { type: string, maxLength: 128 } requestBody: required: true content: @@ -218,6 +225,13 @@ paths: post: operationId: createContractReviewJob summary: Queue a durable contract review model job + parameters: + - $ref: "#/components/parameters/TenantIdHeader" + - $ref: "#/components/parameters/UserIdHeader" + - name: Idempotency-Key + in: header + required: false + schema: { type: string, maxLength: 128 } requestBody: required: true content: @@ -231,6 +245,51 @@ paths: application/json: schema: $ref: "#/components/schemas/ModelInferenceJob" + /api/v1/model-jobs/history: + get: + operationId: listModelInferenceJobHistory + summary: List tenant- and creator-scoped model job history with cursor pagination + parameters: + - name: operation + in: query + schema: { type: string, enum: [legal_qa, contract_review] } + - name: status + in: query + schema: { type: string, enum: [queued, running, succeeded, fallback, failed, awaiting_human, resolved, cancelled] } + - name: cursor + in: query + schema: { type: string, maxLength: 512 } + - name: limit + in: query + schema: { type: integer, minimum: 1, maximum: 100, default: 20 } + - name: mine + in: query + schema: { type: boolean, default: true } + responses: + "200": + description: Cursor-paginated model job history + content: + application/json: + schema: + $ref: "#/components/schemas/ModelInferenceJobHistoryPage" + /api/v1/model-jobs/operations: + get: + operationId: getModelJobOperations + summary: Get task SLO alerts and strict real-provider quality baseline progress + parameters: + - name: window_hours + in: query + schema: { type: integer, minimum: 1, maximum: 2160, default: 168 } + - name: target_sample_count + in: query + schema: { type: integer, minimum: 1, maximum: 1000, default: 30 } + responses: + "200": + description: Model job operational snapshot + content: + application/json: + schema: + $ref: "#/components/schemas/ModelJobOperationsSnapshot" /api/v1/model-jobs/{job_id}: get: operationId: getModelInferenceJob @@ -270,6 +329,27 @@ paths: application/json: schema: $ref: "#/components/schemas/ModelInferenceJob" + /api/v1/model-jobs/{job_id}/cancel: + post: + operationId: cancelModelInferenceJob + summary: Cancel a queued or running model job without deleting its history + parameters: + - name: job_id + in: path + required: true + schema: { type: string } + requestBody: + content: + application/json: + schema: + $ref: "#/components/schemas/ModelInferenceJobCancelRequest" + responses: + "200": + description: Cancelled model job + content: + application/json: + schema: + $ref: "#/components/schemas/ModelInferenceJob" /api/v1/model-jobs/{job_id}/handoff: post: operationId: handoffModelInferenceJob @@ -1569,6 +1649,13 @@ components: reason: type: string nullable: true + ModelInferenceJobCancelRequest: + type: object + properties: + reason: + type: string + maxLength: 1000 + nullable: true ModelInferenceJobHandoffRequest: type: object required: [reason] @@ -1601,6 +1688,8 @@ components: id: { type: string } tenant_id: { type: string } operation: { type: string, enum: [legal_qa, contract_review] } + created_by: { type: string, nullable: true } + display_title: { type: string } status: type: string enum: [queued, running, succeeded, fallback, failed, awaiting_human, resolved, cancelled] @@ -1618,10 +1707,22 @@ components: audit_run_id: { type: string, nullable: true } error_code: { type: string, nullable: true } error_message: { type: string, nullable: true } + worker_name: { type: string, nullable: true } handoff_reason: { type: string, nullable: true } handoff_note: { type: string, nullable: true } handoff_assignee: { type: string, nullable: true } resolution_note: { type: string, nullable: true } + cancel_requested_at: { type: string, format: date-time, nullable: true } + cancelled_by: { type: string, nullable: true } + cancel_reason: { type: string, nullable: true } + quality_status: { type: string, enum: [pending, passed, needs_review, failed, unknown] } + quality_score: { type: integer, minimum: 0, maximum: 100 } + quality_labels: + type: array + items: + $ref: "#/components/schemas/ModelOutputQualityLabel" + quality_summary: { type: string, nullable: true } + quality_gate_version: { type: string } queue_latency_ms: { type: integer } execution_latency_ms: { type: integer } total_latency_ms: { type: integer } @@ -1630,6 +1731,75 @@ components: updated_at: { type: string, format: date-time } started_at: { type: string, format: date-time, nullable: true } completed_at: { type: string, format: date-time, nullable: true } + ModelOutputQualityLabel: + type: object + required: [code, severity, message] + properties: + code: { type: string } + severity: { type: string, enum: [info, warning, error] } + message: { type: string } + ModelInferenceJobHistoryPage: + type: object + required: [items, has_more] + properties: + items: + type: array + items: + $ref: "#/components/schemas/ModelInferenceJob" + next_cursor: { type: string, nullable: true } + has_more: { type: boolean } + ModelInferenceJobMetrics: + type: object + required: [window_hours, total, queued, running, succeeded, fallback, failed, awaiting_human, resolved, cancelled, fallback_rate, failure_rate, manual_handoff_rate, retry_rate, queue_latency_p50_ms, queue_latency_p95_ms, execution_latency_p50_ms, execution_latency_p95_ms, total_latency_p50_ms, total_latency_p95_ms] + properties: + window_hours: { type: integer } + total: { type: integer } + queued: { type: integer } + running: { type: integer } + succeeded: { type: integer } + fallback: { type: integer } + failed: { type: integer } + awaiting_human: { type: integer } + resolved: { type: integer } + cancelled: { type: integer } + fallback_rate: { type: number } + failure_rate: { type: number } + manual_handoff_rate: { type: number } + retry_rate: { type: number } + queue_latency_p50_ms: { type: integer } + queue_latency_p95_ms: { type: integer } + execution_latency_p50_ms: { type: integer } + execution_latency_p95_ms: { type: integer } + total_latency_p50_ms: { type: integer } + total_latency_p95_ms: { type: integer } + ModelJobOperationalAlert: + type: object + required: [code, severity, title, detail, current_value, threshold] + properties: + code: { type: string } + severity: { type: string, enum: [info, warning, error] } + title: { type: string } + detail: { type: string } + current_value: { type: number } + 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] + properties: + metrics: + $ref: "#/components/schemas/ModelInferenceJobMetrics" + alerts: + type: array + items: + $ref: "#/components/schemas/ModelJobOperationalAlert" + completed_sample_count: { type: integer } + target_sample_count: { type: integer } + remaining_sample_count: { type: integer } + baseline_ready: { type: boolean } + quality_passed: { type: integer } + quality_needs_review: { type: integer } + quality_failed: { type: integer } + quality_pass_rate: { type: number } ContractReviewRequest: type: object required: diff --git a/yuqei-ai-platform/services/ai-platform-api/README.md b/yuqei-ai-platform/services/ai-platform-api/README.md index 80d9a75c..c57285b6 100644 --- a/yuqei-ai-platform/services/ai-platform-api/README.md +++ b/yuqei-ai-platform/services/ai-platform-api/README.md @@ -178,7 +178,10 @@ Core MVP endpoints: - `POST /api/v1/legal/qa` - `POST /api/v1/legal/qa/jobs` - `POST /api/v1/legal/contract-review/jobs` +- `GET /api/v1/model-jobs/history` +- `GET /api/v1/model-jobs/operations` - `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` @@ -197,6 +200,10 @@ in-process executor. Shared and production deployments must use `worker`; the worker claims durable jobs, recovers expired worker leases, and records queue, execution, fallback, retry, and manual-handoff outcomes. A fallback response is persisted as `status=fallback`, never relabeled as a successful model call. +Job submission accepts `Idempotency-Key`; reusing a key with a changed payload +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. Use `GET /api/v1/knowledge-chunks/search/explain` to inspect normalized query terms, keyword/vector/hybrid scores, inclusion rank, and filter reasons for the diff --git a/yuqei-ai-platform/services/ai-platform-api/migrations/versions/20260826_0027_add_model_job_operations_quality.py b/yuqei-ai-platform/services/ai-platform-api/migrations/versions/20260826_0027_add_model_job_operations_quality.py new file mode 100644 index 00000000..a4b78657 --- /dev/null +++ b/yuqei-ai-platform/services/ai-platform-api/migrations/versions/20260826_0027_add_model_job_operations_quality.py @@ -0,0 +1,88 @@ +"""add model job operations and quality controls + +Revision ID: 20260826_0027 +Revises: 20260826_0026 +Create Date: 2026-08-26 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "20260826_0027" +down_revision = "20260826_0026" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column("model_inference_jobs", sa.Column("created_by", sa.String(length=128), nullable=True)) + op.add_column( + "model_inference_jobs", + sa.Column("display_title", sa.String(length=256), nullable=False, server_default="模型任务"), + ) + op.add_column("model_inference_jobs", sa.Column("idempotency_key", sa.String(length=128), nullable=True)) + op.add_column("model_inference_jobs", sa.Column("request_fingerprint", sa.String(length=64), nullable=True)) + op.add_column("model_inference_jobs", sa.Column("cancel_requested_at", sa.DateTime(timezone=True), nullable=True)) + op.add_column("model_inference_jobs", sa.Column("cancelled_by", sa.String(length=128), nullable=True)) + op.add_column("model_inference_jobs", sa.Column("cancel_reason", sa.Text(), nullable=True)) + op.add_column( + "model_inference_jobs", + sa.Column("quality_status", sa.String(length=32), nullable=False, server_default="pending"), + ) + op.add_column( + "model_inference_jobs", + sa.Column("quality_score", sa.Integer(), nullable=False, server_default="0"), + ) + op.add_column( + "model_inference_jobs", + sa.Column("quality_labels_json", sa.Text(), nullable=False, server_default="[]"), + ) + op.add_column("model_inference_jobs", sa.Column("quality_summary", sa.Text(), nullable=True)) + op.add_column( + "model_inference_jobs", + sa.Column("quality_gate_version", sa.String(length=32), nullable=False, server_default="model-job-v1"), + ) + op.create_index(op.f("ix_model_inference_jobs_created_by"), "model_inference_jobs", ["created_by"]) + op.create_index(op.f("ix_model_inference_jobs_quality_status"), "model_inference_jobs", ["quality_status"]) + op.create_index( + "ix_model_inference_jobs_tenant_creator_created", + "model_inference_jobs", + ["tenant_id", "created_by", "created_at"], + ) + op.create_index( + "ix_model_inference_jobs_tenant_quality_created", + "model_inference_jobs", + ["tenant_id", "quality_status", "created_at"], + ) + op.create_index( + "ux_model_inference_jobs_tenant_operation_idempotency", + "model_inference_jobs", + ["tenant_id", "operation", "idempotency_key"], + unique=True, + ) + + +def downgrade() -> None: + op.drop_index( + "ux_model_inference_jobs_tenant_operation_idempotency", + table_name="model_inference_jobs", + ) + op.drop_index("ix_model_inference_jobs_tenant_quality_created", table_name="model_inference_jobs") + op.drop_index("ix_model_inference_jobs_tenant_creator_created", table_name="model_inference_jobs") + op.drop_index(op.f("ix_model_inference_jobs_quality_status"), table_name="model_inference_jobs") + op.drop_index(op.f("ix_model_inference_jobs_created_by"), table_name="model_inference_jobs") + op.drop_column("model_inference_jobs", "quality_gate_version") + op.drop_column("model_inference_jobs", "quality_summary") + op.drop_column("model_inference_jobs", "quality_labels_json") + op.drop_column("model_inference_jobs", "quality_score") + op.drop_column("model_inference_jobs", "quality_status") + op.drop_column("model_inference_jobs", "cancel_reason") + op.drop_column("model_inference_jobs", "cancelled_by") + op.drop_column("model_inference_jobs", "cancel_requested_at") + op.drop_column("model_inference_jobs", "request_fingerprint") + op.drop_column("model_inference_jobs", "idempotency_key") + op.drop_column("model_inference_jobs", "display_title") + op.drop_column("model_inference_jobs", "created_by") 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 847a8af6..3f1aa0ff 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 @@ -4,20 +4,24 @@ import asyncio from collections.abc import Callable from typing import Any -from fastapi import BackgroundTasks, FastAPI, HTTPException, Query +from fastapi import BackgroundTasks, FastAPI, Header, HTTPException, Query, Request from yuqei_ai_platform_api.config import AiPlatformSettings 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, + ModelInferenceJobCancelRequest, ModelInferenceJobHandoffRequest, + ModelInferenceJobHistoryPage, ModelInferenceJobMetrics, + ModelJobOperationsSnapshot, ModelInferenceJobResolutionRequest, ModelInferenceJobRetryRequest, ModelInferenceJobStore, ModelInferenceJobRecord, ) +from yuqei_ai_platform_api.model_output_quality import evaluate_model_output_quality from yuqei_ai_platform_api.repository import AiPlatformRepository from yuqei_ai_platform_api.tenant_context import ( current_tenant_id, @@ -39,15 +43,26 @@ def register_model_job_routes( status_code=202, tags=["legal", "model-jobs"], ) - def create_legal_qa_job(request: LegalQaRequest, background_tasks: BackgroundTasks) -> ModelInferenceJob: + def create_legal_qa_job( + request: LegalQaRequest, + background_tasks: BackgroundTasks, + http_request: Request, + idempotency_key: str | None = Header(default=None, alias="Idempotency-Key", max_length=128), + ) -> ModelInferenceJob: tenant_id = current_tenant_id() - job = job_store.create( - tenant_id=tenant_id, - operation="legal_qa", - request_payload=request.model_dump(mode="json"), - max_attempts=settings.model_job_max_attempts, - ) - _dispatch_if_enabled(app, settings, store, job_store, job, background_tasks) + try: + job, created = job_store.create_or_get( + tenant_id=tenant_id, + operation="legal_qa", + request_payload=request.model_dump(mode="json"), + max_attempts=settings.model_job_max_attempts, + created_by=http_request.headers.get("X-User-Id"), + idempotency_key=idempotency_key, + ) + except ValueError as error: + raise HTTPException(status_code=409, detail=str(error)) from error + if created: + _dispatch_if_enabled(app, settings, store, job_store, job, background_tasks) return job @app.post( @@ -59,15 +74,23 @@ def register_model_job_routes( def create_contract_review_job( request: ContractReviewRequest, background_tasks: BackgroundTasks, + http_request: Request, + idempotency_key: str | None = Header(default=None, alias="Idempotency-Key", max_length=128), ) -> ModelInferenceJob: tenant_id = current_tenant_id() - job = job_store.create( - tenant_id=tenant_id, - operation="contract_review", - request_payload=request.model_dump(mode="json"), - max_attempts=settings.model_job_max_attempts, - ) - _dispatch_if_enabled(app, settings, store, job_store, job, background_tasks) + try: + job, created = job_store.create_or_get( + tenant_id=tenant_id, + operation="contract_review", + request_payload=request.model_dump(mode="json"), + max_attempts=settings.model_job_max_attempts, + created_by=http_request.headers.get("X-User-Id"), + idempotency_key=idempotency_key, + ) + except ValueError as error: + raise HTTPException(status_code=409, detail=str(error)) from error + if created: + _dispatch_if_enabled(app, settings, store, job_store, job, background_tasks) return job @app.get( @@ -78,6 +101,39 @@ def register_model_job_routes( def list_model_jobs(limit: int = Query(default=20, ge=1, le=100)) -> list[ModelInferenceJob]: return job_store.list(tenant_id=current_tenant_id(), limit=limit) + @app.get( + f"{settings.api_prefix}/model-jobs/history", + response_model=ModelInferenceJobHistoryPage, + tags=["model-jobs"], + ) + def model_job_history( + request: Request, + operation: str | None = Query(default=None), + status: str | None = Query(default=None), + cursor: str | None = Query(default=None, max_length=512), + limit: int = Query(default=20, ge=1, le=100), + mine: bool = Query(default=True), + ) -> ModelInferenceJobHistoryPage: + if operation not in {None, "legal_qa", "contract_review"}: + raise HTTPException(status_code=422, detail="Unsupported model job operation filter.") + allowed_statuses = { + "queued", "running", "succeeded", "fallback", "failed", "awaiting_human", "resolved", "cancelled", + } + if status not in allowed_statuses | {None}: + raise HTTPException(status_code=422, detail="Unsupported model job status filter.") + requested_by = request.headers.get("X-User-Id") + try: + return job_store.history( + tenant_id=current_tenant_id(), + created_by=(requested_by or "__missing_user_identity__") if mine else None, + operation=operation, # type: ignore[arg-type] + status=status, # type: ignore[arg-type] + cursor=cursor, + limit=limit, + ) + except ValueError as error: + raise HTTPException(status_code=422, detail=str(error)) from error + @app.get( f"{settings.api_prefix}/model-jobs/metrics", response_model=ModelInferenceJobMetrics, @@ -88,6 +144,21 @@ def register_model_job_routes( ) -> ModelInferenceJobMetrics: return job_store.metrics(tenant_id=current_tenant_id(), window_hours=window_hours) + @app.get( + f"{settings.api_prefix}/model-jobs/operations", + response_model=ModelJobOperationsSnapshot, + tags=["model-jobs", "operations"], + ) + 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), + ) -> ModelJobOperationsSnapshot: + return job_store.operations_snapshot( + tenant_id=current_tenant_id(), + window_hours=window_hours, + target_sample_count=target_sample_count, + ) + @app.get( f"{settings.api_prefix}/model-jobs/{{job_id}}", response_model=ModelInferenceJob, @@ -99,6 +170,30 @@ def register_model_job_routes( raise HTTPException(status_code=404, detail="Model job not found.") return job + @app.post( + f"{settings.api_prefix}/model-jobs/{{job_id}}/cancel", + response_model=ModelInferenceJob, + tags=["model-jobs"], + ) + def cancel_model_job( + job_id: str, + payload: ModelInferenceJobCancelRequest, + request: Request, + ) -> ModelInferenceJob: + try: + return job_store.cancel( + job_id, + tenant_id=current_tenant_id(), + requested_by=request.headers.get("X-User-Id"), + reason=payload.reason, + ) + except KeyError as error: + raise HTTPException(status_code=404, detail="Model job not found.") from error + except PermissionError as error: + raise HTTPException(status_code=403, detail=str(error)) from error + except ValueError as error: + raise HTTPException(status_code=409, detail=str(error)) from error + @app.post( f"{settings.api_prefix}/model-jobs/{{job_id}}/retry", response_model=ModelInferenceJob, @@ -200,17 +295,18 @@ def execute_model_job( audit = store.get_ai_run_audit(audit_run_id) if audit_run_id else None audit_status = audit.status if audit is not None else "succeeded" provider_status = audit.provider_status if audit is not None else None - if audit_status == "fallback": - return job_store.finish( - job_id, - tenant_id=tenant_id, - result=result, - status="fallback", - provider_status=provider_status, - audit_run_id=audit_run_id, - error_code="provider_fallback_used", - error_message=audit.provider_error if audit is not None else "The provider fallback path was used.", - ) + quality = evaluate_model_output_quality( + operation=claimed.operation, + request_payload=claimed.request_payload, + result=result, + ) + quality_fields = { + "quality_status": quality.status, + "quality_score": quality.score, + "quality_labels": quality.labels, + "quality_summary": quality.summary, + "quality_gate_version": quality.gate_version, + } if audit_status == "failed": return job_store.finish( job_id, @@ -221,6 +317,43 @@ def execute_model_job( audit_run_id=audit_run_id, error_code="provider_run_failed", error_message=audit.provider_error if audit is not None else "The provider run failed.", + **quality_fields, + ) + if quality.status == "failed": + return job_store.finish( + job_id, + tenant_id=tenant_id, + result=result, + status="failed", + provider_status=provider_status, + audit_run_id=audit_run_id, + error_code="output_quality_gate_failed", + error_message=quality.summary, + **quality_fields, + ) + if audit_status == "fallback": + return job_store.finish( + job_id, + tenant_id=tenant_id, + result=result, + status="fallback", + provider_status=provider_status, + audit_run_id=audit_run_id, + error_code="provider_fallback_used", + error_message=audit.provider_error if audit is not None else "The provider fallback path was used.", + **quality_fields, + ) + if quality.status == "needs_review": + 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="output_quality_review_required", + error_message=quality.summary, + **quality_fields, ) return job_store.finish( job_id, @@ -229,8 +362,12 @@ def execute_model_job( status="succeeded", provider_status=provider_status, audit_run_id=audit_run_id, + **quality_fields, ) except Exception as error: + current = job_store.get(job_id, tenant_id=tenant_id) + if current is not None and current.status == "cancelled": + return current return job_store.fail_unexpected(job_id, tenant_id=tenant_id, error=error) finally: reset_current_tenant_id(tenant_token) 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 5062d66a..ab77dd3c 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 @@ -1,5 +1,8 @@ from __future__ import annotations +import base64 +import binascii +import hashlib import json from datetime import UTC, datetime, timedelta from threading import Lock @@ -7,9 +10,11 @@ from typing import Any, Literal from uuid import uuid4 from pydantic import BaseModel, Field -from sqlalchemy import Engine, select +from sqlalchemy import Engine, and_, or_, select +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 @@ -33,6 +38,8 @@ class ModelInferenceJob(BaseModel): id: str tenant_id: str operation: ModelJobOperation + created_by: str | None = None + display_title: str = "模型任务" status: ModelJobStatus stage: str progress_percent: int = Field(ge=0, le=100) @@ -50,6 +57,14 @@ class ModelInferenceJob(BaseModel): handoff_note: str | None = None handoff_assignee: str | None = None resolution_note: str | None = None + cancel_requested_at: datetime | None = None + cancelled_by: str | None = None + cancel_reason: str | None = None + quality_status: str = "pending" + quality_score: int = Field(default=0, ge=0, le=100) + quality_labels: list[ModelOutputQualityLabel] = Field(default_factory=list) + quality_summary: str | None = None + quality_gate_version: str = "model-job-v1" queue_latency_ms: int = 0 execution_latency_ms: int = 0 total_latency_ms: int = 0 @@ -62,6 +77,8 @@ class ModelInferenceJob(BaseModel): class ModelInferenceJobRecord(ModelInferenceJob): request_payload: dict[str, Any] = Field(default_factory=dict, exclude=True) + idempotency_key: str | None = Field(default=None, exclude=True) + request_fingerprint: str | None = Field(default=None, exclude=True) class ModelInferenceJobRetryRequest(BaseModel): @@ -79,6 +96,16 @@ class ModelInferenceJobResolutionRequest(BaseModel): resolution_note: str = Field(min_length=4, max_length=4000) +class ModelInferenceJobCancelRequest(BaseModel): + reason: str | None = Field(default=None, max_length=1000) + + +class ModelInferenceJobHistoryPage(BaseModel): + items: list[ModelInferenceJob] + next_cursor: str | None = None + has_more: bool = False + + class ModelInferenceJobMetrics(BaseModel): window_hours: int total: int @@ -89,6 +116,7 @@ class ModelInferenceJobMetrics(BaseModel): failed: int awaiting_human: int resolved: int + cancelled: int fallback_rate: float failure_rate: float manual_handoff_rate: float @@ -101,6 +129,28 @@ class ModelInferenceJobMetrics(BaseModel): total_latency_p95_ms: int +class ModelJobOperationalAlert(BaseModel): + code: str + severity: Literal["info", "warning", "error"] + title: str + detail: str + current_value: float + threshold: float + + +class ModelJobOperationsSnapshot(BaseModel): + metrics: ModelInferenceJobMetrics + alerts: list[ModelJobOperationalAlert] + completed_sample_count: int + target_sample_count: int = 30 + remaining_sample_count: int + baseline_ready: bool + quality_passed: int + quality_needs_review: int + quality_failed: int + quality_pass_rate: float + + class ModelInferenceJobStore: """Tenant-scoped durable queue with an in-memory test/development fallback.""" @@ -119,12 +169,50 @@ class ModelInferenceJobStore: operation: ModelJobOperation, request_payload: dict[str, Any], max_attempts: int = 3, + created_by: str | None = None, + idempotency_key: str | None = None, ) -> ModelInferenceJob: + job, _created = self.create_or_get( + tenant_id=tenant_id, + operation=operation, + request_payload=request_payload, + max_attempts=max_attempts, + created_by=created_by, + idempotency_key=idempotency_key, + ) + return job + + def create_or_get( + 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, + ) -> 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}) + if normalized_key: + existing = self._find_by_idempotency( + tenant_id=tenant_id, + operation=operation, + idempotency_key=normalized_key, + ) + if existing is not None: + _ensure_same_fingerprint(existing, fingerprint) + return _view(existing), False record = ModelInferenceJobRecord( id=f"model-job-{uuid4().hex[:16]}", tenant_id=tenant_id, operation=operation, + created_by=normalized_creator, + display_title=_display_title(operation, request_payload), + idempotency_key=normalized_key, + request_fingerprint=fingerprint, status="queued", stage="queued", progress_percent=0, @@ -138,11 +226,69 @@ class ModelInferenceJobStore: ) if self._session_factory is None: with self._lock: + if normalized_key: + existing = next( + ( + item + for item in self._records.values() + if item.tenant_id == tenant_id + and item.operation == operation + and item.idempotency_key == normalized_key + ), + None, + ) + if existing is not None: + _ensure_same_fingerprint(existing, fingerprint) + return _view(existing), False self._records[record.id] = record else: - with self._session_factory.begin() as session: + with self._session_factory() as session: session.add(_record_to_row(record)) - return _view(record) + try: + session.commit() + except IntegrityError: + session.rollback() + if not normalized_key: + raise + existing = self._find_by_idempotency( + tenant_id=tenant_id, + operation=operation, + idempotency_key=normalized_key, + ) + if existing is None: + raise + _ensure_same_fingerprint(existing, fingerprint) + return _view(existing), False + return _view(record), True + + def _find_by_idempotency( + self, + *, + tenant_id: str, + operation: ModelJobOperation, + idempotency_key: str, + ) -> ModelInferenceJobRecord | None: + if self._session_factory is None: + with self._lock: + return next( + ( + item + for item in self._records.values() + if item.tenant_id == tenant_id + and item.operation == operation + and item.idempotency_key == idempotency_key + ), + None, + ) + with self._session_factory() as session: + row = session.scalar( + select(ModelInferenceJobModel).where( + ModelInferenceJobModel.tenant_id == tenant_id, + ModelInferenceJobModel.operation == operation, + ModelInferenceJobModel.idempotency_key == idempotency_key, + ) + ) + return _row_to_record(row) if row is not None else None def get_record(self, job_id: str, *, tenant_id: str | None = None) -> ModelInferenceJobRecord | None: if self._session_factory is None: @@ -175,6 +321,66 @@ class ModelInferenceJobStore: records = [_row_to_record(row) for row in rows] return [_view(record) for record in records[: max(1, min(limit, 100))]] + def history( + self, + *, + tenant_id: str, + created_by: str | None = None, + operation: ModelJobOperation | None = None, + status: ModelJobStatus | None = None, + cursor: str | None = None, + limit: int = 20, + ) -> ModelInferenceJobHistoryPage: + page_size = max(1, min(limit, 100)) + cursor_value = _decode_cursor(cursor) if cursor else None + if self._session_factory is None: + with self._lock: + records = [item for item in self._records.values() if item.tenant_id == tenant_id] + records = [ + item for item in records + if (created_by is None or item.created_by == created_by) + and (operation is None or item.operation == operation) + and (status is None or item.status == status) + and (cursor_value is None or _history_key(item) < cursor_value) + ] + records.sort(key=_history_key, reverse=True) + selected = records[: page_size + 1] + else: + with self._session_factory() as session: + query = select(ModelInferenceJobModel).where(ModelInferenceJobModel.tenant_id == tenant_id) + if created_by is not None: + query = query.where(ModelInferenceJobModel.created_by == created_by) + if operation is not None: + query = query.where(ModelInferenceJobModel.operation == operation) + if status is not None: + query = query.where(ModelInferenceJobModel.status == status) + if cursor_value is not None: + cursor_time, cursor_id = cursor_value + query = query.where( + or_( + ModelInferenceJobModel.created_at < cursor_time, + and_( + ModelInferenceJobModel.created_at == cursor_time, + ModelInferenceJobModel.id < cursor_id, + ), + ) + ) + rows = session.scalars( + query.order_by( + ModelInferenceJobModel.created_at.desc(), + ModelInferenceJobModel.id.desc(), + ).limit(page_size + 1) + ).all() + selected = [_row_to_record(row) for row in rows] + has_more = len(selected) > page_size + page_records = selected[:page_size] + next_cursor = _encode_cursor(page_records[-1]) if has_more and page_records else None + return ModelInferenceJobHistoryPage( + items=[_view(item) for item in page_records], + next_cursor=next_cursor, + has_more=has_more, + ) + def claim(self, job_id: str, *, tenant_id: str, worker_name: str) -> ModelInferenceJobRecord | None: now = datetime.now(UTC) if self._session_factory is None: @@ -302,24 +508,35 @@ class ModelInferenceJobStore: *, tenant_id: str, result: dict[str, Any], - status: Literal["succeeded", "fallback", "failed"], + status: Literal["succeeded", "fallback", "failed", "awaiting_human"], provider_status: str | None, audit_run_id: str | None, error_code: str | None = None, error_message: str | None = None, + quality_status: str = "unknown", + quality_score: int = 0, + quality_labels: list[ModelOutputQualityLabel] | None = None, + quality_summary: str | None = None, + quality_gate_version: str = "model-job-v1", ) -> ModelInferenceJob: now = datetime.now(UTC) existing = self.get_record(job_id, tenant_id=tenant_id) if existing is None: raise KeyError(job_id) - retryable = status in {"fallback", "failed"} and existing.attempt_count < existing.max_attempts + retryable = status in {"fallback", "failed", "awaiting_human"} and existing.attempt_count < existing.max_attempts started_at = existing.started_at or now + stage = { + "succeeded": "completed", + "fallback": "fallback_ready", + "failed": "failed", + "awaiting_human": "manual_review", + }[status] return self._update( job_id, tenant_id=tenant_id, allowed_statuses={"running"}, status=status, - stage="completed" if status == "succeeded" else ("fallback_ready" if status == "fallback" else "failed"), + stage=stage, progress_percent=100, result=result, provider_status=provider_status, @@ -328,6 +545,13 @@ class ModelInferenceJobStore: retryable=retryable, error_code=error_code, error_message=error_message, + quality_status=quality_status, + quality_score=max(0, min(quality_score, 100)), + 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, + 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), completed_at=now, @@ -378,6 +602,47 @@ class ModelInferenceJobStore: total_latency_ms=0, error_code=None, error_message=None, + quality_status="pending", + quality_score=0, + quality_labels=[], + quality_summary=None, + cancel_requested_at=None, + cancelled_by=None, + cancel_reason=None, + ) + + def cancel( + self, + job_id: str, + *, + tenant_id: str, + requested_by: str | None, + reason: str | None, + ) -> ModelInferenceJob: + existing = self.get_record(job_id, tenant_id=tenant_id) + if existing is None: + raise KeyError(job_id) + normalized_actor = (requested_by or "").strip() or None + if existing.created_by and normalized_actor != existing.created_by: + raise PermissionError("Only the task creator can cancel this model job.") + if existing.status not in {"queued", "running"}: + raise ValueError("Only queued or running model jobs can be cancelled.") + now = datetime.now(UTC) + return self._update( + job_id, + tenant_id=tenant_id, + allowed_statuses={"queued", "running"}, + status="cancelled", + stage="cancelled_by_user", + progress_percent=100, + retryable=False, + cancel_requested_at=now, + cancelled_by=normalized_actor, + cancel_reason=(reason or "").strip() or None, + error_code="cancelled_by_user", + error_message=None, + total_latency_ms=_elapsed_ms(existing.created_at, now), + completed_at=now, ) def handoff( @@ -437,7 +702,11 @@ 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] + 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, @@ -449,6 +718,7 @@ class ModelInferenceJobStore: 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( @@ -464,6 +734,84 @@ class ModelInferenceJobStore: total_latency_p95_ms=_percentile([record.total_latency_ms for record in completed], 0.95), ) + def operations_snapshot( + self, + *, + tenant_id: str, + window_hours: int = 24 * 7, + target_sample_count: int = 30, + ) -> 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: + records = [ + item for item in self._records.values() + if item.tenant_id == tenant_id and _aware(item.created_at) >= cutoff + ] + else: + with self._session_factory() as session: + rows = session.scalars( + select(ModelInferenceJobModel).where( + ModelInferenceJobModel.tenant_id == tenant_id, + ModelInferenceJobModel.created_at >= cutoff, + ) + ).all() + records = [_row_to_record(row) for row in rows] + completed = [item for item in records if item.status in TERMINAL_MODEL_JOB_STATUSES] + evaluated = [ + item + for item in completed + if item.provider_status == "succeeded" + and item.quality_status in {"passed", "needs_review", "failed"} + ] + 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) + stale_cutoff = datetime.now(UTC) - timedelta(minutes=10) + stale_active = sum( + item.status in {"queued", "running"} and _aware(item.updated_at) < stale_cutoff + for item in records + ) + alerts: list[ModelJobOperationalAlert] = [] + target = max(1, min(target_sample_count, 1000)) + if len(evaluated) < target: + alerts.append( + _alert( + "baseline_samples_insufficient", + "warning", + "质量基线样本不足", + f"当前只有 {len(evaluated)} 个已评估真实样本,至少需要 {target} 个。", + len(evaluated), + target, + ) + ) + if metrics.queue_latency_p95_ms > 30_000: + alerts.append(_alert("queue_p95_high", "error", "排队耗时异常", "任务排队 P95 超过 30 秒。", metrics.queue_latency_p95_ms, 30_000)) + if metrics.execution_latency_p95_ms > 180_000: + alerts.append(_alert("execution_p95_high", "warning", "模型执行耗时偏高", "模型执行 P95 超过 3 分钟。", metrics.execution_latency_p95_ms, 180_000)) + if metrics.total >= 10 and metrics.fallback_rate > 0.10: + alerts.append(_alert("fallback_rate_high", "warning", "Fallback 比例偏高", "Fallback 比例超过 10%。", metrics.fallback_rate, 0.10)) + if metrics.total >= 10 and metrics.failure_rate > 0.05: + alerts.append(_alert("failure_rate_high", "error", "任务失败率偏高", "失败率超过 5%。", metrics.failure_rate, 0.05)) + if metrics.awaiting_human >= 5: + 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)) + quality_pass_rate = round(quality_passed / max(len(evaluated), 1), 4) + return ModelJobOperationsSnapshot( + metrics=metrics, + alerts=alerts, + completed_sample_count=len(evaluated), + target_sample_count=target, + remaining_sample_count=max(0, target - len(evaluated)), + baseline_ready=len(evaluated) >= target, + quality_passed=quality_passed, + quality_needs_review=quality_needs_review, + quality_failed=quality_failed, + quality_pass_rate=quality_pass_rate, + ) + def _update( self, job_id: str, @@ -493,6 +841,11 @@ class ModelInferenceJobStore: for key, value in updates.items(): if key == "result": row.result_payload_json = json.dumps(value, ensure_ascii=False) + elif key == "quality_labels": + row.quality_labels_json = json.dumps( + [item.model_dump(mode="json") if isinstance(item, BaseModel) else item for item in value], + ensure_ascii=False, + ) else: setattr(row, key, value) session.flush() @@ -508,11 +861,26 @@ def _locked_row(session: Session, job_id: str) -> ModelInferenceJobModel | None: def _record_to_row(record: ModelInferenceJobRecord) -> ModelInferenceJobModel: - data = record.model_dump(exclude={"request_payload", "result", "recommended_poll_after_ms"}) + data = record.model_dump( + exclude={ + "request_payload", + "result", + "recommended_poll_after_ms", + "quality_labels", + "idempotency_key", + "request_fingerprint", + } + ) return ModelInferenceJobModel( **data, + idempotency_key=record.idempotency_key, + request_fingerprint=record.request_fingerprint, request_payload_json=json.dumps(record.request_payload, ensure_ascii=False), result_payload_json=json.dumps(record.result, ensure_ascii=False) if record.result is not None else None, + quality_labels_json=json.dumps( + [item.model_dump(mode="json") for item in record.quality_labels], + ensure_ascii=False, + ), ) @@ -521,6 +889,10 @@ def _row_to_record(row: ModelInferenceJobModel) -> ModelInferenceJobRecord: id=row.id, tenant_id=row.tenant_id, operation=row.operation, + created_by=row.created_by, + display_title=row.display_title, + idempotency_key=row.idempotency_key, + request_fingerprint=row.request_fingerprint, status=row.status, stage=row.stage, progress_percent=row.progress_percent, @@ -539,6 +911,14 @@ def _row_to_record(row: ModelInferenceJobModel) -> ModelInferenceJobRecord: handoff_note=row.handoff_note, handoff_assignee=row.handoff_assignee, resolution_note=row.resolution_note, + cancel_requested_at=_aware(row.cancel_requested_at) if row.cancel_requested_at else None, + cancelled_by=row.cancelled_by, + cancel_reason=row.cancel_reason, + quality_status=row.quality_status, + quality_score=row.quality_score, + 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, queue_latency_ms=row.queue_latency_ms, execution_latency_ms=row.execution_latency_ms, total_latency_ms=row.total_latency_ms, @@ -571,3 +951,70 @@ def _percentile(values: list[int], percentile: float) -> int: return 0 index = round((len(normalized) - 1) * percentile) return normalized[index] + + +def _normalize_idempotency_key(value: str | None) -> str | None: + normalized = (value or "").strip() + if not normalized: + return None + if len(normalized) > 128: + raise ValueError("Idempotency key must not exceed 128 characters.") + return normalized + + +def _request_fingerprint(payload: dict[str, Any]) -> str: + canonical = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +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.") + + +def _display_title(operation: ModelJobOperation, payload: dict[str, Any]) -> str: + candidates = ( + payload.get("question") if operation == "legal_qa" else payload.get("contract_title"), + payload.get("review_focus"), + ) + value = next((str(item).strip() for item in candidates if item and str(item).strip()), "模型任务") + return value[:256] + + +def _history_key(record: ModelInferenceJobRecord) -> tuple[datetime, str]: + return _aware(record.created_at), record.id + + +def _encode_cursor(record: ModelInferenceJobRecord) -> str: + payload = json.dumps( + {"created_at": _aware(record.created_at).isoformat(), "id": record.id}, + separators=(",", ":"), + ).encode("utf-8") + return base64.urlsafe_b64encode(payload).decode("ascii").rstrip("=") + + +def _decode_cursor(value: str) -> tuple[datetime, str]: + try: + padding = "=" * (-len(value) % 4) + payload = json.loads(base64.urlsafe_b64decode(f"{value}{padding}").decode("utf-8")) + return _aware(datetime.fromisoformat(str(payload["created_at"]))), str(payload["id"]) + except (ValueError, KeyError, TypeError, json.JSONDecodeError, binascii.Error) as error: + raise ValueError("Invalid model job history cursor.") from error + + +def _alert( + code: str, + severity: Literal["info", "warning", "error"], + title: str, + detail: str, + current_value: float, + threshold: float, +) -> ModelJobOperationalAlert: + return ModelJobOperationalAlert( + code=code, + severity=severity, + title=title, + detail=detail, + current_value=float(current_value), + threshold=float(threshold), + ) diff --git a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/model_output_quality.py b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/model_output_quality.py new file mode 100644 index 00000000..15b489eb --- /dev/null +++ b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/model_output_quality.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +import re +from typing import Any, Literal + +from pydantic import BaseModel, Field + + +QualitySeverity = Literal["info", "warning", "error"] +QualityStatus = Literal["passed", "needs_review", "failed"] + + +class ModelOutputQualityLabel(BaseModel): + code: str + severity: QualitySeverity + message: str + + +class ModelOutputQualityResult(BaseModel): + status: QualityStatus + score: int = Field(ge=0, le=100) + labels: list[ModelOutputQualityLabel] = Field(default_factory=list) + summary: str + gate_version: str = "model-job-v1" + + +_UNRESOLVED_PLACEHOLDER = re.compile( + r"\{\{?\s*(question|matter_context|review_context|contract_text|contract_title|content)\s*\}?\}", + flags=re.IGNORECASE, +) +_MISSING_QUESTION_RESPONSES = ( + "未提供具体问题", + "没有提供具体问题", + "尚未提供具体问题", + "请提供具体问题", + "请补充具体问题", + "question was not provided", + "no question was provided", +) +_PROMPT_LEAK_MARKERS = ( + "you are chatgpt", + "system prompt", + "developer message", + "忽略之前的指令", +) + + +def evaluate_model_output_quality( + *, + operation: str, + request_payload: dict[str, Any], + result: dict[str, Any], +) -> ModelOutputQualityResult: + labels: list[ModelOutputQualityLabel] = [] + answer = _primary_text(operation, result) + normalized_answer = answer.strip() + + if not normalized_answer: + labels.append(_label("empty_output", "error", "模型没有返回可审阅的正文。")) + else: + if len(normalized_answer) < 60: + labels.append(_label("output_too_short", "warning", "模型正文短于 60 个字符,需要人工确认完整性。")) + if _UNRESOLVED_PLACEHOLDER.search(normalized_answer): + labels.append(_label("placeholder_leak", "error", "模型结果包含未渲染的提示词占位符。")) + lowered = normalized_answer.casefold() + if request_payload.get("question") and any(marker in lowered for marker in _MISSING_QUESTION_RESPONSES): + labels.append(_label("question_context_lost", "error", "已提交问题,但模型结果错误地声称问题缺失。")) + if any(marker in lowered for marker in _PROMPT_LEAK_MARKERS): + labels.append(_label("prompt_leak", "error", "模型结果疑似泄露系统或开发提示词。")) + + if operation == "legal_qa": + citations = result.get("primary_sources") or result.get("citations") or [] + if not isinstance(citations, list) or not citations: + labels.append(_label("missing_citation", "error", "法律问答没有返回引用依据。")) + elif any(not _citation_is_reviewable(item) for item in citations): + labels.append(_label("citation_incomplete", "warning", "部分引用缺少标题、出处或摘录。")) + else: + labels.append(_label("citation_reviewable", "info", f"已返回 {len(citations)} 条可审阅引用。")) + elif operation == "contract_review": + risk_items = result.get("risk_items") + if not isinstance(risk_items, list): + labels.append(_label("risk_items_missing", "error", "合同审查结果缺少结构化风险项。")) + elif not risk_items: + labels.append(_label("risk_items_empty", "warning", "合同审查没有识别风险项,需要抽样复核。")) + + error_count = sum(item.severity == "error" for item in labels) + warning_count = sum(item.severity == "warning" for item in labels) + score = max(0, 100 - error_count * 35 - warning_count * 12) + status: QualityStatus = "failed" if error_count else ("needs_review" if warning_count else "passed") + summary = f"{status}: {error_count} 个阻断项,{warning_count} 个复核项,得分 {score}。" + return ModelOutputQualityResult(status=status, score=score, labels=labels, summary=summary) + + +def _primary_text(operation: str, result: dict[str, Any]) -> str: + candidate = result.get("answer") if operation == "legal_qa" else result.get("summary") + return candidate if isinstance(candidate, str) else "" + + +def _citation_is_reviewable(value: Any) -> bool: + if not isinstance(value, dict): + return False + return all(isinstance(value.get(key), str) and value[key].strip() for key in ("title", "reference", "quote")) + + +def _label(code: str, severity: QualitySeverity, message: str) -> ModelOutputQualityLabel: + return ModelOutputQualityLabel(code=code, severity=severity, message=message) 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 5fb65005..f473e6b2 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 @@ -313,12 +313,24 @@ class KnowledgeIngestionJobModel(Base): class ModelInferenceJobModel(Base): __tablename__ = "model_inference_jobs" __table_args__ = ( + UniqueConstraint( + "tenant_id", + "operation", + "idempotency_key", + name="ux_model_inference_jobs_tenant_operation_idempotency", + ), Index("ix_model_inference_jobs_tenant_status_created", "tenant_id", "status", "created_at"), + Index("ix_model_inference_jobs_tenant_creator_created", "tenant_id", "created_by", "created_at"), + Index("ix_model_inference_jobs_tenant_quality_created", "tenant_id", "quality_status", "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) operation: Mapped[str] = mapped_column(String(64), index=True) + created_by: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True) + display_title: Mapped[str] = mapped_column(String(256), default="模型任务") + idempotency_key: Mapped[str | None] = mapped_column(String(128), nullable=True) + request_fingerprint: Mapped[str | None] = mapped_column(String(64), nullable=True) status: Mapped[str] = mapped_column(String(32), default="queued", index=True) stage: Mapped[str] = mapped_column(String(64), default="queued") progress_percent: Mapped[int] = mapped_column(Integer, default=0) @@ -337,6 +349,14 @@ class ModelInferenceJobModel(Base): handoff_note: Mapped[str | None] = mapped_column(Text, nullable=True) handoff_assignee: Mapped[str | None] = mapped_column(String(128), nullable=True) resolution_note: Mapped[str | None] = mapped_column(Text, nullable=True) + cancel_requested_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + cancelled_by: Mapped[str | None] = mapped_column(String(128), nullable=True) + cancel_reason: Mapped[str | None] = mapped_column(Text, nullable=True) + quality_status: Mapped[str] = mapped_column(String(32), default="pending", index=True) + quality_score: Mapped[int] = mapped_column(Integer, default=0) + 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") 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) diff --git a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/repository.py b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/repository.py index 25b6dd47..49904961 100644 --- a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/repository.py +++ b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/repository.py @@ -16,7 +16,7 @@ from sqlalchemy import bindparam, case, create_engine, delete, func, inspect as from sqlalchemy.engine import Engine from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session, sessionmaker -from sqlalchemy.pool import StaticPool +from sqlalchemy.pool import NullPool, StaticPool from yuqei_ai_platform_api.embeddings import ( EmbeddingProvider, @@ -3520,13 +3520,15 @@ class SqlAlchemyAiPlatformRepository: return {"ai_provider_configs", "prompt_templates", "knowledge_documents"} <= table_names -def build_sqlalchemy_engine(database_url: str) -> Engine: +def build_sqlalchemy_engine(database_url: str, *, use_null_pool: bool = False) -> Engine: ensure_sqlite_parent_dir(database_url) engine_options: dict[str, object] = {"future": True} if database_url.startswith("sqlite"): engine_options["connect_args"] = {"check_same_thread": False} if database_url in {"sqlite://", "sqlite:///:memory:", "sqlite+pysqlite:///:memory:"}: engine_options["poolclass"] = StaticPool + elif database_url.startswith("postgresql") and use_null_pool: + engine_options.update({"pool_pre_ping": True, "poolclass": NullPool}) elif database_url.startswith("postgresql"): engine_options.update( { 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 9c152914..c0ae1585 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 @@ -75,6 +75,22 @@ SECRET_REF_COLUMNS_BY_TABLE = { "ai_provider_configs": ("api_key_secret_ref",), "ocr_provider_configs": ("api_key_secret_ref",), } +MODEL_JOB_OPERATION_COLUMNS_BY_TABLE = { + "model_inference_jobs": ( + "created_by", + "display_title", + "idempotency_key", + "request_fingerprint", + "cancel_requested_at", + "cancelled_by", + "cancel_reason", + "quality_status", + "quality_score", + "quality_labels_json", + "quality_summary", + "quality_gate_version", + ), +} PGVECTOR_REQUIRED_COLUMNS_BY_TABLE = { "knowledge_chunks": ("embedding", "embedding_dimension", "search_vector"), "knowledge_documents": ("search_vector",), @@ -201,6 +217,15 @@ def check_ai_platform_schema( dialect=dialect, ) ) + checks.append( + _check_expected_columns( + "model_job_operation_columns", + MODEL_JOB_OPERATION_COLUMNS_BY_TABLE, + inspector, + table_names, + dialect=dialect, + ) + ) checks.append(_check_extensions(connection, dialect=dialect, strict=strict)) checks.append( _check_pgvector_schema( @@ -485,7 +510,7 @@ def _check_extensions(connection, *, dialect: str, strict: bool) -> RuntimeSchem ) return RuntimeSchemaCheck( name="postgres_extensions", - status="error", + status="error" if strict else "warning", message="Required PostgreSQL extensions are missing.", details={"missing_extensions": ",".join(missing), "installed_extensions": ",".join(sorted(installed))}, ) @@ -515,7 +540,7 @@ def _check_pgvector_schema(connection, inspector, table_names: set[str], *, dial if missing_columns: return RuntimeSchemaCheck( name="pgvector_schema", - status="error", + status="error" if strict else "warning", message="AI Platform pgvector/search columns are missing.", details={ "missing_columns": ",".join(sorted(missing_columns)), 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 27563d7b..a17619a0 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 @@ -100,6 +100,8 @@ def test_runtime_schema_check_warns_for_local_sqlite_auto_created_schema(tmp_pat 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["pgvector_schema"]["status"] == "warning" @@ -126,6 +128,7 @@ def test_runtime_schema_check_errors_for_missing_core_tables(tmp_path) -> None: assert checks["tenant_columns"]["status"] == "error" assert checks["trace_governance_columns"]["status"] == "error" assert checks["secret_ref_columns"]["status"] == "error" + assert checks["model_job_operation_columns"]["status"] == "error" assert checks["pgvector_schema"]["status"] == "error" assert "knowledge_documents" in checks["core_tables"]["details"]["missing_tables"] assert "knowledge_documents.tenant_id" in checks["tenant_columns"]["details"]["missing_columns"] 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 71faa187..1bf91272 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 @@ -133,3 +133,161 @@ def test_model_jobs_are_tenant_isolated() -> None: headers={"X-Tenant-Id": "tenant-a"}, ) assert own_response.status_code == 200 + + +def test_model_job_submission_is_idempotent_and_rejects_payload_or_actor_reuse() -> None: + _app, _repository, client = _app_and_client() + headers = {"X-User-Id": "user-a", "Idempotency-Key": "legal-qa-submit-1"} + payload = {"question": "付款期限如何约定?", "retrieval_strategy": "hybrid"} + + first = client.post("/api/v1/legal/qa/jobs", headers=headers, json=payload) + repeated = client.post("/api/v1/legal/qa/jobs", headers=headers, json=payload) + + assert first.status_code == 202 + assert repeated.status_code == 202 + assert repeated.json()["id"] == first.json()["id"] + changed = client.post( + "/api/v1/legal/qa/jobs", + headers=headers, + json={"question": "验收期限如何约定?", "retrieval_strategy": "hybrid"}, + ) + assert changed.status_code == 409 + other_actor = client.post( + "/api/v1/legal/qa/jobs", + headers={"X-User-Id": "user-b", "Idempotency-Key": "legal-qa-submit-1"}, + json=payload, + ) + assert other_actor.status_code == 409 + + +def test_model_job_history_is_creator_scoped_paginated_and_cancellable() -> None: + _app, _repository, client = _app_and_client() + user_a_headers = {"X-User-Id": "user-a"} + first = client.post( + "/api/v1/legal/qa/jobs", + headers=user_a_headers, + json={"question": "第一条问题是什么?"}, + ).json() + second = client.post( + "/api/v1/legal/qa/jobs", + headers=user_a_headers, + json={"question": "第二条问题是什么?"}, + ).json() + other = client.post( + "/api/v1/legal/qa/jobs", + headers={"X-User-Id": "user-b"}, + json={"question": "另一个用户的问题是什么?"}, + ).json() + + page_one = client.get( + "/api/v1/model-jobs/history?operation=legal_qa&limit=1", + headers=user_a_headers, + ).json() + assert len(page_one["items"]) == 1 + assert page_one["items"][0]["created_by"] == "user-a" + assert page_one["has_more"] is True + page_two = client.get( + f"/api/v1/model-jobs/history?operation=legal_qa&limit=1&cursor={page_one['next_cursor']}", + headers=user_a_headers, + ).json() + assert len(page_two["items"]) == 1 + assert {page_one["items"][0]["id"], page_two["items"][0]["id"]} == {first["id"], second["id"]} + + cancelled = client.post( + f"/api/v1/model-jobs/{first['id']}/cancel", + headers=user_a_headers, + json={"reason": "用户不再需要回答。"}, + ) + assert cancelled.status_code == 200 + assert cancelled.json()["status"] == "cancelled" + assert cancelled.json()["cancelled_by"] == "user-a" + assert client.post( + f"/api/v1/model-jobs/{first['id']}/cancel", + headers=user_a_headers, + json={}, + ).status_code == 409 + assert client.post( + f"/api/v1/model-jobs/{other['id']}/cancel", + headers=user_a_headers, + json={}, + ).status_code == 403 + + +def test_model_job_operations_baseline_only_counts_real_provider_outputs() -> None: + app, _repository, client = _app_and_client() + job_store = app.state.model_job_store + for index in range(3): + job = job_store.create( + tenant_id="default", + operation="legal_qa", + request_payload={"question": f"真实模型样本 {index + 1}"}, + ) + job_store.claim(job.id, tenant_id="default", worker_name="quality-baseline") + job_store.finish( + job.id, + tenant_id="default", + result={"answer": "可审阅回答", "citations": []}, + status="succeeded", + provider_status="succeeded", + audit_run_id=f"audit-{index + 1}", + quality_status="passed", + quality_score=100, + quality_summary="passed", + ) + fallback_job = job_store.create( + tenant_id="default", + operation="legal_qa", + request_payload={"question": "本地 fallback 不应计入真实基线"}, + ) + job_store.claim(fallback_job.id, tenant_id="default", worker_name="quality-baseline") + job_store.finish( + fallback_job.id, + tenant_id="default", + result={"answer": "fallback 回答", "citations": []}, + status="fallback", + provider_status="skipped", + audit_run_id="audit-fallback", + quality_status="passed", + quality_score=100, + quality_summary="passed", + ) + + snapshot = client.get("/api/v1/model-jobs/operations?target_sample_count=3").json() + assert snapshot["completed_sample_count"] == 3 + assert snapshot["baseline_ready"] is True + assert snapshot["remaining_sample_count"] == 0 + + +def test_running_job_cancellation_wins_over_late_worker_result() -> None: + app, repository, client = _app_and_client() + headers = {"X-User-Id": "user-a"} + created = client.post( + "/api/v1/legal/qa/jobs", + headers=headers, + json={"question": "取消中的任务是否会被晚到结果覆盖?"}, + ).json() + claimed = app.state.model_job_store.claim( + created["id"], + tenant_id="default", + worker_name="worker-before-cancel", + ) + assert claimed is not None + cancelled = client.post( + f"/api/v1/model-jobs/{created['id']}/cancel", + headers=headers, + json={"reason": "验证取消优先级"}, + ).json() + assert cancelled["worker_name"] == "worker-before-cancel" + + late_result = execute_model_job( + app=app, + store=repository, + job_store=app.state.model_job_store, + job_id=created["id"], + tenant_id="default", + worker_name="worker-before-cancel", + claimed_job=claimed, + ) + assert late_result is not None + assert late_result.status == "cancelled" + assert late_result.result is None diff --git a/yuqei-ai-platform/services/ai-platform-api/tests/test_model_output_quality.py b/yuqei-ai-platform/services/ai-platform-api/tests/test_model_output_quality.py new file mode 100644 index 00000000..354da55b --- /dev/null +++ b/yuqei-ai-platform/services/ai-platform-api/tests/test_model_output_quality.py @@ -0,0 +1,46 @@ +from yuqei_ai_platform_api.model_output_quality import evaluate_model_output_quality + + +def test_legal_qa_quality_passes_reviewable_answer() -> None: + result = evaluate_model_output_quality( + operation="legal_qa", + request_payload={"question": "知识产权归属需要注意什么?"}, + result={ + "answer": "知识产权归属应明确既有成果与项目成果的边界,并约定权利登记、许可范围及侵权责任。还应区分源代码、文档、数据和第三方素材,明确交付条件、使用地域、许可期限及离职人员成果归属。", + "citations": [ + {"title": "民法典", "reference": "技术合同章", "quote": "技术成果归属由当事人约定。"}, + ], + }, + ) + + assert result.status == "passed" + assert result.score == 100 + + +def test_quality_gate_blocks_placeholder_and_lost_question() -> None: + result = evaluate_model_output_quality( + operation="legal_qa", + request_payload={"question": "验收条款有什么风险?"}, + result={ + "answer": "未提供具体问题,请补充 {question} 后再试。", + "citations": [], + }, + ) + + assert result.status == "failed" + assert {item.code for item in result.labels} >= { + "placeholder_leak", + "question_context_lost", + "missing_citation", + } + + +def test_short_contract_review_is_sent_to_review() -> None: + result = evaluate_model_output_quality( + operation="contract_review", + request_payload={"contract_text": "测试合同"}, + result={"summary": "未发现明显风险。", "risk_items": []}, + ) + + assert result.status == "needs_review" + assert {item.code for item in result.labels} == {"output_too_short", "risk_items_empty"} diff --git a/yuqei-ai-platform/services/worker/src/yuqei_worker/main.py b/yuqei-ai-platform/services/worker/src/yuqei_worker/main.py index c09b997f..a318ba2b 100644 --- a/yuqei-ai-platform/services/worker/src/yuqei_worker/main.py +++ b/yuqei-ai-platform/services/worker/src/yuqei_worker/main.py @@ -10,7 +10,7 @@ from yuqei_ai_platform_api.object_storage import build_object_storage from yuqei_ai_platform_api.api.model_jobs import 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.repository import SqlAlchemyAiPlatformRepository +from yuqei_ai_platform_api.repository import SqlAlchemyAiPlatformRepository, build_sqlalchemy_engine from yuqei_ai_platform_api.tenant_context import reset_current_tenant_id, set_current_tenant_id from yuqei_ai_platform_api.worker_runtime import ( TraceGovernanceWorkerOptions, @@ -108,9 +108,11 @@ def run_once( ) -> WorkerRunSummary: resolved_settings = settings or get_settings() tenant_token = set_current_tenant_id(resolved_settings.tenant_id) + engine = build_sqlalchemy_engine(resolved_settings.database_url, use_null_pool=True) try: repository = SqlAlchemyAiPlatformRepository( resolved_settings.database_url, + engine=engine, auto_create=resolved_settings.database_auto_create, ) processed = 0 @@ -232,6 +234,7 @@ def run_once( ) return summary finally: + engine.dispose() reset_current_tenant_id(tenant_token) diff --git a/yuqei-ai-platform/services/worker/tests/test_health.py b/yuqei-ai-platform/services/worker/tests/test_health.py index ad014db3..970190a6 100644 --- a/yuqei-ai-platform/services/worker/tests/test_health.py +++ b/yuqei-ai-platform/services/worker/tests/test_health.py @@ -1,10 +1,17 @@ from datetime import UTC, datetime, timedelta +from sqlalchemy.pool import NullPool + from yuqei_worker.config import WorkerSettings from yuqei_worker.main import health_snapshot, run_once from yuqei_ai_platform_api.object_storage import build_local_object_storage from yuqei_ai_platform_api.model_jobs import ModelInferenceJobStore -from yuqei_ai_platform_api.repository import AiRunAudit, OcrProviderConfigCreate, SqlAlchemyAiPlatformRepository +from yuqei_ai_platform_api.repository import ( + AiRunAudit, + OcrProviderConfigCreate, + SqlAlchemyAiPlatformRepository, + build_sqlalchemy_engine, +) def test_worker_health_snapshot_returns_worker_status() -> None: @@ -19,6 +26,17 @@ def test_worker_health_snapshot_returns_worker_status() -> None: } +def test_worker_postgres_engine_uses_connection_neutral_pool() -> None: + engine = build_sqlalchemy_engine( + "postgresql+psycopg://worker@example.invalid/worker", + use_null_pool=True, + ) + try: + assert isinstance(engine.pool, NullPool) + finally: + engine.dispose() + + def test_worker_processes_durable_model_job_with_explicit_fallback(tmp_path, monkeypatch) -> None: database_url = f"sqlite:///{tmp_path / 'worker-model-job.db'}" repository = SqlAlchemyAiPlatformRepository(database_url, auto_create=True) diff --git a/yuqei-ai-sdk-python/src/yuqei_sdk/__init__.py b/yuqei-ai-sdk-python/src/yuqei_sdk/__init__.py index 5ef97709..cdf5d55a 100644 --- a/yuqei-ai-sdk-python/src/yuqei_sdk/__init__.py +++ b/yuqei-ai-sdk-python/src/yuqei_sdk/__init__.py @@ -39,6 +39,11 @@ from yuqei_sdk.ai_platform import ( LegalQaRequest, LegalQaResponse, ModelInferenceJob, + ModelInferenceJobHistoryPage, + ModelInferenceJobMetrics, + ModelJobOperationalAlert, + ModelJobOperationsSnapshot, + ModelOutputQualityLabel, PromptTemplate, PromptTemplateChange, PromptTemplateCreate, @@ -97,6 +102,11 @@ __all__ = [ "LegalQaRequest", "LegalQaResponse", "ModelInferenceJob", + "ModelInferenceJobHistoryPage", + "ModelInferenceJobMetrics", + "ModelJobOperationalAlert", + "ModelJobOperationsSnapshot", + "ModelOutputQualityLabel", "PromptTemplate", "PromptTemplateChange", "PromptTemplateCreate", 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 5a705034..4a98f5dd 100644 --- a/yuqei-ai-sdk-python/src/yuqei_sdk/ai_platform.py +++ b/yuqei-ai-sdk-python/src/yuqei_sdk/ai_platform.py @@ -40,10 +40,18 @@ class LegalQaResponse(BaseModel): ai_audit_run_id: str | None = None +class ModelOutputQualityLabel(BaseModel): + code: str + severity: str + message: str + + class ModelInferenceJob(BaseModel): id: str tenant_id: str operation: str + created_by: str | None = None + display_title: str = "模型任务" status: str stage: str progress_percent: int @@ -56,10 +64,19 @@ class ModelInferenceJob(BaseModel): audit_run_id: str | None = None error_code: str | None = None error_message: str | None = None + worker_name: str | None = None handoff_reason: str | None = None handoff_note: str | None = None handoff_assignee: str | None = None resolution_note: str | None = None + cancel_requested_at: datetime | None = None + cancelled_by: str | None = None + cancel_reason: str | None = None + quality_status: str = "pending" + quality_score: int = 0 + quality_labels: list[ModelOutputQualityLabel] = Field(default_factory=list) + quality_summary: str | None = None + quality_gate_version: str = "model-job-v1" queue_latency_ms: int = 0 execution_latency_ms: int = 0 total_latency_ms: int = 0 @@ -70,6 +87,57 @@ class ModelInferenceJob(BaseModel): completed_at: datetime | None = None +class ModelInferenceJobHistoryPage(BaseModel): + items: list[ModelInferenceJob] + next_cursor: str | None = None + has_more: bool = False + + +class ModelInferenceJobMetrics(BaseModel): + window_hours: int + total: int + queued: int + running: int + succeeded: int + fallback: int + failed: int + awaiting_human: int + resolved: int + cancelled: int = 0 + fallback_rate: float + failure_rate: float + manual_handoff_rate: float + retry_rate: float + queue_latency_p50_ms: int + queue_latency_p95_ms: int + execution_latency_p50_ms: int + execution_latency_p95_ms: int + total_latency_p50_ms: int + total_latency_p95_ms: int + + +class ModelJobOperationalAlert(BaseModel): + code: str + severity: str + title: str + detail: str + current_value: float + threshold: float + + +class ModelJobOperationsSnapshot(BaseModel): + metrics: ModelInferenceJobMetrics + alerts: list[ModelJobOperationalAlert] = Field(default_factory=list) + completed_sample_count: int + target_sample_count: int + remaining_sample_count: int + baseline_ready: bool + quality_passed: int + quality_needs_review: int + quality_failed: int + quality_pass_rate: float + + class ContractReviewRequest(BaseModel): contract_title: str = Field(min_length=1) contract_text: str = Field(min_length=1) @@ -721,16 +789,48 @@ class AiPlatformClient: self, request: LegalQaRequest, *, + idempotency_key: str | None = None, context: RequestContext | None = None, ) -> ModelInferenceJob: + headers = (context or RequestContext()).to_headers() + if idempotency_key: + headers["Idempotency-Key"] = idempotency_key response = self._client.post( f"{self._api_prefix}/legal/qa/jobs", json=request.model_dump(exclude_none=True), - headers=(context or RequestContext()).to_headers(), + headers=headers, ) response.raise_for_status() return ModelInferenceJob.model_validate(response.json()) + def list_model_jobs( + self, + *, + operation: str | None = None, + status: str | None = None, + cursor: str | None = None, + limit: int = 20, + mine: bool = True, + context: RequestContext | None = None, + ) -> ModelInferenceJobHistoryPage: + params: dict[str, str | int] = { + "limit": limit, + "mine": str(mine).lower(), + } + if operation is not None: + params["operation"] = operation + if status is not None: + params["status"] = status + if cursor is not None: + params["cursor"] = cursor + response = self._client.get( + f"{self._api_prefix}/model-jobs/history", + params=params, + headers=(context or RequestContext()).to_headers(), + ) + response.raise_for_status() + return ModelInferenceJobHistoryPage.model_validate(response.json()) + def get_model_job( self, job_id: str, @@ -760,6 +860,36 @@ class AiPlatformClient: response.raise_for_status() return ModelInferenceJob.model_validate(response.json()) + def cancel_model_job( + self, + job_id: str, + *, + reason: str | None = None, + context: RequestContext | None = None, + ) -> ModelInferenceJob: + response = self._client.post( + f"{self._api_prefix}/model-jobs/{job_id}/cancel", + json={"reason": reason}, + headers=(context or RequestContext()).to_headers(), + ) + response.raise_for_status() + return ModelInferenceJob.model_validate(response.json()) + + def get_model_job_operations( + self, + *, + window_hours: int = 24 * 7, + target_sample_count: int = 30, + 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}, + headers=(context or RequestContext()).to_headers(), + ) + response.raise_for_status() + return ModelJobOperationsSnapshot.model_validate(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 6e1a22c0..313df3df 100644 --- a/yuqei-ai-sdk-python/tests/test_ai_platform.py +++ b/yuqei-ai-sdk-python/tests/test_ai_platform.py @@ -108,6 +108,82 @@ def test_ai_platform_client_posts_legal_qa_with_context_headers() -> None: assert captured_request.headers["X-AI-Platform-Service-Key"] == "ai-platform-service-key" +def test_ai_platform_client_supports_model_job_operations() -> None: + seen_requests: list[httpx.Request] = [] + + def job_payload(status: str = "queued") -> dict[str, object]: + return { + "id": "model-job-1", + "tenant_id": "tenant-1", + "operation": "legal_qa", + "created_by": "user-1", + "display_title": "付款期限如何约定?", + "status": status, + "stage": "queued" if status == "queued" else "cancelled_by_user", + "progress_percent": 0 if status == "queued" else 100, + "result": None, + "attempt_count": 0, + "max_attempts": 3, + "retryable": status == "queued", + "fallback_used": False, + "quality_status": "pending", + "quality_score": 0, + "quality_labels": [], + "queue_latency_ms": 0, + "execution_latency_ms": 0, + "total_latency_ms": 0, + "recommended_poll_after_ms": 2000 if status == "queued" else 0, + "created_at": "2026-08-26T08:00:00Z", + "updated_at": "2026-08-26T08:00:00Z", + } + + def handler(request: httpx.Request) -> httpx.Response: + seen_requests.append(request) + if request.url.path.endswith("/legal/qa/jobs"): + 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("/cancel"): + return httpx.Response(200, json=job_payload("cancelled")) + if request.url.path.endswith("/model-jobs/operations"): + return httpx.Response(200, json={ + "metrics": { + "window_hours": 168, "total": 1, "queued": 1, "running": 0, "succeeded": 0, + "fallback": 0, "failed": 0, "awaiting_human": 0, "resolved": 0, + "fallback_rate": 0, "failure_rate": 0, "manual_handoff_rate": 0, "retry_rate": 0, + "queue_latency_p50_ms": 0, "queue_latency_p95_ms": 0, + "execution_latency_p50_ms": 0, "execution_latency_p95_ms": 0, + "total_latency_p50_ms": 0, "total_latency_p95_ms": 0, + }, + "alerts": [], "completed_sample_count": 0, "target_sample_count": 30, + "remaining_sample_count": 30, "baseline_ready": False, + "quality_passed": 0, "quality_needs_review": 0, "quality_failed": 0, + "quality_pass_rate": 0, + }) + return httpx.Response(404) + + client = AiPlatformClient("http://ai-platform.test", transport=httpx.MockTransport(handler)) + context = RequestContext(tenant_id="tenant-1", user_id="user-1") + created = client.create_legal_qa_job( + LegalQaRequest(question="付款期限如何约定?"), + idempotency_key="submit-1", + 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) + + assert created.id == "model-job-1" + assert history.items[0].display_title == "付款期限如何约定?" + assert cancelled.status == "cancelled" + assert operations.remaining_sample_count == 30 + assert seen_requests[0].headers["Idempotency-Key"] == "submit-1" + history_query = parse_qs(seen_requests[1].url.query.decode()) + assert history_query["operation"] == ["legal_qa"] + assert "status" not in history_query + assert "cursor" not in history_query + + def test_ai_platform_client_posts_contract_review_with_context_headers() -> None: captured_request: httpx.Request | None = None 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 new file mode 100644 index 00000000..53b67a79 --- /dev/null +++ b/yuqei-contract-platform/apps/contract-web/src/app/admin/model-jobs/page.tsx @@ -0,0 +1,123 @@ +"use client"; + +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"; + +export default function ModelJobOperationsPage() { + const [snapshot, setSnapshot] = useState(null); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(null); + + const loadSnapshot = useCallback(async () => { + setLoading(true); + setError(null); + try { + setSnapshot(await getModelJobOperations()); + } catch (requestError) { + setError(requestError instanceof Error ? requestError.message : "模型任务运营指标加载失败。"); + } finally { + setLoading(false); + } + }, []); + + useEffect(() => { void loadSnapshot(); }, [loadSnapshot]); + + return ( + <> +
+
+

模型任务运营

+

统一观察后台任务积压、失败、人工接管、质量门禁和真实样本基线。

+
+ +
+ + {error &&
{error}
} + {!snapshot && loading &&
正在加载模型任务运营数据…
} + + {snapshot && ( + <> +
+ + + + +
+ +
+
+
+

30 条真实样本基线

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

+ + {snapshot.baseline_ready ? "基线已就绪" : `还差 ${snapshot.remaining_sample_count} 条`} + +
+
+
{snapshot.completed_sample_count}/ {snapshot.target_sample_count} 条
+
+ +
+

通过 {snapshot.quality_passed} · 待人工 {snapshot.quality_needs_review} · 阻断 {snapshot.quality_failed}

+
+
+ +
+

状态与执行质量

最近 {snapshot.metrics.window_hours} 小时
+
+ + + + + + + + + +
排队 / 运行 / 已取消{snapshot.metrics.queued} / {snapshot.metrics.running} / {snapshot.metrics.cancelled}
成功 / fallback / 失败{snapshot.metrics.succeeded} / {snapshot.metrics.fallback} / {snapshot.metrics.failed}
人工复核{snapshot.metrics.awaiting_human}
执行 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.alerts.length} 条
+
+ {snapshot.alerts.length === 0 ?
当前窗口没有触发阈值告警。
: ( +
+ {snapshot.alerts.map((alert) => ( +
+
+ {alert.title} +

{alert.detail}

+
+ + {alert.severity === "error" ? "严重" : alert.severity === "warning" ? "警告" : "提示"} + +
+ ))} +
+ )} +
+
+ + )} + + ); +} + +function Metric({ icon: Icon, label, value }: { icon: typeof Activity; label: string; value: string }) { + return
{value}
{label}
; +} + +function formatRate(value: number) { + return `${(value * 100).toFixed(value > 0 && value < 0.1 ? 1 : 0)}%`; +} + +function formatDuration(value: number) { + if (!value) return "0 秒"; + if (value < 60_000) return `${(value / 1000).toFixed(value < 10_000 ? 1 : 0)} 秒`; + return `${(value / 60_000).toFixed(1)} 分钟`; +} diff --git a/yuqei-contract-platform/apps/contract-web/src/app/admin/page.tsx b/yuqei-contract-platform/apps/contract-web/src/app/admin/page.tsx index 60eda6f6..e8790e1b 100644 --- a/yuqei-contract-platform/apps/contract-web/src/app/admin/page.tsx +++ b/yuqei-contract-platform/apps/contract-web/src/app/admin/page.tsx @@ -135,6 +135,14 @@ const adminSections: AdminSection[] = [ icon: Activity, href: "/admin/ai-quality-dashboard" }, + { + capability: "ai_quality_dashboard", + platformCapability: "platform.operations.view", + title: "模型任务运营", + description: "查看异步任务积压、失败、人工接管、质量门禁和 30 条真实样本基线。", + icon: Gauge, + href: "/admin/model-jobs" + }, { capability: "ai_traces", title: "AI 调用 Trace", diff --git a/yuqei-contract-platform/apps/contract-web/src/app/api/contract-platform/admin/model-jobs/operations/route.ts b/yuqei-contract-platform/apps/contract-web/src/app/api/contract-platform/admin/model-jobs/operations/route.ts new file mode 100644 index 00000000..3efaca5b --- /dev/null +++ b/yuqei-contract-platform/apps/contract-web/src/app/api/contract-platform/admin/model-jobs/operations/route.ts @@ -0,0 +1,6 @@ +import { buildContractApiUrl, proxyContractApiResponse } from "@/lib/contract-api"; + +export async function GET(request: Request) { + const search = new URL(request.url).search; + return proxyContractApiResponse(buildContractApiUrl(`/api/v1/admin/model-jobs/operations${search}`)); +} diff --git a/yuqei-contract-platform/apps/contract-web/src/app/api/contract-platform/legal/qa/jobs/[jobId]/cancel/route.ts b/yuqei-contract-platform/apps/contract-web/src/app/api/contract-platform/legal/qa/jobs/[jobId]/cancel/route.ts new file mode 100644 index 00000000..b97031bf --- /dev/null +++ b/yuqei-contract-platform/apps/contract-web/src/app/api/contract-platform/legal/qa/jobs/[jobId]/cancel/route.ts @@ -0,0 +1,13 @@ +import { buildContractApiUrl, proxyContractApiResponse } from "@/lib/contract-api"; + +export async function POST(request: Request, context: { params: Promise<{ jobId: string }> }) { + const { jobId } = await context.params; + return proxyContractApiResponse( + buildContractApiUrl(`/api/v1/legal/qa/jobs/${encodeURIComponent(jobId)}/cancel`), + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: await request.text(), + }, + ); +} diff --git a/yuqei-contract-platform/apps/contract-web/src/app/api/contract-platform/legal/qa/jobs/route.ts b/yuqei-contract-platform/apps/contract-web/src/app/api/contract-platform/legal/qa/jobs/route.ts index 817d1393..e920218c 100644 --- a/yuqei-contract-platform/apps/contract-web/src/app/api/contract-platform/legal/qa/jobs/route.ts +++ b/yuqei-contract-platform/apps/contract-web/src/app/api/contract-platform/legal/qa/jobs/route.ts @@ -1,9 +1,17 @@ import { buildContractApiUrl, proxyContractApiResponse } from "@/lib/contract-api"; +export async function GET(request: Request) { + const search = new URL(request.url).search; + return proxyContractApiResponse(buildContractApiUrl(`/api/v1/legal/qa/jobs${search}`)); +} + export async function POST(request: Request) { + const headers = new Headers({ "Content-Type": "application/json" }); + const idempotencyKey = request.headers.get("idempotency-key")?.trim(); + if (idempotencyKey) headers.set("Idempotency-Key", idempotencyKey.slice(0, 128)); return proxyContractApiResponse(buildContractApiUrl("/api/v1/legal/qa/jobs"), { method: "POST", - headers: { "Content-Type": "application/json" }, + headers, body: await request.text(), }); } diff --git a/yuqei-contract-platform/apps/contract-web/src/app/legal-qa/legal-qa-client.tsx b/yuqei-contract-platform/apps/contract-web/src/app/legal-qa/legal-qa-client.tsx index 17d7e4bb..ca10d8fc 100644 --- a/yuqei-contract-platform/apps/contract-web/src/app/legal-qa/legal-qa-client.tsx +++ b/yuqei-contract-platform/apps/contract-web/src/app/legal-qa/legal-qa-client.tsx @@ -1,10 +1,17 @@ "use client"; import Link from "next/link"; -import { BookOpenCheck, ClipboardList, Loader2, MessageSquareText, RotateCcw, Send, ShieldCheck } from "lucide-react"; -import { useMemo, useState } from "react"; +import { BookOpenCheck, ClipboardList, History, Loader2, MessageSquareText, RotateCcw, Send, ShieldCheck, XCircle } from "lucide-react"; +import { useEffect, useMemo, useState } from "react"; import type { LegalQaCitation, LegalQaResponse, ModelInferenceJob } from "@/services/contracts"; -import { createLegalQaJob, getLegalQaJob, handoffLegalQaJob, retryLegalQaJob } from "@/services/contracts-client"; +import { + cancelLegalQaJob, + createLegalQaJob, + getLegalQaJob, + handoffLegalQaJob, + listLegalQaJobs, + retryLegalQaJob, +} from "@/services/contracts-client"; type LegalQaClientProps = { initialQuestion: string; @@ -47,12 +54,29 @@ export function LegalQaClient({ const [error, setError] = useState(null); const [activeJob, setActiveJob] = useState(null); const [pendingSubmission, setPendingSubmission] = useState<{ question: string; matterContext: string } | null>(null); + const [jobHistory, setJobHistory] = useState([]); + const [historyCursor, setHistoryCursor] = useState(null); + const [historyLoading, setHistoryLoading] = useState(true); const latestTurn = turns[0] ?? null; const sourceCount = useMemo( () => latestTurn ? Math.max(latestTurn.response.primary_sources.length, latestTurn.response.citations.length) : 0, [latestTurn], ); + useEffect(() => { + let active = true; + void listLegalQaJobs(null).then((page) => { + if (!active) return; + setJobHistory(page.items); + setHistoryCursor(page.next_cursor || null); + }).catch((historyError) => { + if (active) setError(historyError instanceof Error ? historyError.message : "模型任务历史加载失败。"); + }).finally(() => { + if (active) setHistoryLoading(false); + }); + return () => { active = false; }; + }, []); + async function submitQuestion(event: React.FormEvent) { event.preventDefault(); const trimmedQuestion = question.trim(); @@ -65,13 +89,15 @@ export function LegalQaClient({ try { const submission = { question: trimmedQuestion, matterContext: matterContext.trim() }; setPendingSubmission(submission); + const idempotencyKey = `legal-qa-${crypto.randomUUID()}`; const created = await createLegalQaJob({ question: trimmedQuestion, matter_context: matterContext.trim() || null, retrieval_strategy: retrievalStrategy, - }); + }, idempotencyKey); setActiveJob(created); - const completed = await pollLegalQaJob(created.id, setActiveJob); + upsertHistory(created); + const completed = await pollLegalQaJob(created.id, updateActiveJob); appendJobResult(completed, submission); } catch (requestError) { setError(requestError instanceof Error ? requestError.message : "法律问答请求失败。"); @@ -80,22 +106,32 @@ export function LegalQaClient({ } } + function updateActiveJob(job: ModelInferenceJob) { + setActiveJob(job); + upsertHistory(job); + } + + function upsertHistory(job: ModelInferenceJob) { + setJobHistory((items) => [job, ...items.filter((item) => item.id !== job.id)]); + } + function appendJobResult(job: ModelInferenceJob, submission: { question: string; matterContext: string }) { + if (job.status === "cancelled") { + return; + } if (!isLegalQaResponse(job.result)) { setError(job.error_message || "模型任务未返回可用回答,可以重试或转人工复核。"); return; } const response = { ...job.result, ai_audit_run_id: job.audit_run_id || job.result.ai_audit_run_id }; - setTurns((items) => [ - { - id: `${job.id}-${job.attempt_count}`, + const turn = { + id: `${job.id}-${job.attempt_count}`, question: submission.question, matterContext: submission.matterContext, response, createdAt: job.completed_at || new Date().toISOString(), - }, - ...items, - ]); + }; + setTurns((items) => [turn, ...items.filter((item) => item.id !== turn.id)]); } async function retryActiveJob() { @@ -104,8 +140,8 @@ export function LegalQaClient({ setError(null); try { const queued = await retryLegalQaJob(activeJob.id); - setActiveJob(queued); - const completed = await pollLegalQaJob(queued.id, setActiveJob); + updateActiveJob(queued); + const completed = await pollLegalQaJob(queued.id, updateActiveJob); appendJobResult(completed, pendingSubmission); } catch (requestError) { setError(requestError instanceof Error ? requestError.message : "模型任务重试失败。"); @@ -114,6 +150,40 @@ export function LegalQaClient({ } } + async function cancelActiveJob() { + if (!activeJob) return; + setLoading(true); + setError(null); + try { + updateActiveJob(await cancelLegalQaJob(activeJob.id)); + } catch (requestError) { + setError(requestError instanceof Error ? requestError.message : "取消模型任务失败。"); + } finally { + setLoading(false); + } + } + + async function loadMoreHistory() { + if (!historyCursor || historyLoading) return; + setHistoryLoading(true); + try { + const page = await listLegalQaJobs(historyCursor); + setJobHistory((items) => [...items, ...page.items.filter((job) => !items.some((item) => item.id === job.id))]); + setHistoryCursor(page.next_cursor || null); + } catch (historyError) { + setError(historyError instanceof Error ? historyError.message : "更多模型任务加载失败。"); + } finally { + setHistoryLoading(false); + } + } + + function openHistoryJob(job: ModelInferenceJob) { + setActiveJob(job); + if (isLegalQaResponse(job.result)) { + appendJobResult(job, { question: job.display_title || "历史法律问答", matterContext: "" }); + } + } + async function handoffActiveJob() { if (!activeJob) return; setLoading(true); @@ -210,6 +280,12 @@ export function LegalQaClient({ {activeJob.status === "running" && (

任务已在后台执行,可以停留在本页查看进度;页面不再等待单次同步请求。

)} + {activeJob.quality_status !== "pending" && activeJob.quality_status !== "unknown" && ( +

+ 质量门禁:{qualityStatusLabel(activeJob.quality_status)} · {activeJob.quality_score} 分 + {activeJob.quality_summary ? ` · ${activeJob.quality_summary}` : ""} +

+ )} {activeJob.status === "fallback" && (

当前已展示规则 fallback 结果,并明确标记为非模型成功;可重试模型或转人工复核。

)} @@ -228,6 +304,13 @@ export function LegalQaClient({ )} + {(["queued", "running"] as ModelInferenceJob["status"][]).includes(activeJob.status) && ( +
+ +
+ )} {activeJob.status === "awaiting_human" && (

已进入人工复核队列{activeJob.handoff_assignee ? `,负责人:${activeJob.handoff_assignee}` : ""}。

)} @@ -249,6 +332,31 @@ export function LegalQaClient({