feat: add output review kit

This commit is contained in:
yaojingang
2026-06-14 06:56:44 +08:00
parent ecdb16dd90
commit 0cf7f32e32
53 changed files with 1787 additions and 352 deletions
+1
View File
@@ -245,6 +245,7 @@ def build_reviewer_checklist(
"blind_pack_path": display_path(blind_pack_path),
"decisions_path": display_path(decisions_path),
"commands": {
"prepare_review_kit": "python3 scripts/yao.py output-review-kit",
"write_template": "python3 scripts/adjudicate_output_review.py --write-template",
"adjudicate": "python3 scripts/yao.py output-review",
"refresh_review_studio": "python3 scripts/yao.py review-studio .",
+1
View File
@@ -40,6 +40,7 @@ DEFAULT_TARGETS = [
"compiler-check",
"output-eval-check",
"output-execution-check",
"output-review-kit-check",
"output-review-adjudication-check",
"runtime-conformance-check",
"runtime-permission-check",
+316
View File
@@ -0,0 +1,316 @@
#!/usr/bin/env python3
"""Prepare a reviewer-facing blind A/B output review kit."""
import argparse
import json
from pathlib import Path
from typing import Any
from adjudicate_output_review import (
build_decision_template,
confidence_value,
decision_index,
display_path,
load_json,
normalize_variant,
)
ROOT = Path(__file__).resolve().parent.parent
DEFAULT_BLIND_PACK_JSON = ROOT / "reports" / "output_blind_review_pack.json"
DEFAULT_BLIND_PACK_MD = ROOT / "reports" / "output_blind_review_pack.md"
DEFAULT_DECISIONS = ROOT / "reports" / "output_review_decisions.json"
DEFAULT_OUTPUT_JSON = ROOT / "reports" / "output_review_kit.json"
DEFAULT_OUTPUT_MD = ROOT / "reports" / "output_review_kit.md"
def load_optional_decisions(path: Path) -> tuple[dict[str, Any], list[str]]:
if not path.exists():
return {"schema_version": "1.0", "decisions": []}, []
return load_json(path)
def pairs_from_pack(blind_pack: dict[str, Any]) -> list[dict[str, Any]]:
pairs = blind_pack.get("pairs", [])
if not isinstance(pairs, list):
return []
return [item for item in pairs if isinstance(item, dict) and item.get("case_id")]
def decision_state(decision: dict[str, Any] | None) -> dict[str, Any]:
if not decision:
return {
"status": "awaiting-decision",
"winner_variant_recorded": False,
"confidence_recorded": False,
"reason_recorded": False,
"blocking_reason": "No reviewer choice has been recorded yet.",
}
winner = normalize_variant(decision.get("winner_variant", decision.get("reviewer_winner_variant", decision.get("winner", ""))))
confidence, confidence_failure = confidence_value(decision.get("confidence"))
reason = str(decision.get("reason", "")).strip()
if not winner and confidence is None and not reason:
return {
"status": "awaiting-decision",
"winner_variant_recorded": False,
"confidence_recorded": False,
"reason_recorded": False,
"blocking_reason": "Decision template exists but this row is still blank.",
}
failures = []
if winner not in {"A", "B"}:
failures.append("winner_variant must be A or B")
if confidence_failure:
failures.append(confidence_failure)
if not reason:
failures.append("reason should explain the reviewer rationale")
if failures:
return {
"status": "needs-fix",
"winner_variant_recorded": winner in {"A", "B"},
"confidence_recorded": confidence is not None,
"reason_recorded": bool(reason),
"blocking_reason": "; ".join(failures),
}
return {
"status": "ready-for-adjudication",
"winner_variant_recorded": True,
"confidence_recorded": confidence is not None,
"reason_recorded": True,
"blocking_reason": "Reviewer choice is complete enough for adjudication.",
}
def build_cases(blind_pack: dict[str, Any], decisions: dict[str, dict[str, Any]]) -> list[dict[str, Any]]:
cases = []
for pair in pairs_from_pack(blind_pack):
case_id = str(pair.get("case_id", ""))
rubric = []
for item in pair.get("rubric", []) if isinstance(pair.get("rubric", []), list) else []:
if isinstance(item, dict):
rubric.append(
{
"id": str(item.get("id", "")),
"description": str(item.get("description", "")),
"weight": float(item.get("weight", 1) or 0),
}
)
cases.append(
{
"case_id": case_id,
"prompt": str(pair.get("prompt", "")),
"input_files": pair.get("input_files", []) if isinstance(pair.get("input_files", []), list) else [],
"review_instruction": str(pair.get("review_instruction", "")),
"rubric": rubric,
"variant_a": {
"blind_id": str(pair.get("variant_a", {}).get("blind_id", "")) if isinstance(pair.get("variant_a"), dict) else "",
"output": str(pair.get("variant_a", {}).get("output", "")) if isinstance(pair.get("variant_a"), dict) else "",
},
"variant_b": {
"blind_id": str(pair.get("variant_b", {}).get("blind_id", "")) if isinstance(pair.get("variant_b"), dict) else "",
"output": str(pair.get("variant_b", {}).get("output", "")) if isinstance(pair.get("variant_b"), dict) else "",
},
"decision_state": decision_state(decisions.get(case_id)),
}
)
return cases
def build_summary(cases: list[dict[str, Any]], failures: list[str], reviewer: str, reviewed_at: str) -> dict[str, Any]:
status_counts: dict[str, int] = {}
for case in cases:
status = str(case.get("decision_state", {}).get("status", "awaiting-decision"))
status_counts[status] = status_counts.get(status, 0) + 1
ready_count = status_counts.get("ready-for-adjudication", 0)
pending_count = status_counts.get("awaiting-decision", 0)
invalid_count = status_counts.get("needs-fix", 0)
return {
"case_count": len(cases),
"ready_for_adjudication_count": ready_count,
"pending_decision_count": pending_count,
"invalid_decision_count": invalid_count,
"reviewer_metadata_present": bool(str(reviewer).strip() and str(reviewed_at).strip()),
"answer_key_hidden": True,
"answer_key_path_exposed": False,
"ready_to_run_adjudication": bool(cases) and ready_count == len(cases) and invalid_count == 0,
"failure_count": len(failures),
}
def build_contract(blind_pack_md: Path, decisions_path: Path) -> dict[str, Any]:
return {
"reviewer_steps": [
f"Open {display_path(blind_pack_md)} or this kit and compare Variant A vs Variant B for each case.",
f"Record choices in {display_path(decisions_path)} without opening the answer key.",
"Use winner_variant A or B, confidence from 0 to 1, and a short reason for every case.",
"Run python3 scripts/yao.py output-review after choices are recorded.",
"Refresh python3 scripts/yao.py review-studio . before asking for release approval.",
],
"required_fields": {
"reviewer": "Human reviewer name or review group.",
"reviewed_at": "Review date or timestamp.",
"winner_variant": "A or B for every case.",
"confidence": "Optional numeric confidence from 0 to 1.",
"reason": "Short rationale based on the rubric, not on hidden labels.",
},
"privacy_contract": [
"The answer key is intentionally withheld from this kit.",
"Do not inspect hidden labels or expected winners before decisions are recorded.",
"Do not paste private user data into decision reasons.",
"Pending decisions must stay pending instead of being counted as human agreement.",
],
}
def render_rubric(rubric: list[dict[str, Any]]) -> list[str]:
if not rubric:
return ["- No rubric items found."]
return [f"- `{item['id']}` ({item['weight']}): {item['description']}" for item in rubric]
def render_markdown(payload: dict[str, Any]) -> str:
summary = payload["summary"]
contract = payload["review_contract"]
lines = [
"# Output Review Kit",
"",
"This reviewer-facing packet contains the blind A/B cases, decision fields, and command flow. It intentionally does not expose the answer key.",
"",
"## Summary",
"",
f"- cases: `{summary['case_count']}`",
f"- ready for adjudication: `{summary['ready_for_adjudication_count']}`",
f"- pending decisions: `{summary['pending_decision_count']}`",
f"- invalid decisions: `{summary['invalid_decision_count']}`",
f"- reviewer metadata present: `{str(summary['reviewer_metadata_present']).lower()}`",
f"- answer key hidden: `{str(summary['answer_key_hidden']).lower()}`",
f"- answer key path exposed: `{str(summary['answer_key_path_exposed']).lower()}`",
"",
"## Review Flow",
"",
]
lines.extend(f"{index}. {step}" for index, step in enumerate(contract["reviewer_steps"], start=1))
lines.extend(["", "## Required Fields", ""])
lines.extend(f"- `{key}`: {value}" for key, value in contract["required_fields"].items())
lines.extend(["", "## Privacy Contract", ""])
lines.extend(f"- {item}" for item in contract["privacy_contract"])
lines.extend(
[
"",
"## Decision States",
"",
"| Case | State | Winner | Confidence | Reason | Blocking Reason |",
"| --- | --- | --- | --- | --- | --- |",
]
)
for case in payload["cases"]:
state = case["decision_state"]
lines.append(
f"| `{case['case_id']}` | `{state['status']}` | `{str(state['winner_variant_recorded']).lower()}` | "
f"`{str(state['confidence_recorded']).lower()}` | `{str(state['reason_recorded']).lower()}` | {state['blocking_reason']} |"
)
lines.extend(["", "## Blind Cases", ""])
for case in payload["cases"]:
lines.extend(
[
f"### {case['case_id']}",
"",
f"Prompt: {case['prompt']}",
"",
"Rubric:",
*render_rubric(case["rubric"]),
"",
"#### Variant A",
"",
case["variant_a"]["output"] or "_No output found._",
"",
"#### Variant B",
"",
case["variant_b"]["output"] or "_No output found._",
"",
]
)
if payload.get("failures"):
lines.extend(["## Failures", ""])
lines.extend(f"- {failure}" for failure in payload["failures"])
return "\n".join(lines).strip() + "\n"
def prepare_output_review_kit(
blind_pack_json: Path,
blind_pack_md: Path,
decisions_path: Path,
output_json: Path,
output_md: Path,
write_template: bool = False,
) -> dict[str, Any]:
blind_pack, failures = load_json(blind_pack_json)
decisions_payload, decision_failures = load_optional_decisions(decisions_path)
failures.extend(decision_failures)
template_written = False
if write_template and blind_pack and not decisions_path.exists():
decisions_path.parent.mkdir(parents=True, exist_ok=True)
decisions_path.write_text(
json.dumps(build_decision_template(blind_pack), ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
template_written = True
decisions_payload, decision_failures = load_optional_decisions(decisions_path)
failures.extend(decision_failures)
decisions, index_failures = decision_index(decisions_payload)
failures.extend(index_failures)
case_ids = {str(pair.get("case_id", "")) for pair in pairs_from_pack(blind_pack)}
for case_id in decisions:
if case_id not in case_ids:
failures.append(f"decision references unknown case_id: {case_id}")
cases = build_cases(blind_pack, decisions)
summary = build_summary(cases, failures, str(decisions_payload.get("reviewer", "")), str(decisions_payload.get("reviewed_at", "")))
payload = {
"schema_version": "1.0",
"ok": not failures,
"summary": summary,
"artifacts": {
"reviewer_kit_json": display_path(output_json),
"reviewer_kit_markdown": display_path(output_md),
"blind_pack_json": display_path(blind_pack_json),
"blind_pack_markdown": display_path(blind_pack_md),
"decisions": display_path(decisions_path),
"adjudication_json": "reports/output_review_adjudication.json",
"adjudication_markdown": "reports/output_review_adjudication.md",
},
"template_written": template_written,
"review_contract": build_contract(blind_pack_md, decisions_path),
"cases": cases,
"failures": failures,
}
output_json.parent.mkdir(parents=True, exist_ok=True)
output_md.parent.mkdir(parents=True, exist_ok=True)
output_json.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
output_md.write_text(render_markdown(payload), encoding="utf-8")
return payload
def main() -> None:
parser = argparse.ArgumentParser(description="Prepare a reviewer-facing blind A/B output review kit.")
parser.add_argument("--blind-pack-json", default=str(DEFAULT_BLIND_PACK_JSON))
parser.add_argument("--blind-pack-md", default=str(DEFAULT_BLIND_PACK_MD))
parser.add_argument("--decisions", default=str(DEFAULT_DECISIONS))
parser.add_argument("--output-json", default=str(DEFAULT_OUTPUT_JSON))
parser.add_argument("--output-md", default=str(DEFAULT_OUTPUT_MD))
parser.add_argument("--write-template", action="store_true")
args = parser.parse_args()
payload = prepare_output_review_kit(
Path(args.blind_pack_json).resolve(),
Path(args.blind_pack_md).resolve(),
Path(args.decisions).resolve(),
Path(args.output_json).resolve(),
Path(args.output_md).resolve(),
write_template=args.write_template,
)
print(json.dumps(payload, ensure_ascii=False, indent=2))
raise SystemExit(0 if payload["ok"] else 2)
if __name__ == "__main__":
main()
+3 -1
View File
@@ -41,8 +41,9 @@ TASK_TEMPLATES: dict[str, dict[str, Any]] = {
"owner": "human reviewer",
"objective": "Record real blind A/B reviewer decisions before claiming human output review completion.",
"runbook": [
"python3 scripts/yao.py output-review-kit --write-template",
"Open reports/output_review_kit.md and choose A or B for each pair without opening the answer key.",
"python3 scripts/adjudicate_output_review.py --write-template",
"Open reports/output_blind_review_pack.md and choose A or B for each pair without opening the answer key.",
"Edit reports/output_review_decisions.json with winner_variant values and reviewer metadata.",
"python3 scripts/yao.py output-review",
"python3 scripts/yao.py skill-os2-audit . --generated-at <YYYY-MM-DD>",
@@ -55,6 +56,7 @@ TASK_TEMPLATES: dict[str, dict[str, Any]] = {
],
"evidence_artifacts": [
"reports/output_blind_review_pack.md",
"reports/output_review_kit.md",
"reports/output_review_decisions.json",
"reports/output_review_adjudication.json",
"reports/output_review_adjudication.md",
+8
View File
@@ -63,6 +63,7 @@ def evidence_paths(skill_dir: Path) -> dict[str, str]:
"output_eval": "reports/output_quality_scorecard.md",
"output_execution": "reports/output_execution_runs.md",
"output_blind_review": "reports/output_blind_review_pack.md",
"output_review_kit": "reports/output_review_kit.md",
"output_review_decisions": "reports/output_review_decisions.json",
"output_review_adjudication": "reports/output_review_adjudication.md",
"benchmark_reproducibility": "reports/benchmark_reproducibility.md",
@@ -103,6 +104,7 @@ def load_review_data(skill_dir: Path) -> dict[str, dict[str, Any]]:
"output_quality": load_json(reports / "output_quality_scorecard.json"),
"output_execution": load_json(reports / "output_execution_runs.json"),
"output_blind_review": load_json(reports / "output_blind_review_pack.json"),
"output_review_kit": load_json(reports / "output_review_kit.json"),
"output_review_adjudication": load_json(reports / "output_review_adjudication.json"),
"skill_os2_coverage": load_json(reports / "skill_os2_coverage.json"),
"compiled_targets": load_json(reports / "compiled_targets.json"),
@@ -136,6 +138,7 @@ def insight_cards(data: dict[str, dict[str, Any]]) -> list[dict[str, str]]:
output = data["output_quality"].get("summary", {})
output_execution = data["output_execution"].get("summary", {})
output_blind = data["output_blind_review"].get("summary", {})
output_review_kit = data["output_review_kit"].get("summary", {})
output_review = data["output_review_adjudication"].get("summary", {})
blueprint = data["skill_os2_coverage"].get("summary", {})
compiled = data["compiled_targets"].get("summary", {})
@@ -184,6 +187,11 @@ def insight_cards(data: dict[str, dict[str, Any]]) -> list[dict[str, str]]:
"value": str(output_blind.get("pair_count", 0)),
"detail": "review pairs hide baseline vs with-skill labels",
},
{
"label": "Review Kit",
"value": f"{output_review_kit.get('ready_for_adjudication_count', 0)}/{output_review_kit.get('case_count', 0)}",
"detail": f"pending {output_review_kit.get('pending_decision_count', 0)}; answer key hidden",
},
{
"label": "Review A/B",
"value": f"{output_review.get('judgment_count', 0)}/{output_review.get('pair_count', 0)}",
+1 -1
View File
@@ -78,7 +78,7 @@ def render_output_review_section(adjudication: dict[str, Any]) -> str:
return (
"<section>"
"<h2>评审清单</h2>"
"<p class='muted'>每张卡片对应一个 blind A/B case,说明当前是否待判、答案是否仍隐藏、应填写的决策文件、有效字段和复跑命令。</p>"
"<p class='muted'>先打开 reports/output_review_kit.md每张卡片对应一个 blind A/B case,说明当前是否待判、答案是否仍隐藏、应填写的决策文件、有效字段和复跑命令。</p>"
f"{render_output_review_checklist(adjudication)}"
"</section>"
)
+11 -1
View File
@@ -475,6 +475,7 @@ def build_report_model(skill_dir: Path) -> dict:
output_quality = load_json(skill_dir / "reports" / "output_quality_scorecard.json")
output_execution = load_json(skill_dir / "reports" / "output_execution_runs.json")
output_blind_review = load_json(skill_dir / "reports" / "output_blind_review_pack.json")
output_review_kit = load_json(skill_dir / "reports" / "output_review_kit.json")
output_review_adjudication = load_json(skill_dir / "reports" / "output_review_adjudication.json")
benchmark_reproducibility = load_json(skill_dir / "reports" / "benchmark_reproducibility.json")
conformance = load_json(skill_dir / "reports" / "conformance_matrix.json")
@@ -550,8 +551,11 @@ def build_report_model(skill_dir: Path) -> dict:
if (skill_dir / "reports" / "output_blind_answer_key.json").exists():
insert_after = deliverables.index("reports/output_blind_review_pack.md") + 1 if "reports/output_blind_review_pack.md" in deliverables else deliverables.index("reports/output_quality_scorecard.md") + 1
deliverables.insert(insert_after, "reports/output_blind_answer_key.json")
if (skill_dir / "reports" / "output_review_kit.md").exists():
insert_after = deliverables.index("reports/output_blind_review_pack.md") + 1 if "reports/output_blind_review_pack.md" in deliverables else deliverables.index("reports/output_quality_scorecard.md") + 1
deliverables.insert(insert_after, "reports/output_review_kit.md")
if (skill_dir / "reports" / "output_review_adjudication.md").exists():
insert_after = deliverables.index("reports/output_blind_answer_key.json") + 1 if "reports/output_blind_answer_key.json" in deliverables else deliverables.index("reports/output_quality_scorecard.md") + 1
insert_after = deliverables.index("reports/output_review_kit.md") + 1 if "reports/output_review_kit.md" in deliverables else deliverables.index("reports/output_quality_scorecard.md") + 1
deliverables.insert(insert_after, "reports/output_review_adjudication.md")
if (skill_dir / "reports" / "benchmark_reproducibility.md").exists():
insert_after = deliverables.index("reports/output_review_adjudication.md") + 1 if "reports/output_review_adjudication.md" in deliverables else deliverables.index("reports/output_quality_scorecard.md") + 1
@@ -680,6 +684,12 @@ def build_report_model(skill_dir: Path) -> dict:
"pair_count": output_blind_review.get("summary", {}).get("pair_count", 0),
"answer_key_separate": output_blind_review.get("summary", {}).get("answer_key_separate", False),
},
"output_review_kit": {
"ok": output_review_kit.get("ok", False),
"summary": output_review_kit.get("summary", {}),
"artifacts": output_review_kit.get("artifacts", {}),
"failures": output_review_kit.get("failures", []),
},
"output_review_adjudication": {
"ok": output_review_adjudication.get("ok", False),
"summary": output_review_adjudication.get("summary", {}),
+21
View File
@@ -167,6 +167,7 @@ def command_report(args: argparse.Namespace) -> int:
run_script("compile_skill.py", [str(ROOT)]),
run_script("run_output_eval.py", []),
run_script("run_output_execution.py", ["--runner-command", local_output_runner_command()]),
run_script("prepare_output_review_kit.py", []),
run_script("adjudicate_output_review.py", []),
run_script("render_adoption_drift_report.py", [str(ROOT)]),
run_script("render_telemetry_hook_recipes.py", [str(ROOT)]),
@@ -201,6 +202,7 @@ def command_report(args: argparse.Namespace) -> int:
"system_model": "reports/system-model.json",
"compiled_targets": "reports/compiled_targets.json",
"output_execution": "reports/output_execution_runs.json",
"output_review_kit": "reports/output_review_kit.json",
"output_review_adjudication": "reports/output_review_adjudication.json",
"adoption_drift": "reports/adoption_drift_report.json",
"telemetry_hooks": "reports/telemetry_hook_recipes.json",
@@ -501,6 +503,25 @@ def command_output_execution(args: argparse.Namespace) -> int:
return 0 if result["ok"] else 2
def command_output_review_kit(args: argparse.Namespace) -> int:
cmd = []
if args.blind_pack_json:
cmd.extend(["--blind-pack-json", args.blind_pack_json])
if args.blind_pack_md:
cmd.extend(["--blind-pack-md", args.blind_pack_md])
if args.decisions:
cmd.extend(["--decisions", args.decisions])
if args.output_json:
cmd.extend(["--output-json", args.output_json])
if args.output_md:
cmd.extend(["--output-md", args.output_md])
if args.write_template:
cmd.append("--write-template")
result = run_script("prepare_output_review_kit.py", cmd)
print(json.dumps(result["payload"] if result["payload"] is not None else result, ensure_ascii=False, indent=2))
return 0 if result["ok"] else 2
def command_output_review(args: argparse.Namespace) -> int:
cmd = []
if args.blind_pack:
+12
View File
@@ -587,6 +587,18 @@ def build_parser(command_handlers: dict[str, Callable[[argparse.Namespace], int]
output_eval_cmd.add_argument("--blind-answer-key-json")
output_eval_cmd.set_defaults(func=_handler(command_handlers, "command_output_eval"))
output_review_kit_cmd = subparsers.add_parser(
"output-review-kit",
help="Prepare a reviewer-facing blind A/B output review kit without exposing the answer key.",
)
output_review_kit_cmd.add_argument("--blind-pack-json")
output_review_kit_cmd.add_argument("--blind-pack-md")
output_review_kit_cmd.add_argument("--decisions")
output_review_kit_cmd.add_argument("--output-json")
output_review_kit_cmd.add_argument("--output-md")
output_review_kit_cmd.add_argument("--write-template", action="store_true")
output_review_kit_cmd.set_defaults(func=_handler(command_handlers, "command_output_review_kit"))
output_execution_cmd = subparsers.add_parser(
"output-exec",
help="Record output-eval execution evidence, timing, and token usage.",