"""兴邦预审 P0 模拟闭环演示脚本（T11）。 依据：docs/v2/08-data-pilot-and-demo.md §7 的八步演示顺序。 **诚实性要求**（08 §7 末、06 §6）：脚本只驱动本仓库已经实现并验证过的入口； 没有实现的步骤打印"未实现"及其原因，不伪造屏幕状态；模型效果与工程演示 分列，模拟标签不得包装成真实业务验证。 用法：backend/venv/Scripts/python.exe scripts/xingbang_prescreen_demo.py 全程在临时目录内使用临时数据库，不触碰真实 xingbang_data。 """ from __future__ import annotations import sys import tempfile from pathlib import Path ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT / "backend")) from contracts.prescreen import ErrorCode # noqa: E402 from services import (assistant_store as store, improvements, judgment_adapters, # noqa: E402 judgment_provider as jp, prescreen, prescreen_policy, prescreen_sandbox, prescreen_store, review_plan, trajectory) DONE = "已执行" NOT_DONE = "未实现" MODEL_UNVERIFIED = "未验证（无权重与授权）" FINDINGS = [ {"code": "corporate.cashflow", "result": "unknown", "stage": 1, "evidence_ids": ["EV-1"]}, {"code": "corporate.title", "result": "unknown", "stage": 1, "evidence_ids": ["EV-2"]}, {"code": "corporate.deal", "result": "checked", "stage": 2, "evidence_ids": ["EV-3"]}, ] DIMENSIONS = [ {"id": "DC-CASH", "component_refs": [{"id": "corporate.cashflow"}]}, {"id": "DC-TITLE", "component_refs": [{"id": "corporate.title"}]}, {"id": "DC-DEAL", "component_refs": [{"id": "corporate.deal"}]}, ] PLAN = { "plan": {"id": "corporate.default", "version": "1.0.0"}, "skills": [{"id": "corporate.cashflow", "version": "1.0.0"}], "rules": [], "template": {"id": "template.ai-pre-review", "version": "1.0.0"}, "content_hash": "demo-hash", "dimensions": DIMENSIONS, } STEPS = [ (1, "3,000 条候选清单、生成版本与多结果分布", NOT_DONE, "T04 的服务已在单测中验证，但本脚本不重复驱动批量生成；清单需真实数据授权"), (2, "选开发集案例，展开问询/补件/会议事件，提炼 Skill 与 advisory Rule", NOT_DONE, "历史提炼走 T05 的治理入口，需要真实历史材料；此处不伪造提炼结果"), (3, "自然语言改适用范围 + 人工新增卡片，统一审核发布", NOT_DONE, "卡片治理入口属 T05；演示脚本不代替审核动作"), (4, "数据卡 + Skills + Rules 组方案，维度随方案变化，Tools 只在依据中展示", DONE, None), (5, "缺证 → 主审核查 → 问题与补件 → 演示概率来源", DONE, None), (6, "勾选动作比较情景、撤销恢复基线", DONE, None), (7, "回放完整 trajectory，对齐模拟会议记录，查看材料版本差异", DONE, None), (8, "从漏检提出改进候选，隔离样本对照后人工审核发布，旧版仍可回放", DONE, None), ] def run_demo() -> dict: facts: dict = {"steps": [], "sandbox": {}, "trajectory": {}, "improvement": {}} with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmp: db_path = Path(tmp) / "demo.sqlite3" store.initialize(db_path) # 步骤 4：方案维度（去重与刻度判定） dimensions = review_plan.resolve_dimensions([ {"id": "DC-CASH", "label": "回款核验", "scale_id": "evidence-stage", "scale_version": "1.0.0", "order": 1, "component_refs": [{"id": "corporate.cashflow", "version": "1.0.0"}], "levels": [{"value": i, "label": name, "criterion": f"证据阶段：{name}"} for i, name in enumerate(("未提供", "仅有说明", "资料待核", "交叉核验", "人工复核"))]}, {"id": "DC-TITLE", "label": "权属核验", "scale_id": "evidence-stage", "scale_version": "1.0.0", "order": 2, "component_refs": [{"id": "corporate.title", "version": "1.0.0"}], "levels": [{"value": i, "label": name, "criterion": f"证据阶段：{name}"} for i, name in enumerate(("未提供", "仅有说明", "资料待核", "交叉核验", "人工复核"))]}, {"id": "DC-DEAL", "label": "交易要素", "scale_id": "evidence-stage", "scale_version": "1.0.0", "order": 3, "component_refs": [{"id": "corporate.deal", "version": "1.0.0"}], "levels": [{"value": i, "label": name, "criterion": f"证据阶段：{name}"} for i, name in enumerate(("未提供", "仅有说明", "资料待核", "交叉核验", "人工复核"))]}, ]) presentation = review_plan.presentation(dimensions) facts["presentation"] = presentation["layout"] # 步骤 5：预审执行（专项关闭，无校准概率） config = jp.resolve_config(False) decision = prescreen_policy.decide(FINDINGS) decision["findings"] = FINDINGS decision["missing_inputs"] = ["近六个月银行流水", "设备权属材料"] screening = prescreen.assemble( {"id": "run-demo", "project_id": "DEMO-026", "revision": 1, "created_at": store.now(), "input_snapshot_id": "snap-demo"}, PLAN, decision, config) with store.transaction(db_path) as db: prescreen_store.save(db, screening) facts["recommendation"] = screening["recommendation"] facts["probability_is_null"] = screening["probability"] is None # 步骤 6：沙盘（演示值只在演示域、组合查表） with store.transaction(db_path) as db: stored = prescreen_store.load(db, "DEMO-026", "run-demo") baseline = prescreen_sandbox.simulate(stored, []) both = prescreen_sandbox.simulate(stored, ["assume_bank_flow", "assume_title"]) facts["sandbox"] = { "baseline": baseline["probability_view"]["demo_estimate"]["value"], "combined": both["probability_view"]["demo_estimate"]["value"], "label": baseline["probability_view"]["demo_estimate"]["label"], "unchanged_dimensions": [item["id"] for item in both["dimensions"] if not item["assumed"]], } # 步骤 7：轨迹回放与模拟会议结果 with store.transaction(db_path) as db: store.put(db, "run", {"id": "run-demo", "project_id": "DEMO-026", "status": "completed", "input_snapshot_id": "snap-demo", "created_at": store.now(), "steps": [ {"run_id": "run-demo", "step_id": "corporate.cashflow", "skill": {"id": "corporate.cashflow", "version": "1.0.0"}, "status": "completed", "input_snapshot_id": "snap-demo", "started_at": store.now(), "outputs": {"note": "已核查"}, "findings": [{"code": "cashflow", "result": "unknown", "evidence_ids": ["EV-1"]}], "errors": [], "missing_inputs": []}, {"run_id": "run-demo", "step_id": "corporate.title", "skill": {"id": "corporate.title", "version": "1.0.0"}, "status": "failed", "input_snapshot_id": "snap-demo", "started_at": store.now(), "outputs": {}, "findings": [], "missing_inputs": [], "errors": [{"code": "SKILL_TIMEOUT", "message": "演示中的失败步骤"}]}]}) replay = trajectory.replay(db, "DEMO-026", "run-demo") trajectory.record(db, "DEMO-026", {"raw_code": "附条件通过", "meeting_id": "M-DEMO", "occurred_at": store.now()}, "demo-reviewer", known_snapshots=["snap-demo"]) outcomes = trajectory.listing(db, "DEMO-026") facts["trajectory"] = { "steps": [item["status"] for item in replay["steps"]], "failed_preserved": any(item["status"] == "failed" for item in replay["steps"]), "unmatched": len(outcomes["unmatched"]), "observed_denominator": outcomes["counts_toward_denominator"]["observed"], } # 步骤 8：改进候选（合成样本、独立评测、可回放旧版） with store.transaction(db_path) as db: candidate = improvements.propose(db, "DEMO-026", { "target": "knowledge", "target_ref": {"id": "card-demo", "version": "1.0.0"}, "proposed_revision_ref": {"id": "card-demo", "version": "1.1.0"}, "change_summary": "把回款核验经验扩到设备直租", "source_run_ids": ["run-demo"]}, "demo-manager", "demo") evaluation = {"baseline_ref": {"id": "current", "version": "1.0.0"}, "candidate_ref": {"id": "prescreen.v2", "version": "0.2.0"}, "dataset_ref": "demo/synthetic-manifest.json", "split_ref": "demo-split.v1", "independent": True, "samples": [ {"id": "S-1", "split": "dev", "observed": True, "baseline_pass": True, "candidate_pass": True, "predicted_at": "2026-09-29T01:00:00+00:00", "label_revealed_at": "2026-09-29T03:00:00+00:00"}, {"id": "S-2", "split": "dev", "observed": True, "baseline_pass": False, "candidate_pass": True, "predicted_at": "2026-09-29T01:00:00+00:00", "label_revealed_at": "2026-09-29T03:00:00+00:00"}]} evaluated = improvements.transition(db, candidate["id"], "evaluate", "demo-reviewer", evaluation=evaluation) improvements.transition(db, evaluated["id"], "submit", "demo-reviewer") approved = improvements.transition(db, evaluated["id"], "approve", "demo-reviewer") released = improvements.transition( db, approved["id"], "release", "demo-reviewer", publisher=lambda body: {"id": "card-demo", "version": "1.1.0"}) facts["improvement"] = {"verdict": evaluated["contrast"]["verdict"], "status": released["status"], "released_ref": released["released_ref"]} # 专项：离线 fixture 可跑，live 未接通 facts["judgment_offline"] = judgment_adapters.load_fixture( "question_bank_extraction.json")["rubric_version"] facts["demo_table_version"] = prescreen_sandbox.DEMO_TABLE_VERSION facts["budget_code"] = ErrorCode.JUDGMENT_CONTEXT_EXCEEDED.value return facts def main() -> int: facts = run_demo() print("兴邦预审 P0 模拟闭环演示（合成数据 · 不连接真实数据与模型）") print("=" * 78) for index, title, status, reason in STEPS: print(f"{index}. [{status}] {title}") if reason: print(f" 原因：{reason}") print("-" * 78) print("工程演示（本仓库已实现并验证的入口）") print(f" · 维度呈现：{facts['presentation']}") print(f" · 主审意见：{facts['recommendation']} · 真实概率为空：{facts['probability_is_null']}") sandbox = facts["sandbox"] print(f" · 沙盘：基线 {sandbox['baseline']:.0%} → 组合 {sandbox['combined']:.0%}" f"（{sandbox['label']}）") print(f" · 未变化的维度仍列出：{sandbox['unchanged_dimensions']}") trace = facts["trajectory"] print(f" · 轨迹步骤状态：{trace['steps']} · 失败步骤保留：{trace['failed_preserved']}") print(f" · 会议结果：unmatched {trace['unmatched']} 条 · 已观察分母 {trace['observed_denominator']}") imp = facts["improvement"] print(f" · 改进对照：{imp['verdict']} → {imp['status']}（{imp['released_ref']}）") print("-" * 78) print("模型效果（与上面的工程演示分开计）") print(f" · 本地专项：{MODEL_UNVERIFIED}；当前只有离线 fixture（{facts['judgment_offline']}）") print(f" · Jev：未接通（无端点与凭据）") print(f" · 真实审批概率：未校准，恒为空；界面上的百分数来自演示表 " f"{facts['demo_table_version']}，不是模型输出") print(f" · 超预算时的拒绝码（{facts['budget_code']}）证明预算守卫生效，与概率无关") print("=" * 78) print("以上均为合成数据与演示值，不能作为真实业务效果或审批依据。") return 0 if __name__ == "__main__": raise SystemExit(main())