ai-agent-book/chapter10/voice-werewolf/evaluate_audio_run.py

#!/usr/bin/env python3
"""Direct-audio independent evaluation of simulator utterances through OpenRouter."""

from __future__ import annotations

import argparse
import base64
import hashlib
import json
import os
import re
import time
from pathlib import Path
from typing import Any

from openai import OpenAI


WORDS = {
    1: "one", 2: "two", 3: "three", 4: "four", 5: "five",
    6: "six", 7: "seven", 8: "eight", 9: "nine", 10: "ten",
}


def sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def expected_text(tool_event: dict[str, Any]) -> str:
    arguments = tool_event.get("arguments") or {}
    if tool_event.get("tool") == "speak_publicly":
        return str(arguments.get("utterance", "")).strip()
    target = str(arguments.get("target", "")).strip()
    if target == "none":
        return "I choose to abstain."
    number = int(target.removeprefix("P"))
    return f"I choose player {WORDS.get(number, str(number))}."


def parse_json(text: str) -> dict[str, Any]:
    text = text.strip()
    if text.startswith("```"):
        text = re.sub(r"^```(?:json)?\s*|\s*```$", "", text, flags=re.S)
    return json.loads(text)


def evaluate(report_path: Path, output: Path, model: str) -> dict[str, Any]:
    if not os.getenv("OPENROUTER_API_KEY"):
        raise RuntimeError("OPENROUTER_API_KEY is required")
    raw = report_path.read_bytes()
    report = json.loads(raw)
    events = report.get("voice_events") or []
    client = OpenAI(
        api_key=os.environ["OPENROUTER_API_KEY"],
        base_url="https://openrouter.ai/api/v1",
        timeout=120,
        max_retries=2,
    )
    rows = []
    for index, event in enumerate(events):
        if event.get("type") != "simulator_llm_tool":
            continue
        seat = event.get("seat")
        # Treat each tool -> TTS -> ASR sequence as a transaction.  Searching to
        # the end of the report can pair a failed turn with a later player's audio
        # and produce a false positive independent evaluation.
        transaction = []
        for item in events[index + 1:]:
            if item.get("type") == "simulator_llm_tool":
                break
            transaction.append(item)
        tts = next(
            (item for item in transaction
             if item.get("type") == "tts_ready" and item.get("speaker") == seat),
            None,
        )
        asr = next(
            (item for item in transaction if item.get("type") == "simulator_asr"),
            None,
        )
        if not tts or not asr:
            raise ValueError(f"tool event {event.get('sequence')} lacks TTS/ASR evidence")
        if transaction.index(tts) > transaction.index(asr):
            raise ValueError(f"tool event {event.get('sequence')} has ASR before TTS")
        if asr.get("source_audio_sha256") != tts.get("audio_sha256"):
            raise ValueError(f"tool event {event.get('sequence')} ASR/TTS audio hash mismatch")
        audio_path = Path(str(tts["file"]))
        if not audio_path.is_absolute():
            project_root = next(
                parent for parent in report_path.resolve().parents
                if parent.name == "voice-werewolf"
            )
            candidates = [Path.cwd() / audio_path, project_root / audio_path]
            audio_path = next((path for path in candidates if path.is_file()), candidates[0])
        if not audio_path.is_file():
            raise FileNotFoundError(audio_path)
        if sha256(audio_path) != tts.get("audio_sha256"):
            raise ValueError(f"audio hash mismatch: {audio_path}")
        expected = expected_text(event)
        started = time.perf_counter()
        response = client.chat.completions.create(
            model=model,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "text",
                            "text": (
                                "Independently evaluate this synthetic Werewolf-game speech audio. "
                                f"The intended text is: {expected!r}. Return one JSON object only with "
                                "keys transcript (string), intelligibility_1_to_5 (integer), "
                                "semantic_fidelity_1_to_5 (integer), naturalness_1_to_5 (integer), "
                                "action_or_seat_preserved (boolean), and rationale (short string). "
                                "Judge the waveform itself. A robotic voice may score low on naturalness "
                                "without losing intelligibility or semantic fidelity."
                            ),
                        },
                        {
                            "type": "input_audio",
                            "input_audio": {
                                "data": base64.b64encode(audio_path.read_bytes()).decode("ascii"),
                                "format": audio_path.suffix.lstrip(".").lower(),
                            },
                        },
                    ],
                }
            ],
            temperature=0,
            max_tokens=600,
        )
        judgment = parse_json(response.choices[0].message.content or "")
        usage = response.usage.model_dump() if response.usage else None
        row = {
            "tool_sequence": event.get("sequence"),
            "tts_sequence": tts.get("sequence"),
            "asr_sequence": asr.get("sequence"),
            "tool": event.get("tool"),
            "target": (event.get("arguments") or {}).get("target"),
            "expected_text": expected,
            "original_asr_transcript": asr.get("transcript"),
            "audio_file": str(audio_path.relative_to(Path.cwd())),
            "audio_bytes": audio_path.stat().st_size,
            "audio_sha256": sha256(audio_path),
            "judge": judgment,
            "request_id": response.id,
            "provider_reported_model": response.model,
            "usage": usage,
            "latency_seconds": round(time.perf_counter() - started, 3),
        }
        row["pass"] = bool(
            int(judgment.get("intelligibility_1_to_5", 0)) >= 4
            and int(judgment.get("semantic_fidelity_1_to_5", 0)) >= 4
            and judgment.get("action_or_seat_preserved") is True
        )
        rows.append(row)

    result = {
        "schema_version": 1,
        "experiment": "10-6-independent-audio-evaluation",
        "source_report": str(report_path),
        "source_report_sha256": hashlib.sha256(raw).hexdigest(),
        "provider": "OpenRouter multimodal audio API",
        "requested_model": model,
        "evaluations": rows,
        "gates": {
            "all_simulator_audio_evaluated": len(rows)
            == sum(event.get("type") == "simulator_llm_tool" for event in events),
            "unique_real_request_ids": len({row["request_id"] for row in rows}) == len(rows),
            "all_audio_hashes_match": bool(rows),
            "all_intelligible_and_semantically_faithful": bool(rows)
            and all(row["pass"] for row in rows),
        },
    }
    result["status"] = "pass" if all(result["gates"].values()) else "fail"
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(
        json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
    )
    return result


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("report", type=Path)
    parser.add_argument("--output", type=Path, required=True)
    parser.add_argument("--model", default="google/gemini-3-flash-preview")
    args = parser.parse_args()
    result = evaluate(args.report, args.output, args.model)
    print(json.dumps({"status": result["status"], "gates": result["gates"]}, indent=2))
    return 0 if result["status"] == "pass" else 1


if __name__ == "__main__":
    raise SystemExit(main())