from abc import ABC, abstractmethod from dataclasses import dataclass import json import os from pathlib import Path import queue import threading import time from typing import Any from pydantic import ValidationError from puppet_theater.models import Actor, ActorResponse, DirectorDecision, TheaterSession, ToolRequest from puppet_theater.prompts import ACTOR_LINE_PROMPT from puppet_theater import zerogpu MAX_ACTOR_LINE_CHARS = 220 DEFAULT_OPENBMB_MODEL_ID = "openbmb/MiniCPM5-1B" DEFAULT_HF_API_MODEL = "Qwen/Qwen3-4B-Instruct-2507:nscale" DEFAULT_HF_API_MODEL_ID = DEFAULT_HF_API_MODEL OPENBMB_MAX_NEW_TOKENS = 80 OPENBMB_TEMPERATURE = 0.8 HF_API_ACTOR_MAX_TOKENS = 120 # Show-bible JSON (title, setting, 3 roles, URLs) needs a much larger completion budget than one actor line. HF_API_SHOW_BIBLE_MAX_TOKENS = int(os.getenv("HF_API_SHOW_BIBLE_MAX_TOKENS", "1024")) HF_API_SHOW_BIBLE_MAX_TOKENS = max(256, min(HF_API_SHOW_BIBLE_MAX_TOKENS, 4096)) # Summoned single-actor JSON is smaller than full show bible but still needs headroom vs actor-line defaults. HF_API_SUMMON_ACTOR_MAX_TOKENS = int(os.getenv("HF_API_SUMMON_ACTOR_MAX_TOKENS", "512")) HF_API_SUMMON_ACTOR_MAX_TOKENS = max(128, min(HF_API_SUMMON_ACTOR_MAX_TOKENS, 1024)) HF_API_BACKDROP_URL_MAX_TOKENS = int(os.getenv("HF_API_BACKDROP_URL_MAX_TOKENS", "256")) HF_API_BACKDROP_URL_MAX_TOKENS = max(64, min(HF_API_BACKDROP_URL_MAX_TOKENS, 512)) HF_API_DIRECTOR_MAX_TOKENS = 180 HF_API_ACTOR_TEMPERATURE = 0.75 HF_API_DIRECTOR_TEMPERATURE = 0.35 HF_API_TOP_P = 0.9 HF_API_TIMEOUT_SECONDS = 30.0 DEFAULT_ACTOR_LORA_BASE_MODEL = "openbmb/MiniCPM5-1B" DEFAULT_ACTOR_LORA_ADAPTER = "build-small-hackathon/AI-Puppet-Theater-MiniCPM5-Actor-LoRA" DEFAULT_ACTOR_GGUF_REPO_ID = "build-small-hackathon/AI-Puppet-Theater-MiniCPM5-Actor-GGUF" DEFAULT_ACTOR_GGUF_FILENAME = "minicpm5-actor-q4_k_m.gguf" ACTOR_LORA_MAX_NEW_TOKENS = 160 ACTOR_LORA_TIMEOUT_SECONDS = 120.0 ACTOR_GGUF_MAX_TOKENS = 160 ACTOR_GGUF_TIMEOUT_SECONDS = 120.0 ACTOR_GGUF_N_GPU_LAYERS = -1 LOCAL_ACTOR_REPETITION_PENALTY = 1.12 LOCAL_ACTOR_TEMPERATURE = 0.35 LOCAL_ACTOR_TOP_P = 0.9 ACTOR_JSON_FIELDS = ( "intent", "line", "emotion", "gesture", "stage_effect", "memory_update", "tool_request", ) ACTOR_SYSTEM_MESSAGE = ( "You are an Actor agent in AI Puppet Theater. Return only one valid JSON object. " "No markdown. No commentary. Keep the puppet line short, theatrical, and speakable." ) ACTOR_RESPONSE_SUFFIX = ( "Return exactly one JSON object with only these top-level keys: " "intent, line, emotion, gesture, stage_effect, memory_update, tool_request. " "Do not include speaking_style, show_state, recent_transcript, or any copied input fields. " "Stop after the JSON object." ) @dataclass(frozen=True) class BackendGeneration: response: ActorResponse backend_name: str model_id: str | None fallback_used: bool validation_status: str load_status: str latency_ms: int | None = None error: str | None = None @dataclass(frozen=True) class BackendRuntimeStatus: backend_name: str model_id: str | None load_status: str token_configured: bool | None = None latest_latency_ms: int | None = None latest_validation_status: str | None = None latest_fallback_used: bool | None = None latest_fallback_reason: str | None = None max_new_tokens: int | None = None temperature: float | None = None zerogpu_enabled: bool = False spaces_available: bool = False zerogpu_gpu_active: bool = False torch_version: str | None = None cuda_available_in_gpu_fn: bool | None = None class ModelBackend(ABC): name: str = "base" model_id: str | None = None @abstractmethod def generate_actor_response( self, session: TheaterSession, decision: DirectorDecision, speaker: Actor, prop: str | None, ) -> ActorResponse | dict[str, Any] | str: """Return raw or structured actor output for one beat.""" def repair_actor_response( self, session: TheaterSession, decision: DirectorDecision, speaker: Actor, prop: str | None, invalid_output: ActorResponse | dict[str, Any] | str, validation_status: str, ) -> ActorResponse | dict[str, Any] | str | None: return None class DeterministicBackend(ModelBackend): name = "deterministic" load_status = "loaded" def generate_actor_response( self, session: TheaterSession, decision: DirectorDecision, speaker: Actor, prop: str | None, ) -> ActorResponse: return deterministic_actor_response(session, decision, speaker, prop) class OpenBMBTransformersBackend(ModelBackend): name = "openbmb" def __init__( self, model_id: str | None = None, max_new_tokens: int = OPENBMB_MAX_NEW_TOKENS, temperature: float = OPENBMB_TEMPERATURE, ) -> None: self.model_id = model_id or os.getenv("OPENBMB_MODEL_ID", DEFAULT_OPENBMB_MODEL_ID) self.max_new_tokens = max_new_tokens self.temperature = temperature self.load_status = "unloaded" self.latest_latency_ms: int | None = None self.latest_validation_status: str | None = None self.latest_fallback_used: bool | None = None self.latest_fallback_reason: str | None = None self._tokenizer = None self._model = None self._torch = None def configure(self, max_new_tokens: int | None = None, temperature: float | None = None) -> None: if max_new_tokens is not None: self.max_new_tokens = _clamp_int(max_new_tokens, 16, 160) if temperature is not None: self.temperature = _clamp_float(temperature, 0.0, 1.5) def generate_actor_response( self, session: TheaterSession, decision: DirectorDecision, speaker: Actor, prop: str | None, ) -> str: prompt = build_actor_line_prompt(session, decision, speaker, prop) return self._generate_text(prompt) def repair_actor_response( self, session: TheaterSession, decision: DirectorDecision, speaker: Actor, prop: str | None, invalid_output: ActorResponse | dict[str, Any] | str, validation_status: str, ) -> str | None: prompt = build_actor_line_prompt(session, decision, speaker, prop) repair_prompt = ( f"{prompt}\n\nThe previous output failed validation with status {validation_status}.\n" "Return only valid compact JSON. Do not include markdown, commentary, or extra keys.\n" f"Previous output: {invalid_output}" ) return self._generate_text(repair_prompt) def _load(self) -> None: if self._tokenizer is not None and self._model is not None: self.load_status = "loaded" return self.load_status = "loading" try: import torch from transformers import AutoModelForCausalLM, AutoTokenizer except ImportError as exc: self.load_status = "error" self.latest_fallback_reason = "OpenBMB backend dependencies are not installed" raise RuntimeError("OpenBMB backend dependencies are not installed") from exc try: self._torch = torch self._tokenizer = AutoTokenizer.from_pretrained(self.model_id) self._model = AutoModelForCausalLM.from_pretrained( self.model_id, torch_dtype="auto", device_map="auto", ) self._model.eval() except Exception as exc: self._tokenizer = None self._model = None self.load_status = "error" self.latest_fallback_reason = _summarize_error(exc) raise self.load_status = "loaded" def _generate_text(self, prompt: str) -> str: if zerogpu.USE_ZEROGPU: if not zerogpu.ZEROGPU_GPU_ACTIVE: reason = "USE_ZEROGPU=true but the spaces package is not available" self.load_status = "zerogpu_unavailable" self.latest_fallback_reason = reason zerogpu.record_gpu_failure(reason) raise RuntimeError(reason) try: self.load_status = "zerogpu_ready" return zerogpu.generate_openbmb_text_on_zerogpu( self.model_id, prompt, self.max_new_tokens, self.temperature, ) except Exception as exc: reason = _summarize_error(exc) self.load_status = "error" self.latest_fallback_reason = reason zerogpu.record_gpu_failure(reason) raise RuntimeError(f"ZeroGPU local generation failed: {reason}") from exc self._load() messages = [{"role": "user", "content": prompt}] tokenizer = self._tokenizer model = self._model try: inputs = tokenizer.apply_chat_template( messages, tokenize=True, add_generation_prompt=True, enable_thinking=False, return_dict=True, return_tensors="pt", ) except TypeError: inputs = tokenizer.apply_chat_template( messages, tokenize=True, add_generation_prompt=True, return_dict=True, return_tensors="pt", ) inputs = inputs.to(model.device) eos_token_id = tokenizer.eos_token_id pad_token_id = tokenizer.pad_token_id or eos_token_id do_sample = self.temperature > 0 generation_kwargs = { "max_new_tokens": self.max_new_tokens, "do_sample": do_sample, "pad_token_id": pad_token_id, "eos_token_id": eos_token_id, } if do_sample: generation_kwargs["temperature"] = self.temperature with self._torch.inference_mode(): outputs = model.generate(**inputs, **generation_kwargs) new_tokens = outputs[0][inputs["input_ids"].shape[-1] :] return tokenizer.decode(new_tokens, skip_special_tokens=True).strip() class HFAPIBackend(ModelBackend): name = "hf_api" def __init__( self, model_id: str | None = None, max_new_tokens: int = HF_API_ACTOR_MAX_TOKENS, temperature: float = HF_API_ACTOR_TEMPERATURE, top_p: float = HF_API_TOP_P, timeout_seconds: float = HF_API_TIMEOUT_SECONDS, ) -> None: self.model_id = model_id or os.getenv("HF_API_MODEL_ID", DEFAULT_HF_API_MODEL_ID) self.max_new_tokens = max_new_tokens self.temperature = temperature self.top_p = top_p self.timeout_seconds = timeout_seconds self.load_status = "remote_ready" if self.token_configured else "missing_token" self.latest_latency_ms: int | None = None self.latest_validation_status: str | None = None self.latest_fallback_used: bool | None = None self.latest_fallback_reason: str | None = None @property def token_configured(self) -> bool: return bool(_hf_api_token()) def configure( self, max_new_tokens: int | None = None, temperature: float | None = None, top_p: float | None = None, ) -> None: if max_new_tokens is not None: self.max_new_tokens = _clamp_int(max_new_tokens, 16, 240) if temperature is not None: self.temperature = _clamp_float(temperature, 0.0, 1.5) if top_p is not None: self.top_p = _clamp_float(top_p, 0.0, 1.0) self.load_status = "remote_ready" if self.token_configured else "missing_token" def generate_actor_response( self, session: TheaterSession, decision: DirectorDecision, speaker: Actor, prop: str | None, ) -> str: prompt = build_actor_line_prompt(session, decision, speaker, prop) return self._generate_text( prompt, max_tokens=HF_API_ACTOR_MAX_TOKENS, temperature=HF_API_ACTOR_TEMPERATURE, ) def repair_actor_response( self, session: TheaterSession, decision: DirectorDecision, speaker: Actor, prop: str | None, invalid_output: ActorResponse | dict[str, Any] | str, validation_status: str, ) -> str | None: repair_prompt = ( f"{build_actor_line_prompt(session, decision, speaker, prop)}\n\n" f"The previous output failed validation with status {validation_status}.\n" "Return only valid compact JSON. Do not include markdown, commentary, or extra keys.\n" f"Previous output: {invalid_output}" ) return self._generate_text( repair_prompt, max_tokens=HF_API_ACTOR_MAX_TOKENS, temperature=HF_API_ACTOR_TEMPERATURE, ) def _generate_text( self, prompt: str, *, max_tokens: int | None = None, temperature: float | None = None, system_message: str | None = None, ) -> str: token = _hf_api_token() if not token: self.load_status = "missing_token" self.latest_fallback_reason = "HF API token is not configured" raise RuntimeError("HF API token is not configured; set HF_TOKEN or HUGGINGFACEHUB_API_TOKEN") self.load_status = "remote_ready" try: from huggingface_hub import InferenceClient except ImportError as exc: self.load_status = "error" self.latest_fallback_reason = "huggingface_hub is not installed" raise RuntimeError("huggingface_hub is not installed") from exc client = InferenceClient(token=token, timeout=self.timeout_seconds) sys_content = system_message or "You are a puppet theater generation backend. Return valid JSON only." try: output = client.chat.completions.create( model=self.model_id, messages=[ { "role": "system", "content": sys_content, }, {"role": "user", "content": prompt}, ], max_tokens=max_tokens or self.max_new_tokens, temperature=self.temperature if temperature is None else temperature, top_p=self.top_p, ) return _extract_chat_completion_text(output) except Exception as exc: self.load_status = "error" self.latest_fallback_reason = _summarize_error(exc) raise RuntimeError(f"HF API request failed: {_summarize_error(exc)}") from exc class LocalLoRAActorBackend(ModelBackend): name = "local_lora" def __init__(self) -> None: self.base_model_id = os.getenv("ACTOR_LORA_BASE_MODEL", DEFAULT_ACTOR_LORA_BASE_MODEL) self.adapter_id = os.getenv("ACTOR_LORA_ADAPTER", DEFAULT_ACTOR_LORA_ADAPTER) self.model_id = self.adapter_id self.max_new_tokens = _env_int("ACTOR_LORA_MAX_NEW_TOKENS", ACTOR_LORA_MAX_NEW_TOKENS, 16, 320) self.timeout_seconds = _env_float("ACTOR_LORA_TIMEOUT_SECONDS", ACTOR_LORA_TIMEOUT_SECONDS, 1.0, 300.0) self.device = (os.getenv("ACTOR_LORA_DEVICE") or "auto").strip().lower() self.load_in_4bit = _env_bool("ACTOR_LORA_LOAD_IN_4BIT", False) self.repetition_penalty = _env_float("ACTOR_REPETITION_PENALTY", LOCAL_ACTOR_REPETITION_PENALTY, 1.0, 2.0) self.temperature = _env_float("ACTOR_LOCAL_TEMPERATURE", LOCAL_ACTOR_TEMPERATURE, 0.0, 1.5) self.top_p = _env_float("ACTOR_LOCAL_TOP_P", LOCAL_ACTOR_TOP_P, 0.05, 1.0) self.load_status = "unloaded" self.latest_latency_ms: int | None = None self.latest_validation_status: str | None = None self.latest_fallback_used: bool | None = None self.latest_fallback_reason: str | None = None self._tokenizer = None self._model = None self._torch = None def configure(self, max_new_tokens: int | None = None, temperature: float | None = None) -> None: if max_new_tokens is not None: self.max_new_tokens = _clamp_int(max_new_tokens, 16, 320) def generate_actor_response( self, session: TheaterSession, decision: DirectorDecision, speaker: Actor, prop: str | None, ) -> str: messages = build_actor_chat_messages(session, decision, speaker, prop) self._load() return _run_with_timeout(lambda: self._generate_text(messages), self.timeout_seconds, "Local LoRA generation timed out") def repair_actor_response( self, session: TheaterSession, decision: DirectorDecision, speaker: Actor, prop: str | None, invalid_output: ActorResponse | dict[str, Any] | str, validation_status: str, ) -> str | None: messages = build_actor_chat_messages(session, decision, speaker, prop) messages[-1]["content"] += ( f"\n\nPrevious output failed validation with status {validation_status}.\n" "Return only one valid compact JSON object with the exact Actor schema.\n" f"Previous output: {invalid_output}" ) self._load() return _run_with_timeout(lambda: self._generate_text(messages), self.timeout_seconds, "Local LoRA repair timed out") def _load(self) -> None: if self._tokenizer is not None and self._model is not None: self.load_status = "loaded" return self.load_status = "loading" try: import torch from peft import PeftModel from transformers import AutoModelForCausalLM, AutoTokenizer except ImportError as exc: self.load_status = "not_configured" self.latest_fallback_reason = "Local LoRA dependencies are not installed" raise RuntimeError("Local LoRA dependencies are not installed; install torch, transformers, and peft") from exc try: self._torch = torch self._tokenizer = AutoTokenizer.from_pretrained(self.base_model_id, trust_remote_code=True) model_kwargs: dict[str, object] = {"torch_dtype": "auto"} if self.load_in_4bit: model_kwargs["load_in_4bit"] = True model_kwargs["device_map"] = "auto" elif self.device == "auto": model_kwargs["device_map"] = "auto" base_model = AutoModelForCausalLM.from_pretrained( self.base_model_id, trust_remote_code=True, **model_kwargs, ) self._model = PeftModel.from_pretrained(base_model, self.adapter_id) if self.device in {"cpu", "cuda", "mps"} and not self.load_in_4bit: self._model.to(self.device) self._model.eval() except Exception as exc: self._tokenizer = None self._model = None self.load_status = "error" self.latest_fallback_reason = _summarize_error(exc) raise self.load_status = "loaded" def _generate_text(self, messages: list[dict[str, str]]) -> str: self._load() tokenizer = self._tokenizer model = self._model try: inputs = tokenizer.apply_chat_template( messages, tokenize=True, add_generation_prompt=True, enable_thinking=False, return_dict=True, return_tensors="pt", ) except (AttributeError, TypeError): prompt = format_chatml(messages) inputs = tokenizer(prompt, return_tensors="pt") inputs = inputs.to(model.device) eos_token_id = tokenizer.eos_token_id pad_token_id = tokenizer.pad_token_id or eos_token_id with self._torch.inference_mode(): do_sample = self.temperature > 0 generation_kwargs = { "max_new_tokens": self.max_new_tokens, "do_sample": do_sample, "repetition_penalty": self.repetition_penalty, "pad_token_id": pad_token_id, "eos_token_id": eos_token_id, } if do_sample: generation_kwargs["temperature"] = self.temperature generation_kwargs["top_p"] = self.top_p outputs = model.generate( **inputs, **generation_kwargs, ) new_tokens = outputs[0][inputs["input_ids"].shape[-1] :] return tokenizer.decode(new_tokens, skip_special_tokens=True).strip() class LocalGGUFActorBackend(ModelBackend): name = "local_gguf" def __init__(self) -> None: self.model_path = (os.getenv("ACTOR_GGUF_MODEL_PATH") or "").strip() self.repo_id = (os.getenv("ACTOR_GGUF_REPO_ID") or DEFAULT_ACTOR_GGUF_REPO_ID).strip() self.filename = (os.getenv("ACTOR_GGUF_FILENAME") or DEFAULT_ACTOR_GGUF_FILENAME).strip() self.model_id = Path(self.model_path).name if self.model_path else f"{self.repo_id}/{self.filename}" self.n_ctx = _env_int("ACTOR_GGUF_N_CTX", 4096, 512, 32768) self.n_gpu_layers = _env_int("ACTOR_GGUF_N_GPU_LAYERS", ACTOR_GGUF_N_GPU_LAYERS, -1, 999) self.max_new_tokens = _env_int("ACTOR_GGUF_MAX_TOKENS", ACTOR_GGUF_MAX_TOKENS, 16, 320) self.timeout_seconds = _env_float("ACTOR_GGUF_TIMEOUT_SECONDS", ACTOR_GGUF_TIMEOUT_SECONDS, 1.0, 300.0) self.repetition_penalty = _env_float("ACTOR_REPETITION_PENALTY", LOCAL_ACTOR_REPETITION_PENALTY, 1.0, 2.0) self.temperature = _env_float("ACTOR_LOCAL_TEMPERATURE", LOCAL_ACTOR_TEMPERATURE, 0.0, 1.5) self.top_p = _env_float("ACTOR_LOCAL_TOP_P", LOCAL_ACTOR_TOP_P, 0.05, 1.0) self.load_status = "unloaded" self.latest_latency_ms: int | None = None self.latest_validation_status: str | None = None self.latest_fallback_used: bool | None = None self.latest_fallback_reason: str | None = None self._llama = None def configure(self, max_new_tokens: int | None = None, temperature: float | None = None) -> None: if max_new_tokens is not None: self.max_new_tokens = _clamp_int(max_new_tokens, 16, 320) def generate_actor_response( self, session: TheaterSession, decision: DirectorDecision, speaker: Actor, prop: str | None, ) -> str: prompt = format_chatml(build_actor_chat_messages(session, decision, speaker, prop)) self._load() return _run_with_timeout(lambda: self._generate_text(prompt), self.timeout_seconds, "Local GGUF generation timed out") def repair_actor_response( self, session: TheaterSession, decision: DirectorDecision, speaker: Actor, prop: str | None, invalid_output: ActorResponse | dict[str, Any] | str, validation_status: str, ) -> str | None: messages = build_actor_chat_messages(session, decision, speaker, prop) messages[-1]["content"] += ( f"\n\nPrevious output failed validation with status {validation_status}.\n" "Return only one valid compact JSON object with the exact Actor schema.\n" f"Previous output: {invalid_output}" ) prompt = format_chatml(messages) self._load() return _run_with_timeout(lambda: self._generate_text(prompt), self.timeout_seconds, "Local GGUF repair timed out") def _load(self) -> None: if self._llama is not None: self.load_status = "loaded" return model_path = self._resolve_model_path() if not Path(model_path).exists(): self.load_status = "not_configured" self.latest_fallback_reason = "Local GGUF file was not found" raise RuntimeError("Local GGUF file was not found") self.load_status = "loading" try: from llama_cpp import Llama except ImportError as exc: self.load_status = "not_configured" self.latest_fallback_reason = "llama-cpp-python is not installed" raise RuntimeError("llama-cpp-python is not installed") from exc try: self._llama = Llama( model_path=model_path, n_ctx=self.n_ctx, n_gpu_layers=self.n_gpu_layers, verbose=False, ) except Exception as exc: self._llama = None self.load_status = "error" self.latest_fallback_reason = _summarize_error(exc) raise self.load_status = "loaded" def _resolve_model_path(self) -> str: if self.model_path: return self.model_path if not self.repo_id or not self.filename: self.load_status = "not_configured" self.latest_fallback_reason = "ACTOR_GGUF_REPO_ID or ACTOR_GGUF_FILENAME is not configured" raise RuntimeError("ACTOR_GGUF_REPO_ID or ACTOR_GGUF_FILENAME is not configured") try: from huggingface_hub import hf_hub_download except ImportError as exc: self.load_status = "not_configured" self.latest_fallback_reason = "huggingface_hub is not installed" raise RuntimeError("huggingface_hub is not installed") from exc try: self.load_status = "downloading" resolved_path = hf_hub_download(repo_id=self.repo_id, filename=self.filename) except Exception as exc: self.load_status = "error" self.latest_fallback_reason = _summarize_error(exc) raise RuntimeError(f"GGUF download failed: {_summarize_error(exc)}") from exc self.model_path = resolved_path self.model_id = f"{self.repo_id}/{self.filename}" return resolved_path def _generate_text(self, prompt: str) -> str: self._load() output = self._llama( prompt, max_tokens=self.max_new_tokens, temperature=self.temperature, top_p=self.top_p, repeat_penalty=self.repetition_penalty, stop=["<|im_end|>", "<|im_start|>", "\nUSER:", "\nSYSTEM:", "\nASSISTANT:"], ) if isinstance(output, dict): choices = output.get("choices") or [] if choices and isinstance(choices[0], dict): return str(choices[0].get("text", "")).strip() return str(output).strip() def deterministic_actor_response( session: TheaterSession, decision: DirectorDecision, speaker: Actor, prop: str | None, ) -> ActorResponse: return ActorResponse( intent=_intent_for_beat(decision.beat_type, prop), line=_line_for_beat(session, decision, speaker, prop), emotion=_emotion_for_beat(decision.beat_type), gesture=_gesture_for_beat(decision.beat_type), stage_effect=decision.stage_effect or _effect_for_beat(decision.beat_type), memory_update=_memory_for_beat(session, decision, speaker, prop), tool_request=_tool_request_for_beat(decision, prop), ) def generate_actor_response( session: TheaterSession, decision: DirectorDecision, speaker: Actor, prop: str | None, backend: ModelBackend | None = None, ) -> BackendGeneration: active_backend = backend or get_backend( session.backend_name, max_new_tokens=session.backend_max_new_tokens, temperature=session.backend_temperature, ) start_time = time.perf_counter() raw_output: ActorResponse | dict[str, Any] | str | None = None try: raw_output = active_backend.generate_actor_response(session, decision, speaker, prop) except Exception as exc: latency_ms = _elapsed_ms(start_time) return _fallback_generation( session=session, decision=decision, speaker=speaker, prop=prop, backend=active_backend, validation_status="backend_error", latency_ms=latency_ms, error=_summarize_error(exc), ) response, validation_status = parse_actor_output(raw_output) if response is not None: if _is_repeated_actor_line(response, session): validation_status = ( f"{validation_status};repeated_line" if validation_status != "valid" else "repeated_line" ) else: generation = BackendGeneration( response=response, backend_name=active_backend.name, model_id=active_backend.model_id, fallback_used=False, validation_status=validation_status, load_status=getattr(active_backend, "load_status", "loaded"), latency_ms=_elapsed_ms(start_time), ) _record_generation_status(active_backend, generation) return generation if response is not None and validation_status.endswith("repeated_line"): raw_output = response.model_dump() if response is not None and not validation_status.endswith("repeated_line"): generation = BackendGeneration( response=response, backend_name=active_backend.name, model_id=active_backend.model_id, fallback_used=False, validation_status=validation_status, load_status=getattr(active_backend, "load_status", "loaded"), latency_ms=_elapsed_ms(start_time), ) _record_generation_status(active_backend, generation) return generation try: repair_output = active_backend.repair_actor_response( session=session, decision=decision, speaker=speaker, prop=prop, invalid_output=raw_output, validation_status=validation_status, ) except Exception as exc: return _fallback_generation( session=session, decision=decision, speaker=speaker, prop=prop, backend=active_backend, validation_status=f"{validation_status};repair_backend_error", latency_ms=_elapsed_ms(start_time), error=_summarize_error(exc), ) if repair_output is not None: response, repair_status = parse_actor_output(repair_output) if response is not None: if _is_repeated_actor_line(response, session): validation_status = f"{validation_status};repair_{repair_status};repair_repeated_line" else: generation = BackendGeneration( response=response, backend_name=active_backend.name, model_id=active_backend.model_id, fallback_used=False, validation_status=f"repair_{repair_status}", load_status=getattr(active_backend, "load_status", "loaded"), latency_ms=_elapsed_ms(start_time), ) _record_generation_status(active_backend, generation) return generation validation_status = f"{validation_status};repair_{repair_status}" return _fallback_generation( session=session, decision=decision, speaker=speaker, prop=prop, backend=active_backend, validation_status=validation_status, latency_ms=_elapsed_ms(start_time), ) def get_backend( backend_name: str | None, max_new_tokens: int | None = None, temperature: float | None = None, ) -> ModelBackend: normalized_name = (backend_name or "deterministic").strip().lower() if normalized_name in {"local_lora", "lora", "actor_lora"}: cache_key = ( "local_lora:" f"{os.getenv('ACTOR_LORA_BASE_MODEL', DEFAULT_ACTOR_LORA_BASE_MODEL)}:" f"{os.getenv('ACTOR_LORA_ADAPTER', DEFAULT_ACTOR_LORA_ADAPTER)}" ) if cache_key not in _BACKEND_CACHE: _BACKEND_CACHE[cache_key] = LocalLoRAActorBackend() backend = _BACKEND_CACHE[cache_key] if isinstance(backend, LocalLoRAActorBackend): backend.configure(max_new_tokens=max_new_tokens, temperature=temperature) return backend if normalized_name in {"local_gguf", "gguf", "actor_gguf"}: model_path = os.getenv("ACTOR_GGUF_MODEL_PATH", "").strip() repo_id = os.getenv("ACTOR_GGUF_REPO_ID", DEFAULT_ACTOR_GGUF_REPO_ID).strip() filename = os.getenv("ACTOR_GGUF_FILENAME", DEFAULT_ACTOR_GGUF_FILENAME).strip() cache_key = f"local_gguf:{model_path or repo_id}:{filename}" if cache_key not in _BACKEND_CACHE: _BACKEND_CACHE[cache_key] = LocalGGUFActorBackend() backend = _BACKEND_CACHE[cache_key] if isinstance(backend, LocalGGUFActorBackend): backend.configure(max_new_tokens=max_new_tokens, temperature=temperature) return backend if normalized_name in {"hf_api", "huggingface_api", "hf-inference-api"}: model_id = os.getenv("HF_API_MODEL_ID", DEFAULT_HF_API_MODEL_ID) cache_key = f"hf_api:{model_id}" if cache_key not in _BACKEND_CACHE: _BACKEND_CACHE[cache_key] = HFAPIBackend( model_id=model_id, max_new_tokens=max_new_tokens or HF_API_ACTOR_MAX_TOKENS, temperature=temperature if temperature is not None else HF_API_ACTOR_TEMPERATURE, ) backend = _BACKEND_CACHE[cache_key] if isinstance(backend, HFAPIBackend): backend.configure(max_new_tokens=max_new_tokens, temperature=temperature) return backend if normalized_name == "openbmb": model_id = os.getenv("OPENBMB_MODEL_ID", DEFAULT_OPENBMB_MODEL_ID) cache_key = f"openbmb:{model_id}" if cache_key not in _BACKEND_CACHE: _BACKEND_CACHE[cache_key] = OpenBMBTransformersBackend( model_id=model_id, max_new_tokens=max_new_tokens or OPENBMB_MAX_NEW_TOKENS, temperature=temperature if temperature is not None else OPENBMB_TEMPERATURE, ) backend = _BACKEND_CACHE[cache_key] if isinstance(backend, OpenBMBTransformersBackend): backend.configure(max_new_tokens=max_new_tokens, temperature=temperature) return backend return _BACKEND_CACHE["deterministic"] def warm_up_openbmb( max_new_tokens: int = OPENBMB_MAX_NEW_TOKENS, temperature: float = OPENBMB_TEMPERATURE, ) -> BackendRuntimeStatus: backend = get_backend("openbmb", max_new_tokens=max_new_tokens, temperature=temperature) if not isinstance(backend, OpenBMBTransformersBackend): return get_backend_status("deterministic") start_time = time.perf_counter() try: if zerogpu.USE_ZEROGPU: backend._generate_text( "Return only this JSON: " '{"intent":"Confirm readiness.","line":"Ready.","emotion":"ready","gesture":"wave","stage_effect":"spotlight","memory_update":"Ready for the cue.","tool_request":null}' ) else: backend._load() backend.latest_latency_ms = _elapsed_ms(start_time) backend.latest_fallback_reason = None except Exception as exc: backend.latest_latency_ms = _elapsed_ms(start_time) backend.latest_fallback_reason = _summarize_error(exc) return get_backend_status("openbmb") def get_backend_status(backend_name: str | None = None) -> BackendRuntimeStatus: normalized_name = (backend_name or "deterministic").strip().lower() if normalized_name in {"local_lora", "lora", "actor_lora"}: base_model_id = os.getenv("ACTOR_LORA_BASE_MODEL", DEFAULT_ACTOR_LORA_BASE_MODEL) adapter_id = os.getenv("ACTOR_LORA_ADAPTER", DEFAULT_ACTOR_LORA_ADAPTER) cache_key = f"local_lora:{base_model_id}:{adapter_id}" backend = _BACKEND_CACHE.get(cache_key) if backend is None: return BackendRuntimeStatus( backend_name="local_lora", model_id=adapter_id, load_status="unloaded", max_new_tokens=_env_int("ACTOR_LORA_MAX_NEW_TOKENS", ACTOR_LORA_MAX_NEW_TOKENS, 16, 320), temperature=0, **_zerogpu_status_kwargs(), ) return _runtime_status_from_backend(backend) if normalized_name in {"local_gguf", "gguf", "actor_gguf"}: model_path = (os.getenv("ACTOR_GGUF_MODEL_PATH") or "").strip() repo_id = os.getenv("ACTOR_GGUF_REPO_ID", DEFAULT_ACTOR_GGUF_REPO_ID).strip() filename = os.getenv("ACTOR_GGUF_FILENAME", DEFAULT_ACTOR_GGUF_FILENAME).strip() cache_key = f"local_gguf:{model_path or repo_id}:{filename}" backend = _BACKEND_CACHE.get(cache_key) if backend is None: return BackendRuntimeStatus( backend_name="local_gguf", model_id=Path(model_path).name if model_path else f"{repo_id}/{filename}", load_status="unloaded", max_new_tokens=_env_int("ACTOR_GGUF_MAX_TOKENS", ACTOR_GGUF_MAX_TOKENS, 16, 320), temperature=0, **_zerogpu_status_kwargs(), ) return _runtime_status_from_backend(backend) if normalized_name in {"hf_api", "huggingface_api", "hf-inference-api"}: model_id = os.getenv("HF_API_MODEL_ID", DEFAULT_HF_API_MODEL_ID) backend = _BACKEND_CACHE.get(f"hf_api:{model_id}") if backend is None: return BackendRuntimeStatus( backend_name="hf_api", model_id=model_id, load_status="remote_ready" if _hf_api_token() else "missing_token", token_configured=bool(_hf_api_token()), max_new_tokens=HF_API_ACTOR_MAX_TOKENS, temperature=HF_API_ACTOR_TEMPERATURE, **_zerogpu_status_kwargs(), ) return _runtime_status_from_backend(backend) if normalized_name == "openbmb": model_id = os.getenv("OPENBMB_MODEL_ID", DEFAULT_OPENBMB_MODEL_ID) backend = _BACKEND_CACHE.get(f"openbmb:{model_id}") if backend is None: return BackendRuntimeStatus( backend_name="openbmb", model_id=model_id, load_status="unloaded", max_new_tokens=OPENBMB_MAX_NEW_TOKENS, temperature=OPENBMB_TEMPERATURE, **_zerogpu_status_kwargs(), ) return _runtime_status_from_backend(backend) return _runtime_status_from_backend(_BACKEND_CACHE["deterministic"]) def build_actor_line_prompt( session: TheaterSession, decision: DirectorDecision, speaker: Actor, prop: str | None, ) -> str: recent_transcript = "\n".join( f"{beat.speaker}: {beat.line}" for beat in session.transcript[-3:] ) or "No lines yet." recent_memory = "; ".join(speaker.recent_memory[-3:]) or "None" held_props = ", ".join(speaker.held_props or ([speaker.held_prop] if speaker.held_prop else [])) or "none" prompt = ACTOR_LINE_PROMPT.format( show_title=session.show_title, premise=session.premise, setting=session.setting, beat_type=decision.beat_type, speaker_name=speaker.name, speaker_goal=speaker.goal, speaker_mood=speaker.mood, speaker_current_goal=speaker.current_goal or speaker.goal, speaker_goal_progress=speaker.goal_progress, speaker_held_props=held_props, speaker_tools=", ".join(speaker.tools) or "None", speaker_secret_status=speaker.secret_status, speaker_recent_memory=recent_memory, speaker_style=speaker.speaking_style, recent_tool_results=_format_recent_tool_results(session), audience_action=session.latest_audience_action or "None", latest_prop=prop or session.latest_prop or "None", director_instruction=decision.instruction, reveal_secret=decision.reveal_secret, stage_effect=decision.stage_effect, ) return ( f"{prompt}\nRecent transcript:\n{recent_transcript}\n\n" "Keep the line under 25 words. Keep intent and memory_update short. Return JSON only." ) def build_actor_chat_messages( session: TheaterSession, decision: DirectorDecision, speaker: Actor, prop: str | None, ) -> list[dict[str, str]]: show_state = { "show_title": session.show_title, "setting": session.setting, "beat_index": session.beat_index, "min_beats": session.min_beats, "target_beats": session.target_beats, "max_beats": session.max_beats, "story_phase": _story_phase_for_prompt(session), "latest_audience_action": session.latest_audience_action, "latest_prop": prop or session.latest_prop, "stage_lighting": session.stage_lighting, "recent_transcript": [ {"speaker": beat.speaker, "line": beat.line} for beat in session.transcript[-4:] ], "recent_tool_results": [ { "actor_name": result.actor_name, "tool_name": result.tool_name, "result": result.result, } for result in session.recent_tool_results[-3:] ], "finale_requested": session.finale_requested, } actor_payload = { "avatar": speaker.avatar, "name": speaker.name, "goal": speaker.goal, "secret": speaker.secret, "speaking_style": speaker.speaking_style, "tools": speaker.tools, "mood": speaker.mood, "current_goal": speaker.current_goal or speaker.goal, "goal_progress": speaker.goal_progress, "held_props": speaker.held_props or ([speaker.held_prop] if speaker.held_prop else []), "secret_status": speaker.secret_status, "recent_memory": speaker.recent_memory[-3:], } instruction_parts = [decision.instruction] if decision.beat_type: instruction_parts.append(f"Beat type: {decision.beat_type}.") if decision.reveal_secret: instruction_parts.append("Reveal or hint the actor secret in the line.") if prop: instruction_parts.append(f"Use the latest prop: {prop}.") if session.transcript: instruction_parts.append("Do not repeat any exact line from recent_transcript; advance the scene with new wording.") instruction_parts.append(f"Suggested stage effect: {decision.stage_effect}.") user = "\n".join( [ f"premise: {session.premise}", f"show_state JSON: {json.dumps(show_state, sort_keys=True, separators=(',', ':'))}", f"actor JSON: {json.dumps(actor_payload, sort_keys=True, separators=(',', ':'))}", f"director_instruction: {' '.join(instruction_parts)}", "", ACTOR_RESPONSE_SUFFIX, ] ) return [{"role": "system", "content": ACTOR_SYSTEM_MESSAGE}, {"role": "user", "content": user}] def format_chatml(messages: list[dict[str, str]]) -> str: chunks = [] for message in messages: role = message.get("role", "user") content = message.get("content", "") chunks.append(f"<|im_start|>{role}\n{content}\n<|im_end|>") chunks.append("<|im_start|>assistant\n") return "\n".join(chunks) def _fallback_generation( session: TheaterSession, decision: DirectorDecision, speaker: Actor, prop: str | None, backend: ModelBackend, validation_status: str, latency_ms: int | None, error: str | None = None, ) -> BackendGeneration: fallback_response = deterministic_actor_response(session, decision, speaker, prop) if _is_repeated_actor_line(fallback_response, session): fallback_response = _vary_repeated_fallback_response(fallback_response, session, decision, speaker) generation = BackendGeneration( response=fallback_response, backend_name=backend.name, model_id=backend.model_id, fallback_used=True, validation_status=validation_status, load_status=getattr(backend, "load_status", "loaded"), latency_ms=latency_ms, error=error, ) _record_generation_status(backend, generation) return generation def _is_repeated_actor_line(response: ActorResponse, session: TheaterSession) -> bool: line = " ".join(response.line.lower().strip().split()) if not line: return False recent_lines = { " ".join(beat.line.lower().strip().split()) for beat in session.transcript[-4:] if beat.line.strip() } return line in recent_lines def _vary_repeated_fallback_response( response: ActorResponse, session: TheaterSession, decision: DirectorDecision, speaker: Actor, ) -> ActorResponse: premise_word = next( (word.strip(".,!?;:()[]{}\"'") for word in session.premise.split() if len(word.strip(".,!?;:()[]{}\"'")) > 4), "mystery", ) beat_label = decision.beat_type.replace("_", " ") line = f"{speaker.name.split()[0]} turns the {premise_word.lower()} clue into a new {beat_label} wobble." return response.model_copy( update={ "line": _cap_words(line, 25), "memory_update": f"Varied a repeated {beat_label} beat.", } ) def parse_actor_output(raw_output: ActorResponse | dict[str, Any] | str) -> tuple[ActorResponse | None, str]: parsed = _coerce_actor_output(raw_output) if parsed is None: return None, "invalid_schema" parsed, validation_status = _normalize_actor_payload(parsed) try: response = ActorResponse.model_validate(parsed) except ValidationError: return None, "invalid_required_fields" if not response.line.strip(): return None, "invalid_empty_line" if len(response.line) > MAX_ACTOR_LINE_CHARS or len(response.line.split()) > 25: capped_line = _cap_words(response.line, 25) response = response.model_copy(update={"line": capped_line}) capped_status = "valid_line_capped" return response, f"{validation_status};{capped_status}" if validation_status != "valid" else capped_status return response, validation_status def _coerce_actor_output(raw_output: ActorResponse | dict[str, Any] | str) -> ActorResponse | dict[str, Any] | None: if isinstance(raw_output, ActorResponse): return raw_output if isinstance(raw_output, dict): return raw_output if isinstance(raw_output, str): text = raw_output.strip() if not text: return None if text.startswith("```"): text = text.strip("`") if "\n" in text: text = text.split("\n", maxsplit=1)[1] extracted_json = extract_first_balanced_json(text) if extracted_json is not None: text = extracted_json try: decoded = json.loads(text) except json.JSONDecodeError: return None return decoded if isinstance(decoded, dict) else None return None def _normalize_actor_payload(parsed: ActorResponse | dict[str, Any]) -> tuple[ActorResponse | dict[str, Any], str]: if isinstance(parsed, ActorResponse): return parsed, "valid" normalized = {field: parsed[field] for field in ACTOR_JSON_FIELDS if field in parsed} notes: list[str] = [] if "intent" not in normalized: normalized["intent"] = "Respond to the Director's cue." notes.append("compat_intent_defaulted") if "emotion" not in normalized: normalized["emotion"] = "focused" notes.append("compat_emotion_defaulted") if "gesture" not in normalized: normalized["gesture"] = "leans toward the tiny spotlight" notes.append("compat_gesture_defaulted") if "stage_effect" not in normalized: normalized["stage_effect"] = "warm_spotlight" notes.append("compat_stage_effect_defaulted") if "memory_update" not in normalized: normalized["memory_update"] = "" notes.append("compat_memory_defaulted") if "tool_request" not in normalized: normalized["tool_request"] = None notes.append("compat_tool_defaulted") elif isinstance(normalized["tool_request"], str | list): normalized["tool_request"] = None notes.append("compat_tool_ignored") elif isinstance(normalized["tool_request"], dict): tool_request = dict(normalized["tool_request"]) if "tool_name" not in tool_request and "tool" in tool_request: tool_request["tool_name"] = tool_request.pop("tool") notes.append("compat_tool_name_mapped") if "arguments" not in tool_request and "args" in tool_request: tool_request["arguments"] = tool_request.pop("args") notes.append("compat_tool_args_mapped") if "arguments" not in tool_request: tool_request["arguments"] = {} notes.append("compat_tool_arguments_defaulted") if "reason" not in tool_request: tool_request["reason"] = "Requested theatrical help." notes.append("compat_tool_reason_defaulted") tool_request, tool_status = _sanitize_tool_request(tool_request) normalized["tool_request"] = tool_request if tool_status != "valid": notes.append(tool_status) if isinstance(normalized.get("line"), str): line = " ".join(str(normalized["line"]).strip().split()) if len(line) > MAX_ACTOR_LINE_CHARS or len(line.split()) > 25: normalized["line"] = _cap_words(line, 25) notes.append("line_capped") if isinstance(normalized.get("intent"), str) and len(str(normalized["intent"])) > 90: normalized["intent"] = str(normalized["intent"])[:87].rstrip() + "..." notes.append("intent_capped") if isinstance(normalized.get("memory_update"), str) and len(str(normalized["memory_update"])) > 140: normalized["memory_update"] = str(normalized["memory_update"])[:137].rstrip() + "..." notes.append("memory_capped") return normalized, ";".join(notes) if notes else "valid" def extract_first_balanced_json(text: str) -> str | None: start = text.find("{") if start == -1: return None depth = 0 in_string = False escape_next = False for index, char in enumerate(text[start:], start=start): if escape_next: escape_next = False continue if char == "\\" and in_string: escape_next = True continue if char == '"': in_string = not in_string continue if in_string: continue if char == "{": depth += 1 elif char == "}": depth -= 1 if depth == 0: return text[start : index + 1] return None def _sanitize_tool_request(raw_request: dict[str, Any]) -> tuple[dict[str, Any] | None, str]: try: from puppet_theater.tools import validate_tool_request request, validation_status = validate_tool_request(raw_request) except Exception: return None, "tool_request_ignored" if request is None: return None, f"tool_request_{validation_status}" return request.model_dump(), "valid" def _cap_words(value: str, max_words: int) -> str: words = " ".join(value.strip().split()).split() capped = " ".join(words[:max_words]).rstrip(" ,;:") if len(capped) > MAX_ACTOR_LINE_CHARS: capped = capped[: MAX_ACTOR_LINE_CHARS - 3].rstrip() if len(words) > max_words or len(value) > MAX_ACTOR_LINE_CHARS: return f"{capped}..." return capped def _elapsed_ms(start_time: float) -> int: return round((time.perf_counter() - start_time) * 1000) def _run_with_timeout(callback, timeout_seconds: float, timeout_message: str) -> str: result_queue: queue.Queue[tuple[str, object]] = queue.Queue(maxsize=1) def run_callback() -> None: try: result_queue.put(("result", callback()), block=False) except Exception as exc: result_queue.put(("error", exc), block=False) worker = threading.Thread(target=run_callback, daemon=True) worker.start() try: status, payload = result_queue.get(timeout=timeout_seconds) except queue.Empty as exc: raise RuntimeError(timeout_message) from exc if status == "error": raise payload return str(payload) def _summarize_error(exc: Exception) -> str: message = " ".join(str(exc).split()) if not message: message = exc.__class__.__name__ return _redact_secrets(message)[:180] def _clamp_int(value: int, minimum: int, maximum: int) -> int: return max(minimum, min(maximum, int(value))) def _clamp_float(value: float, minimum: float, maximum: float) -> float: return max(minimum, min(maximum, float(value))) def _env_int(name: str, default: int, minimum: int, maximum: int) -> int: raw_value = os.getenv(name) if raw_value is None: return default try: return _clamp_int(int(raw_value), minimum, maximum) except ValueError: return default def _env_float(name: str, default: float, minimum: float, maximum: float) -> float: raw_value = os.getenv(name) if raw_value is None: return default try: return _clamp_float(float(raw_value), minimum, maximum) except ValueError: return default def _env_bool(name: str, default: bool) -> bool: raw_value = os.getenv(name) if raw_value is None: return default return raw_value.strip().lower() in {"1", "true", "yes", "on"} def _story_phase_for_prompt(session: TheaterSession) -> str: target_beats = max(1, session.target_beats) progress = min(1.0, max(0.0, session.beat_index / target_beats)) if progress < 0.20: return "opening" if progress < 0.45: return "complication" if progress < 0.65: return "reveal" if progress < 0.85: return "chaos" return "finale" def _record_generation_status(backend: ModelBackend, generation: BackendGeneration) -> None: if hasattr(backend, "latest_latency_ms"): backend.latest_latency_ms = generation.latency_ms if hasattr(backend, "latest_validation_status"): backend.latest_validation_status = generation.validation_status if hasattr(backend, "latest_fallback_used"): backend.latest_fallback_used = generation.fallback_used if hasattr(backend, "latest_fallback_reason"): backend.latest_fallback_reason = generation.error def _runtime_status_from_backend(backend: ModelBackend) -> BackendRuntimeStatus: fallback_reason = getattr(backend, "latest_fallback_reason", None) if fallback_reason is None: fallback_reason = zerogpu.LAST_GPU_FALLBACK_REASON return BackendRuntimeStatus( backend_name=backend.name, model_id=backend.model_id, load_status=getattr(backend, "load_status", "loaded"), token_configured=getattr(backend, "token_configured", None), latest_latency_ms=getattr(backend, "latest_latency_ms", None), latest_validation_status=getattr(backend, "latest_validation_status", None), latest_fallback_used=getattr(backend, "latest_fallback_used", None), latest_fallback_reason=fallback_reason, max_new_tokens=getattr(backend, "max_new_tokens", None), temperature=getattr(backend, "temperature", None), **_zerogpu_status_kwargs(), ) def _zerogpu_status_kwargs() -> dict[str, object]: return { "zerogpu_enabled": zerogpu.USE_ZEROGPU, "spaces_available": zerogpu.SPACES_AVAILABLE, "zerogpu_gpu_active": zerogpu.ZEROGPU_GPU_ACTIVE, "torch_version": zerogpu.torch_version(), "cuda_available_in_gpu_fn": zerogpu.LAST_GPU_CUDA_AVAILABLE, } def _line_for_beat( session: TheaterSession, decision: DirectorDecision, speaker: Actor, prop: str | None, ) -> str: if prop is not None: return f"This {prop} is evidence, comfort, and possibly our smallest witness." beat_type = decision.beat_type if session.latest_tool_result is not None and beat_type not in {"setup", "finale"}: clue = _cap_words(session.latest_tool_result.result, 13).rstrip(".") return f"The last stage clue says: {clue}, so I am changing tactics." if beat_type == "setup": premise = _cap_words(session.premise, 12).rstrip(".") return f"I see it clearly: {premise}, and somehow I am in charge." if beat_type == "denial_or_contradiction": return "Absolutely not. The premise is innocent, which is exactly what makes it suspicious." if beat_type == "evidence_or_prop": return "I found a prop with fingerprints, glitter, and a very dramatic attitude." if beat_type == "secret_reveal": if decision.reveal_secret: return f"I confess: {speaker.secret}" return "I nearly confessed something, but the spotlight blinked and I lost my nerve." if beat_type == "chaos_or_intervention": return "The audience has interrupted with imaginary confetti, so everyone must panic gracefully." return "Curtain call! We solved nothing, learned everything, and bowed before the wobble got worse." def _intent_for_beat(beat_type: str, prop: str | None) -> str: if prop is not None: return f"Make the {prop} matter." return { "setup": "Establish the scene.", "denial_or_contradiction": "Challenge the premise.", "evidence_or_prop": "Turn a clue into momentum.", "secret_reveal": "Reveal pressure without derailing.", "chaos_or_intervention": "React and escalate briefly.", "finale": "Close the show cleanly.", }[beat_type] def _memory_for_beat( session: TheaterSession, decision: DirectorDecision, speaker: Actor, prop: str | None, ) -> str: if prop is not None: return f"Used {prop} as important evidence." if decision.beat_type == "secret_reveal": if decision.reveal_secret: return "Shared a secret with the audience." return "Almost revealed a secret under pressure." if decision.beat_type == "finale": return "Reached the curtain call." if session.latest_audience_action: return "Reacted to the audience interruption." return f"Advanced the {decision.beat_type.replace('_', ' ')} beat." def _tool_request_for_beat(decision: DirectorDecision, prop: str | None) -> ToolRequest | None: if prop is None or not decision.uses_prop: return None return ToolRequest( tool_name="inspect_prop", arguments={"prop": prop}, reason="The prop should reveal a theatrical clue.", ) def _format_recent_tool_results(session: TheaterSession) -> str: if not session.recent_tool_results: return "None" return "\n".join( f"- {result.actor_name} used {result.tool_name}: {result.result}" for result in session.recent_tool_results[-3:] ) def _emotion_for_beat(beat_type: str) -> str: return { "setup": "curious", "denial_or_contradiction": "defensive", "evidence_or_prop": "suspicious", "secret_reveal": "confessional", "chaos_or_intervention": "frantic", "finale": "triumphant", }[beat_type] def _gesture_for_beat(beat_type: str) -> str: return { "setup": "raise_curtain", "denial_or_contradiction": "shake_head", "evidence_or_prop": "present_prop", "secret_reveal": "lean_to_audience", "chaos_or_intervention": "flail_politely", "finale": "deep_bow", }[beat_type] def _effect_for_beat(beat_type: str) -> str: return { "setup": "warm_spotlight", "denial_or_contradiction": "quick_blackout", "evidence_or_prop": "prop_table_glow", "secret_reveal": "single_spotlight", "chaos_or_intervention": "confetti_rustle", "finale": "curtain_fall", }[beat_type] _BACKEND_CACHE: dict[str, ModelBackend] = { "deterministic": DeterministicBackend(), } def _hf_api_token() -> str | None: token = os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACEHUB_API_TOKEN") return token.strip() if token and token.strip() else None def _redact_secrets(message: str) -> str: redacted = message for secret in (os.getenv("HF_TOKEN"), os.getenv("HUGGINGFACEHUB_API_TOKEN")): if secret: redacted = redacted.replace(secret, "[redacted]") return redacted def _extract_chat_completion_text(output: Any) -> str: choices = getattr(output, "choices", None) if choices: message = getattr(choices[0], "message", None) content = getattr(message, "content", None) if content is not None: return str(content).strip() if isinstance(output, dict): try: return str(output["choices"][0]["message"]["content"]).strip() except (KeyError, IndexError, TypeError): pass return str(output).strip() def _extract_text_generation_output(output: Any) -> str: if isinstance(output, str): return output.strip() generated_text = getattr(output, "generated_text", None) if generated_text is not None: return str(generated_text).strip() if isinstance(output, dict): text = output.get("generated_text") or output.get("text") if text is not None: return str(text).strip() return str(output).strip()