2b414d13b7
Live test found: deepseek returned 14/15 personas -> whole analyze 500'd, breaking the 'กดสร้าง -> auto-analyze' flow. Now generate() retries up to 3x with a nudge, and accepts a short result (>=8 personas) instead of crashing — admin can top up the rest with 'สร้างบุคคลต้นแบบเพิ่มเติม'. Also default tolerance/recontact on generated personas. All backend suites pass.
93 lines
3.6 KiB
Python
93 lines
3.6 KiB
Python
"""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
|
|
|
|
|
|
class PersonaGenerator:
|
|
def __init__(self, llm: LLMClient) -> None:
|
|
self.llm = llm
|
|
|
|
def generate(
|
|
self,
|
|
*,
|
|
sales_kit: dict[str, Any],
|
|
language: str = "en",
|
|
channel: str = "facebook",
|
|
) -> 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) >= 15) 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 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/13), we ACCEPT what we
|
|
# got rather than crashing the whole analyze — the caller/UI can top up with
|
|
# "create more personas". A retry loop lives in generate().
|
|
return normalized
|