File size: 4,123 Bytes
2219798 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 | """Luna writes both words and direction; manual prompts bypass the director."""
import json
import time
import httpx
import settings
from actor_prompt import SYSTEM
class Director:
def __init__(self):
self.client = httpx.AsyncClient(timeout=httpx.Timeout(30, connect=10))
async def close(self):
await self.client.aclose()
async def direct(self, message, fmt, language, mode, history):
key = settings.luna_key()
if not key:
raise ValueError("Luna API-Key fehlt. Manuellen Prompt verwenden oder SMALL_LUNA_API_KEY konfigurieren.")
context = [{"role": "system", "content": SYSTEM}]
for item in history[-16:]:
if item.get("role") in {"user", "assistant"} and isinstance(item.get("content"), str):
context.append({"role": item["role"], "content": item["content"][:3000]})
task = ("The entire input below is fixed spoken text. Perform these words EXACTLY, without any extra words."
if mode == "perform" else
"Answer conversationally, or actually enact the user's acting brief. Interpret the scene; do not echo the assignment or describe your performance. Preserve words explicitly supplied to be spoken.")
context.append({"role": "user", "content": f"Format: {fmt}. Spoken language: {language}. {task}\n{message}"})
started = time.monotonic()
for attempt in range(2):
try:
response = await self.client.post(settings.LUNA_BASE.rstrip("/") + "/v1/chat/completions",
headers={"Authorization": f"Bearer {key}"},
json={"model": settings.LUNA_MODEL, "messages": context,
"reasoning_effort": "none", "max_completion_tokens": 2400,
"response_format": {"type": "json_object"}})
except httpx.TimeoutException as exc:
raise RuntimeError("Luna hat nicht rechtzeitig geantwortet. Bitte erneut senden oder einen manuellen Prompt verwenden.") from exc
except httpx.RequestError as exc:
raise RuntimeError("Luna API ist momentan nicht erreichbar. Bitte erneut versuchen.") from exc
if response.status_code >= 400:
raise RuntimeError(f"Luna API antwortet mit HTTP {response.status_code}.")
raw = response.json()["choices"][0]["message"].get("content") or ""
try:
result = json.loads(raw)
if not all(k in result for k in ("text", "general", "script", "caption", "tags", "language")):
raise ValueError("Unvollständige Regie-Antwort")
if not all(isinstance(result[k], str) for k in ("text", "general", "script", "caption", "language")) or not isinstance(result["tags"], list) or not all(isinstance(tag, str) for tag in result["tags"]):
raise ValueError("Ungültige Regie-Antwort")
from prompting import plain_script
spoken = plain_script(result["script"]) if fmt == "structured" else " ".join(result["text"].split())
if not spoken:
raise ValueError("Leeres gesprochenes Skript")
if mode == "perform" and spoken != " ".join(message.split()):
raise ValueError("Luna hat die vorgegebenen Worte geändert")
# As in the existing demo, SCRIPT is authoritative for chat.
# A duplicate text field drifting in punctuation must not
# reject an otherwise valid acting script.
result["text"] = spoken
return result, round((time.monotonic() - started) * 1000)
except (ValueError, TypeError):
if attempt:
raise ValueError("Luna lieferte zweimal keine gültige, wortgetreue Regie. Bitte erneut versuchen.")
context.append({"role": "assistant", "content": raw[:5000]})
context.append({"role": "user", "content": "Correct the JSON. Keep text and spoken script identical; preserve any exact requested transcript."})
|