realtime/openai_client.py
realtime/openai_client.pyBrowse 18 files
2,052 tokens
9,027 bytes
Token encoding: o200k_base
Snapshot 24fd22b
← Back to SKILL.md
1"""OpenAI Realtime API WebSocket client + file-queue speaker.2 3text → OpenAI Realtime → audio deltas appended as PCM to a file the audio bridge streams into4Chrome's fake mic. One sync WebSocket per session; ``websockets`` is imported lazily.5"""6 7from __future__ import annotations8 9import base6410import contextlib11import json12import threading13import time14import uuid15from pathlib import Path16from typing import Any, Callable, Optional17 18 19REALTIME_URL = "wss://api.openai.com/v1/realtime"20 21_TERMINAL_FRAMES = {"response.done", "response.completed", "response.cancelled"}22 23 24def _decode_audio(b64: str) -> bytes:25 try:26 return base64.b64decode(b64) if b64 else b""27 except (ValueError, TypeError):28 return b""29 30 31class RealtimeSession:32 """Minimal sync client for the OpenAI Realtime WebSocket API; ``speak`` and ``cancel_response``33 may run on different threads — a lock serializes WebSocket writes."""34 35 def __init__(self, api_key: str, model: str = "gpt-realtime", voice: str = "alloy",36 instructions: str = "", audio_sink_path: Optional[Path] = None, sample_rate: int = 24000) -> None:37 self.api_key = api_key38 self.model = model39 self.voice = voice40 self.instructions = instructions41 self.audio_sink_path = Path(audio_sink_path) if audio_sink_path else None42 self.sample_rate = sample_rate43 self._ws: Any = None44 self._send_lock = threading.Lock()45 self.audio_bytes_out: int = 0 # public counters for status reporting46 self.last_audio_out_at: Optional[float] = None47 48 def connect(self) -> None:49 """Open the WS and send ``session.update`` with voice + instructions."""50 try:51 from websockets.sync.client import connect # type: ignore52 except ImportError as exc: # pragma: no cover - exercised via test53 raise RuntimeError("websockets package is required for OpenAI Realtime; "54 "install with: pip install websockets") from exc55 url = f"{REALTIME_URL}?model={self.model}"56 headers = [("Authorization", f"Bearer {self.api_key}"), ("OpenAI-Beta", "realtime=v1")]57 # Newer websockets takes additional_headers=, older extra_headers=.58 try:59 self._ws = connect(url, additional_headers=headers)60 except TypeError:61 self._ws = connect(url, extra_headers=headers)62 self._send_json({"type": "session.update", "session": {63 "voice": self.voice, "instructions": self.instructions, "modalities": ["audio", "text"],64 "output_audio_format": "pcm16", "input_audio_format": "pcm16"}})65 66 def close(self) -> None:67 if self._ws is not None:68 with contextlib.suppress(Exception):69 self._ws.close()70 self._ws = None71 72 def speak(self, text: str, timeout: float = 30.0) -> dict:73 """Send ``text`` and append the audio response to ``audio_sink_path`` (opened 'ab' per call74 so a streaming reader can consume it). Frames other than audio deltas/terminal/error are ignored."""75 if self._ws is None:76 raise RuntimeError("RealtimeSession.connect() must be called first")77 start = time.monotonic()78 self._send_json({"type": "conversation.item.create", "item": {79 "type": "message", "role": "user", "content": [{"type": "input_text", "text": text}]}})80 self._send_json({"type": "response.create", "response": {"modalities": ["audio"]}})81 bytes_written = 082 with contextlib.ExitStack() as stack:83 sink_fp = None84 if self.audio_sink_path is not None:85 self.audio_sink_path.parent.mkdir(parents=True, exist_ok=True)86 sink_fp = stack.enter_context(open(self.audio_sink_path, "ab"))87 while True:88 frame = self._recv_frame(start + timeout, timeout)89 if frame is None or frame.get("type") in _TERMINAL_FRAMES: # peer closed / response done90 break91 ftype = frame.get("type")92 if ftype == "error":93 raise RuntimeError(f"realtime error: {frame.get('error') or frame}")94 chunk = _decode_audio(frame.get("delta") or frame.get("audio") or "") if (95 ftype == "response.audio.delta" and sink_fp is not None) else b""96 if chunk:97 sink_fp.write(chunk)98 sink_fp.flush()99 bytes_written += len(chunk)100 self.audio_bytes_out += len(chunk)101 self.last_audio_out_at = time.time()102 return {"ok": True, "bytes_written": bytes_written, "duration_ms": (time.monotonic() - start) * 1000.0}103 104 def cancel_response(self) -> bool:105 """Barge-in: send ``response.cancel``. True if sent, False if nothing to cancel / socket closed."""106 if self._ws is None:107 return False108 try:109 self._send_json({"type": "response.cancel"})110 return True111 except Exception:112 return False113 114 def _send_json(self, payload: dict) -> None:115 assert self._ws is not None116 with self._send_lock:117 self._ws.send(json.dumps(payload))118 119 def _recv_frame(self, deadline: float, timeout: float) -> Optional[dict]:120 """Next dict frame before *deadline* (monotonic), ``None`` once the peer closes.121 Non-dict / unparseable frames are skipped; TimeoutError past the deadline."""122 assert self._ws is not None123 while True:124 remaining = deadline - time.monotonic()125 if remaining <= 0:126 raise TimeoutError(f"realtime response did not complete within {timeout}s")127 try:128 raw = self._ws.recv(timeout=remaining)129 except TypeError: # older websockets: no timeout kwarg130 raw = self._ws.recv()131 if raw is None:132 return None133 with contextlib.suppress(TypeError, ValueError):134 frame = json.loads(raw) if isinstance(raw, (str, bytes, bytearray)) else raw135 if isinstance(frame, dict):136 return frame137 138 139class RealtimeSpeaker:140 """JSONL queue (``{"id", "text"}`` per line) wrapper around :class:`RealtimeSession`; processed141 lines are appended to ``processed_path`` (if set) and removed from the queue."""142 143 def __init__(self, session: RealtimeSession, queue_path: Path, processed_path: Optional[Path] = None) -> None:144 self.session = session145 self.queue_path = Path(queue_path)146 self.processed_path = Path(processed_path) if processed_path else None147 148 def _read_queue(self) -> list[dict]:149 """Parse the JSONL queue, skipping blank/malformed lines; entries lacking an ``id`` get one."""150 if not self.queue_path.exists():151 return []152 out: list[dict] = []153 for line in self.queue_path.read_text(encoding="utf-8").splitlines():154 with contextlib.suppress(ValueError):155 entry = json.loads(line) if line.strip() else None156 if isinstance(entry, dict):157 entry.setdefault("id", str(uuid.uuid4()))158 out.append(entry)159 return out160 161 def _rewrite_queue(self, remaining: list[dict]) -> None:162 # Always keep the file (empty when drained): consumers may watch its mtime.163 body = "".join(json.dumps(e) + "\n" for e in remaining)164 self.queue_path.write_text(body, encoding="utf-8")165 166 def _append_processed(self, entry: dict, result: dict) -> None:167 if self.processed_path is None:168 return169 self.processed_path.parent.mkdir(parents=True, exist_ok=True)170 record = {"id": entry.get("id"), "text": entry.get("text", ""), "result": result}171 with open(self.processed_path, "a", encoding="utf-8") as fp:172 fp.write(json.dumps(record) + "\n")173 174 def run_until_stopped(self, stop_fn: Callable[[], bool], poll_interval: float = 0.5) -> None:175 while not stop_fn():176 entries = self._read_queue()177 if not entries:178 time.sleep(poll_interval)179 continue180 # One entry per iteration: the queue may grow while we speak.181 head = entries[0]182 text = (head.get("text") or "").strip()183 result = {"ok": True, "bytes_written": 0, "duration_ms": 0.0}184 if text:185 try:186 result = self.session.speak(text)187 except Exception as exc:188 result = {"ok": False, "error": str(exc)}189 self._append_processed(head, result)190 # Re-read (new entries may have arrived), then drop the head by position or id.191 latest = self._read_queue()192 self._rewrite_queue(latest[1:] if latest and latest[0].get("id") == head.get("id")193 else [e for e in latest if e.get("id") != head.get("id")])194