Repository navigation
fix(voice): keep agent busy until pending tool replies finish - #8379
Epsilon006 wants to merge 9 commits into
Conversation
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
|
I found several GPT-Live cases where the pending-reply tracking either releases too early or stays busy after the reply is finished. I checked PR head
The batch and interruption cases change from the release; the other rows are existing gaps that remain with this change. The main mismatch appears to be in Could these cases be covered before merging? The tracking needs to distinguish an audible answer from other generations, follow the provider's actual continuation behavior, and release replies that fail or are genuinely abandoned. The reproduction below drives the real SDK over a loopback WebSocket and asserts on the observed states. It does not mock SDK internals. Both runs recorded seven loopback connections and zero external connections. Real-provider error paths were not exercised. Standalone reproduction and observed outputSave the script below as uv run --no-project --with livekit-agents==1.8.10 --with aiohttp python -I repro_gpt_live_pending.py
uv run --no-project --with livekit-agents==1.8.10 --with aiohttp python -I repro_gpt_live_pending.py /path/to/agents-pr-checkoutBoth commands exit with status 1 because the script asserts on the expected states. The JSON rows show the actual state, pending-entry count, and whether the answer was delivered. Release 1.8.10: {"case": "control", "before": "thinking", "between": "thinking", "pending_before_audio": 0, "after": "listening", "pending_after_audio": 0, "answer_delivered": true}
{"case": "batch", "before": "thinking", "between": "thinking", "pending_before_audio": 0, "after": "listening", "pending_after_audio": 0, "answer_delivered": true}
{"case": "interrupt", "before": "thinking", "between": "thinking", "pending_before_audio": 0, "after": "listening", "pending_after_audio": 0, "answer_delivered": true}
{"case": "failed", "state": "thinking", "pending": 0}
{"case": "late_transcript", "before": "thinking", "between": "listening", "pending_before_audio": 0, "after": "listening", "pending_after_audio": 0, "answer_delivered": true}
{"case": "orphan", "state": "thinking", "pending": 0}
{"case": "no_reply", "before": "listening", "between": "listening", "pending_before_audio": 0, "after": "listening", "pending_after_audio": 0, "answer_delivered": true}
{"failed_cases": ["failed", "late_transcript", "orphan", "no_reply"], "local_connections": 7, "external_connections": 0}PR head: {"case": "control", "before": "thinking", "between": "thinking", "pending_before_audio": 1, "after": "listening", "pending_after_audio": 0, "answer_delivered": true}
{"case": "batch", "before": "thinking", "between": "thinking", "pending_before_audio": 2, "after": "thinking", "pending_after_audio": 1, "answer_delivered": true}
{"case": "interrupt", "before": "thinking", "between": "listening", "pending_before_audio": 0, "after": "listening", "pending_after_audio": 0, "answer_delivered": true}
{"case": "failed", "state": "thinking", "pending": 1}
{"case": "late_transcript", "before": "thinking", "between": "listening", "pending_before_audio": 0, "after": "listening", "pending_after_audio": 0, "answer_delivered": true}
{"case": "orphan", "state": "thinking", "pending": 1}
{"case": "no_reply", "before": "listening", "between": "listening", "pending_before_audio": 0, "after": "listening", "pending_after_audio": 0, "answer_delivered": true}
{"failed_cases": ["batch", "interrupt", "failed", "late_transcript", "orphan", "no_reply"], "local_connections": 7, "external_connections": 0}"""Reproduce GPT-Live reply-state edges against an installed SDK or a source checkout."""
import asyncio
import base64
import json
import sys
from pathlib import Path
if len(sys.argv) > 1:
sys.path.insert(0, str(Path(sys.argv[1]).resolve() / "livekit-agents"))
connections = []
def audit(event, args):
if event == "socket.connect":
address = args[1]
connections.append(address)
if not isinstance(address, tuple) or address[0] != "127.0.0.1":
raise AssertionError(f"Unexpected external connection: {address}")
sys.addaudithook(audit)
import aiohttp
from aiohttp import web
from livekit import rtc
from livekit.agents import Agent, AgentSession, function_tool, llm
from livekit.agents.llm._realtime import gpt_live_types as wire
from livekit.agents.llm._realtime.gpt_live import GPTLiveModel, SAMPLE_RATE
from livekit.agents.voice import io
DEADLINE_SECONDS = TIMEOUT = 6
ANSWER = "The fixture result is verified."
RESPONSE_CREATED = "response.created"
RESPONSE_COMPLETED = "response.completed"
RESPONSE_ITEM_DONE = "response.output_item.done"
OUTPUT_TRANSCRIPT_DELTA = "session.output_transcript.delta"
AUDIBLE_SAMPLES = SAMPLE_RATE // 10
class MemorySpeaker(io.AudioOutput):
def __init__(self):
super().__init__(
label="memory",
next_in_chain=None,
sample_rate=SAMPLE_RATE,
capabilities=io.AudioOutputCapabilities(pause=False),
)
self.duration = 0.0
async def capture_frame(self, frame):
await super().capture_frame(frame)
self.duration += frame.duration
def flush(self):
super().flush()
duration, self.duration = self.duration, 0.0
if duration:
self.on_playback_finished(playback_position=duration, interrupted=False)
def clear_buffer(self):
self.duration = 0.0
class ProviderPeer:
def __init__(self):
self.ws = None
self.commands = asyncio.Queue()
self.processed = asyncio.Queue()
self.runner = None
async def handler(self, request):
self.ws = web.WebSocketResponse()
await self.ws.prepare(request)
async for message in self.ws:
if message.type != aiohttp.WSMsgType.TEXT:
continue
event = json.loads(message.data)
self.commands.put_nowait(event)
if event["type"] == wire.SessionStartEvent.model_fields["type"].default:
await self.send(
wire.SessionStartedEvent(session=wire.SessionResource(id="session"))
)
elif event["type"] == wire.SessionCloseEvent.model_fields["type"].default:
await self.send(wire.SessionClosedEvent(reason="close_requested"))
return self.ws
async def start(self):
app = web.Application()
app.router.add_get("/v1/live/sessions", self.handler)
self.runner = web.AppRunner(app)
await self.runner.setup()
site = web.TCPSite(self.runner, "127.0.0.1", 0)
await site.start()
port = site._server.sockets[0].getsockname()[1]
return f"http://127.0.0.1:{port}/v1"
async def send(self, event):
await self.ws.send_str(event.model_dump_json(exclude_none=True))
async def receive(self, event_type):
async with asyncio.timeout(DEADLINE_SECONDS):
while True:
event = await self.commands.get()
if event["type"] == event_type:
return event
async def backend(self, event_type, *, item=None):
event = wire.ResponseEventEnvelope(
delegation_id="delegation",
event=wire.ResponsesEvent(type=event_type, item=item),
)
await self.send(event)
async with asyncio.timeout(DEADLINE_SECONDS):
while True:
received = await self.processed.get()
if (
received.get("type") == event.type
and received.get("event", {}).get("type") == event_type
):
return
async def speech(self, session, speeches, text, **timestamps):
index = len(speeches)
created = asyncio.Event()
def observe(event):
if len(speeches) > index:
created.set()
session.on("speech_created", observe)
try:
await self.send(
wire.TranscriptDeltaEvent(
type=OUTPUT_TRANSCRIPT_DELTA, delta=text, **timestamps
)
)
for pcm in (b"\x00\x10" * AUDIBLE_SAMPLES, b"\x00\x00" * SAMPLE_RATE):
await self.send(
wire.OutputAudioDeltaEvent(delta=base64.b64encode(pcm).decode())
)
await asyncio.wait_for(created.wait(), DEADLINE_SECONDS)
await asyncio.wait_for(speeches[index].wait_for_playout(), DEADLINE_SECONDS)
finally:
session.off("speech_created", observe)
async def close(self):
if self.ws is not None:
await self.ws.close()
await self.runner.cleanup()
async def run_case(case):
peer = ProviderPeer()
url = await peer.start()
started = asyncio.Event()
release = asyncio.Event()
speeches = []
count = 2 if case == "batch" else 1
running = 0
@function_tool
async def lookup():
"""Read a deterministic local fixture."""
nonlocal running
running += 1
if running == count:
started.set()
await release.wait()
return (
llm.ToolResult(ANSWER, reply_required=False)
if case == "no_reply"
else ANSWER
)
async with aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=TIMEOUT)
) as http:
model = GPTLiveModel(
model="gpt-live-1",
delegation="responses",
api_key="local-peer-not-a-key",
base_url=url,
http_session=http,
)
session = AgentSession(llm=model)
session.output.audio = MemorySpeaker()
session.on("speech_created", lambda event: speeches.append(event.speech_handle))
try:
await session.start(
Agent(instructions="Answer briefly.", llm=model, tools=[lookup])
)
await peer.receive(wire.SessionStartEvent.model_fields["type"].default)
activity = session._activity
live = activity.realtime_llm_session.duplex_session
live.on("openai_server_event_received", peer.processed.put_nowait)
await peer.backend(RESPONSE_CREATED)
for index in range(count):
await peer.backend(
RESPONSE_ITEM_DONE,
item=wire.OutputItem(
id=f"item-{index}",
type="function_call",
status="completed",
call_id=f"call-{index}",
name="lookup",
arguments="{}",
),
)
if case != "orphan":
await peer.backend(RESPONSE_COMPLETED)
await asyncio.wait_for(started.wait(), TIMEOUT)
await peer.speech(session, speeches, "Checking. ", start_ms=0, end_ms=100)
assert session.agent_state == "thinking"
if case == "orphan":
await peer.backend("response.failed")
release.set()
if case == "orphan":
await peer.receive(
wire.ThinkingAppendEvent.model_fields["type"].default
)
else:
for _ in range(count):
await peer.receive(
wire.ResponseItemCreateEvent.model_fields["type"].default
)
await peer.receive(
wire.ResponseCreateEvent.model_fields["type"].default
)
await peer.backend(RESPONSE_CREATED)
await peer.backend(
"response.failed" if case == "failed" else RESPONSE_COMPLETED
)
await asyncio.gather(
*(asyncio.wait_for(speech, TIMEOUT) for speech in speeches)
)
before = session.agent_state
if case in ("failed", "orphan"):
print(
json.dumps(
{
"case": case,
"state": before,
"pending": len(
getattr(activity, "_pending_auto_tool_replies", ())
),
}
),
flush=True,
)
assert before == "listening"
return
if case == "interrupt":
await asyncio.wait_for(session.interrupt(), TIMEOUT)
elif case == "late_transcript":
event = wire.TranscriptDeltaEvent(
type=OUTPUT_TRANSCRIPT_DELTA,
delta="Late old words.",
start_ms=50,
end_ms=100,
)
await peer.send(event)
async with asyncio.timeout(TIMEOUT):
while (await peer.processed.get()).get("type") != event.type:
pass
await asyncio.wait_for(speeches[-1], TIMEOUT)
between = session.agent_state
pending_before_audio = len(
getattr(activity, "_pending_auto_tool_replies", ())
)
await peer.speech(session, speeches, ANSWER)
after = session.agent_state
print(
json.dumps(
{
"case": case,
"before": before,
"between": between,
"pending_before_audio": pending_before_audio,
"after": after,
"pending_after_audio": len(
getattr(activity, "_pending_auto_tool_replies", ())
),
"answer_delivered": any(
getattr(item, "text_content", None) == ANSWER
for item in session.history.items
),
}
),
flush=True,
)
assert before == "thinking"
assert between == "thinking"
assert after == "listening"
finally:
release.set()
await asyncio.wait_for(session.aclose(), TIMEOUT)
await model.aclose()
await peer.close()
async def main():
failed = []
for case in (
"control",
"batch",
"interrupt",
"failed",
"late_transcript",
"orphan",
"no_reply",
):
try:
await run_case(case)
except AssertionError:
failed.append(case)
external = sum(not isinstance(a, tuple) or a[0] != "127.0.0.1" for a in connections)
print(
json.dumps(
{
"failed_cases": failed,
"local_connections": len(connections) - external,
"external_connections": external,
}
),
flush=True,
)
assert external == 0
return bool(failed)
raise SystemExit(asyncio.run(main())) |
|
Thanks for the thorough reproduction and detailed comparison against 1.8.10! You’re right that a generation does not necessarily correspond to a single pending tool reply. The batched-response and transcript-only cases expose assumptions in the current tracking logic. I’ll use your loopback reproduction to verify the lifecycle behavior, address the regressions, and extend the tests to cover these cases before requesting another review. Really appreciate the concrete test cases! |
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
|
Quick update: I've pushed additional fixes following the latest review feedback. The changes now cover fallback transitions, unrelated speech during pending tool continuations, and mixed backend responses containing both tool calls and answers. The loopback regression tests and CI are passing. I'm doing a final check on the fallback event-forwarding path before considering the PR ready for merge. Thanks again for the detailed feedback and reproductions — they've been really helpful in tightening up the implementation. |
|
Thanks for the updates. I tested One inline-tool case still emits Recorded SDK events, in seconds from session creation: The The playout state update happens before waiting for inline tool execution. The pending-work check excludes the current speech task. The provider's pending-reply flag does not cover this interval because the result has not been submitted yet. Could this decision also account for inline tool work, both while execution is running and after execution until an owed reply is handed off? That would avoid publishing This run covered the normal inline-tool path. I did not exercise microphone barge-in or provider failures. |
|
@zek Thanks for the detailed trace. I've pushed 2a4aa87, targeting the inline-tool gap you observed. The existing local work ownership now covers preamble playout through tool execution and result submission, until the continuation takes over. Four loopback regressions reproduce the listening pulse on 984643a and pass after this change: tool completion before/after preamble playout, each with/without an active RunResult. No timeout policy or new public API was added. Full Linux unit/audio_eot CI passed (7,249 passed, 5 skipped), as did Ruff, Python 3.10/3.13 type checks, and blockguard on all three platforms. The unchanged seven-case reproduction and #8362 reproduction also passed locally. This is not a claim that every GPT-Live boundary is solved. Local diagnostics still expose two limitations: a completed backend response with no subsequent audio can retain its pending wait, and unrelated subsequent audio can release an eligible wait. The optional Could you retest 2a4aa87 in the same application setup, with the application-level state hold disabled, and check whether the premature listening pulse between the inline preamble and continuation is gone? I haven't independently retested the real service. I'd like to use that result and maintainer feedback to decide whether the remaining association limits need treatment in this PR. |
|
We ran the #8362 repro scenario against both sides of this PR, same environment (Windows, text-only scripted LLM, no network; livekit-protocol 1.1.27 site-packages). The scenario: one turn calls a slow tool that waits on an external event, a second user message arrives while the tool reply is still owed, then the answer lands.
So on our side the fix covers the original #8362 path where the user speaks again before the deferred reply lands. It complements the handoff/task cases in tests/test_deferred_tool_reply.py, which don't hit this second-user-message shape directly. One environment note: current main (7c9d40b) needs livekit-protocol>=1.1.31 (the pin moved up from 1.1.29 in this branch), so we could not run the repro against latest main with our env; the comparison above is merge-base vs PR head, which both import cleanly on 1.1.27. Happy to re-run against main once we bump the protocol package. |
Fixes #8362: keep the agent busy while tool work or a tool answer is still owed, including across another speech. Return to listening after the answer plays or the work is abandoned, without arming the user-away timer in between.
Implementation
finally.These follow-up refinements add no public API or generic state machine. The GPT-Live refinement consolidated two call maps and a ready set. The latest inline-tool fix reuses the existing background-speech set; it changes one production module by a net three physical lines.
Regression evidence
Zek's latest real-service report exposed a gap before result submission: a tool called inside an audible preamble emitted
speaking → listening → thinking. Four deterministic loopback cases reproduce this on 984643a, then pass on this commit: execution finishing before/after preamble playout, each with/without an active RunResult. They also hold the real chat-context lock after execution, verify no listening/away timer before submission and continuation, and verify final answer delivery and return to listening.The existing fixture now reports playback start, so these cases exercise a call in the same audible generation rather than an independent tool-only generation. All 54 prior loopback cases remain, alongside the four new cases.
Validation
Commit 2a4aa87: full Linux unit/audio_eot CI passes: 7,249 passed, 5 skipped. Ruff, Python 3.10/3.13 type checks, and blockguard on all three platforms also pass. Provider integration/evaluation jobs are skipped by workflow conditions.
test_the_adapter_keeps_the_context_under_the_ids_the_framework_uses.git diff --checkpass. Strict mypy passes for all eight PR source modules using explicit paths and--explicit-package-bases --follow-imports=silent --platform linux.Real-provider integration/evals were not run here. The latest real-service reproduction still needs verification in the application environment. Audio events do not identify individual tool answers: handoff uses backend completion and subsequent audible output, not proof of semantic answer completeness.