Files
ai-podcast/backend/services/caller_gen.py
T
lukeandClaude Opus 4.6 d0bb50fe12 Fix Silas identity/voice leak, avatar gender, and add pre-warm batch gen
session.caller was reading name/voice from CALLER_BASES randomized defaults
instead of the slim bg dict — causing Silas to appear as "Earl" with wrong
voice. Now prefers bg data when populated.

Also: avatar endpoint infers gender from voice pool, /api/callers returns bg
name, regulars get canonical voice force-locked, batch gen splits into two
parallel calls with pre-warm so subsequent resets are instant.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-05 19:08:46 -06:00

186 lines
7.7 KiB
Python

from dataclasses import dataclass
from typing import Optional
import json
import httpx
from ..config import settings
from .cost_tracker import cost_tracker
BATCH_MODEL = "anthropic/claude-sonnet-4.6"
REQUIRED_FIELDS = {
"name", "age", "voice_suggestion", "location", "identity",
"situation", "reason_calling", "opening_line", "secret_want",
"specific_details", "emotional_register"
}
@dataclass
class CallerIdentity:
name: str
age: int
voice_suggestion: str
location: str
identity: str
situation: str
reason_calling: str
opening_line: str
secret_want: str
specific_details: list[str]
emotional_register: str
# set after voice validation
voice_resolved: Optional[str] = None
def parse_batch_response(raw: str) -> list[CallerIdentity]:
stripped = raw.strip()
if stripped.startswith("```") and stripped.endswith("```"):
lines = stripped.splitlines()
stripped = "\n".join(lines[1:-1])
data = json.loads(stripped)
callers = data.get("callers", [])
result = []
for c in callers:
missing = REQUIRED_FIELDS - set(c.keys())
if missing:
raise ValueError(f"CallerIdentity missing fields: {missing}")
result.append(CallerIdentity(**{k: c[k] for k in REQUIRED_FIELDS}))
return result
def resolve_voice(suggestion: str, roster: list[str]) -> str:
"""Map sonnet's voice suggestion to a real voice in the roster.
Case-insensitive exact match; deterministic fallback to first roster entry."""
if not suggestion or not roster:
return roster[0] if roster else ""
lower_map = {v.lower(): v for v in roster}
return lower_map.get(suggestion.lower(), roster[0])
BATCH_SYSTEM_PROMPT = """You are writing a roster of callers for Luke's late-night radio show in New Mexico.
CREATIVE RANGE: Your callers must span the emotional range of Howard Stern (chaos, strong characters), Coast to Coast AM (earnest weirdos, sincere believers), Loveline (real problems, real advice-seeking), Delilah (emotional vulnerability, connection), and Opie and Anthony (sharp, irreverent, specific people).
Maximum character distance between callers. No two callers should feel like siblings. Do not default to sitcom plots. Real humans are specific and strange. Give each caller details that could only belong to them.
You will output strict JSON with a "callers" array. Each caller has exactly these fields: name, age, voice_suggestion, location, identity, situation, reason_calling, opening_line, secret_want, specific_details (array of 2-3 strings), emotional_register."""
def build_batch_prompt(ctx: dict) -> str:
lines = [
f"Tonight is {ctx['date']}. {ctx['weather']}.",
"",
"Today's news headlines (ground callers in real context, but do not force topicality):",
]
for h in ctx["headlines"]:
lines.append(f"- {h}")
lines.append("")
if ctx["recent_caller_summaries"]:
lines.append("Recent callers (DO NOT repeat these archetypes or situations):")
for s in ctx["recent_caller_summaries"]:
lines.append(f"- {s}")
lines.append("")
if ctx["regulars_included"]:
lines.append("RECURRING CHARACTERS IN TONIGHT'S LINEUP:")
lines.append("")
for r in ctx["regulars_included"]:
lines.append(f"### {r['name']}")
lines.append(r["lore"])
lines.append(f"Current arc state: {r['arc_state']}")
lines.append("")
lines.append(f"For {r['name']}: invent a fresh reason he is calling tonight — a new development, grievance, or specific recent event. DO NOT alter his voice, personality, or core traits. Write a new scene for an existing character.")
lines.append(f"CRITICAL — this caller MUST appear in the output JSON with EXACTLY these fields locked: name=\"{r['name']}\", voice_suggestion=\"{r['voice']}\", age={r['age']}. Do NOT rename this caller. Do NOT give him a different voice. Do NOT change his age. Only invent: location, identity, situation, reason_calling, opening_line, secret_want, specific_details, emotional_register.")
lines.append("")
lines.append(f"Available voices (voice_suggestion must match one of these exactly):")
lines.append(", ".join(ctx["voice_roster"]))
lines.append("")
lines.append(f"Generate {ctx['caller_count']} callers. Output JSON only, no prose.")
return BATCH_SYSTEM_PROMPT + "\n\n" + "\n".join(lines)
async def generate_batch(ctx: dict) -> list[CallerIdentity]:
"""Call sonnet-4.6 with the batch prompt, parse + voice-resolve the response."""
prompt = build_batch_prompt(ctx)
async with httpx.AsyncClient(timeout=120.0) as client:
resp = await client.post(
"https://openrouter.ai/api/v1/chat/completions",
headers={"Authorization": f"Bearer {settings.openrouter_api_key}"},
json={
"model": BATCH_MODEL,
"messages": [{"role": "user", "content": prompt}],
"response_format": {"type": "json_object"},
"max_tokens": 8000,
"temperature": 0.9,
},
)
resp.raise_for_status()
data = resp.json()
content = data["choices"][0]["message"]["content"]
usage = data.get("usage", {})
cost_tracker.record_llm_call(
category="background_gen",
model=BATCH_MODEL,
usage_data=usage,
)
callers = parse_batch_response(content)
for c in callers:
c.voice_resolved = resolve_voice(c.voice_suggestion, ctx["voice_roster"])
return callers
REGULAR_SYSTEM_PROMPT = """You are writing a single caller — a recurring character on Luke's late-night radio show. The character's identity is fixed (name, voice, age, core lore). You only invent a fresh reason he's calling tonight, grounded in his current arc state.
Output strict JSON with these fields only: location, identity, situation, reason_calling, opening_line, secret_want, specific_details (array of 2-3 strings), emotional_register. Do NOT include name, voice_suggestion, or age — those are locked."""
async def generate_regular_situation(regular: dict, ctx: dict) -> dict:
"""Generate a fresh situation for a locked regular. Returns a dict matching
the CallerIdentity schema. One small LLM call (~$0.01)."""
lines = [
f"Tonight is {ctx['date']}. {ctx['weather']}.",
"",
f"### {regular['name']}",
regular["lore"],
f"Current arc state: {regular['arc_state']}",
"",
f"Invent a fresh reason {regular['name']} is calling tonight — a new development, grievance, or specific recent event. DO NOT alter his voice, personality, or core traits.",
"",
"Output JSON only, no prose.",
]
prompt = REGULAR_SYSTEM_PROMPT + "\n\n" + "\n".join(lines)
async with httpx.AsyncClient(timeout=60.0) as client:
resp = await client.post(
"https://openrouter.ai/api/v1/chat/completions",
headers={"Authorization": f"Bearer {settings.openrouter_api_key}"},
json={
"model": BATCH_MODEL,
"messages": [{"role": "user", "content": prompt}],
"response_format": {"type": "json_object"},
"max_tokens": 1500,
"temperature": 0.9,
},
)
resp.raise_for_status()
data = resp.json()
content = data["choices"][0]["message"]["content"]
usage = data.get("usage", {})
cost_tracker.record_llm_call(
category="background_gen",
model=BATCH_MODEL,
usage_data=usage,
)
stripped = content.strip()
if stripped.startswith("```") and stripped.endswith("```"):
lines = stripped.splitlines()
stripped = "\n".join(lines[1:-1])
return json.loads(stripped)