fix(voice): answer every tool call

livekit-agents/livekit/agents/llm/chat_context.py

@@ -376,6 +376,11 @@ class FunctionCallOutput(BaseModel):
     output: str
     is_error: bool
     created_at: float = Field(default_factory=time.time)
+    reply_required: bool = Field(default=True)
+    """Whether the model should answer once it receives this output.
+
+    Only realtime models read it, since they answer a result on their own.
+    """
 
 
 class AgentHandoff(BaseModel):

livekit-agents/livekit/agents/llm/llm.py

@@ -482,8 +482,7 @@ async def collect(self) -> CollectedResponse:
             for tc in response.tool_calls:
                 result = await llm.execute_function_call(tc, tool_ctx)
                 ctx.insert(result.fnc_call)
-                if result.fnc_call_out:
-                    ctx.insert(result.fnc_call_out)
+                ctx.insert(result.fnc_call_out)
             ```
         """
         text_parts: list[str] = []

livekit-agents/livekit/agents/llm/utils.py

@@ -897,7 +897,7 @@ def _is_valid_function_output(value: Any) -> bool:
 @dataclass
 class FunctionCallResult:
     fnc_call: FunctionCall
-    fnc_call_out: FunctionCallOutput | None
+    fnc_call_out: FunctionCallOutput
     raw_output: Any
     raw_exception: BaseException | None
     fnc_call_updates: list[tuple[FunctionCall, FunctionCallOutput]] = field(default_factory=list)
@@ -933,9 +933,16 @@ def make_function_call_output(
         )
 
     if isinstance(exception, StopResponse):
+        # StopResponse asks for silence, not for the call to go unanswered
         return FunctionCallResult(
             fnc_call=fnc_call,
-            fnc_call_out=None,
+            fnc_call_out=FunctionCallOutput(
+                name=fnc_call.name,
+                call_id=fnc_call.call_id,
+                output="",
+                is_error=False,
+                reply_required=False,
+            ),
             raw_output=output,
             raw_exception=exception,
         )
@@ -960,7 +967,12 @@ def make_function_call_output(
         )
         return FunctionCallResult(
             fnc_call=fnc_call,
-            fnc_call_out=None,
+            fnc_call_out=FunctionCallOutput(
+                name=fnc_call.name,
+                call_id=fnc_call.call_id,
+                output="the tool returned an invalid output",
+                is_error=True,
+            ),
             raw_output=output,
             raw_exception=None,
         )

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

@@ -71,6 +71,7 @@
     _AudioOutput,
     _ForwardOutput,
     _inject_running_tool_calls,
+    _interrupted_tool_output,
     _strip_assistant_markup,
     _strip_running_tool_calls,
     _TextOutput,
@@ -3320,8 +3321,7 @@ def _tool_execution_started_cb(fnc_call: llm.FunctionCall) -> None:
             speech_handle._item_added([fnc_call])
 
         def _tool_execution_completed_cb(out: ToolExecutionOutput) -> None:
-            if out.fnc_call_out:
-                speech_handle._item_added([out.fnc_call_out])
+            speech_handle._item_added([out.fnc_call_out])
 
         # start to execute tools (only after play())
         exe_task, tool_output = perform_tool_executions(
@@ -3477,14 +3477,13 @@ async def _next_segment() -> _SpeechSegment | None:
         if speech_handle.interrupted:
             await utils.aio.cancel_and_wait(exe_task)
 
-            # commit results of tools that finished despite the interruption (#3702);
-            # handoffs excluded: not applied when interrupted, must stay retryable
+            # commit results of tools that finished despite the interruption (#3702), so
+            # the next inference doesn't run them again
             interrupted_calls: list[llm.FunctionCall] = []
             interrupted_fnc_outputs: list[llm.FunctionCallOutput] = []
             for sanitized_out in tool_output.output:
-                if sanitized_out.fnc_call_out is not None and sanitized_out.agent_task is None:
-                    interrupted_calls.append(sanitized_out.fnc_call)
-                    interrupted_fnc_outputs.append(sanitized_out.fnc_call_out)
+                interrupted_calls.append(sanitized_out.fnc_call)
+                interrupted_fnc_outputs.append(_interrupted_tool_output(sanitized_out))
 
             if interrupted_tool_messages := interrupted_calls + interrupted_fnc_outputs:
                 self._agent._chat_ctx.insert(interrupted_tool_messages)
@@ -3520,11 +3519,10 @@ async def _next_segment() -> _SpeechSegment | None:
                 function_calls=[], function_call_outputs=[]
             )
             for sanitized_out in tool_output.output:
-                if sanitized_out.fnc_call_out is not None:
-                    new_calls.append(sanitized_out.fnc_call)
-                    new_fnc_outputs.append(sanitized_out.fnc_call_out)
-                    if sanitized_out.reply_required:
-                        fnc_executed_ev._reply_required = True
+                new_calls.append(sanitized_out.fnc_call)
+                new_fnc_outputs.append(sanitized_out.fnc_call_out)
+                if sanitized_out.fnc_call_out.reply_required:
+                    fnc_executed_ev._reply_required = True
 
                 # add the function call and output to the event, including the None outputs
                 fnc_executed_ev.function_calls.append(sanitized_out.fnc_call)
@@ -3979,8 +3977,7 @@ def _tool_execution_started_cb(fnc_call: llm.FunctionCall) -> None:
             self._session._tool_items_added([fnc_call])
 
         def _tool_execution_completed_cb(out: ToolExecutionOutput) -> None:
-            if out.fnc_call_out:
-                speech_handle._item_added([out.fnc_call_out])
+            speech_handle._item_added([out.fnc_call_out])
 
         exe_task, tool_output = perform_tool_executions(
             session=self._session,
@@ -4115,6 +4112,27 @@ def _create_assistant_message(
 
         if speech_handle.interrupted:
             await utils.aio.cancel_and_wait(exe_task)
+
+            # commit results of tools that finished despite the interruption, as the pipeline
+            # task does. the calls are already recorded, so each one answers or the model waits
+            interrupted_fnc_outputs = [
+                _interrupted_tool_output(sanitized_out) for sanitized_out in tool_output.output
+            ]
+
+            if interrupted_fnc_outputs:
+                self._agent._chat_ctx.insert(interrupted_fnc_outputs)
+                self._session._tool_items_added(interrupted_fnc_outputs)
+
+                # unlike the pipeline, a realtime model holds the call open server-side
+                chat_ctx = self._rt_session.chat_ctx.copy()
+                chat_ctx.items.extend(interrupted_fnc_outputs)
+                try:
+                    await self._rt_session.update_chat_ctx(chat_ctx)
+                except llm.RealtimeError as e:
+                    logger.warning(
+                        "failed to sync the tool results of an interrupted generation",
+                        extra={"error": str(e)},
+                    )
             return
 
         # wait for the tool execution to complete
@@ -4134,7 +4152,6 @@ def _create_assistant_message(
             speech_handle._num_steps += 1
 
             new_fnc_outputs: list[llm.FunctionCallOutput] = []
-            generate_tool_reply: bool = False
             fnc_executed_ev = FunctionToolsExecutedEvent(
                 function_calls=[], function_call_outputs=[]
             )
@@ -4146,15 +4163,13 @@ def _create_assistant_message(
                 fnc_executed_ev.function_calls.append(sanitized_out.fnc_call)
                 fnc_executed_ev.function_call_outputs.append(sanitized_out.fnc_call_out)
 
-                if sanitized_out.fnc_call_out is not None:
-                    new_fnc_outputs.append(sanitized_out.fnc_call_out)
-                    if sanitized_out.reply_required:
-                        generate_tool_reply = True
-                        fnc_executed_ev._reply_required = True
+                new_fnc_outputs.append(sanitized_out.fnc_call_out)
+                if sanitized_out.fnc_call_out.reply_required:
+                    fnc_executed_ev._reply_required = True
 
-                    # add tool output to the chat context
-                    self._agent._chat_ctx._upsert_item(sanitized_out.fnc_call_out)
-                    self._session._tool_items_added([sanitized_out.fnc_call_out])
+                # add tool output to the chat context
+                self._agent._chat_ctx._upsert_item(sanitized_out.fnc_call_out)
+                self._session._tool_items_added([sanitized_out.fnc_call_out])
 
                 if new_agent_task is not None and sanitized_out.agent_task is not None:
                     logger.error(
@@ -4169,6 +4184,11 @@ def _create_assistant_message(
 
             self._session.emit("function_tools_executed", fnc_executed_ev)
 
+            if not fnc_executed_ev._reply_required:
+                # a handler can withdraw the reply the tools asked for
+                for fnc_call_out in new_fnc_outputs:
+                    fnc_call_out.reply_required = False
+
             draining = self.scheduling_paused
             if fnc_executed_ev._handoff_required and new_agent_task and not ignore_task_switch:
                 self._session.update_agent(new_agent_task)
@@ -4254,14 +4274,6 @@ async def _wait_for_auto_tool_reply() -> None:
                 self._schedule_speech(
                     speech_handle, SpeechHandle.SPEECH_PRIORITY_NORMAL, force=True
                 )
-            elif (
-                self._rt_session.capabilities.auto_tool_reply_generation
-                and not fnc_executed_ev._reply_required
-                and generate_tool_reply
-            ):
-                logger.warning(
-                    f"Tool reply cannot be prevented when using {self.llm._label}, it generates reply automatically."
-                )
 
     def _update_paused_speech(self, speech_handle: SpeechHandle, timeout: float) -> None:
         """Record that ``speech_handle`` is paused.

livekit-agents/livekit/agents/voice/events.py

@@ -282,17 +282,7 @@ def _make_update_pair(
             extra=dict(self.function_call.extra),
         )
         tool_output = make_tool_output(fnc_call=fnc_call, output=message, exception=None)
-        # fall back to a stub when the message isn't a valid tool output (e.g. raw object)
-        if tool_output.fnc_call_out is None:
-            fnc_call_out = FunctionCallOutput(
-                name=fnc_call.name,
-                call_id=fnc_call.call_id,
-                output=str(message or ""),
-                is_error=False,
-            )
-        else:
-            fnc_call_out = tool_output.fnc_call_out
-        return (fnc_call, fnc_call_out)
+        return (fnc_call, tool_output.fnc_call_out)
 
 
 EventTypes = Literal[
@@ -431,21 +421,19 @@ class FunctionToolsExecutedEvent(BaseModel):
     """Emitted after a batch of function tools finishes executing.
 
     ``function_calls`` and ``function_call_outputs`` are parallel lists: the
-    output at a given index belongs to the call at the same index. When an
-    output is present, its ``call_id`` matches the paired function call's
-    ``call_id``. A ``None`` output means the function call did not produce a
-    value that should be sent back to the LLM, such as when a tool raises
-    ``StopResponse`` or returns an invalid output.
+    output at a given index belongs to the call at the same index and carries
+    the same ``call_id``. Every call has one output, even one whose tool raised
+    ``StopResponse``; such an output asks for no reply with ``reply_required``.
     """
 
     type: Literal["function_tools_executed"] = "function_tools_executed"
     function_calls: list[FunctionCall]
-    function_call_outputs: list[FunctionCallOutput | None]
+    function_call_outputs: list[FunctionCallOutput]
     created_at: float = Field(default_factory=time.time)
     _reply_required: bool = PrivateAttr(default=False)
     _handoff_required: bool = PrivateAttr(default=False)
 
-    def zipped(self) -> list[tuple[FunctionCall, FunctionCallOutput | None]]:
+    def zipped(self) -> list[tuple[FunctionCall, FunctionCallOutput]]:
         """Return calls paired with outputs by list position."""
         return list(zip(self.function_calls, self.function_call_outputs, strict=False))
 

livekit-agents/livekit/agents/voice/generation.py

@@ -884,13 +884,12 @@ async def _traceable_fnc_tool(
 
                         output = make_tool_output(fnc_call=fnc_call, output=None, exception=e)
 
-                    if fnc_call_out := output.fnc_call_out:
-                        current_span.set_attribute(
-                            trace_types.ATTR_FUNCTION_TOOL_OUTPUT, fnc_call_out.output
-                        )
-                        current_span.set_attribute(
-                            trace_types.ATTR_FUNCTION_TOOL_IS_ERROR, fnc_call_out.is_error
-                        )
+                    current_span.set_attribute(
+                        trace_types.ATTR_FUNCTION_TOOL_OUTPUT, output.fnc_call_out.output
+                    )
+                    current_span.set_attribute(
+                        trace_types.ATTR_FUNCTION_TOOL_IS_ERROR, output.fnc_call_out.is_error
+                    )
 
                     # TODO(theomonnom): Add the agent handoff inside the current_span
                     _tool_completed(output)
@@ -944,11 +943,10 @@ async def _traceable_fnc_tool(
 @dataclass
 class ToolExecutionOutput:
     fnc_call: llm.FunctionCall
-    fnc_call_out: llm.FunctionCallOutput | None
+    fnc_call_out: llm.FunctionCallOutput
     agent_task: Agent | None
     raw_output: Any
     raw_exception: BaseException | None
-    reply_required: bool = field(default=True)
 
 
 def make_tool_output(
@@ -989,7 +987,12 @@ def make_tool_output(
             )
             return ToolExecutionOutput(
                 fnc_call=fnc_call.model_copy(),
-                fnc_call_out=None,
+                fnc_call_out=llm.FunctionCallOutput(
+                    name=fnc_call.name,
+                    call_id=fnc_call.call_id,
+                    output="the tool returned more than one agent",
+                    is_error=True,
+                ),
                 agent_task=None,
                 raw_output=output,
                 raw_exception=exception,
@@ -1013,17 +1016,32 @@ def make_tool_output(
     base_result = llm_utils.make_function_call_output(
         fnc_call=fnc_call, output=fnc_out, exception=None
     )
+    # a tool with nothing to say, such as a bare handoff, expects no reply
+    base_result.fnc_call_out.reply_required = fnc_out is not None
 
     return ToolExecutionOutput(
         fnc_call=fnc_call.model_copy(),
         fnc_call_out=base_result.fnc_call_out,
-        reply_required=fnc_out is not None,  # require a reply if the tool returned an output
         agent_task=task,
         raw_output=output,
         raw_exception=exception,
     )
 
 
+def _interrupted_tool_output(out: ToolExecutionOutput) -> llm.FunctionCallOutput:
+    """The output to record for a tool that finished on an interrupted turn.
+
+    A handoff answers as a failure, since the interruption left it unapplied.
+    """
+    fnc_call_out = out.fnc_call_out
+    if out.agent_task is not None:
+        fnc_call_out.output = "the agent handoff was interrupted and did not happen"
+        fnc_call_out.is_error = True
+
+    fnc_call_out.reply_required = False
+    return fnc_call_out
+
+
 INSTRUCTIONS_MESSAGE_ID = "lk.agent_task.instructions"  #  value must not change
 """
 The ID of the instructions message in the chat context. (only for stateless LLMs)

livekit-agents/livekit/agents/voice/remote_session.py

@@ -629,7 +629,6 @@ def _on_function_tools_executed(self, event: FunctionToolsExecutedEvent) -> None
                 is_error=fco.is_error,
             )
             for fco in event.function_call_outputs
-            if fco is not None
         ]
         self._send_event(
             agent_pb.AgentSessionEvent(

livekit-plugins/livekit-plugins-aws/livekit/plugins/aws/experimental/realtime/realtime_model.py

@@ -1770,6 +1770,13 @@ async def update_chat_ctx(self, chat_ctx: llm.ChatContext) -> None:
                 logger.debug(f"function call output: {item}")
                 self._pending_tools.discard(item.call_id)
 
+                if not item.reply_required:
+                    logger.warning(
+                        "a tool result wants no reply, but Nova Sonic will answer it anyway. "
+                        "Sending it regardless, since an unanswered tool use keeps the turn open.",
+                        extra={"function": item.name, "call_id": item.call_id},
+                    )
+
                 # Format tool result as proper JSON
                 if item.is_error:
                     tool_result = json.dumps({"error": str(item.output)})