scripts/validate_silicon_llm_api.py

#!/usr/bin/env python3
"""Run small real-API validation for dynamic silicon sampling."""

from __future__ import annotations

import argparse
import json
import sys
import time
from urllib import request
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from silicon_llm import resolve_openai_api_key, run_silicon_llm_sampling  # noqa: E402


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--agents", type=int, default=2)
    parser.add_argument("--max-cases", type=int, default=3)
    parser.add_argument("--model", default="gpt-4o-mini")
    parser.add_argument("--endpoint", default="")
    parser.add_argument("--out", default="")
    args = parser.parse_args()
    if not args.endpoint and not resolve_openai_api_key():
        print(json.dumps({"ok": False, "error": "OPENAI_API_KEY or .env OPENAI_API_KEY is required"}, ensure_ascii=False, indent=2))
        return 1
    out_dir = Path(args.out) if args.out else ROOT / "runs" / f"silicon_llm_api_validation_{timestamp()}"
    out_dir.mkdir(parents=True, exist_ok=True)
    report: dict[str, Any] = {"ok": False, "model": args.model, "agents": args.agents, "maxCases": args.max_cases, "endpoint": args.endpoint, "cases": []}
    for case in validation_cases()[: max(1, args.max_cases)]:
        payload = base_payload(args.agents, case["questions"], model=args.model)
        started = time.time()
        try:
            result = run_payload(payload, endpoint=args.endpoint)
            checks = validate_result(case["case_id"], result, case["questions"], args.agents)
            report["cases"].append(
                {
                    "case_id": case["case_id"],
                    "ok": all(check["ok"] for check in checks),
                    "elapsed_seconds": round(time.time() - started, 2),
                    "questions": [{"id": item["id"], "kind": item["kind"], "scale": item.get("scale"), "options": item.get("options")} for item in case["questions"]],
                    "checks": checks,
                    "llmTrace": {
                        key: value
                        for key, value in result.get("llmTrace", {}).items()
                        if key != "rawResponses"
                    },
                    "sample_categorical_answers": result.get("categoricalAnswers", [])[:3],
                    "sample_open_answers": result.get("openAnswers", [])[:3],
                    "questionStats": result.get("questionStats", []),
                    "runLog": result.get("runLog"),
                    "runLogError": result.get("runLogError"),
                }
            )
        except Exception as exc:  # noqa: BLE001
            report["cases"].append({"case_id": case["case_id"], "ok": False, "error": f"{type(exc).__name__}: {exc}"})
    report["ok"] = bool(report["cases"]) and all(case.get("ok") for case in report["cases"])
    (out_dir / "report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps({"ok": report["ok"], "out": str(out_dir / "report.json"), "cases": report["cases"]}, ensure_ascii=False, indent=2))
    return 0 if report["ok"] else 1


def run_payload(payload: dict[str, Any], *, endpoint: str) -> dict[str, Any]:
    if not endpoint:
        return run_silicon_llm_sampling(payload)
    url = endpoint.rstrip("/") + "/api/silicon/llm-run"
    data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    req = request.Request(url, data=data, headers={"content-type": "application/json"}, method="POST")
    with request.urlopen(req, timeout=180) as response:  # noqa: S310
        result = json.loads(response.read().decode("utf-8"))
    if "error" in result:
        raise RuntimeError(str(result["error"]))
    return result


def validation_cases() -> list[dict[str, Any]]:
    return [
        {
            "case_id": "mixed_policy_trust",
            "questions": [
                likert("approval_5", "현 정부 국정 운영을 어떻게 평가하십니까?", 5),
                likert("medical_access_4", "거주 지역 의료 접근성에 만족하십니까?", 4),
                categorical("life_priority", "삶에서 가장 중요한 것을 하나만 고르십시오.", ["돈", "명예", "건강"]),
                open_q("local_problem", "거주 지역에서 가장 먼저 해결해야 할 문제는 무엇입니까?"),
            ],
        },
        {
            "case_id": "open_only_priorities",
            "questions": [
                categorical("social_priority", "한국 사회가 가장 우선해야 할 가치를 고르십시오.", ["경제 성장", "사회 안전", "공정성", "환경"]),
                open_q("top_issue", "한국 사회가 가장 먼저 해결해야 할 문제는 무엇이라고 보십니까?"),
                open_q("policy_request", "새 정부나 지방자치단체에 가장 바라는 정책을 적어주십시오."),
            ],
        },
        {
            "case_id": "multi_scale_likert",
            "questions": [
                likert("vote_turnout_4", "다음 지방선거에 투표할 가능성이 얼마나 높습니까?", 4),
                likert("party_reflects_5", "주요 정당이 국민 의견을 잘 반영한다고 보십니까?", 5),
                categorical("vote_factor", "투표할 때 가장 중요하게 보는 기준을 고르십시오.", ["후보 역량", "정당", "공약", "지역 현안"]),
                likert("climate_cost_7", "탄소 감축 비용 부담 정책에 어느 정도 동의하십니까?", 7),
            ],
        },
        {
            "case_id": "custom_korean_mix",
            "questions": [
                likert("custom_transport_safety", "출퇴근길 대중교통 안전에 만족하십니까?", 5),
                categorical("custom_life_choice", "지금 생활에서 가장 크게 필요한 것을 고르십시오.", ["시간", "소득", "건강", "관계"]),
                open_q("custom_one_sentence", "본인의 생활에서 가장 크게 체감되는 정책 문제를 한 문장으로 적어주십시오."),
                likert("custom_media_trust", "온라인 뉴스와 유튜브 정치 정보를 신뢰하십니까?", 5),
            ],
        },
        {
            "case_id": "many_questions_mix",
            "questions": [
                likert("economy_next_year", "향후 1년 한국 경제가 좋아질 것이라고 보십니까?", 5),
                likert("household_life", "현재 가계 생활 형편에 얼마나 만족하십니까?", 5),
                likert("institution_trust", "국회, 언론, 행정부 등 공공기관을 전반적으로 신뢰하십니까?", 5),
                categorical("public_spend_priority", "정부 재정이 우선 투입되어야 할 분야를 고르십시오.", ["복지", "일자리", "교육", "안보", "기후"]),
                open_q("free_reason", "그렇게 응답한 가장 큰 이유를 적어주십시오."),
            ],
        },
    ]


def likert(question_id: str, title: str, scale: int) -> dict[str, Any]:
    return {"id": question_id, "title": title, "source": "API validation", "category": "검증", "kind": "likert", "scale": scale}


def open_q(question_id: str, title: str) -> dict[str, Any]:
    return {"id": question_id, "title": title, "source": "API validation", "category": "검증", "kind": "open"}


def categorical(question_id: str, title: str, labels: list[str]) -> dict[str, Any]:
    return {
        "id": question_id,
        "title": title,
        "source": "API validation",
        "category": "검증",
        "kind": "categorical",
        "options": [{"id": f"opt_{index + 1}", "label": label} for index, label in enumerate(labels)],
    }


def base_payload(agent_count: int, questions: list[dict[str, Any]], *, model: str) -> dict[str, Any]:
    return {
        "config": {
            "sampleSize": agent_count,
            "genders": [{"id": "male", "enabled": True, "weight": agent_count // 2}, {"id": "female", "enabled": True, "weight": agent_count - agent_count // 2}],
            "ages": [{"id": "30s", "enabled": True, "weight": 2}, {"id": "50s", "enabled": True, "weight": agent_count - 2}],
            "locations": [{"id": "seoul", "enabled": True, "weight": 3}, {"id": "busan", "enabled": True, "weight": agent_count - 3}],
            "locationOptions": [
                {"id": "seoul", "label": "서울", "parentRegion": "seoul", "level": "sido", "defaultWeight": 60, "short": "서울", "group": "시도"},
                {"id": "busan", "label": "부산", "parentRegion": "busan", "level": "sido", "defaultWeight": 40, "short": "부산", "group": "시도"},
            ],
            "personaAttributes": [
                {"id": "occ_office", "enabled": True, "weight": 2},
                {"id": "occ_student", "enabled": True, "weight": agent_count - 2},
                {"id": "edu_bachelor", "enabled": True, "weight": agent_count},
            ],
            "nemotronFields": ["persona", "sex", "age", "province", "occupation", "education_level"],
            "questions": questions,
            "seed": 20260630,
        },
        "execution": {"max_agents": agent_count, "model": model, "timeout_seconds": 90},
    }


def validate_result(case_id: str, result: dict[str, Any], questions: list[dict[str, Any]], agent_count: int) -> list[dict[str, Any]]:
    checks: list[dict[str, Any]] = []
    likert_questions = [item for item in questions if item["kind"] == "likert"]
    categorical_questions = [item for item in questions if item["kind"] == "categorical"]
    open_questions = [item for item in questions if item["kind"] == "open"]
    checks.append(check("respondent_count", len(result.get("respondents", [])) == agent_count, {"count": len(result.get("respondents", [])), "expected": agent_count}))
    checks.append(check("one_llm_call_per_agent", result.get("llmTrace", {}).get("calls") == agent_count, result.get("llmTrace", {})))
    checks.append(check("all_questions_in_config", [item["id"] for item in result.get("config", {}).get("questions", [])] == [item["id"] for item in questions], None))
    checks.append(check("likert_answer_count", len(result.get("likertAnswers", [])) == agent_count * len(likert_questions), {"count": len(result.get("likertAnswers", [])), "expected": agent_count * len(likert_questions)}))
    checks.append(check("categorical_answer_count", len(result.get("categoricalAnswers", [])) == agent_count * len(categorical_questions), {"count": len(result.get("categoricalAnswers", [])), "expected": agent_count * len(categorical_questions)}))
    checks.append(check("open_answer_count", len(result.get("openAnswers", [])) == agent_count * len(open_questions), {"count": len(result.get("openAnswers", [])), "expected": agent_count * len(open_questions)}))
    checks.append(check("question_stats_count", len(result.get("questionStats", [])) == len(questions), {"count": len(result.get("questionStats", [])), "expected": len(questions)}))
    checks.append(check("parse_issues_empty", not result.get("llmTrace", {}).get("parseIssues"), result.get("llmTrace", {}).get("parseIssues")))
    if result.get("runLog") or result.get("runLogError"):
        checks.append(check("run_log_recorded", bool(result.get("runLog")) and not result.get("runLogError"), result.get("runLog") or result.get("runLogError")))
    for question in likert_questions:
        question_id = question["id"]
        scale = int(question.get("scale") or 5)
        values = [int(answer["value"]) for answer in result.get("likertAnswers", []) if answer.get("questionId") == question_id]
        stat = next((item for item in result.get("questionStats", []) if item.get("questionId") == question_id), {})
        checks.append(check(f"{case_id}:{question_id}:value_range", len(values) == agent_count and all(1 <= value <= scale for value in values), {"values": values, "scale": scale}))
        checks.append(check(f"{case_id}:{question_id}:distribution_sum", sum(item.get("count", 0) for item in stat.get("distribution", [])) == agent_count, stat.get("distribution")))
        rationales = [answer.get("rationale") for answer in result.get("likertAnswers", []) if answer.get("questionId") == question_id]
        checks.append(check(f"{case_id}:{question_id}:likert_rationale", len(rationales) == agent_count and all(rationales), rationales[:2]))
    for question in categorical_questions:
        question_id = question["id"]
        allowed = {option["id"] for option in question.get("options", [])}
        answers = [answer for answer in result.get("categoricalAnswers", []) if answer.get("questionId") == question_id]
        stat = next((item for item in result.get("questionStats", []) if item.get("questionId") == question_id), {})
        checks.append(check(f"{case_id}:{question_id}:categorical_option", len(answers) == agent_count and all(answer.get("optionId") in allowed for answer in answers), answers[:2]))
        checks.append(check(f"{case_id}:{question_id}:categorical_distribution_sum", sum(item.get("count", 0) for item in stat.get("distribution", [])) == agent_count, stat.get("distribution")))
        checks.append(check(f"{case_id}:{question_id}:categorical_rationale", len(answers) == agent_count and all(answer.get("rationale") for answer in answers), answers[:2]))
    for question in open_questions:
        answers = [answer for answer in result.get("openAnswers", []) if answer.get("questionId") == question["id"]]
        checks.append(check(f"{case_id}:{question['id']}:open_text_theme", len(answers) == agent_count and all(answer.get("text") and answer.get("theme") for answer in answers), answers[:2]))
        checks.append(check(f"{case_id}:{question['id']}:open_rationale", len(answers) == agent_count and all(answer.get("rationale") for answer in answers), answers[:2]))
    expected_primary = next((item["id"] for item in likert_questions), None)
    checks.append(check("primary_question_dynamic", result.get("primaryQuestionId") == expected_primary, {"actual": result.get("primaryQuestionId"), "expected": expected_primary}))
    return checks


def check(name: str, ok: bool, detail: Any) -> dict[str, Any]:
    return {"name": name, "ok": bool(ok), "detail": detail}


def timestamp() -> str:
    return time.strftime("%Y%m%d%H%M%S")


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