from __future__ import annotations import re from dataclasses import dataclass from typing import TYPE_CHECKING from ... import llm, stt, tts, vad from ...llm.chat_context import Instructions from ...llm.tool_context import ToolError, ToolFlag, function_tool from ...log import logger from ...types import NOT_GIVEN, NotGivenOr from ...utils import is_given from ...voice.agent import AgentTask from ...voice.events import RunContext from .utils import WorkflowInstructions if TYPE_CHECKING: from ...voice.turn import TurnDetectionMode @dataclass class GetEmailResult: email_address: str class GetEmailTask(AgentTask[GetEmailResult]): def __init__( self, *, instructions: NotGivenOr[WorkflowInstructions | 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 = WorkflowInstructions(persona=PERSONA, extra=extra_instructions) elif extra_instructions: logger.warning("`extra_instructions` will be ignored when `instructions` is provided") if isinstance(instructions, WorkflowInstructions): instructions = instructions.resolve( template=INSTRUCTIONS_TEMPLATE, default_persona=PERSONA, _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 isinstance(instructions, (str, Instructions)) # for type checking self._current_email = "" 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_email_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 an email address.") def _build_update_email_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_email_address(email: str, ctx: RunContext) -> str | None: """Update the email address provided by the user. Args: email: The email address provided by the user """ return await self._update_email_impl(email, ctx) return update_email_address async def _update_email_impl(self, email: str, ctx: RunContext) -> str | None: email = email.strip() if not re.match(EMAIL_REGEX, email): raise ToolError(f"Invalid email address provided: {email}") self._current_email = email separated_email = " ".join(email) if not self._confirmation_required(ctx): if not self.done(): self.complete(GetEmailResult(email_address=self._current_email)) return None # no need to continue the conversation confirm_tool = self._build_confirm_tool(email=email) current_tools = [t for t in self.tools if t.id != "confirm_email_address"] current_tools.append(confirm_tool) await self.update_tools(current_tools) return ( f"The email has been updated to {email}\n" f"Repeat the email character by character: {separated_email} if needed\n" f"Prompt the user for confirmation, do not call `confirm_email_address` directly" ) def _build_confirm_tool(self, *, email: str) -> llm.FunctionTool: @function_tool() async def confirm_email_address() -> None: """Call after the user confirms the email address is correct.""" if email != self._current_email: self.session.generate_reply( instructions="The email has changed since confirmation was requested, ask the user to confirm the updated email." ) return if not self.done(): self.complete(GetEmailResult(email_address=email)) return confirm_email_address @function_tool(flags=ToolFlag.IGNORE_ON_ENTER) async def decline_email_capture(self, reason: str) -> None: """Handles the case when the user explicitly declines to provide an email address. Args: reason: A short explanation of why the user declined to provide the email address """ if not self.done(): self.complete(ToolError(f"couldn't get the email 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" EMAIL_REGEX = ( r"^[A-Za-z0-9][A-Za-z0-9._%+\-]*@(?:[A-Za-z0-9](?:[A-Za-z0-9\-]*[A-Za-z0-9])?\.)+[A-Za-z]{2,}$" ) # instructions PERSONA = "You are only a single step in a broader system, responsible solely for capturing an email address." AUDIO_SPECIFIC = """\ Handle input as noisy voice transcription. Expect that users will say emails aloud with formats like: - 'john dot doe at gmail dot com' - 'susan underscore smith at yahoo dot co dot uk' - 'dave dash b at protonmail dot com' - 'jane at example' (partial—prompt for the domain) - 'theo t h e o at livekit dot io' (name followed by spelling) Normalize common spoken patterns silently: - Convert words like 'dot', 'underscore', 'dash', 'plus' into symbols: `.`, `_`, `-`, `+`. - Convert 'at' to `@`. - Recognize patterns where users speak their name or a word, followed by spelling: e.g., 'john j o h n'. - Filter out filler words or hesitations. - Assume some spelling if contextually obvious (e.g. 'mike b two two' → mikeb22). Don't mention corrections. Treat inputs as possibly imperfect but fix them silently.""" TEXT_SPECIFIC = """\ Handle input as typed text. Expect users to type their email address directly in standard format. If the address looks almost correct but has minor typos (e.g. missing '@' or domain), prompt for clarification.""" CONFIRMATION_INSTRUCTION = """\ Call `confirm_email_address` after the user confirmed the email address is correct.""" INSTRUCTIONS_TEMPLATE = """\ {persona} {_modality_specific} Call `update_email_address` at the first opportunity whenever you form a new hypothesis about the email. (before asking any questions or providing any answers.) Don't invent new email addresses, stick strictly to what the user said. {_confirmation} If the email is unclear or invalid, or it takes too much back-and-forth, prompt for it in parts: first the part before the '@', then the domain—only if needed. Ignore unrelated input and avoid going off-topic. Do not generate markdown, greetings, or unnecessary commentary. Always explicitly invoke a tool when applicable. Do not simulate tool usage, no real action is taken unless the tool is explicitly called. {extra} """