Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions livekit-agents/livekit/agents/voice/agent_activity.py
Original file line number Diff line number Diff line change
Expand Up @@ -3494,6 +3494,13 @@ async def _next_segment() -> _SpeechSegment | None:
interrupted_fnc_outputs.append(sanitized_out.fnc_call_out)

if interrupted_tool_messages := interrupted_calls + interrupted_fnc_outputs:
self._session.emit(
"function_tools_executed",
FunctionToolsExecutedEvent(
function_calls=interrupted_calls,
function_call_outputs=interrupted_fnc_outputs,
),
)
self._agent._chat_ctx.insert(interrupted_tool_messages)
self._session._tool_items_added(interrupted_tool_messages)
return
Expand Down
27 changes: 26 additions & 1 deletion tests/test_tool_results_preserved_on_interruption.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
from livekit.agents import Agent, AgentSession, function_tool
from livekit.agents.llm import FunctionToolCall
from livekit.agents.voice.agent_activity import AgentActivity
from livekit.agents.voice.events import FunctionToolsExecutedEvent
from livekit.agents.voice.speech_handle import SpeechHandle

from .fake_session import FakeActions, create_session, run_session
Expand Down Expand Up @@ -116,19 +117,40 @@ def _interrupt_after_forced_schedule(


async def test_tool_results_preserved_when_interrupted_during_playout() -> None:
"""Interruption lands while the agent is still speaking the tool turn."""
"""Completed tool results stay observable when their parent speech is interrupted."""
actions = FakeActions()
_weather_tool_turn(actions, tts_duration=10.0) # playout 3.5s -> 13.5s
actions.add_user_speech(5.0, 6.0, "Stop!", stt_delay=0.2) # interrupts at 5.5s
actions.add_llm(content="Okay, stopping.")
actions.add_tts(1.0)
actions.add_tts(1.0, input="The weather in Tokyo is sunny today.")

session = create_session(actions)
agent = WeatherAgent() # the tool completes at ~3.4s, before the interruption
tool_executed_events: list[FunctionToolsExecutedEvent] = []
tool_speeches: list[SpeechHandle] = []

def on_function_tools_executed(event: FunctionToolsExecutedEvent) -> None:
tool_executed_events.append(event)
output = event.function_call_outputs[0]
if output is not None:
tool_speeches.append(session.say(output.output, allow_interruptions=False))

session.on("function_tools_executed", on_function_tools_executed)

await asyncio.wait_for(run_session(session, agent), timeout=SESSION_TIMEOUT)

_assert_weather_tool_preserved(agent, session)
assert len(tool_executed_events) == 1
event = tool_executed_events[0]
assert event.function_calls[0].name == "get_weather"
output = event.function_call_outputs[0]
assert output is not None
assert output.output == "The weather in Tokyo is sunny today."
assert len(tool_speeches) == 1
assert tool_speeches[0].done()
assert not tool_speeches[0].interrupted
assert not tool_speeches[0].allow_interruptions


async def test_tool_results_preserved_when_tool_in_flight_at_interruption() -> None:
Expand Down Expand Up @@ -175,8 +197,11 @@ async def transfer_to_billing(self) -> Agent:

session = create_session(actions)
agent = TransferAgent()
tool_executed_events: list[FunctionToolsExecutedEvent] = []
session.on("function_tools_executed", tool_executed_events.append)

await asyncio.wait_for(run_session(session, agent), timeout=SESSION_TIMEOUT)

assert not tool_executed_events
for items in (agent.chat_ctx.items, session.history.items):
assert not any(i.type in ("function_call", "function_call_output") for i in items)
Loading