Spaces:
Sleeping
Sleeping
File size: 3,107 Bytes
03760d5 643994f 03760d5 | 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 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | """Commentary runtime — turns vision into spoken co-streamer reactions.
Bridges the (now active) vision pipeline to the (now active) voice layer:
when Byte sees an interesting scene it generates one short, punchy line and
speaks it via voice_runtime. Heavily throttled so it never spams or burns
inference, and fully fire-and-forget so it can never break frame ingest.
"""
from __future__ import annotations
import threading
import time
from bot.runtime_config import logger
from bot.voice_runtime import voice_runtime
COMMENTARY_MIN_INTERVAL_SECONDS = 25.0
COMMENTARY_MAX_CHARS = 180
_BORING_MARKERS = ("cena_normal", "tela_inicial", "menu", "loading")
_COMMENTARY_PROMPT = (
"Voce e o Byte, co-streamer de Twitch carismatico. Em UMA frase curta "
"(maximo 120 caracteres), empolgada e natural em portugues, comente esta "
"cena de gameplay para o chat. Sem aspas, sem emojis em excesso.\n"
"Cena observada: {scene}"
)
class CommentaryRuntime:
"""Throttled vision->voice commentary."""
def __init__(self) -> None:
self._lock = threading.Lock()
self._last_comment_at = 0.0
def _is_interesting(self, analysis: str) -> bool:
lowered = analysis.lower()
return not any(marker in lowered for marker in _BORING_MARKERS)
def maybe_comment(self, analysis: str) -> dict:
"""Maybe generate and speak a comment about the scene. Never raises."""
clean = (analysis or "").strip()
if not clean or not self._is_interesting(clean):
return {"ok": False, "reason": "uninteresting"}
now = time.time()
with self._lock:
if now - self._last_comment_at < COMMENTARY_MIN_INTERVAL_SECONDS:
return {"ok": False, "reason": "throttled"}
self._last_comment_at = now # reserve the slot before the slow call
try:
comment = self._generate(clean)
except Exception as error: # inference must never break vision ingest
logger.error("Commentary generation failed: %s", error)
return {"ok": False, "reason": "generation_error", "error": str(error)}
if not comment:
return {"ok": False, "reason": "empty_comment"}
voice_runtime.say(comment, source="commentary")
return {"ok": True, "comment": comment}
def _generate(self, scene: str) -> str:
from bot.runtime_config import LLM_MODEL_DEFAULT, client
response = client.chat.completions.create(
model=LLM_MODEL_DEFAULT,
# Gemini 2.5-flash spends part of the budget on internal "thinking";
# a small max_tokens yields empty/truncated output, so give it room
# (the visible line is trimmed to COMMENTARY_MAX_CHARS anyway).
max_tokens=400,
messages=[{"role": "user", "content": _COMMENTARY_PROMPT.format(scene=scene[:300])}],
)
text = (response.choices[0].message.content or "").strip()
return text[:COMMENTARY_MAX_CHARS]
commentary_runtime = CommentaryRuntime()
__all__ = ["CommentaryRuntime", "commentary_runtime"]
|