openai: keep GPT-Live's input clock running when no audio is pushed

livekit-agents/livekit/agents/voice/agent_activity.py

@@ -308,14 +308,6 @@ def __init__(self, agent: Agent, sess: AgentSession) -> None:
             )
             self._session._warned_realtime_audio_redaction = True
 
-        # a duplex model has no text modality, and the adapter resolves each reply from the audio
-        # the model produces; without audio a text simulation would only time out on turn one
-        if self._text_only and isinstance(self.llm, llm.DuplexRealtimeAdapter):
-            raise RuntimeError(
-                "a DuplexModel speaks only through audio, so it cannot run under a text "
-                "simulation; run `lk agent simulate audio` instead"
-            )
-
         if self._rt_turn_detection_enabled and not self.allow_interruptions:
             raise ValueError(
                 "the RealtimeModel uses a server-side turn detection, "

livekit-plugins/livekit-plugins-openai/livekit/plugins/openai/realtime/gpt_live_model.py

@@ -4,6 +4,7 @@
 import base64
 import contextlib
 import json
+import math
 import os
 import time
 from collections.abc import AsyncIterable
@@ -39,6 +40,16 @@
 DEFAULT_BACKEND_MODEL = "gpt-5.6-luna"
 OPENAI_BASE_URL = "https://api.openai.com/v1"
 
+# the model speaks only while its input clock runs, and that clock is the audio the client appends:
+# a session with no microphone keeps it running with silence
+_INPUT_IDLE_S = 0.2
+_SILENCE_100MS = rtc.AudioFrame(
+    data=b"\x00" * (SAMPLE_RATE // 10 * 2),
+    sample_rate=SAMPLE_RATE,
+    num_channels=NUM_CHANNELS,
+    samples_per_channel=SAMPLE_RATE // 10,
+)
+
 # the service also caps startup history at 8192 tokens and an append at 500; there is no tokenizer
 # here, so those two are the service's to enforce
 _MAX_INPUT_ITEMS = 128
@@ -302,6 +313,10 @@ def __init__(self, duplex_model: GPTLiveModel) -> None:
         )
 
         self._main_atask = asyncio.create_task(self._main_task(), name="GPTLiveSession._main")
+        self._last_audio_at = -math.inf
+        self._silence_atask = asyncio.create_task(
+            self._silence_task(), name="GPTLiveSession._silence"
+        )
 
     # outbound
 
@@ -901,6 +916,21 @@ def tools(self) -> llm.ToolContext:
         return self._tools.copy()
 
     def push_audio(self, frame: rtc.AudioFrame) -> None:
+        self._last_audio_at = asyncio.get_running_loop().time()
+        self._append_audio(frame)
+
+    async def _silence_task(self) -> None:
+        loop = asyncio.get_running_loop()
+        while True:
+            await asyncio.sleep(_INPUT_IDLE_S / 2)
+            started = self._session_started_fut
+            if not started.done() or started.cancelled():
+                continue
+            if loop.time() - self._last_audio_at < _INPUT_IDLE_S:
+                continue
+            self._append_audio(_SILENCE_100MS)
+
+    def _append_audio(self, frame: rtc.AudioFrame) -> None:
         # the caller's turn ends on their own audio: this much pushed since their last fragment
         if (speech := self._speech.get("user")) is not None:
             speech.quiet_ms += round(frame.duration * 1000)
@@ -961,6 +991,7 @@ def unmute_input(self) -> None:
 
     async def aclose(self) -> None:
         await super().aclose()
+        await utils.aio.cancel_and_wait(self._silence_atask)
         if not self._session_started_fut.done():
             self._session_started_fut.cancel()
         self._msg_ch.close()

tests/test_duplex_adapter.py

@@ -934,25 +934,3 @@ def deactivate(self) -> None:
     assert [e.recoverable for e in errors] == [False]
     assert isinstance(errors[0].error, _Boom)
     assert errors[0].label == fake.duplex_model.label
-
-
-# a duplex model has no text modality: it hears and speaks only audio, and the adapter resolves a
-# reply from the sound the model produces. under a text simulation there is no audio, so the
-# session refuses to start rather than time out on the first turn
-
-
-async def test_a_duplex_model_refuses_a_text_simulation(monkeypatch: pytest.MonkeyPatch) -> None:
-    from livekit.agents.voice import Agent, AgentSession
-
-    monkeypatch.setattr(AgentSession, "_text_only", property(lambda self: True))
-    session = AgentSession(llm=_FakeDuplexModel())
-    with pytest.raises(RuntimeError, match="text simulation"):
-        await session.start(Agent(instructions="hi"))
-
-
-async def test_a_duplex_model_starts_outside_a_text_simulation() -> None:
-    from livekit.agents.voice import Agent, AgentSession
-
-    session = AgentSession(llm=_FakeDuplexModel())
-    await session.start(Agent(instructions="hi"))
-    await session.aclose()

tests/test_gpt_live_model.py

@@ -34,14 +34,16 @@ class _FakeWS:
     ``session.start``; tests about that hold turn the ack off.
     """
 
-    def __init__(self, *, auto_start: bool) -> None:
+    def __init__(self, *, auto_start: bool, record_silence: bool = False) -> None:
         self.sent: list[dict[str, Any]] = []
         self.auto_start = auto_start
+        self.record_silence = record_silence
         self.session: GPTLiveSession | None = None
 
     async def send_str(self, data: str) -> None:
         event = json.loads(data)
-        self.sent.append(event)
+        if self.record_silence or not _is_silence(event):
+            self.sent.append(event)
         if self.session is None or not self.auto_start:
             return
         if event["type"] == "session.start":
@@ -58,9 +60,18 @@ async def close(self) -> None:
         pass
 
 
-def _connect_hook(monkeypatch: pytest.MonkeyPatch, *, auto_start: bool = True) -> _FakeWS:
+def _is_silence(event: dict[str, Any]) -> bool:
+    """The session pads an idle microphone with silence; sequence assertions ignore that padding."""
+    return event["type"] == "session.input_audio.append" and not base64.b64decode(
+        event["audio"]
+    ).strip(b"\x00")
+
+
+def _connect_hook(
+    monkeypatch: pytest.MonkeyPatch, *, auto_start: bool = True, record_silence: bool = False
+) -> _FakeWS:
     """Replace the handshake before any session exists, so nothing can reach the network."""
-    ws = _FakeWS(auto_start=auto_start)
+    ws = _FakeWS(auto_start=auto_start, record_silence=record_silence)
 
     async def _create_ws_conn(self: GPTLiveSession) -> _FakeWS:
         ws.session = self
@@ -71,9 +82,12 @@ async def _create_ws_conn(self: GPTLiveSession) -> _FakeWS:
 
 
 class _LifecycleWS:
-    def __init__(self, *, start: bool = True, disconnect_on_start: bool = False) -> None:
+    def __init__(
+        self, *, start: bool = True, disconnect_on_start: bool = False, record_silence: bool = False
+    ) -> None:
         self.start = start
         self.disconnect_on_start = disconnect_on_start
+        self.record_silence = record_silence
         self.sent: list[dict[str, Any]] = []
         self.incoming: asyncio.Queue[aiohttp.WSMessage] = asyncio.Queue()
         self.started = asyncio.Event()
@@ -85,7 +99,8 @@ def emit(self, event: dict[str, Any]) -> None:
 
     async def send_str(self, data: str) -> None:
         event = json.loads(data)
-        self.sent.append(event)
+        if self.record_silence or not _is_silence(event):
+            self.sent.append(event)
         if event["type"] == "session.start":
             if self.start:
                 self.emit({"type": "session.started", "session": {"id": "live_test"}})
@@ -324,7 +339,7 @@ async def connect(self: GPTLiveSession) -> _LifecycleWS:
 async def test_reconnect_discards_partial_input_audio(
     monkeypatch: pytest.MonkeyPatch, input_rate: int
 ) -> None:
-    sockets = [_LifecycleWS(), _LifecycleWS()]
+    sockets = [_LifecycleWS(), _LifecycleWS(record_silence=True)]
     connections = iter(sockets)
 
     async def connect(self: GPTLiveSession) -> _LifecycleWS:
@@ -803,7 +818,7 @@ async def test_delayed_context_receipts_do_not_block_commands_or_finish_speech(
         session.append_instructions("Be concise.")
         session.append_thinking("The caller is returning a chair.")
         session.append_commentary("Ask for the order number.")
-        session.push_audio(_silence(100))
+        session.push_audio(_pcm(0.2))
 
         # Context injection can outlast the connection timeout without holding later commands.
         await asyncio.sleep(model._opts.conn_options.timeout * 2)
@@ -1222,3 +1237,44 @@ async def test_a_typed_message_rides_in_the_ask_while_it_is_the_newest_thing_sai
     finally:
         await session.aclose()
         await model.aclose()
+
+
+# GPT-Live speaks only while its input clock is running, and that clock is the audio the client
+# appends. A session with no microphone (text simulation, muted input, text console) would
+# otherwise never hear a reply to anything it asks.
+@pytest.mark.virtual_time
+async def test_silence_keeps_the_input_clock_running_without_a_microphone(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    ws = _connect_hook(monkeypatch, record_silence=True)
+    model = GPTLiveModel(api_key="sk-test")
+    session = model.session()
+    try:
+        await session._update_session()
+        await asyncio.sleep(1.0)
+        appends = [e for e in ws.sent if e["type"] == "session.input_audio.append"]
+        assert len(appends) >= 5
+        assert all(base64.b64decode(e["audio"]) == b"\x00" * 4800 for e in appends)
+    finally:
+        await session.aclose()
+        await model.aclose()
+
+
+@pytest.mark.virtual_time
+async def test_a_live_microphone_is_not_padded_with_silence(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    ws = _connect_hook(monkeypatch)
+    model = GPTLiveModel(api_key="sk-test")
+    session = model.session()
+    try:
+        await session._update_session()
+        await asyncio.sleep(0.05)
+        for _ in range(10):
+            session.push_audio(_pcm(0.2))
+            await asyncio.sleep(0.1)
+        appends = [e for e in ws.sent if e["type"] == "session.input_audio.append"]
+        assert len(appends) == 10
+    finally:
+        await session.aclose()
+        await model.aclose()