"""Local Nemotron-Personas-Korea dataset support for the simulation UI. The dataset is intentionally treated as a local asset after download. Runtime sampling reads downloaded parquet shards through DuckDB; it does not call the Hugging Face dataset viewer for rows. """ from __future__ import annotations import json import math import os import re import time import urllib.request from pathlib import Path from typing import Any APP_DIR = Path(__file__).resolve().parent DATASET_ID = "nvidia/Nemotron-Personas-Korea" DATASET_CONFIG = "default" DATASET_SPLIT = "train" DATA_DIR = Path(os.getenv("NEMOTRON_KOREA_DATA_DIR", APP_DIR / "data" / "nemotron_personas_korea")).expanduser() PARQUET_DIR = DATA_DIR / "parquet" MANIFEST_PATH = DATA_DIR / "manifest.json" DUCKDB_PATH = DATA_DIR / "personas.duckdb" DATASET_COLUMNS = [ "uuid", "professional_persona", "sports_persona", "arts_persona", "travel_persona", "culinary_persona", "family_persona", "persona", "cultural_background", "skills_and_expertise", "skills_and_expertise_list", "hobbies_and_interests", "hobbies_and_interests_list", "career_goals_and_ambitions", "sex", "age", "marital_status", "military_status", "family_type", "housing_type", "education_level", "bachelors_field", "occupation", "district", "province", "country", ] TEXT_FIELDS = [ "persona", "professional_persona", "family_persona", "cultural_background", "skills_and_expertise", "hobbies_and_interests", "career_goals_and_ambitions", "sports_persona", "arts_persona", "travel_persona", "culinary_persona", ] DEFAULT_SELECTED_FIELDS = [ "persona", "professional_persona", "family_persona", "hobbies_and_interests", "skills_and_expertise", "career_goals_and_ambitions", ] SEX_ALIASES = { "여성": "여자", "여": "여자", "여자": "여자", "female": "여자", "f": "여자", "woman": "여자", "women": "여자", "남성": "남자", "남": "남자", "남자": "남자", "male": "남자", "m": "남자", "man": "남자", "men": "남자", } PROVINCE_ALIASES = { "seoul": "서울", "서울시": "서울", "서울특별시": "서울", "busan": "부산", "부산시": "부산", "부산광역시": "부산", "gyeonggi": "경기", "gyeonggi-do": "경기", "경기도": "경기", "incheon": "인천", "daegu": "대구", "daejeon": "대전", "gwangju": "광주", "ulsan": "울산", "sejong": "세종", "jeju": "제주", "jeju-do": "제주", "강원도": "강원", "chungcheongnam-do": "충청남", "chungcheongbuk-do": "충청북", "jeollanam-do": "전라남", "jeollabuk-do": "전북", "gyeongsangnam-do": "경상남", "gyeongsangbuk-do": "경상북", } CATEGORICAL_FIELDS = [ "sex", "marital_status", "military_status", "family_type", "housing_type", "education_level", "bachelors_field", "occupation", "district", "province", "country", ] SAMPLING_FIELDS = [ "uuid", "sex", "age", "marital_status", "military_status", "family_type", "housing_type", "education_level", "bachelors_field", "occupation", "district", "province", "country", *TEXT_FIELDS, ] class PersonaDatasetError(RuntimeError): """Raised when the local dataset is unavailable or cannot be queried.""" def _duckdb_module() -> Any: try: import duckdb # type: ignore except ModuleNotFoundError as exc: raise PersonaDatasetError( "duckdb is required for local Nemotron sampling. Install it in the " "Python environment that runs social-sim-ui: python -m pip install duckdb" ) from exc return duckdb def _json_url(url: str, timeout: int = 60) -> Any: request = urllib.request.Request(url, headers={"User-Agent": "social-sim-ui/0.1"}) with urllib.request.urlopen(request, timeout=timeout) as response: return json.loads(response.read().decode("utf-8")) def _parquet_glob() -> str: return str(PARQUET_DIR / "*.parquet") def _safe_sql_literal(value: str) -> str: return "'" + value.replace("'", "''") + "'" def _read_manifest() -> dict[str, Any] | None: if not MANIFEST_PATH.exists(): return None try: return json.loads(MANIFEST_PATH.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None def fetch_manifest() -> dict[str, Any]: """Fetch and persist a parquet URL manifest from Hugging Face.""" url = f"https://huggingface.co/api/datasets/{DATASET_ID}/parquet" payload = _json_url(url) urls = payload.get(DATASET_CONFIG, {}).get(DATASET_SPLIT) if not isinstance(urls, list) or not urls: raise PersonaDatasetError(f"Could not find {DATASET_CONFIG}/{DATASET_SPLIT} parquet URLs for {DATASET_ID}") manifest = { "dataset_id": DATASET_ID, "config": DATASET_CONFIG, "split": DATASET_SPLIT, "fetched_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "parquet_urls": urls, "columns": DATASET_COLUMNS, "license": "cc-by-4.0", } DATA_DIR.mkdir(parents=True, exist_ok=True) MANIFEST_PATH.write_text(json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8") return manifest def download_dataset(*, force: bool = False, limit_shards: int | None = None) -> dict[str, Any]: """Download parquet shards locally. ``limit_shards`` is for smoke tests. Production use should leave it unset. """ manifest = _read_manifest() or fetch_manifest() urls = list(manifest.get("parquet_urls") or []) if limit_shards is not None: urls = urls[: max(0, int(limit_shards))] if not urls: raise PersonaDatasetError("manifest contains no parquet URLs") PARQUET_DIR.mkdir(parents=True, exist_ok=True) downloaded: list[dict[str, Any]] = [] skipped: list[dict[str, Any]] = [] for idx, url in enumerate(urls): target = PARQUET_DIR / f"{idx}.parquet" if target.exists() and target.stat().st_size > 0 and not force: skipped.append({"shard": idx, "path": str(target), "bytes": target.stat().st_size}) continue tmp = target.with_suffix(".parquet.tmp") request = urllib.request.Request(url, headers={"User-Agent": "social-sim-ui/0.1"}) with urllib.request.urlopen(request, timeout=120) as response, tmp.open("wb") as handle: while True: chunk = response.read(1024 * 1024) if not chunk: break handle.write(chunk) tmp.replace(target) downloaded.append({"shard": idx, "path": str(target), "bytes": target.stat().st_size}) return { "dataset_id": DATASET_ID, "data_dir": str(DATA_DIR), "manifest": str(MANIFEST_PATH), "downloaded": downloaded, "skipped": skipped, "status": dataset_status(), } def dataset_status() -> dict[str, Any]: files = sorted(PARQUET_DIR.glob("*.parquet")) if PARQUET_DIR.exists() else [] total_bytes = sum(path.stat().st_size for path in files if path.exists()) duckdb_available = True duckdb_error = "" try: _duckdb_module() except PersonaDatasetError as exc: duckdb_available = False duckdb_error = str(exc) manifest = _read_manifest() expected_shards = len(manifest.get("parquet_urls", [])) if manifest else None return { "dataset_id": DATASET_ID, "config": DATASET_CONFIG, "split": DATASET_SPLIT, "license": "cc-by-4.0", "data_dir": str(DATA_DIR), "manifest_exists": MANIFEST_PATH.exists(), "parquet_dir": str(PARQUET_DIR), "parquet_shards": len(files), "expected_shards": expected_shards, "download_complete": bool(files) and expected_shards is not None and len(files) >= expected_shards, "downloaded_bytes": total_bytes, "duckdb_available": duckdb_available, "duckdb_error": duckdb_error, "duckdb_path": str(DUCKDB_PATH), "columns": DATASET_COLUMNS, "default_selected_fields": DEFAULT_SELECTED_FIELDS, } def _connect_duckdb() -> Any: duckdb = _duckdb_module() if not PARQUET_DIR.exists() or not list(PARQUET_DIR.glob("*.parquet")): raise PersonaDatasetError( f"No local parquet shards found in {PARQUET_DIR}. Run scripts/download_nemotron_personas_korea.py first." ) DATA_DIR.mkdir(parents=True, exist_ok=True) conn = duckdb.connect(str(DUCKDB_PATH)) conn.execute( f"CREATE OR REPLACE VIEW personas AS SELECT * FROM read_parquet({_safe_sql_literal(_parquet_glob())})" ) return conn def dataset_metadata() -> dict[str, Any]: status = dataset_status() if not status["parquet_shards"]: return {**status, "available": False, "error": "dataset parquet files are not downloaded"} conn = _connect_duckdb() try: row_count = int(conn.execute("SELECT count(*) FROM personas").fetchone()[0]) sex_counts = _top_counts(conn, "sex", 20) province_counts = _top_counts(conn, "province", 40) district_counts_by_province = _district_counts_by_province(conn) education_counts = _top_counts(conn, "education_level", 30) occupation_counts = _top_counts(conn, "occupation", 40) age_buckets = conn.execute( """ SELECT CASE WHEN TRY_CAST(age AS INTEGER) IS NULL THEN 'unknown' WHEN TRY_CAST(age AS INTEGER) < 20 THEN 'under_20' WHEN TRY_CAST(age AS INTEGER) < 30 THEN '20s' WHEN TRY_CAST(age AS INTEGER) < 40 THEN '30s' WHEN TRY_CAST(age AS INTEGER) < 50 THEN '40s' WHEN TRY_CAST(age AS INTEGER) < 60 THEN '50s' ELSE '60_plus' END AS bucket, count(*) AS count FROM personas GROUP BY bucket ORDER BY count DESC """ ).fetchall() finally: conn.close() return { **status, "available": True, "row_count": row_count, "sex_counts": sex_counts, "province_counts": province_counts, "district_counts_by_province": district_counts_by_province, "education_counts": education_counts, "occupation_counts": occupation_counts, "age_buckets": [{"value": value, "count": int(count)} for value, count in age_buckets], } def _top_counts(conn: Any, field: str, limit: int) -> list[dict[str, Any]]: if field not in CATEGORICAL_FIELDS: return [] rows = conn.execute( f""" SELECT {field} AS value, count(*) AS count FROM personas WHERE {field} IS NOT NULL AND CAST({field} AS VARCHAR) != '' GROUP BY {field} ORDER BY count DESC, value LIMIT ? """, [int(limit)], ).fetchall() return [{"value": str(value), "count": int(count)} for value, count in rows] def _district_counts_by_province(conn: Any) -> dict[str, list[dict[str, Any]]]: rows = conn.execute( """ SELECT province, district, count(*) AS count FROM personas WHERE province IS NOT NULL AND district IS NOT NULL AND CAST(province AS VARCHAR) != '' AND CAST(district AS VARCHAR) != '' GROUP BY province, district ORDER BY province, count DESC, district """ ).fetchall() grouped: dict[str, list[dict[str, Any]]] = {} for province, district, count in rows: province_text = str(province) grouped.setdefault(province_text, []).append({"value": str(district), "count": int(count)}) return grouped def build_index() -> dict[str, Any]: """Create a small DuckDB database with a persistent parquet-backed view.""" conn = _connect_duckdb() try: row_count = int(conn.execute("SELECT count(*) FROM personas").fetchone()[0]) conn.execute("CREATE TABLE IF NOT EXISTS dataset_cache_metadata(key VARCHAR PRIMARY KEY, value VARCHAR)") conn.execute( "INSERT OR REPLACE INTO dataset_cache_metadata VALUES (?, ?)", ["row_count", str(row_count)], ) conn.execute( "INSERT OR REPLACE INTO dataset_cache_metadata VALUES (?, ?)", ["built_at", time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())], ) finally: conn.close() return {"duckdb_path": str(DUCKDB_PATH), "row_count": row_count, "status": dataset_status()} def sample_personas(spec: dict[str, Any] | None = None) -> dict[str, Any]: spec = spec or {} total = _bounded_int(spec.get("total_agents"), 10, minimum=1, maximum=_max_dataset_agents(spec)) seed = str(spec.get("seed", 7)) selected_fields = _selected_fields(spec.get("selected_fields")) base_filters = _normalize_filters(spec.get("filters") if isinstance(spec.get("filters"), dict) else {}) groups = _normalize_sampling_groups(spec.get("groups") if isinstance(spec.get("groups"), list) else []) conn = _connect_duckdb() try: rows: list[dict[str, Any]] = [] seen: set[str] = set() allocations = _allocate_group_counts(total, groups) if allocations: for idx, group in enumerate(allocations): merged_filters = {**base_filters, **group["filters"]} sampled = _query_sample( conn, count=group["count"], seed=f"{seed}:group:{idx}:{group['label']}", filters=merged_filters, selected_fields=selected_fields, exclude_uuids=seen, ) for row in sampled: row["_sampling_group"] = group["label"] seen.add(str(row.get("uuid", ""))) rows.extend(sampled) if len(rows) < total: top_up = _query_sample( conn, count=total - len(rows), seed=f"{seed}:topup", filters=base_filters, selected_fields=selected_fields, exclude_uuids=seen, ) rows.extend(top_up) finally: conn.close() agents = project_rows_to_agents( rows[:total], scenario_text=str(spec.get("scenario_text") or ""), policy_text=str(spec.get("policy_text") or ""), decision_task=spec.get("decision_task") if isinstance(spec.get("decision_task"), dict) else None, selected_fields=selected_fields, ) return { "dataset_id": DATASET_ID, "requested_agents": total, "sampled_rows": len(rows[:total]), "selected_fields": selected_fields, "filters": base_filters, "groups": groups, "agents": agents, "population": { "total_agents": len(agents), "groups": _groups_from_agents(agents), "persona_source": { "type": "nemotron_korea", "dataset_id": DATASET_ID, "selected_fields": selected_fields, "filters": base_filters, "groups": groups, }, }, } def _max_dataset_agents(spec: dict[str, Any]) -> int: try: value = int(spec.get("max_agents", os.getenv("SOCIAL_SIM_DATASET_AGENT_LIMIT", "5000"))) except (TypeError, ValueError): value = 5000 return max(1, min(100000, value)) def _selected_fields(raw: Any) -> list[str]: if not isinstance(raw, list): return list(DEFAULT_SELECTED_FIELDS) fields = [str(field) for field in raw if str(field) in TEXT_FIELDS] return fields or list(DEFAULT_SELECTED_FIELDS) def _normalize_filters(filters: dict[str, Any]) -> dict[str, Any]: normalized: dict[str, Any] = {} for key, value in filters.items(): normalized[str(key)] = _normalize_filter_value(str(key), value) return normalized def _normalize_sampling_groups(groups: list[Any]) -> list[Any]: normalized: list[Any] = [] for item in groups: if not isinstance(item, dict): continue group = dict(item) if isinstance(group.get("filters"), dict): group["filters"] = _normalize_filters(group["filters"]) normalized.append(group) return normalized def _normalize_filter_value(field: str, value: Any) -> Any: if isinstance(value, list): return [_normalize_filter_value(field, item) for item in value] text = str(value).strip() if value is not None else "" if not text: return value lowered = text.lower() if field == "sex": return SEX_ALIASES.get(lowered, SEX_ALIASES.get(text, text)) if field == "province": return PROVINCE_ALIASES.get(lowered, PROVINCE_ALIASES.get(text, text)) return value def _allocate_group_counts(total: int, groups: list[Any]) -> list[dict[str, Any]]: valid: list[dict[str, Any]] = [] for idx, item in enumerate(groups): if not isinstance(item, dict): continue filters = _normalize_filters(item.get("filters") if isinstance(item.get("filters"), dict) else {}) label = str(item.get("label") or f"group_{idx + 1}") count = _int_or_none(item.get("count")) ratio = _float_or_none(item.get("ratio")) valid.append({"label": label, "filters": filters, "count": count, "ratio": ratio}) if not valid: return [] fixed_total = sum(item["count"] for item in valid if isinstance(item.get("count"), int) and item["count"] > 0) ratio_items = [item for item in valid if not isinstance(item.get("count"), int) or item["count"] <= 0] remaining = max(0, total - fixed_total) ratio_sum = sum(max(0.0, float(item.get("ratio") or 0.0)) for item in ratio_items) allocated: list[dict[str, Any]] = [] for item in valid: if isinstance(item.get("count"), int) and item["count"] > 0: count = min(item["count"], total) fraction = 0.0 elif ratio_sum > 0: raw = remaining * max(0.0, float(item.get("ratio") or 0.0)) / ratio_sum count = int(math.floor(raw)) fraction = raw - count else: raw = remaining / max(1, len(ratio_items)) count = int(math.floor(raw)) fraction = raw - count allocated.append({"label": item["label"], "filters": item["filters"], "count": count, "_fraction": fraction}) while sum(item["count"] for item in allocated) < total: target = max(allocated, key=lambda item: item.get("_fraction", 0.0)) target["count"] += 1 target["_fraction"] = 0.0 while sum(item["count"] for item in allocated) > total: target = max(allocated, key=lambda item: item["count"]) target["count"] -= 1 return [{key: value for key, value in item.items() if key != "_fraction" and item["count"] > 0} for item in allocated] def _query_sample( conn: Any, *, count: int, seed: str, filters: dict[str, Any], selected_fields: list[str], exclude_uuids: set[str], ) -> list[dict[str, Any]]: if count <= 0: return [] columns = _query_columns(selected_fields) where_sql, params = _where_clause(filters, exclude_uuids) params.extend([str(seed), int(count)]) rows = conn.execute( f""" SELECT {", ".join(columns)} FROM personas {where_sql} ORDER BY hash(CAST(uuid AS VARCHAR) || CAST(? AS VARCHAR)) LIMIT ? """, params, ).fetchall() names = [column.split(" AS ")[-1] if " AS " in column else column for column in columns] return [dict(zip(names, row)) for row in rows] def _query_columns(selected_fields: list[str]) -> list[str]: base = [ "uuid", "sex", "age", "marital_status", "military_status", "family_type", "housing_type", "education_level", "bachelors_field", "occupation", "district", "province", "country", ] for field in selected_fields: if field not in base: base.append(field) return base def _where_clause(filters: dict[str, Any], exclude_uuids: set[str]) -> tuple[str, list[Any]]: clauses: list[str] = [] params: list[Any] = [] for field in CATEGORICAL_FIELDS: if field not in filters: continue value = filters.get(field) if isinstance(value, list): clean_values = [str(item) for item in value if str(item).strip()] if clean_values: placeholders = ", ".join("?" for _ in clean_values) clauses.append(f"CAST({field} AS VARCHAR) IN ({placeholders})") params.extend(clean_values) elif value not in {None, ""}: clauses.append(f"CAST({field} AS VARCHAR) = ?") params.append(str(value)) age_min = _int_or_none(filters.get("age_min")) age_max = _int_or_none(filters.get("age_max")) if age_min is not None: clauses.append("TRY_CAST(age AS INTEGER) >= ?") params.append(age_min) if age_max is not None: clauses.append("TRY_CAST(age AS INTEGER) <= ?") params.append(age_max) occupation_query = str(filters.get("occupation_query") or "").strip().lower() if occupation_query: clauses.append("(lower(CAST(occupation AS VARCHAR)) LIKE ? OR lower(CAST(professional_persona AS VARCHAR)) LIKE ?)") pattern = f"%{occupation_query}%" params.extend([pattern, pattern]) keyword_query = str(filters.get("keyword_query") or "").strip().lower() if keyword_query: text_expr = " || ' ' || ".join(f"coalesce(CAST({field} AS VARCHAR), '')" for field in TEXT_FIELDS[:7]) clauses.append(f"lower({text_expr}) LIKE ?") params.append(f"%{keyword_query}%") if exclude_uuids: placeholders = ", ".join("?" for _ in exclude_uuids) clauses.append(f"CAST(uuid AS VARCHAR) NOT IN ({placeholders})") params.extend(sorted(exclude_uuids)) if not clauses: return "", params return "WHERE " + " AND ".join(clauses), params def project_rows_to_agents( rows: list[dict[str, Any]], *, scenario_text: str = "", policy_text: str = "", decision_task: dict[str, Any] | None = None, selected_fields: list[str] | None = None, ) -> list[dict[str, Any]]: selected_fields = selected_fields or DEFAULT_SELECTED_FIELDS agents: list[dict[str, Any]] = [] for idx, row in enumerate(rows, start=1): text = " ".join(str(row.get(field) or "") for field in TEXT_FIELDS) dimensions = _infer_behavior_dimensions(row, text) baseline = _baseline_from_dimensions(dimensions) location = " ".join(str(row.get(key) or "").strip() for key in ["province", "district"] if row.get(key)).strip() occupation = str(row.get("occupation") or "").strip() age = str(row.get("age") or "").strip() sex = str(row.get("sex") or "").strip() role_parts = [part for part in [location, f"{age}세" if age else "", sex, occupation] if part] role = " ".join(role_parts) or "Nemotron persona participant" persona_type = _slugify( "_".join(str(part) for part in [row.get("province"), row.get("occupation"), row.get("age")] if part), f"nemotron_persona_{idx}", ) prompt_card = build_prompt_card(row, selected_fields=selected_fields) agents.append( { "agent_id": f"agent_{idx:04d}", "name": f"Persona {idx:04d}", "persona_type": persona_type, "role": role, "goal": _goal_from_context(row, scenario_text, policy_text, decision_task), "persona_description": prompt_card, "baseline_usage": baseline, "behavior_dimensions": dimensions, "traits": _traits_from_row(row, dimensions), "state": { "activation": "eligible", "last_numeric_decision": None, "memory_summary": "", }, "persona_source": { "type": "nemotron_korea", "dataset_id": DATASET_ID, "uuid": str(row.get("uuid") or ""), "sampling_group": str(row.get("_sampling_group") or ""), "selected_fields": selected_fields, }, } ) return agents def build_prompt_card(row: dict[str, Any], *, selected_fields: list[str] | None = None, field_limit: int = 360) -> str: selected_fields = selected_fields or DEFAULT_SELECTED_FIELDS lines = [ f"성별: {_clip(row.get('sex'), 80)}", f"연령: {_clip(row.get('age'), 80)}", f"지역: {_clip(' '.join(str(row.get(key) or '') for key in ['country', 'province', 'district']).strip(), 140)}", f"학력/전공: {_clip(' / '.join(str(row.get(key) or '') for key in ['education_level', 'bachelors_field']).strip(' /'), 180)}", f"직업: {_clip(row.get('occupation'), 180)}", f"가족/주거: {_clip(' / '.join(str(row.get(key) or '') for key in ['marital_status', 'family_type', 'housing_type']).strip(' /'), 180)}", ] for field in selected_fields: value = _clip(row.get(field), field_limit) if value: lines.append(f"{field}: {value}") return "\n".join(line for line in lines if not line.endswith(": ")) def _infer_behavior_dimensions(row: dict[str, Any], text: str) -> dict[str, float]: lowered = text.lower() age = _int_or_none(row.get("age")) or 40 occupation = str(row.get("occupation") or "").lower() education = str(row.get("education_level") or "").lower() family = str(row.get("family_type") or "").lower() cooperation_terms = ["봉사", "지역", "가족", "교육", "상담", "간호", "의료", "협력", "community", "team", "mentor"] self_terms = ["창업", "투자", "영업", "관리자", "경영", "ambition", "entrepreneur", "business", "sales", "finance"] risk_terms = ["안전", "의료", "법", "회계", "보험", "품질", "security", "compliance", "risk"] authority_terms = ["군", "공무", "교사", "관리", "법", "행정", "manager", "government", "military", "teacher"] social_terms = ["예술", "스포츠", "여행", "요리", "가족", "커뮤니티", "media", "arts", "sports", "travel"] cooperation = 0.45 + 0.08 * _contains_any(lowered, cooperation_terms) + 0.06 * bool(family) self_interest = 0.45 + 0.1 * _contains_any(lowered + " " + occupation, self_terms) risk_aversion = 0.42 + 0.12 * _contains_any(lowered + " " + occupation, risk_terms) + min(0.16, max(0, age - 35) / 200) authority = 0.42 + 0.12 * _contains_any(lowered + " " + occupation + " " + education, authority_terms) + min(0.12, max(0, age - 30) / 250) social = 0.42 + 0.12 * _contains_any(lowered, social_terms) if age < 30: social += 0.05 risk_aversion -= 0.03 if "single" in family or "미혼" in family: self_interest += 0.04 return { "cooperation": _clamp01(cooperation), "self_interest": _clamp01(self_interest), "risk_aversion": _clamp01(risk_aversion), "authority_respect": _clamp01(authority), "social_influence": _clamp01(social), } def _traits_from_row(row: dict[str, Any], dimensions: dict[str, float]) -> dict[str, Any]: return { "demographic": { "sex": row.get("sex"), "age": row.get("age"), "province": row.get("province"), "district": row.get("district"), "occupation": row.get("occupation"), "education_level": row.get("education_level"), }, "behavior_projection": dimensions, } def _goal_from_context( row: dict[str, Any], scenario_text: str, policy_text: str, decision_task: dict[str, Any] | None, ) -> str: task_text = "" if decision_task: task_text = str(decision_task.get("action_prompt") or decision_task.get("metric_name") or "") occupation = str(row.get("occupation") or "participant").strip() context = _clip(task_text or scenario_text or policy_text, 220) if context: return f"{occupation} 관점에서 {context}에 대해 개인 상황과 사회적 결과를 함께 고려한다." return f"{occupation} 관점에서 개인 상황과 사회적 결과를 함께 고려한다." def _baseline_from_dimensions(dimensions: dict[str, float]) -> int: raw = 6.0 + 3.2 * dimensions["self_interest"] - 2.0 * dimensions["cooperation"] - 1.4 * dimensions["risk_aversion"] return max(0, min(12, int(round(raw)))) def _groups_from_agents(agents: list[dict[str, Any]]) -> dict[str, int]: groups: dict[str, int] = {} for agent in agents: key = str(agent.get("persona_type") or "nemotron_persona") groups[key] = groups.get(key, 0) + 1 return groups def _contains_any(text: str, terms: list[str]) -> int: return 1 if any(term.lower() in text for term in terms) else 0 def _clip(value: Any, limit: int) -> str: text = str(value or "").strip() if len(text) <= limit: return text return text[: limit - 3].rstrip() + "..." def _slugify(value: Any, fallback: str) -> str: text = str(value or "").strip().lower() text = re.sub(r"[^0-9a-zA-Z가-힣]+", "_", text).strip("_") if not text: return fallback return text[:80] def _bounded_int(value: Any, default: int, *, minimum: int, maximum: int) -> int: parsed = _int_or_none(value) if parsed is None: parsed = default return max(minimum, min(maximum, parsed)) def _int_or_none(value: Any) -> int | None: try: return int(float(value)) except (TypeError, ValueError): return None def _float_or_none(value: Any) -> float | None: try: return float(value) except (TypeError, ValueError): return None def _clamp01(value: float) -> float: return round(max(0.0, min(1.0, float(value))), 3)