Spaces:
Running
Running
| """Authenticated realtime turns over binary PCM frames.""" | |
| from __future__ import annotations | |
| import json | |
| from uuid import uuid4 | |
| from fastapi import APIRouter, WebSocket, WebSocketDisconnect | |
| from starlette.websockets import WebSocketDisconnected | |
| from sonicagent_rt.audio.pcm import pcm16_to_floats | |
| from sonicagent_rt.auth.service_token import enforce_origin | |
| from sonicagent_rt.observability.stt_metrics import REALTIME_INFLIGHT, observe_realtime_turn | |
| from sonicagent_rt.pipeline.cancellation import CancellationToken | |
| from sonicagent_rt.pipeline.orchestrator import run_turn | |
| from sonicagent_rt.protocol.binary_frame import BinaryFrameError, decode_frame | |
| from sonicagent_rt.protocol.constants import FLAG_END_OF_UTTERANCE, MessageType | |
| from sonicagent_rt.protocol.sequence import SequenceError, SequenceValidator | |
| from sonicagent_rt.stt.errors import SpeechError | |
| from sonicagent_rt.vad.factory import create_endpoint, vad_backend | |
| router = APIRouter() | |
| _MAX_FRAME_BYTES = 64 * 1024 | |
| async def realtime(websocket: WebSocket, session_id: str) -> None: | |
| settings = websocket.app.state.settings | |
| token = websocket.query_params.get("resume_token", "") | |
| try: | |
| if not settings.voice_enabled: | |
| raise SpeechError("voice endpoints are disabled") | |
| enforce_origin(websocket.headers.get("origin"), settings) | |
| session, resumed = websocket.app.state.sessions.acquire(session_id, token) | |
| except SpeechError: | |
| await websocket.close(code=1008) | |
| return | |
| await websocket.accept() | |
| hello = "session.updated" if resumed else "session.created" | |
| await websocket.send_json( | |
| { | |
| "type": hello, | |
| "session_id": session.session_id, | |
| "status": "ready", | |
| "vad": vad_backend(settings), | |
| "resumed": resumed, | |
| } | |
| ) | |
| endpoint = create_endpoint(settings) | |
| sequences = SequenceValidator() | |
| turn: object | None = None | |
| cancellation: CancellationToken | None = None | |
| async def cancel_turn(reason: str) -> None: | |
| nonlocal cancellation | |
| if cancellation is not None and not cancellation.cancelled: | |
| cancellation.cancel() | |
| return | |
| await websocket.send_json({"type": "response.cancelled", "reason": reason}) | |
| async def start_turn(samples: list[float]) -> None: | |
| nonlocal turn, cancellation | |
| if cancellation is not None: | |
| cancellation.cancel() | |
| cancellation = CancellationToken() | |
| token_box = cancellation | |
| turn_id = f"turn_{uuid4().hex}" | |
| async def pump() -> None: | |
| queued_ms = 0 | |
| recorded = False | |
| REALTIME_INFLIGHT.inc() | |
| try: | |
| async for item in run_turn( | |
| samples=samples, | |
| sample_rate=settings.input_sample_rate, | |
| language=session.language, | |
| voice_id=session.voice_id, | |
| stt=websocket.app.state.stt, | |
| llm=websocket.app.state.llm, | |
| tts=websocket.app.state.tts, | |
| settings=settings, | |
| cancellation=token_box, | |
| turn_id=turn_id, | |
| speech_ms=len(samples) * 1000 / settings.input_sample_rate, | |
| ): | |
| if token_box.cancelled and not isinstance(item, dict): | |
| break | |
| terminal = {"metrics.turn", "response.cancelled", "error"} | |
| if isinstance(item, dict) and not recorded and item.get("type") in terminal: | |
| recorded = True | |
| kind = item.get("type") | |
| result = {"metrics.turn": "ok", "response.cancelled": "cancelled"}.get(str(kind), "error") | |
| observe_realtime_turn(result, item) | |
| if isinstance(item, bytes): | |
| queued_ms += settings.audio_frame_ms | |
| if queued_ms > settings.max_outbound_queue_ms: | |
| token_box.cancel() | |
| await websocket.send_json( | |
| { | |
| "type": "error", | |
| "turn_id": turn_id, | |
| "code": "rate_limited", | |
| "message": "outbound audio queue is full", | |
| } | |
| ) | |
| break | |
| await websocket.send_bytes(item) | |
| else: | |
| await websocket.send_json(item) | |
| except Exception: | |
| if not recorded: | |
| observe_realtime_turn("error", {}) | |
| await websocket.send_json( | |
| {"type": "error", "turn_id": turn_id, "code": "inference_failed", "message": "turn failed"} | |
| ) | |
| finally: | |
| REALTIME_INFLIGHT.dec() | |
| import asyncio | |
| turn = asyncio.create_task(pump()) | |
| try: | |
| while True: | |
| message = await websocket.receive() | |
| if message.get("text"): | |
| await _on_text(websocket, message["text"], cancel_turn) | |
| elif message.get("bytes") is not None: | |
| payload = message["bytes"] | |
| if len(payload) > _MAX_FRAME_BYTES: | |
| await websocket.send_json( | |
| {"type": "error", "code": "audio_too_large", "message": "audio frame is too large"} | |
| ) | |
| continue | |
| try: | |
| frame = decode_frame(payload) | |
| if frame.message_type != int(MessageType.AUDIO_INPUT): | |
| raise BinaryFrameError("client frames must be microphone audio") | |
| sequences.accept(frame.sequence) | |
| samples = pcm16_to_floats(frame.payload) | |
| except (BinaryFrameError, SequenceError, ValueError) as exc: | |
| await websocket.send_json({"type": "error", "code": "invalid_audio", "message": str(exc)}) | |
| continue | |
| if ( | |
| settings.barge_in_enabled | |
| and turn is not None | |
| and not turn.done() | |
| and _rms(samples) >= settings.vad_threshold | |
| ): | |
| await cancel_turn("barge-in") | |
| for event in endpoint.push(samples): | |
| if event.kind == "started": | |
| await websocket.send_json({"type": "speech.started"}) | |
| elif event.kind == "utterance": | |
| await websocket.send_json({"type": "speech.stopped"}) | |
| await start_turn(event.samples) | |
| if frame.flags & FLAG_END_OF_UTTERANCE: | |
| for event in endpoint.flush(): | |
| if event.kind == "utterance": | |
| await websocket.send_json({"type": "speech.stopped"}) | |
| await start_turn(event.samples) | |
| except (WebSocketDisconnect, WebSocketDisconnected): | |
| if cancellation is not None: | |
| cancellation.cancel() | |
| finally: | |
| websocket.app.state.sessions.release(session_id) | |
| async def _on_text(websocket: WebSocket, raw: str, cancel_turn) -> None: | |
| try: | |
| event = json.loads(raw) | |
| except json.JSONDecodeError: | |
| await websocket.send_json({"type": "error", "code": "invalid_request", "message": "control event is not JSON"}) | |
| return | |
| kind = event.get("type") | |
| if kind == "ping": | |
| await websocket.send_json({"type": "pong"}) | |
| elif kind == "response.cancel": | |
| await cancel_turn("client") | |
| elif kind == "call.end": | |
| await websocket.close(code=1000) | |
| else: | |
| await websocket.send_json({"type": "error", "code": "invalid_request", "message": "unsupported control event"}) | |
| def _rms(samples: list[float]) -> float: | |
| if not samples: | |
| return 0.0 | |
| total = sum(sample * sample for sample in samples) | |
| return (total / len(samples)) ** 0.5 | |