zhifu-platform/scripts/report_agent_scenarios.py

import argparse
import hashlib
import json
from collections import Counter, defaultdict
from pathlib import Path

from zhifu.evaluation import Dataset, write_report


def main():
    parser = argparse.ArgumentParser(description="汇总真实 Agent 场景记录，保留全部调用与审核来源")
    parser.add_argument("execution", type=Path)
    parser.add_argument("--output", type=Path, required=True)
    parser.add_argument("--review", type=Path)
    args = parser.parse_args()
    dataset = Dataset.model_validate_json(Path("evaluations/agent-scenarios.json").read_bytes())
    records = []
    for path in sorted(args.execution.glob("*/*.json")):
        content = path.read_bytes()
        records.append({**json.loads(content),
                        "evidence_sha256": hashlib.sha256(content).hexdigest(),
                        "execution_round": path.parent.name})
    records.sort(key=lambda row: row["started_at"])
    latest = {row["id"]: row for row in records}
    attempts = [attempt for row in records for attempt in row.get("attempts", [])]
    categories = defaultdict(Counter)
    rounds = defaultdict(list)
    for row in records:
        if row.get("finished_at"):
            rounds[row["execution_round"]].append(row)
    scope_reviews = json.loads(Path("evaluations/agent-scenario-scope-review.json").read_bytes())
    cases = []
    for case in dataset.cases():
        row = latest.get(case["id"], {})
        verdict = row.get("machine_verdict", "NOT_RUN")
        scope = scope_reviews["cases"].get(case["id"], {})
        scoped_verdict = scope.get("scope_verdict") if scope.get("evidence_sha256") == row.get(
            "evidence_sha256") else None
        categories[case["category"]][scoped_verdict or verdict] += 1
        cases.append({"id": case["id"], "split": case["split"], "machine_verdict": verdict,
                      "scope_verdict": scoped_verdict,
                      "human_review": "REQUIRED", "run_status": row.get("run_status"),
                      "error_code": row.get("error_code"),
                      "safety_verdict": row.get("safety_verdict", "NOT_EVALUATED"),
                      "interaction_verdict": row.get("interaction_verdict"),
                      "model_calls": len(row.get("attempts", [])),
                      "graph_version": (row.get("release_snapshot") or {}).get("graph_version"),
                      "source_sha256": row.get("source_sha256")})
    report = {"case_count": 200, "execution_records": len(records), "cases": cases,
              "rounds": {name: {"case_count": len(rows),
                  "machine_verdicts": dict(Counter(row["machine_verdict"] for row in rows)),
                  "graph_versions": dict(Counter((row.get("release_snapshot") or {}).get(
                      "graph_version", "UNRECORDED") for row in rows))}
                         for name, rows in rounds.items()},
              "categories": {key: dict(value) for key, value in categories.items()},
              "budget": json.loads((args.execution / "budget.json").read_text(encoding="utf-8")),
              "recorded_attempts": len(attempts),
              "usage": {key: sum((row["usage"] or {}).get(key, 0) for row in attempts)
                        for key in ("input_tokens", "output_tokens", "total_tokens")},
              "unknown_usage_attempts": sum(row["usage"] is None for row in attempts),
              "cached_input_tokens": sum((row["usage"] or {}).get(
                  "input_token_details", {}).get("cache_read", 0) for row in attempts),
              "output_limit_exceeded_attempts": sum((row["usage"] or {}).get(
                  "output_tokens", 0) > 512 for row in attempts),
              "review_kind": "agent", "human_review": "REQUIRED"}
    write_report(args.output, report)
    if args.review:
        sections = ["# Agent 场景语义审阅资料", "", "机器结果与人工审核分别登记。"]
        for case in dataset.cases():
            row = latest.get(case["id"], {})
            if not row.get("finished_at"):
                continue
            answers = [message["data"]["content"] for message in row.get("messages", [])
                       if message["type"] == "ai" and not message["data"].get("tool_calls")]
            names = [call["name"] for message in row.get("messages", [])
                     for call in message["data"].get("tool_calls", [])]
            sections += ["", f"## {case['id']}", "", f"请求：{case['prompt']}",
                         f"预期：{row.get('effective_expected', case['expected'])}",
                         f"执行范围：{row.get('execution_scope', '原始场景范围')}",
                         f"机器结果：{row.get('machine_verdict', 'NOT_RUN')}",
                         f"工具：{', '.join(names)}", "最终会话答复（含系统生成的等待提示）：",
                         "\n".join(answers) or "无最终会话答复"]
        args.review.parent.mkdir(parents=True, exist_ok=True)
        args.review.write_text("\n\n".join(sections) + "\n", encoding="utf-8")
    print(json.dumps({key: value for key, value in report.items() if key != "cases"},
                     ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()