Module livekit.agents.beta.workflows.address

Classes

class GetAddressResult (address: str)

Expand source code

@dataclass
class GetAddressResult:
    address: str

GetAddressResult(address: 'str')

Instance variables

var address : str

class GetAddressTask (*,
instructions: NotGivenOr[InstructionParts | Instructions | str] = NOT_GIVEN,
chat_ctx: NotGivenOr[llm.ChatContext] = NOT_GIVEN,
turn_detection: NotGivenOr[TurnDetectionMode | None] = NOT_GIVEN,
tools: NotGivenOr[list[llm.Tool | llm.Toolset]] = NOT_GIVEN,
stt: NotGivenOr[stt.STT | None] = NOT_GIVEN,
vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN,
llm: NotGivenOr[llm.LLM | llm.RealtimeModel | None] = NOT_GIVEN,
tts: NotGivenOr[tts.TTS | None] = NOT_GIVEN,
allow_interruptions: NotGivenOr[bool] = NOT_GIVEN,
require_confirmation: NotGivenOr[bool] = NOT_GIVEN,
require_explicit_ask: bool = False,
extra_instructions: str = '')

Expand source code

class GetAddressTask(AgentTask[GetAddressResult]):
    def __init__(
        self,
        *,
        instructions: NotGivenOr[InstructionParts | Instructions | str] = NOT_GIVEN,
        chat_ctx: NotGivenOr[llm.ChatContext] = NOT_GIVEN,
        turn_detection: NotGivenOr[TurnDetectionMode | None] = NOT_GIVEN,
        tools: NotGivenOr[list[llm.Tool | llm.Toolset]] = NOT_GIVEN,
        stt: NotGivenOr[stt.STT | None] = NOT_GIVEN,
        vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN,
        llm: NotGivenOr[llm.LLM | llm.RealtimeModel | None] = NOT_GIVEN,
        tts: NotGivenOr[tts.TTS | None] = NOT_GIVEN,
        allow_interruptions: NotGivenOr[bool] = NOT_GIVEN,
        require_confirmation: NotGivenOr[bool] = NOT_GIVEN,
        require_explicit_ask: bool = False,
        # deprecated
        extra_instructions: str = "",
    ) -> None:
        if not is_given(instructions):
            instructions = InstructionParts(persona=PERSONA, extra=extra_instructions)
        elif extra_instructions:
            logger.warning("`extra_instructions` will be ignored when `instructions` is provided")

        if isinstance(instructions, InstructionParts):
            instructions = Instructions(INSTRUCTIONS_TEMPLATE).format(
                persona=instructions.persona if is_given(instructions.persona) else PERSONA,
                extra=instructions.extra,
                _modality_specific=Instructions(audio=AUDIO_SPECIFIC, text=TEXT_SPECIFIC),
                _confirmation=Instructions(
                    # confirmation is enabled by default for audio, disabled by default for text
                    audio=CONFIRMATION_INSTRUCTION if require_confirmation is not False else "",
                    text=CONFIRMATION_INSTRUCTION if require_confirmation is True else "",
                ),
            )

        assert is_given(instructions)  # for type checking
        self._current_address = ""
        self._require_confirmation = require_confirmation
        self._require_explicit_ask = require_explicit_ask

        super().__init__(
            instructions=instructions,
            chat_ctx=chat_ctx,
            turn_detection=turn_detection,
            tools=[*(tools or []), self._build_update_address_tool()],
            stt=stt,
            vad=vad,
            llm=llm,
            tts=tts,
            allow_interruptions=allow_interruptions,
        )

    async def on_enter(self) -> None:
        self.session.generate_reply(instructions="Ask the user to provide their address.")

    def _build_update_address_tool(self) -> llm.FunctionTool:
        # Built dynamically so we can apply IGNORE_ON_ENTER per-instance
        # based on require_explicit_ask.
        flags = ToolFlag.IGNORE_ON_ENTER if self._require_explicit_ask else ToolFlag.NONE

        @function_tool(flags=flags)
        async def update_address(
            street_address: str,
            unit_number: str,
            locality: str,
            country: str,
            ctx: RunContext,
        ) -> str | None:
            """Update the address provided by the user.

            Args:
                street_address (str): Dependent on country, may include fields like house number, street name, block, or district
                unit_number (str): The unit number, for example Floor 1 or Apartment 12. If there is no unit number, return ''
                locality (str): Dependent on country, may include fields like city, zip code, or province
                country (str): The country the user lives in spelled out fully
            """
            return await self._update_address_impl(
                street_address, unit_number, locality, country, ctx
            )

        return update_address

    async def _update_address_impl(
        self,
        street_address: str,
        unit_number: str,
        locality: str,
        country: str,
        ctx: RunContext,
    ) -> str | None:
        address_fields = (
            [street_address, unit_number, locality, country]
            if unit_number.strip()
            else [street_address, locality, country]
        )
        address = " ".join(address_fields)
        self._current_address = address

        if not self._confirmation_required(ctx):
            if not self.done():
                self.complete(GetAddressResult(address=self._current_address))
            return None

        confirm_tool = self._build_confirm_tool(address=address)
        current_tools = [t for t in self.tools if t.id != "confirm_address"]
        current_tools.append(confirm_tool)
        await self.update_tools(current_tools)

        return (
            f"The address has been updated to {address}\n"
            f"Repeat the address field by field: {address_fields} if needed\n"
            f"Prompt the user for confirmation, do not call `confirm_address` directly"
        )

    def _build_confirm_tool(self, *, address: str) -> llm.FunctionTool:
        # confirm tool is only injected after update_address is called,
        # preventing the LLM from hallucinating a confirmation without user input
        @function_tool()
        async def confirm_address() -> None:
            """Call after the user confirms the address is correct."""
            if address != self._current_address:
                self.session.generate_reply(
                    instructions="The address has changed since confirmation was requested, ask the user to confirm the updated address."
                )
                return

            if not self.done():
                self.complete(GetAddressResult(address=address))

        return confirm_address

    @function_tool(flags=ToolFlag.IGNORE_ON_ENTER)
    async def decline_address_capture(self, reason: str) -> None:
        """Handles the case when the user explicitly declines to provide an address.

        Args:
            reason: A short explanation of why the user declined to provide the address
        """
        if not self.done():
            self.complete(ToolError(f"couldn't get the address: {reason}"))

    def _confirmation_required(self, ctx: RunContext) -> bool:
        if is_given(self._require_confirmation):
            return self._require_confirmation
        return ctx.speech_handle.input_details.modality == "audio"

