"""Persona generator: builds 15 personas (5 per tier) from a Sales Kit + scenario.""" from __future__ import annotations import json from typing import Any from ..llm import LLMClient from .persona_prompts import PERSONA_SYSTEM TIERS = ["A", "B", "C"] PER_TIER = 5 # 5 per tier = 15 total TARGET = 15 # total personas the system generates (no "add more" button needed) class PersonaGenerator: def __init__(self, llm: LLMClient) -> None: self.llm = llm def generate( self, *, sales_kit: dict[str, Any], language: str = "en", channel: str = "social", ) -> list[dict[str, Any]]: kit_json = json.dumps(sales_kit, ensure_ascii=False)[:12000] lang_name = "Thai" if language == "th" else "English" scenario = (sales_kit.get("scenarioFrame") or "").strip() or "a general product sale" user_prompt = ( f"Platform/channel preference: {channel}\n" f"Language: {lang_name} (all persona text in {lang_name})\n" f"Sales Kit:\n{kit_json}\n\n" f"Generate exactly 15 personas (5 per tier A/B/C) as JSON." ) # Real LLMs sometimes return fewer than 15 (truncation / merge). Retry up to 2 extra # times with a nudge; the body below already ACCEPTS short results (>= 8) instead of # hard-failing, so these retries are just best-effort to reach a fuller set. attempt = 0 while True: attempt += 1 prompt = user_prompt + ( "" if attempt == 1 else "\n\n(Note: you left some personas out — please output all 15, one JSON object per persona, no extra prose.)" ) result = self.llm.complete_json( PERSONA_SYSTEM, prompt, temperature=0.8, max_tokens=14000 ) personas = result.get("personas") or [] if (isinstance(personas, list) and len(personas) >= TARGET) or attempt >= 3: break if not isinstance(personas, list) or not personas: raise ValueError("persona generator returned no personas") normalized, counts = [], {"A": 0, "B": 0, "C": 0} for idx, p in enumerate(personas, start=1): if len(normalized) >= TARGET: break # already reached 20 total if not isinstance(p, dict): continue tier = p.get("tier", p.get("intent_tier")) if tier not in TIERS: tier = "B" if counts[tier] >= PER_TIER: continue # skip overflow per tier counts[tier] += 1 p["id"] = f"persona-{idx:02d}" p["tier"] = tier p["channel"] = p.get("channel", channel) p.setdefault("initiation_mode", "customer") p.setdefault("special", "") p.setdefault("difficulty", 1) p.setdefault("pains", []) p.setdefault("negotiation_levers", []) p.setdefault("objections", []) p.setdefault("tolerance", 3) p.setdefault("recontact", False) normalized.append(p) # Wrap tier-C: ensure at least one wrong_text persona if "C" in counts and not any( p.get("special") == "wrong_text" for p in normalized ): # find first tier-C and mark it for p in normalized: if p["tier"] == "C": p["special"] = "wrong_text" break if len(normalized) < 8: raise ValueError(f"expected ~15 personas, generated only {len(normalized)}") # NOTE: if we're short of 15 (real LLMs occasionally return 14), we ACCEPT what we # got rather than crashing the whole analyze — with TARGET=15 there's normally no gap. return normalized