Module livekit.agents.beta.workflows.phone_number

Classes

class GetPhoneNumberResult (phone_number: str)

Expand source code

@dataclass
class GetPhoneNumberResult:
    phone_number: str

GetPhoneNumberResult(phone_number: 'str')

Instance variables

var phone_number : str

class GetPhoneNumberTask (extra_instructions: str = '',
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)

Expand source code

class GetPhoneNumberTask(AgentTask[GetPhoneNumberResult]):
    def __init__(
        self,
        extra_instructions: str = "",
        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,
    ) -> None:
        confirmation_instructions = (
            "Call `confirm_phone_number` after the user confirmed the phone number is correct."
        )
        extra = extra_instructions if extra_instructions else ""

        super().__init__(
            instructions=Instructions(
                _BASE_INSTRUCTIONS.format(
                    modality_specific=_AUDIO_SPECIFIC,
                    confirmation_instructions=(
                        confirmation_instructions if require_confirmation is not False else ""
                    ),
                    extra_instructions=extra,
                ),
                text=_BASE_INSTRUCTIONS.format(
                    modality_specific=_TEXT_SPECIFIC,
                    confirmation_instructions=(
                        confirmation_instructions if require_confirmation is True else ""
                    ),
                    extra_instructions=extra,
                ),
            ),
            chat_ctx=chat_ctx,
            turn_detection=turn_detection,
            tools=tools or [],
            stt=stt,
            vad=vad,
            llm=llm,
            tts=tts,
            allow_interruptions=allow_interruptions,
        )

        self._current_phone_number = ""
        self._require_confirmation = require_confirmation

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

    @function_tool()
    async def update_phone_number(self, phone_number: str, ctx: RunContext) -> str | None:
        """Update the phone number provided by the user.

        Args:
            phone_number: The phone number provided by the user, digits only with optional leading +
        """
        cleaned = re.sub(r"[\s\-().]+", "", phone_number.strip())

        if not re.match(PHONE_REGEX, cleaned):
            raise ToolError(f"Invalid phone number provided: {phone_number}")

        self._current_phone_number = cleaned

        if not self._confirmation_required(ctx):
            if not self.done():
                self.complete(GetPhoneNumberResult(phone_number=self._current_phone_number))
            return None  # no need to continue the conversation

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

        return (
            f"The phone number has been updated to {cleaned}\n"
            f"Read the number back to the user in groups.\n"
            f"Prompt the user for confirmation, do not call `confirm_phone_number` directly"
        )

    def _build_confirm_tool(self, *, phone_number: str) -> llm.FunctionTool:
        @function_tool()
        async def confirm_phone_number() -> None:
            """Call after the user confirms the phone number is correct."""
            if phone_number != self._current_phone_number:
                self.session.generate_reply(
                    instructions="The phone number has changed since confirmation was requested, ask the user to confirm the updated number."
                )
                return

            if not self.done():
                self.complete(GetPhoneNumberResult(phone_number=phone_number))

        return confirm_phone_number

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

        Args:
            reason: A short explanation of why the user declined to provide the phone number
        """
        if not self.done():
            self.complete(ToolError(f"couldn't get the phone number: {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_phone_number_capture(self, reason: str) ‑> None

Expand source code

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

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

Handles the case when the user explicitly declines to provide a phone number.

Args

reason

A short explanation of why the user declined to provide the phone number

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 phone number.")

Called when the task is entered

async def update_phone_number(self, phone_number: str, ctx: RunContext) ‑> str | None

Expand source code

@function_tool()
async def update_phone_number(self, phone_number: str, ctx: RunContext) -> str | None:
    """Update the phone number provided by the user.

    Args:
        phone_number: The phone number provided by the user, digits only with optional leading +
    """
    cleaned = re.sub(r"[\s\-().]+", "", phone_number.strip())

    if not re.match(PHONE_REGEX, cleaned):
        raise ToolError(f"Invalid phone number provided: {phone_number}")

    self._current_phone_number = cleaned

    if not self._confirmation_required(ctx):
        if not self.done():
            self.complete(GetPhoneNumberResult(phone_number=self._current_phone_number))
        return None  # no need to continue the conversation

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

    return (
        f"The phone number has been updated to {cleaned}\n"
        f"Read the number back to the user in groups.\n"
        f"Prompt the user for confirmation, do not call `confirm_phone_number` directly"
    )

Update the phone number provided by the user.

Args

phone_number

The phone number provided by the user, digits only with optional leading +