"""Окно голосовых точек: ESP32 шлёт json на /voice, отвечает дешёвый агент.""" from __future__ import annotations import asyncio import json import logging from typing import TYPE_CHECKING, Any from uuid import uuid4 from beaver_gateway.agents.claude import ClaudeAgent from beaver_gateway.conversations.turns import content_of from beaver_gateway.frontends.accumulate import StreamAccumulator from beaver_gateway.frontends.base import Frontend from beaver_gateway.frontends.bearer import require_token from beaver_gateway.frontends.sse import ( KEEPALIVE, SSE_HEADERS, events_with_heartbeat, sse_pack, ) from beaver_gateway.security import audit from fastapi import FastAPI, HTTPException, Request, status from fastapi.responses import JSONResponse, StreamingResponse from beaver_agent.voice import texts from beaver_agent.voice.thread import FRONTEND, current, open_thread if TYPE_CHECKING: from collections.abc import AsyncIterator, Mapping from beaver_gateway.frontends.base import GatewayRuntime from beaver_gateway.storage.models import Conversation _log = logging.getLogger("beaver_agent.voice") FIELDS = ("room", "device", "content") class VoiceFrontend(Frontend): """Одна ветка на всю квартиру, свой агент, свои ключи. Тёрн идёт мимо очереди, а значит и мимо конверта: справка по vault в этот разговор не попадает даже случайно. В телеграм-ветку вопрос кладётся отдельной репликой, ответ пишется черновиком, как обычно. """ name = FRONTEND kinds = ("branch",) path = "/voice" scope = "voice" def __init__(self, *, agent: str, timeout: float = 90.0) -> None: self.agent = agent self.timeout = timeout """Сколько ждать ответа; дальше 504, а тёрн договаривает в телеграм.""" self._runtime: GatewayRuntime | None = None self._app: FastAPI | None = None self._lock = asyncio.Lock() self._late: set[asyncio.Task[Any]] = set() def configure(self, runtime: GatewayRuntime) -> None: if runtime.conversations is None or runtime.bus is None: msg = "VoiceFrontend: нужны runtime.conversations и runtime.bus" raise RuntimeError(msg) agent = runtime.agents.get(self.agent) if not isinstance(agent, ClaudeAgent) or not agent.serves("branch"): msg = f"VoiceFrontend: агент {self.agent!r} не обслуживает ветки" raise RuntimeError(msg) self._runtime = runtime self._app = self._build_app(runtime) def app(self) -> FastAPI | None: return self._app def _build_app(self, runtime: GatewayRuntime) -> FastAPI: app = FastAPI(title="beaver-agent / голосовой") @app.get("/healthz") async def healthz() -> dict[str, str]: return {"status": "ok", "frontend": self.name} @app.post("/") async def ask(request: Request) -> Any: token = await require_token(request, runtime, scope=self.scope) body = await _body(request) room, device, content = (_field(body, name) for name in FIELDS) conv = await self.thread() await audit.log( runtime, actor=f"token:{token}", kind="voice_ask", agent_name=conv.agent_name, conversation=conv.external_id, room=room, device=device, request_id=body.get("request_id"), ) _log.info("голос: %s/%s -> %s", room, device, conv.external_id) await runtime.conversations.say( conv, texts.ASKED.format(room=room, device=device, text=content) ) prompt = await self._prompt( conv, texts.HEARD.format(room=room, device=device, text=content) ) if bool(body.get("stream")): return StreamingResponse( self._sse(conv, prompt, body), media_type="text/event-stream", headers=SSE_HEADERS, ) return JSONResponse(await self._answer(conv, prompt, body)) @app.exception_handler(HTTPException) async def http_error(_request: Request, exc: HTTPException) -> JSONResponse: return JSONResponse( status_code=exc.status_code, content={"error": exc.detail}, headers=exc.headers, ) return app async def thread(self) -> Conversation: conversations = self.runtime.conversations async with self._lock: conv = await current(conversations) if conv is not None: return conv return await open_thread(conversations, agent=self.agent) @property def runtime(self) -> GatewayRuntime: if self._runtime is None: msg = "VoiceFrontend: configure() не вызывали" raise RuntimeError(msg) return self._runtime async def _prompt(self, conv: Conversation, text: str) -> str: seed = await self.runtime.conversations.pending_seed(conv) return f"{seed}\n\n{text}" if seed else text async def _answer( self, conv: Conversation, prompt: str, body: Mapping[str, Any] ) -> dict[str, Any]: turn_id = _turn_id() task = asyncio.create_task(self._run(conv, prompt, turn_id)) self._late.add(task) task.add_done_callback(self._late.discard) done, _ = await asyncio.wait({task}, timeout=self.timeout) if not done: _log.warning("голос: %s не уложился в %.0f c", turn_id, self.timeout) raise HTTPException( status.HTTP_504_GATEWAY_TIMEOUT, "агент не успел; ответ уйдёт в ветку" ) return _body_out(body, reply=task.result(), conv=conv) async def _run(self, conv: Conversation, prompt: str, turn_id: str) -> str: reply, _capture = await self.runtime.conversations.run_text_turn( conv, prompt, origin="user", turn_id=turn_id ) self._published(conv, reply=reply, turn_id=turn_id) return reply async def _sse( self, conv: Conversation, prompt: str, body: Mapping[str, Any] ) -> AsyncIterator[bytes]: turn_id = _turn_id() acc = StreamAccumulator() model = self.runtime.agents[conv.agent_name].model try: events = self.runtime.conversations.turn( conv, messages=[{"role": "user", "content": content_of(prompt, None)}], origin="user", turn_id=turn_id, ) async for event in events_with_heartbeat(events): if event is None: yield KEEPALIVE continue acc.feed(event) chunk = _text_delta(event) if chunk: yield sse_pack("delta", {"text": chunk}) except Exception as exc: # noqa: BLE001 _log.exception("голос: тёрн %s упал", turn_id) yield sse_pack("error", {"error": str(exc)}) return reply = _text_of(acc.finalize(model=model)) self._published(conv, reply=reply, turn_id=turn_id) yield sse_pack("done", _body_out(body, reply=reply, conv=conv)) def _published(self, conv: Conversation, *, reply: str, turn_id: str) -> None: self.runtime.bus.publish( "reply", conversation_id=conv.external_id, turn_id=turn_id, source=self.name, text=reply, ) async def _body(request: Request) -> Mapping[str, Any]: try: data = await request.json() except json.JSONDecodeError as exc: raise HTTPException(status.HTTP_400_BAD_REQUEST, f"кривой json: {exc}") from exc if not isinstance(data, dict): raise HTTPException(status.HTTP_400_BAD_REQUEST, "тело должно быть объектом") return data def _field(body: Mapping[str, Any], name: str) -> str: value = body.get(name) if not isinstance(value, str) or not value.strip(): raise HTTPException( status.HTTP_400_BAD_REQUEST, f"нужно непустое строковое поле {name!r}" ) return value.strip() def _body_out( body: Mapping[str, Any], *, reply: str, conv: Conversation ) -> dict[str, Any]: return { "request_id": body.get("request_id"), "room": body.get("room"), "device": body.get("device"), "reply": reply, "conversation": conv.external_id, } def _turn_id() -> str: return f"turn_{uuid4().hex[:12]}" def _text_delta(event: Any) -> str: if getattr(event, "type", "") != "content_block_delta": return "" delta = getattr(event, "delta", None) if getattr(delta, "type", "") != "text_delta": return "" return getattr(delta, "text", "") def _text_of(message: Any) -> str: return "\n\n".join( getattr(b, "text", "") for b in message.content if getattr(b, "type", "") == "text" ).strip()