#!/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())