199 lines
9.7 KiB
Python
199 lines
9.7 KiB
Python
"""Build and validate the immutable W4 multiscene evaluation plan."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from hashlib import sha256
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from yuqei_contract_api.contract_templates import template_document_kind
|
|
from yuqei_contract_api.drafting_workspace import (
|
|
DraftingCandidateRequest, DraftingCreateRequest, DraftingEvaluationCreateRequest,
|
|
DraftingWorkspaceStore,
|
|
)
|
|
from yuqei_contract_api.in_memory_repository import InMemoryContractRepository
|
|
from yuqei_sdk.contract_drafting import ContractDraftRequest, draft_digest
|
|
|
|
from multiscene_eval_cases import build_cases
|
|
from multiscene_supplemental_cases import build_supplemental_cases
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SCHEMA_VERSION = "multiscene-evaluation-plan.v1"
|
|
EXPECTED_OPERATIONS = {
|
|
"draft": 120, "semantic_review": 30, "repair_draft": 8,
|
|
"repair_review": 8, "specialist_review": 106, "intake": 8,
|
|
}
|
|
EXPECTED_BUDGET = 280
|
|
PIN_ROOTS = (
|
|
"yuqei-ai-sdk-python/src/yuqei_sdk",
|
|
"yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api",
|
|
"yuqei-contract-platform/services/contract-api/src/yuqei_contract_api",
|
|
)
|
|
PIN_FILES = (
|
|
"yuqei-infra/scripts/multiscene_eval_cases.py",
|
|
"yuqei-infra/scripts/multiscene_supplemental_cases.py",
|
|
"yuqei-infra/scripts/multiscene_call_ledger.py",
|
|
"yuqei-infra/scripts/multiscene_eval_plan.py",
|
|
"yuqei-infra/scripts/multiscene_eval_runner.py",
|
|
"yuqei-infra/scripts/run-core-drafting-smoke.py",
|
|
"yuqei-contract-platform/services/contract-api/tests/test_semantic_review_cross_service_flow.py",
|
|
"yuqei-ai-sdk-python/pyproject.toml",
|
|
"yuqei-ai-platform/services/ai-platform-api/pyproject.toml",
|
|
"yuqei-contract-platform/services/contract-api/pyproject.toml",
|
|
)
|
|
PIN_SUFFIXES = {".py", ".json", ".md", ".toml", ".yaml", ".yml"}
|
|
|
|
|
|
def _hash_bytes(value: bytes) -> str:
|
|
return sha256(value).hexdigest()
|
|
|
|
|
|
def _canonical(value: Any) -> str:
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True,
|
|
separators=(",", ":"), allow_nan=False)
|
|
|
|
|
|
def draft_request_projection(request: ContractDraftRequest | dict[str, Any]) -> dict[str, Any]:
|
|
"""Return stable model facts and template identities; omit workspace/lifecycle data."""
|
|
request = ContractDraftRequest.model_validate(request)
|
|
data = request.model_dump(mode="json")
|
|
snapshot = data["template_snapshot"]
|
|
return {
|
|
"document_kind": data["document_kind"], "title": data["contract_title"],
|
|
"contract_type": data["contract_type"], "industry": data["industry"],
|
|
"party_position": data["party_position"], "jurisdiction": data["jurisdiction"],
|
|
"transaction_stage": data["transaction_stage"], "facts": data["facts"],
|
|
"missing_facts": data["missing_facts"], "base_document": data["base_document"],
|
|
"instructions": data["drafting_instructions"],
|
|
"template_content_sha256": snapshot.get("content_sha256"),
|
|
"template_schema_sha256": snapshot.get("schema_sha256"),
|
|
}
|
|
|
|
|
|
def _source_pins(runner) -> dict[str, str]:
|
|
paths: set[Path] = set()
|
|
for relative in PIN_ROOTS:
|
|
root = ROOT / relative
|
|
paths.update(path for path in root.rglob("*") if path.is_file()
|
|
and path.suffix.lower() in PIN_SUFFIXES
|
|
and "__pycache__" not in path.parts)
|
|
paths.update(ROOT / relative for relative in PIN_FILES)
|
|
runner_file = Path(getattr(runner, "__file__", ""))
|
|
if runner_file.is_file():
|
|
paths.add(runner_file.resolve())
|
|
cross = getattr(runner, "CROSS", None)
|
|
cross_file = Path(getattr(cross, "__file__", ""))
|
|
if cross_file.is_file():
|
|
paths.add(cross_file.resolve())
|
|
missing = [str(path.relative_to(ROOT)) for path in paths if not path.is_file()]
|
|
if missing:
|
|
raise ValueError(f"source pin file missing: {missing}")
|
|
return {path.relative_to(ROOT).as_posix(): _hash_bytes(path.read_bytes())
|
|
for path in sorted(paths)}
|
|
|
|
|
|
def _prepared_projection(case: dict[str, Any], repository: InMemoryContractRepository) -> tuple[dict[str, Any], Any]:
|
|
# Base cases need the real contract policy (including auto repair); scene
|
|
# release-gate cases must remain in the isolated evaluation entry point.
|
|
evaluation = case["mode"] == "scene_evaluation"
|
|
store = DraftingWorkspaceStore(repository, mode="scene_evaluation" if evaluation else "contract")
|
|
template, _ = (store.evaluation_source(case["template_id"]) if evaluation
|
|
else repository.get_template_source(case["template_id"]))
|
|
request_type = DraftingEvaluationCreateRequest if evaluation else DraftingCreateRequest
|
|
context = case.get("scene_request_context", {})
|
|
created = request_type(
|
|
template_id=case["template_id"], expected_template_content_sha256=template.content_sha256,
|
|
title=case["title"], values=case["values"], industry=context.get("industry", "未指定"),
|
|
party_position=context.get("party_position", "中立"),
|
|
jurisdiction=context.get("jurisdiction", "中国大陆"),
|
|
transaction_stage=context.get("transaction_stage", "drafting"),
|
|
instructions=case["instructions"], **({"synthetic_only": True} if evaluation else {}),
|
|
)
|
|
record = store.create(created, tenant="w4-isolated", owner="w4-builder",
|
|
key=f"create-{case['case_id']}")
|
|
policy = {"track_changes": True, "semantic_review": not evaluation,
|
|
"auto_repair": case["auto_repair"]}
|
|
_, candidate = store.prepare(record.id, DraftingCandidateRequest(
|
|
expected_revision=record.revision, idempotency_key=f"candidate-{draft_digest(case['case_id'])}",
|
|
**policy), tenant="w4-isolated", owner="w4-builder")
|
|
projection = draft_request_projection(candidate.model_request)
|
|
if projection["document_kind"] != case["document_kind"]:
|
|
raise ValueError(f"document kind drift: {case['case_id']}")
|
|
actual_missing = [item["field"] for item in projection["missing_facts"]]
|
|
expected_missing = case["expectation_profile"]["expected_missing_fields"]
|
|
actual_status = "incomplete" if actual_missing else "candidate_ready"
|
|
if actual_missing != expected_missing or actual_status != case["expectation_profile"]["expected_status"]:
|
|
raise ValueError(f"prepared missing/status drift: {case['case_id']}")
|
|
return projection, template
|
|
|
|
|
|
def _normalized_cases() -> list[dict[str, Any]]:
|
|
repository = InMemoryContractRepository(contracts=[])
|
|
repository.allow_internal_trial_templates = True
|
|
drafting = []
|
|
for case in build_cases():
|
|
projection, template = _prepared_projection(case, repository)
|
|
policy = {"track_changes": True, "semantic_review": case["mode"] == "contract",
|
|
"auto_repair": case["auto_repair"]}
|
|
drafting.append({**case, "candidate_policy": policy,
|
|
"template_content_sha256": projection["template_content_sha256"],
|
|
"template_input_schema_sha256": projection["template_schema_sha256"],
|
|
"values_sha256": draft_digest(case["values"]),
|
|
"draft_request_projection": projection})
|
|
supplemental = []
|
|
for case in build_supplemental_cases():
|
|
supplemental.append({
|
|
"case_id": case["case_id"], "source_case_id": case["case_id"],
|
|
"kind": case["operation"], "mode": case["operation"], "operation": case["operation"],
|
|
"operations": [case["operation"]], "provider_call_budget": 1,
|
|
"request": case["request"].model_dump(mode="json"),
|
|
"request_sha256": case["request_sha256"], "expectations": case["expectations"],
|
|
**({"module": case["module"], "party_position": case["party_position"]}
|
|
if case["operation"] == "specialist_review" else {}),
|
|
})
|
|
return [*drafting, *supplemental]
|
|
|
|
|
|
def _operation_counts(cases: list[dict[str, Any]]) -> dict[str, int]:
|
|
result: dict[str, int] = {}
|
|
for case in cases:
|
|
for operation in case["operations"]:
|
|
result[operation] = result.get(operation, 0) + 1
|
|
return result
|
|
|
|
|
|
def build_plan(runner) -> dict[str, Any]:
|
|
cases = _normalized_cases()
|
|
counts = _operation_counts(cases)
|
|
if len(cases) != 144 or counts != EXPECTED_OPERATIONS:
|
|
raise ValueError(f"case or operation inventory mismatch: {len(cases)}, {counts}")
|
|
if sum(case["provider_call_budget"] for case in cases) != EXPECTED_BUDGET:
|
|
raise ValueError("global provider call budget mismatch")
|
|
return {
|
|
"schema_version": SCHEMA_VERSION, "prepared_at": datetime.now(UTC).isoformat(),
|
|
"model": "gpt-5.6-sol", "reasoning_effort": "medium", "timeout_seconds": 300,
|
|
"case_count": 144, "drafting_case_count": 120, "supplemental_case_count": 24,
|
|
"operation_limits": counts, "total_provider_call_budget": EXPECTED_BUDGET,
|
|
"source_pins": _source_pins(runner), "cases": cases,
|
|
"limitations": ["合成隔离评测，不是生产发布或法律批准。",
|
|
"deterministic projection equality does not establish semantic or legal correctness."],
|
|
}
|
|
|
|
|
|
def validate_plan(runner, path: Path, expected_sha: str | None = None) -> tuple[dict[str, Any], str]:
|
|
raw = Path(path).read_bytes()
|
|
actual = _hash_bytes(raw)
|
|
if expected_sha is not None and actual != expected_sha:
|
|
raise ValueError("plan raw SHA-256 mismatch")
|
|
plan = json.loads(raw)
|
|
expected = build_plan(runner)
|
|
if plan.get("schema_version") != SCHEMA_VERSION:
|
|
raise ValueError("unsupported multiscene plan")
|
|
left = {key: value for key, value in plan.items() if key != "prepared_at"}
|
|
right = {key: value for key, value in expected.items() if key != "prepared_at"}
|
|
if _canonical(left) != _canonical(right):
|
|
raise ValueError("multiscene plan content or source pins drift")
|
|
return plan, actual
|