OpenAIRealtimeSession(
*,
api_key: Optional[str] = None,
model: str = _DEFAULT_MODEL,
voice: str = "alloy",
instructions: str = "",
language: str = "",
enable_tools: bool = True,
max_tool_steps: int = 12,
allowed_tools: Optional[Sequence[str]] = None,
)
Bases: RealtimeVoiceSession
Duplex audio session against OpenAI Realtime WebSocket API.
Source code in src/diapason/speech/realtime/openai_realtime.py
| def __init__(
self,
*,
api_key: Optional[str] = None,
model: str = _DEFAULT_MODEL,
voice: str = "alloy",
instructions: str = "",
language: str = "",
enable_tools: bool = True,
max_tool_steps: int = 12,
allowed_tools: Optional[Sequence[str]] = None,
) -> None:
from diapason.core.cloud_keys import get_cloud_key
self._api_key = api_key or get_cloud_key("OPENAI_API_KEY")
self._model = model or _DEFAULT_MODEL
self._voice = voice or "alloy"
self._instructions = instructions
self._language = language
self._enable_tools = enable_tools
self._allowed_tools = list(allowed_tools) if allowed_tools else None
self._budget = VoiceToolBudget(max_tool_steps)
self._ws: Any = None
self._queue: asyncio.Queue[Optional[SessionEvent]] = asyncio.Queue()
self._reader_task: Optional[asyncio.Task[None]] = None
self._closed = False
self._pending_args: dict[str, str] = {}
|