livekit--agents
wehub-resource-sync
136 行
5.4 KiB
Python
136 行
5.4 KiB
Python
import asyncio
|
|
from collections.abc import Awaitable, Callable
|
|
from typing import Any
|
|
|
|
from ...job import get_job_context
|
|
from ...llm import RealtimeModel, ToolFlag, Toolset, function_tool
|
|
from ...log import logger
|
|
from ...voice.events import CloseEvent, RunContext, SpeechCreatedEvent
|
|
from ...voice.speech_handle import SpeechHandle
|
|
|
|
END_CALL_DESCRIPTION = """
|
|
Ends the current call and disconnects immediately.
|
|
|
|
Call when:
|
|
- The user clearly indicates they are done (e.g., “that’s all, bye”).
|
|
|
|
Do not call when:
|
|
- The user asks to pause, hold, or transfer.
|
|
- Intent is unclear.
|
|
|
|
This is the final action the agent can take.
|
|
Once called, no further interaction is possible with the user.
|
|
Don't generate any other text or response when the tool is called.
|
|
"""
|
|
|
|
|
|
class EndCallTool(Toolset):
|
|
def __init__(
|
|
self,
|
|
*,
|
|
extra_description: str = "",
|
|
delete_room: bool = True,
|
|
end_instructions: str | None = "say goodbye to the user",
|
|
ignore_on_enter: bool = False,
|
|
on_tool_called: Callable[[Toolset.ToolCalledEvent], Awaitable[None]] | None = None,
|
|
on_tool_completed: Callable[[Toolset.ToolCompletedEvent], Awaitable[None]] | None = None,
|
|
):
|
|
"""
|
|
This tool allows the agent to end the call and disconnect from the room.
|
|
|
|
Args:
|
|
extra_description: Additional description to add to the end call tool.
|
|
delete_room: Whether to delete the room when the user ends the call. deleting the room disconnects all remote users, including SIP callers.
|
|
end_instructions: Tool output to the LLM for generating the tool response.
|
|
ignore_on_enter: Hide the tool during ``on_enter`` so the model can't end the call while greeting.
|
|
on_tool_called: Callback to call when the tool is called.
|
|
on_tool_completed: Callback to call when the tool is completed.
|
|
"""
|
|
end_call_tool = function_tool(
|
|
self._end_call,
|
|
name="end_call",
|
|
description=f"{END_CALL_DESCRIPTION}\n{extra_description}",
|
|
flags=ToolFlag.IGNORE_ON_ENTER if ignore_on_enter else ToolFlag.NONE,
|
|
)
|
|
super().__init__(id="end_call", tools=[end_call_tool])
|
|
|
|
self._delete_room = delete_room
|
|
self._extra_description = extra_description
|
|
|
|
self._end_instructions = end_instructions
|
|
self._on_tool_called = on_tool_called
|
|
self._on_tool_completed = on_tool_completed
|
|
|
|
self._shutdown_session_task: asyncio.Task[None] | None = None
|
|
|
|
async def _end_call(self, ctx: RunContext) -> Any | None:
|
|
logger.debug("end_call tool called")
|
|
activity = ctx.session.current_agent._get_activity_or_raise()
|
|
llm_v = activity.llm
|
|
|
|
def _on_speech_done(_: SpeechHandle) -> None:
|
|
# read auto_tool_reply_generation from the active realtime session so a fallback
|
|
# adapter reports the model actually in use
|
|
rt_session = activity.realtime_llm_session
|
|
auto_tool_reply = (
|
|
rt_session is not None and rt_session.capabilities.auto_tool_reply_generation
|
|
)
|
|
if not isinstance(llm_v, RealtimeModel) or not auto_tool_reply:
|
|
# tool reply will reuse the same speech handle, so we can shutdown the session
|
|
# directly after this speech handle is done
|
|
ctx.session.shutdown()
|
|
else:
|
|
self._shutdown_session_task = asyncio.create_task(
|
|
self._delayed_session_shutdown(ctx)
|
|
)
|
|
|
|
ctx.speech_handle.add_done_callback(_on_speech_done)
|
|
ctx.session.once("close", self._on_session_close)
|
|
|
|
if self._on_tool_called:
|
|
await self._on_tool_called(Toolset.ToolCalledEvent(ctx=ctx, arguments={}))
|
|
|
|
completed_ev = Toolset.ToolCompletedEvent(ctx=ctx, output=self._end_instructions)
|
|
if self._on_tool_completed:
|
|
await self._on_tool_completed(completed_ev)
|
|
|
|
return completed_ev.output
|
|
|
|
async def _delayed_session_shutdown(self, ctx: RunContext) -> None:
|
|
"""Shutdown the session after the tool reply is played out"""
|
|
speech_created_fut = asyncio.Future[SpeechHandle]()
|
|
|
|
@ctx.session.once("speech_created")
|
|
def _on_speech_created(ev: SpeechCreatedEvent) -> None:
|
|
if not speech_created_fut.done():
|
|
speech_created_fut.set_result(ev.speech_handle)
|
|
|
|
try:
|
|
speech_handle = await asyncio.wait_for(speech_created_fut, timeout=5.0)
|
|
await speech_handle
|
|
except asyncio.TimeoutError:
|
|
logger.warning("tool reply timed out, shutting down session")
|
|
finally:
|
|
ctx.session.off("speech_created", _on_speech_created)
|
|
ctx.session.shutdown()
|
|
|
|
def _on_session_close(self, ev: CloseEvent) -> None:
|
|
"""Close the job process when AgentSession is closed"""
|
|
if self._shutdown_session_task:
|
|
# cleanup
|
|
self._shutdown_session_task.cancel()
|
|
self._shutdown_session_task = None
|
|
|
|
job_ctx = get_job_context()
|
|
|
|
if self._delete_room:
|
|
|
|
async def _on_shutdown() -> None:
|
|
logger.info("deleting the room because the user ended the call")
|
|
await job_ctx.delete_room()
|
|
|
|
job_ctx.add_shutdown_callback(_on_shutdown)
|
|
|
|
# shutdown the job process
|
|
job_ctx.shutdown(reason=ev.reason.value)
|