Spaces:
Sleeping
Sleeping
| """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"] | |