feat: add dependency-aware tool execution

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

@@ -199,6 +199,19 @@ async def check_order(ctx: RunContext, order_id: str) -> str: ...
                      guarantee; put that in the tool body.
 """
 
+DependencyNames = tuple[str, ...]
+"""Tool names used as prerequisites within one function-stream batch."""
+
+
+def _normalize_dependency_names(after: Sequence[str]) -> DependencyNames:
+    """Validate dependency metadata and return an immutable sequence of tool names."""
+    if isinstance(after, str):
+        raise TypeError("function_tool(after=...) must be a sequence of tool names")
+    names = tuple(after)
+    if any(not isinstance(name, str) for name in names):
+        raise TypeError("function_tool(after=...) must contain only tool names")
+    return names
+
 
 @dataclass
 class FunctionToolInfo:
@@ -207,6 +220,8 @@ class FunctionToolInfo:
     flags: ToolFlag
     on_duplicate: DuplicateMode = "allow"
     duplicate_scope: DuplicateScope = "name"
+    after: DependencyNames = ()
+    """Batch-local prerequisites; progress updates do not count as terminal completion."""
 
 
 class RawFunctionDescription(TypedDict):
@@ -232,6 +247,8 @@ class RawFunctionToolInfo:
     flags: ToolFlag
     on_duplicate: DuplicateMode = "allow"
     duplicate_scope: DuplicateScope = "name"
+    after: DependencyNames = ()
+    """Batch-local prerequisites; progress updates do not count as terminal completion."""
 
 
 CONFIRM_DUPLICATE_PARAM = "lk_agents_confirm_duplicate"
@@ -311,6 +328,7 @@ def function_tool(
     flags: ToolFlag = ToolFlag.NONE,
     on_duplicate: DuplicateMode = "allow",
     duplicate_scope: DuplicateScope = "name",
+    after: Sequence[str] = (),
 ) -> RawFunctionTool[_P, _R]: ...
 
 
@@ -322,6 +340,7 @@ def function_tool(
     flags: ToolFlag = ToolFlag.NONE,
     on_duplicate: DuplicateMode = "allow",
     duplicate_scope: DuplicateScope = "name",
+    after: Sequence[str] = (),
 ) -> Callable[[Callable[_P, _R]], RawFunctionTool[_P, _R]]: ...
 
 
@@ -334,6 +353,7 @@ def function_tool(
     flags: ToolFlag = ToolFlag.NONE,
     on_duplicate: DuplicateMode = "allow",
     duplicate_scope: DuplicateScope = "name",
+    after: Sequence[str] = (),
 ) -> FunctionTool[_P, _R]: ...
 
 
@@ -346,6 +366,7 @@ def function_tool(
     flags: ToolFlag = ToolFlag.NONE,
     on_duplicate: DuplicateMode = "allow",
     duplicate_scope: DuplicateScope = "name",
+    after: Sequence[str] = (),
 ) -> Callable[[Callable[_P, _R]], FunctionTool[_P, _R]]: ...
 
 
@@ -358,11 +379,31 @@ def function_tool(
     flags: ToolFlag = ToolFlag.NONE,
     on_duplicate: DuplicateMode = "allow",
     duplicate_scope: DuplicateScope = "name",
+    after: Sequence[str] = (),
 ) -> (
     FunctionTool[_P, _R]
     | RawFunctionTool[_P, _R]
     | Callable[[Callable[_P, _R]], FunctionTool[_P, _R] | RawFunctionTool[_P, _R]]
 ):
+    """Create a function tool from a callable.
+
+    Args:
+        after: Tool names that must finish before this tool is admitted. Dependencies
+            are local to one streamed function-call batch: every matching call is a
+            prerequisite. Independent root calls may start before end-of-stream
+            (EOF), while dependent calls wait until EOF before admission. If no
+            matching call is present in the batch, no dependency edge is created.
+            ``ctx.update()`` reports progress but does not satisfy a dependency;
+            terminal completion is required. Failed or individually cancelled
+            prerequisites also satisfy the ordering constraint; dependents still run.
+
+            Cycles are validated against the available tool snapshot before dispatch.
+            This metadata orders execution only; it does not regenerate arguments or
+            pass results between tools. Cancellation, session close, and permanent
+            handoff abandon queued dependents without running them.
+    """
+    normalized_after = _normalize_dependency_names(after)
+
     def deco_raw(
         func: Callable[_P, _R],
     ) -> RawFunctionTool[_P, _R]:
@@ -385,6 +426,7 @@ def deco_raw(
             flags=flags,
             on_duplicate=on_duplicate,
             duplicate_scope=duplicate_scope,
+            after=normalized_after,
         )
         return RawFunctionTool(func, info)
 
@@ -402,6 +444,7 @@ def deco_func(func: Callable[_P, _R]) -> FunctionTool[_P, _R]:
             flags=flags,
             on_duplicate=on_duplicate,
             duplicate_scope=duplicate_scope,
+            after=normalized_after,
         )
         return FunctionTool(wrapped, info)
 

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

@@ -78,6 +78,7 @@
 from .generation import (
     ToolExecutionOutput,
     _AudioOutput,
+    _DependencyScheduler,
     _ForwardOutput,
     _inject_running_tool_calls,
     _interrupted_tool_output,
@@ -382,6 +383,10 @@ def __init__(self, agent: Agent, sess: AgentSession) -> None:
         self._tool_executor = _ToolExecutor(
             owning_activity=self, async_tool_options=activity_options
         )
+        # Dependency schedulers outlive the first tool progress result.  Keep them
+        # activity-owned so close/force-interrupt can settle queued nodes before the
+        # executor begins draining admitted work.
+        self._dependency_schedulers: set[_DependencyScheduler] = set()
 
         self._user_turn_exceeded_atask: asyncio.Task[None] | None = None
         self._user_turn_exceeded_locked: bool = False
@@ -1621,6 +1626,8 @@ async def aclose(self) -> None:
             self._cancel_preemptive_generation()
             await self._session._keyterm_detector.aclose()
 
+            await self._abandon_dependency_schedulers()
+
             # on_exit_task should be awaited in `drain`
             self._on_exit_task = None
 
@@ -1648,6 +1655,20 @@ async def aclose(self) -> None:
 
             self._agent._activity = None
 
+    async def _abandon_dependency_schedulers(self) -> None:
+        """Abandon response dependency graphs before this activity finishes closing."""
+        schedulers = tuple(self._dependency_schedulers)
+        if not schedulers:
+            return
+        await asyncio.gather(
+            *(
+                scheduler.abandon(ToolError("agent activity is closing"))
+                for scheduler in schedulers
+            ),
+            return_exceptions=True,
+        )
+        self._dependency_schedulers.difference_update(schedulers)
+
     def push_audio(self, frame: rtc.AudioFrame) -> None:
         if not self._started:
             return
@@ -1906,6 +1927,7 @@ def interrupt(
         self._cancel_preemptive_generation()
 
         future = asyncio.Future[None]()
+        abandon_task = asyncio.create_task(self._abandon_dependency_schedulers()) if force else None
 
         interrupted_speeches = self._interrupt_background_speeches(force=force, source=source)
 
@@ -1932,16 +1954,21 @@ def interrupt(
 
             interrupted_speeches.append(speech)
 
-        if not interrupted_speeches:
+        def maybe_done(_: Any = None) -> None:
+            if future.done():
+                return
+            if interrupted_speeches and not all(speech.done() for speech in interrupted_speeches):
+                return
+            if abandon_task is not None and not abandon_task.done():
+                return
             future.set_result(None)
-        else:
-
-            def on_playout_done(_: SpeechHandle) -> None:
-                if not future.done() and all(speech.done() for speech in interrupted_speeches):
-                    future.set_result(None)
 
+        if abandon_task is not None:
+            abandon_task.add_done_callback(maybe_done)
+        if interrupted_speeches:
             for speech in interrupted_speeches:
-                speech.add_done_callback(on_playout_done)
+                speech.add_done_callback(maybe_done)
+        maybe_done()
 
         return future
 
@@ -3883,13 +3910,14 @@ async def _next_segment() -> _SpeechSegment | None:
         speech_handle._mark_generation_done()  # mark the playout done before waiting for the tool execution  # noqa: E501
 
         if speech_handle.interrupted:
+            tool_output.interrupted = True
             await utils.aio.cancel_and_wait(exe_task)
 
             # 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:
+            for sanitized_out in tool_output.interrupted_outputs():
                 interrupted_calls.append(sanitized_out.fnc_call)
                 interrupted_fnc_outputs.append(_interrupted_tool_output(sanitized_out))
 
@@ -3908,7 +3936,13 @@ async def _next_segment() -> _SpeechSegment | None:
         # wait for the tool execution to complete
         self._background_speeches.add(speech_handle)
         try:
-            await exe_task
+            try:
+                await exe_task
+            except BaseException as error:
+                if isinstance(error, asyncio.CancelledError):
+                    raise
+                speech_handle._mark_done(error=error)
+                return
         finally:
             self._background_speeches.discard(speech_handle)
 
@@ -3955,6 +3989,7 @@ async def _next_segment() -> _SpeechSegment | None:
                 fnc_executed_ev._handoff_required = True
 
             self._session.emit("function_tools_executed", fnc_executed_ev)
+            tool_output.resolve_handoffs(fnc_executed_ev.has_agent_handoff)
 
             draining = self.scheduling_paused
             if fnc_executed_ev._handoff_required and new_agent_task and not ignore_task_switch:
@@ -3972,6 +4007,7 @@ async def _next_segment() -> _SpeechSegment | None:
             if tool_messages:
                 self._agent._chat_ctx.insert(tool_messages)
                 self._session._tool_items_added(tool_messages)
+            tool_output.release_initial_batch()
 
             if fnc_executed_ev.has_tool_reply and not speech_handle.interrupted:
                 # forwarding chat_ctx to the tool reply: drop the in-progress placeholders
@@ -4579,13 +4615,14 @@ def _create_assistant_message(
         speech_handle._mark_generation_done()
 
         if speech_handle.interrupted:
+            tool_output.interrupted = True
             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_calls: list[llm.FunctionCall] = []
             interrupted_fnc_outputs: list[llm.FunctionCallOutput] = []
-            for sanitized_out in tool_output.output:
+            for sanitized_out in tool_output.interrupted_outputs():
                 interrupted_calls.append(sanitized_out.fnc_call)
                 interrupted_fnc_outputs.append(_interrupted_tool_output(sanitized_out))
 
@@ -4619,7 +4656,13 @@ def _create_assistant_message(
 
         self._background_speeches.add(speech_handle)
         try:
-            await exe_task
+            try:
+                await exe_task
+            except BaseException as error:
+                if isinstance(error, asyncio.CancelledError):
+                    raise
+                speech_handle._mark_done(error=error)
+                return
         finally:
             self._background_speeches.discard(speech_handle)
 
@@ -4663,6 +4706,7 @@ def _create_assistant_message(
                 fnc_executed_ev._handoff_required = True
 
             self._session.emit("function_tools_executed", fnc_executed_ev)
+            tool_output.resolve_handoffs(fnc_executed_ev.has_agent_handoff)
 
             draining = self.scheduling_paused
             if fnc_executed_ev._handoff_required and new_agent_task and not ignore_task_switch:
@@ -4725,6 +4769,8 @@ async def _wait_for_auto_tool_reply() -> None:
                             self._pending_auto_tool_reply_fut = None
                         auto_reply_fut.set_result(None)
 
+                tool_output.release_initial_batch()
+
             tool_reply_expected = fnc_executed_ev.has_tool_reply
             if tool_reply_expected and not self._rt_session.capabilities.auto_tool_reply_generation:
                 self._rt_session.interrupt()