"""LLM-backed silicon sampling runtime. Each sampled persona is treated as one respondent-agent. The agent receives all selected survey questions in one prompt and must return one structured answer per question. The parser and aggregation are question-driven, so Likert/open question mixes can change without adding scenario-specific code. """ from __future__ import annotations import json import math import os import re from dataclasses import dataclass from pathlib import Path from typing import Any GENDER_LABELS = {"male": "남성", "female": "여성", "no_response": "응답 안 함"} AGE_LABELS = {"20s": "20대", "30s": "30대", "40s": "40대", "50s": "50대", "60plus": "60대 이상"} PERSONA_LABELS = { "occ_office": "사무/관리직", "occ_service": "서비스/판매직", "occ_professional": "전문직", "occ_self_employed": "자영업", "occ_student": "학생", "occ_homemaker": "전업주부", "occ_technical": "기술/생산직", "occ_retired": "은퇴/무직", "edu_high_school": "고졸 이하", "edu_college": "전문대/대학 재학", "edu_bachelor": "대졸", "edu_graduate": "대학원 이상", "housing_apartment": "아파트", "housing_house": "단독/다가구", "housing_officetel": "오피스텔/원룸", "housing_other": "기타 주거", "marital_single": "미혼", "marital_married": "기혼", "marital_divorced": "이혼/별거", "marital_widowed": "사별", "family_single": "1인 가구", "family_couple": "부부 가구", "family_children": "자녀 동거", "family_extended": "확대 가족", } PERSONA_DIMENSIONS = { "occupation": ["occ_office", "occ_service", "occ_professional", "occ_self_employed", "occ_student", "occ_homemaker", "occ_technical", "occ_retired"], "education": ["edu_high_school", "edu_college", "edu_bachelor", "edu_graduate"], "housing": ["housing_apartment", "housing_house", "housing_officetel", "housing_other"], "marital": ["marital_single", "marital_married", "marital_divorced", "marital_widowed"], "family": ["family_single", "family_couple", "family_children", "family_extended"], } def build_silicon_llm_system_prompt() -> str: return ( "당신은 한국 설문조사의 가상 응답자입니다. " "주어진 persona와 인구통계 정보를 일관되게 반영해 모든 문항에 답하십시오. " "반드시 JSON만 반환하십시오. Likert 문항은 해당 척도 범위의 정수 value를, " "categorical 문항은 제공된 options 중 하나의 optionId와 label을, " "open-ended 문항은 한국어 text와 짧은 theme를 반환하십시오. " "모든 답변에는 rationale을 포함하십시오. rationale은 응답자의 1인칭 관점으로 쓰고, " "성별, 연령, 지역, 직업, 학력, 주거, 혼인, 가구 등 제공된 배경 중 최소 두 가지를 구체적으로 언급하십시오. " "rationale은 '나는 ... 배경/상황이라서 ...라고 판단했다'처럼 개인 맥락이 보이게 작성하고, " "문항을 반복하거나 일반론만 말하지 마십시오." ) def build_silicon_llm_user_prompt(agent: dict[str, Any], questions: list[dict[str, Any]], response_contract: dict[str, Any], *, pretty: bool = False) -> str: return json.dumps( { "agent": agent, "questions": questions, "required_json_contract": response_contract, }, ensure_ascii=False, indent=2 if pretty else None, ) def build_silicon_llm_messages(agent: dict[str, Any], questions: list[dict[str, Any]], response_contract: dict[str, Any]) -> list[dict[str, str]]: return [ {"role": "system", "content": build_silicon_llm_system_prompt()}, {"role": "user", "content": build_silicon_llm_user_prompt(agent, questions, response_contract)}, ] @dataclass class OpenAILLM: model: str = "gpt-4o-mini" temperature: float = 0.2 timeout: float = 60.0 def complete_agent(self, *, agent: dict[str, Any], questions: list[dict[str, Any]], response_contract: dict[str, Any]) -> str: from openai import OpenAI api_key = resolve_openai_api_key() if not api_key: raise RuntimeError("OPENAI_API_KEY or OPENAI_API_KEY_FILE is required for silicon LLM sampling") client = OpenAI(api_key=api_key) messages = build_silicon_llm_messages(agent, questions, response_contract) kwargs: dict[str, Any] = {"max_completion_tokens": 1800} if self.model.startswith("gpt-5"): kwargs["reasoning_effort"] = "minimal" else: kwargs["temperature"] = self.temperature response = client.chat.completions.create( model=self.model, messages=messages, response_format={"type": "json_object"}, timeout=self.timeout, **kwargs, ) return response.choices[0].message.content or "{}" def resolve_openai_api_key() -> str | None: key = os.getenv("OPENAI_API_KEY") if key: return key.strip() path = os.getenv("OPENAI_API_KEY_FILE") if path: try: return Path(path).expanduser().read_text(encoding="utf-8").strip().splitlines()[0].strip() except (OSError, IndexError): pass for env_path in [Path(__file__).resolve().parent / ".env", Path.cwd() / ".env"]: try: if not env_path.exists(): continue for line in env_path.read_text(encoding="utf-8").splitlines(): stripped = line.strip() if not stripped or stripped.startswith("#") or "=" not in stripped: continue name, value = stripped.split("=", 1) if name.strip() == "OPENAI_API_KEY": return value.strip().strip('"').strip("'") except OSError: continue return None def build_silicon_llm_response_contract(questions: list[dict[str, Any]]) -> dict[str, Any]: answer_properties: dict[str, Any] = {} for question in questions: question_id = str(question.get("id") or "") if not question_id: continue if question.get("kind") == "open": answer_properties[question_id] = { "type": "object", "properties": { "text": {"type": "string"}, "theme": {"type": "string"}, "rationale": {"type": "string"}, }, "required": ["text", "theme", "rationale"], "additionalProperties": True, } elif question.get("kind") == "categorical": options = [item for item in question.get("options", []) if isinstance(item, dict)] option_ids = [str(item.get("id")) for item in options if item.get("id")] answer_properties[question_id] = { "type": "object", "properties": { "optionId": {"type": "string", "enum": option_ids or ["opt_1"]}, "label": {"type": "string"}, "rationale": {"type": "string"}, }, "required": ["optionId", "label", "rationale"], "additionalProperties": True, } else: scale = int(question.get("scale") or 5) answer_properties[question_id] = { "type": "object", "properties": { "value": {"type": "integer", "minimum": 1, "maximum": scale}, "rationale": {"type": "string"}, }, "required": ["value", "rationale"], "additionalProperties": True, } return { "name": "silicon_agent_answers", "schema": { "type": "object", "properties": { "answers": { "type": "object", "properties": answer_properties, "required": list(answer_properties), "additionalProperties": False, } }, "required": ["answers"], "additionalProperties": False, }, } def parse_silicon_agent_response(raw: str, questions: list[dict[str, Any]], *, respondent_id: str) -> dict[str, Any]: payload = extract_json(raw) answers = payload.get("answers") if isinstance(payload, dict) else None if isinstance(answers, list): answers = {str(item.get("questionId") or item.get("id") or ""): item for item in answers if isinstance(item, dict)} if not isinstance(answers, dict): answers = {} likert_answers: list[dict[str, Any]] = [] categorical_answers: list[dict[str, Any]] = [] open_answers: list[dict[str, Any]] = [] issues: list[dict[str, Any]] = [] for question in questions: question_id = str(question.get("id") or "") answer = answers.get(question_id) if not isinstance(answer, dict): issues.append({"questionId": question_id, "type": "missing_answer"}) answer = {} if question.get("kind") == "open": text = str(answer.get("text") or answer.get("answer") or answer.get("response") or "").strip() theme = str(answer.get("theme") or answer.get("topic") or infer_theme(text)).strip() or "기타" rationale = str(answer.get("rationale") or answer.get("reason") or answer.get("explanation") or "").strip() if not text: text = "응답을 생성하지 못했습니다." issues.append({"questionId": question_id, "type": "empty_open_text"}) if not rationale: rationale = "해당 persona와 인구통계 조건을 기준으로 자유응답을 생성했습니다." issues.append({"questionId": question_id, "type": "missing_rationale"}) open_answers.append({"respondentId": respondent_id, "questionId": question_id, "text": text, "theme": theme[:40], "rationale": rationale[:500]}) elif question.get("kind") == "categorical": options = [item for item in question.get("options", []) if isinstance(item, dict)] option_by_id = {str(item.get("id")): str(item.get("label") or item.get("id")) for item in options if item.get("id")} fallback_id = next(iter(option_by_id), "opt_1") option_id = str(answer.get("optionId") or answer.get("option_id") or answer.get("value") or "").strip() if option_id not in option_by_id: matched = next((oid for oid, label in option_by_id.items() if label == str(answer.get("label") or answer.get("answer") or "").strip()), "") option_id = matched or fallback_id issues.append({"questionId": question_id, "type": "invalid_categorical_option"}) label = option_by_id.get(option_id, str(answer.get("label") or option_id)) rationale = str(answer.get("rationale") or answer.get("reason") or answer.get("explanation") or "").strip() if not rationale: rationale = "해당 persona와 인구통계 조건에 가장 가까운 선택지를 골랐습니다." issues.append({"questionId": question_id, "type": "missing_rationale"}) categorical_answers.append({"respondentId": respondent_id, "questionId": question_id, "optionId": option_id, "label": label, "rationale": rationale[:500]}) else: scale = int(question.get("scale") or 5) value = coerce_int(answer.get("value", answer.get("score", answer.get("rating"))), default=math.ceil(scale / 2)) value = max(1, min(scale, value)) rationale = str(answer.get("rationale") or answer.get("reason") or answer.get("explanation") or "").strip() if not rationale: rationale = "해당 persona와 인구통계 조건을 기준으로 척도 응답을 선택했습니다." issues.append({"questionId": question_id, "type": "missing_rationale"}) likert_answers.append({"respondentId": respondent_id, "questionId": question_id, "value": value, "rationale": rationale[:500]}) return {"likertAnswers": likert_answers, "categoricalAnswers": categorical_answers, "openAnswers": open_answers, "issues": issues} def run_silicon_llm_sampling(payload: dict[str, Any], *, llm: Any | None = None) -> dict[str, Any]: config = dict(payload.get("config") or payload) questions = [dict(item) for item in config.get("questions", []) if isinstance(item, dict)] if not questions: raise ValueError("config.questions is required") provided_respondents = config.get("respondents") if isinstance(config.get("respondents"), list) else [] sample_size = len(provided_respondents) if provided_respondents else bounded_int(config.get("sampleSize"), 5, minimum=1, maximum=100) execution = payload.get("execution") if isinstance(payload.get("execution"), dict) else {} max_agents = bounded_int(execution.get("max_agents"), int(os.getenv("SILICON_LLM_MAX_AGENTS", "100")), minimum=1, maximum=100) if sample_size > max_agents: raise ValueError(f"LLM silicon sampling requested {sample_size} agents but max_agents is {max_agents}") respondents = provided_respondents if provided_respondents else build_respondents(config, sample_size) response_contract = build_silicon_llm_response_contract(questions) model = str(execution.get("model") or os.getenv("SILICON_LLM_MODEL", "gpt-4o-mini")) llm = llm or OpenAILLM(model=model, timeout=float(execution.get("timeout_seconds") or 60)) likert_answers: list[dict[str, Any]] = [] categorical_answers: list[dict[str, Any]] = [] open_answers: list[dict[str, Any]] = [] parse_issues: list[dict[str, Any]] = [] raw_responses: list[dict[str, str]] = [] for respondent in respondents: raw = llm.complete_agent(agent=respondent, questions=questions, response_contract=response_contract) raw_responses.append({"respondentId": respondent["id"], "text": raw[:4000]}) parsed = parse_silicon_agent_response(raw, questions, respondent_id=respondent["id"]) likert_answers.extend(parsed["likertAnswers"]) categorical_answers.extend(parsed["categoricalAnswers"]) open_answers.extend(parsed["openAnswers"]) parse_issues.extend({"respondentId": respondent["id"], **issue} for issue in parsed["issues"]) primary_question_id = next((str(item.get("id")) for item in questions if item.get("kind") == "likert"), None) result = { "config": config, "respondents": respondents, "likertAnswers": likert_answers, "categoricalAnswers": categorical_answers, "openAnswers": open_answers, "regionStats": build_region_stats(config, respondents, likert_answers, open_answers, questions), "questionStats": build_question_stats(questions, likert_answers, categorical_answers), "primaryQuestionId": primary_question_id, "llmTrace": { "engine": "openai_chat_completions" if isinstance(llm, OpenAILLM) else "injected_llm", "model": model, "calls": len(respondents), "questionsPerCall": len(questions), "parseIssues": parse_issues, "rawResponses": raw_responses, }, } return result def build_silicon_prompt_preview(payload: dict[str, Any]) -> dict[str, Any]: config = dict(payload.get("config") or payload) questions = [dict(item) for item in config.get("questions", []) if isinstance(item, dict)] provided_respondents = config.get("respondents") if isinstance(config.get("respondents"), list) else [] respondents = provided_respondents if provided_respondents else build_respondents(config, 1) agent = dict(respondents[0]) if respondents else build_respondents(config, 1)[0] response_contract = build_silicon_llm_response_contract(questions) return { "systemContent": build_silicon_llm_system_prompt(), "userContent": build_silicon_llm_user_prompt(agent, questions, response_contract, pretty=True), "sampleRespondent": agent, "questionCount": len(questions), } def build_respondents(config: dict[str, Any], sample_size: int) -> list[dict[str, Any]]: genders = allocation_pool(config.get("genders"), sample_size, "female") ages = allocation_pool(config.get("ages"), sample_size, "40s") locations = allocation_pool(config.get("locations"), sample_size, "seoul") persona_pools = {dimension: allocation_pool([pick for pick in config.get("personaAttributes", []) if pick.get("id") in ids], sample_size, "") for dimension, ids in PERSONA_DIMENSIONS.items()} location_options = {str(item.get("id")): item for item in config.get("locationOptions", []) if isinstance(item, dict)} respondents: list[dict[str, Any]] = [] for index in range(sample_size): location_id = str(locations[index] or "seoul") location = location_options.get(location_id, {"id": location_id, "label": location_id, "parentRegion": "seoul"}) persona_attributes: dict[str, str] = {} persona_labels: dict[str, str] = {} for dimension, pool in persona_pools.items(): picked = str(pool[index] or "") if not picked: continue persona_attributes[dimension] = picked persona_labels[dimension] = PERSONA_LABELS.get(picked, picked) respondents.append( { "id": f"R{index + 1:04d}", "gender": str(genders[index] or "female"), "genderLabel": GENDER_LABELS.get(str(genders[index]), str(genders[index])), "age": str(ages[index] or "40s"), "ageLabel": AGE_LABELS.get(str(ages[index]), str(ages[index])), "region": str(location.get("parentRegion") or location_id), "location": location_id, "locationLabel": str(location.get("label") or location_id), "personaAttributes": persona_attributes, "personaLabels": persona_labels, "segment": ", ".join(persona_labels.values()) or "일반 응답자", "trust": 0.5, "economicAnxiety": 0.5, "participation": 0.5, } ) return respondents def allocation_pool(items: Any, target_size: int, fallback: str) -> list[str]: source = [item for item in items or [] if isinstance(item, dict) and item.get("enabled") and float_or_zero(item.get("weight")) > 0] if not source: return [fallback for _ in range(target_size)] total = sum(float_or_zero(item.get("weight")) for item in source) rows = [] for item in source: raw = float_or_zero(item.get("weight")) / total * target_size rows.append({"id": str(item.get("id")), "floor": math.floor(raw), "remainder": raw - math.floor(raw)}) used = sum(row["floor"] for row in rows) for row in sorted(rows, key=lambda item: item["remainder"], reverse=True): if used >= target_size: break row["floor"] += 1 used += 1 pool: list[str] = [] for row in rows: pool.extend([row["id"]] * int(row["floor"])) while len(pool) < target_size: pool.append(fallback) return pool[:target_size] def build_question_stats(questions: list[dict[str, Any]], likert_answers: list[dict[str, Any]], categorical_answers: list[dict[str, Any]]) -> list[dict[str, Any]]: stats: list[dict[str, Any]] = [] for question in questions: question_id = str(question.get("id")) if question.get("kind") == "open": stats.append({"questionId": question_id, "title": question.get("title", question_id), "kind": "open"}) continue if question.get("kind") == "categorical": answers = [answer for answer in categorical_answers if answer.get("questionId") == question_id] options = [item for item in question.get("options", []) if isinstance(item, dict)] distribution = [] for option in options: option_id = str(option.get("id")) count = len([answer for answer in answers if answer.get("optionId") == option_id]) distribution.append( { "optionId": option_id, "label": str(option.get("label") or option_id), "count": count, "share": count / len(answers) if answers else 0, } ) stats.append({"questionId": question_id, "title": question.get("title", question_id), "kind": "categorical", "distribution": distribution}) continue scale = int(question.get("scale") or 5) values = [int(answer["value"]) for answer in likert_answers if answer.get("questionId") == question_id] distribution = [{"value": value, "count": values.count(value), "share": values.count(value) / len(values) if values else 0} for value in range(1, scale + 1)] positive_cut = max(3, math.ceil(scale * 0.7)) stats.append( { "questionId": question_id, "title": question.get("title", question_id), "kind": "likert", "scale": scale, "mean": sum(values) / len(values) if values else 0, "positiveShare": len([value for value in values if value >= positive_cut]) / len(values) if values else 0, "distribution": distribution, } ) return stats def build_region_stats(config: dict[str, Any], respondents: list[dict[str, Any]], likert_answers: list[dict[str, Any]], open_answers: list[dict[str, Any]], questions: list[dict[str, Any]]) -> list[dict[str, Any]]: primary = next((question for question in questions if question.get("kind") == "likert"), None) primary_id = str(primary.get("id")) if primary else None scale = int(primary.get("scale") or 5) if primary else 5 positive_cut = max(3, math.ceil(scale * 0.7)) stats: list[dict[str, Any]] = [] for pick in config.get("locations", []): if not isinstance(pick, dict) or not pick.get("enabled"): continue location_id = str(pick.get("id")) people = [respondent for respondent in respondents if respondent.get("location") == location_id] ids = {respondent["id"] for respondent in people} values = [int(answer["value"]) for answer in likert_answers if answer.get("questionId") == primary_id and answer.get("respondentId") in ids] label = next((str(item.get("label")) for item in config.get("locationOptions", []) if isinstance(item, dict) and str(item.get("id")) == location_id), location_id) parent = next((str(item.get("parentRegion")) for item in config.get("locationOptions", []) if isinstance(item, dict) and str(item.get("id")) == location_id), location_id) stats.append( { "region": location_id, "parentRegion": parent, "label": label, "respondents": len(people), "mean": sum(values) / len(values) if values else 0, "scale": scale, "positiveShare": len([value for value in values if value >= positive_cut]) / len(values) if values else 0, "openCount": len([answer for answer in open_answers if answer.get("respondentId") in ids]), } ) return stats def extract_json(text: str) -> Any: text = (text or "").strip() if not text: raise ValueError("empty model response") try: return json.loads(text) except json.JSONDecodeError: pass fenced = re.search(r"```(?:json)?\s*(.*?)```", text, flags=re.DOTALL | re.IGNORECASE) if fenced: return json.loads(fenced.group(1).strip()) start = text.find("{") end = text.rfind("}") if start != -1 and end != -1 and end > start: return json.loads(text[start : end + 1]) raise ValueError(f"could not extract JSON from model response: {text[:200]}") def infer_theme(text: str) -> str: for token in ["물가", "의료", "교통", "주거", "돌봄", "일자리", "교육", "환경", "경제"]: if token in text: return token return "기타" def coerce_int(value: Any, *, default: int) -> int: try: return int(round(float(value))) except (TypeError, ValueError): return default def bounded_int(value: Any, default: int, *, minimum: int, maximum: int) -> int: try: parsed = int(float(value)) except (TypeError, ValueError): parsed = default return max(minimum, min(maximum, parsed)) def float_or_zero(value: Any) -> float: try: return float(value) except (TypeError, ValueError): return 0.0