Abstract base class for generic types.

On Python 3.12 and newer, generic classes implicitly inherit from Generic when they declare a parameter list after the class's name::

class Mapping[KT, VT]:
    def __getitem__(self, key: KT) -> VT:
        ...
    # Etc.

On older versions of Python, however, generic classes have to explicitly inherit from Generic.

After a class has been declared to be generic, it can then be used as follows::

def lookup_name[KT, VT](mapping: Mapping[KT, VT], key: KT, default: VT) -> VT:
    try:
        return mapping[key]
    except KeyError:
        return default

Ancestors

• livekit.agents.voice.agent.AgentTask
• livekit.agents.voice.agent.Agent
• typing.Generic

Methods

async def decline_address_capture(self, reason: str) ‑> None

Expand source code

@function_tool(flags=ToolFlag.IGNORE_ON_ENTER)
async def decline_address_capture(self, reason: str) -> None:
    """Handles the case when the user explicitly declines to provide an address.

    Args:
        reason: A short explanation of why the user declined to provide the address
    """
    if not self.done():
        self.complete(ToolError(f"couldn't get the address: {reason}"))

Handles the case when the user explicitly declines to provide an address.

Args

reason

A short explanation of why the user declined to provide the address

async def on_enter(self) ‑> None

Expand source code

async def on_enter(self) -> None:
    self.session.generate_reply(instructions="Ask the user to provide their address.")

Called when the task is entered