LiveKit Agent: Credit card collection Task

livekit_agent_credit_card_collection_task.py

"""
This LiveKit Agent showcases how to create a Task that collects credit card information.

This Agent was built from the agent-starter-python template and customized with a Task that collects credit card data 
one step at a time.

To get started, first setup the starter as instructed here: https://github.com/livekit-examples/agent-starter-python 
then replace the content of agent.py with this gist.

The conversation flow should go as follows:
- The agent introduces itself and checks if the user is ready (to check if the user is connected and can hear the agent).
- Then, the agent informs the user that it needs their credit card information and asks if they're ok with it.
  - If they aren't, the call ends
- If they are, the agent asks in turn:
  - Cardholder name
  - Card number (16 digits)
  - Card expiration date
  - Card security code
- If all data is collected successfully, the agent then starts an open-ended conversation.
- If at any point the user fails to provide the requested data, the agent terminates the call.

This version was tested with OpenAI's Realtime (`gpt-realtime`), with its pros and cons.
"""

from __future__ import annotations
import logging

import datetime
from dataclasses import dataclass
from typing import TYPE_CHECKING

from dotenv import load_dotenv
from livekit import api
from livekit.agents import (
    Agent,
    AgentSession,
    JobContext,
    JobProcess,
    MetricsCollectedEvent,
    RoomInputOptions,
    WorkerOptions,
    cli,
    metrics,
    function_tool, 
    get_job_context,
    llm, 
    stt, 
    tts, 
    vad,
    RunContext
)
from livekit.plugins import noise_cancellation, silero
from livekit.plugins import openai

from livekit.agents.llm.tool_context import ToolError
from livekit.agents.types import NOT_GIVEN, NotGivenOr
from livekit.agents.voice.agent import AgentTask
from livekit.agents.voice.speech_handle import SpeechHandle

if TYPE_CHECKING:
    from livekit.agents.voice.agent_session import TurnDetectionMode

logger = logging.getLogger("agent")

load_dotenv(".env.local")


@dataclass
class CreditCardData:
    holder_name: CardHolderName
    number: CardNumber
    expiration_date: CardExpirationDate
    security_code: CardSecurityCode


@dataclass
class CardHolderName:
    full_name: str  # Not wise to try handling all the name formats out there, a single string works best


@dataclass
class CardNumber:
    number: str


@dataclass
class CardExpirationDate:
    month: int
    year: int


@dataclass
class CardSecurityCode:
    code: str


class GetCreditCardData(AgentTask[CreditCardData]):
    """
    Task that collects credit card information. 
    
    Note: I considered not having this abstraction and then ended up keeping it.

    Pros:
    - Easy to reuse in bigger workflows.
    - Produces a single dataclass with all the data (helps preventing the agent from returning partial results).
    - Easy to abort the whole procedure without aborting the call when used in a larger workflow.

    Cons:
    - It's just a shell task that may introduce unnecessary latency.

    TBH in this example it looks a bit overkill to have nested tasks, but if I were designing this for reusability I'd 
    keep this structure to isolate the credit card info collection from the rest of the conversation. It makes the LLM
    less likely to stray way mid-process.
    """
    def __init__(
        self,
        extra_instructions: str = "",
        chat_ctx: NotGivenOr[llm.ChatContext] = NOT_GIVEN,
        turn_detection: NotGivenOr[TurnDetectionMode | None] = NOT_GIVEN,
        tools: NotGivenOr[list[llm.FunctionTool | llm.RawFunctionTool]] = 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,
    ) -> None:
        super().__init__(
            instructions=("""
                You are only a single step in a broader system, responsible solely for capturing a credit card's data.
                Handle input as noisy voice transcription.
                Don't invent the content of any of the fields, stick strictly to what the user said. 
                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.
                After the first message, call the `collect_data` tool. If the user declines to share the data, call `decline_data_capture` instead.
                """
                + extra_instructions
            ),
            chat_ctx=chat_ctx,
            turn_detection=turn_detection,
            tools=tools,
            stt=stt,
            vad=vad,
            llm=llm,
            tts=tts,
            allow_interruptions=allow_interruptions,
        )

    async def on_enter(self) -> None:
        await self.session.generate_reply(instructions="Tell them that before you proceed, you will ask them for their credit card information. Ask if they're ok with it.")

    @function_tool
    async def collect_data(self, context: RunContext):
        """
        Collects the user's card information. Always call this tool as soon as the user has been informed they need to share their credit card data.
        """
        if not (holder_name := await GetCreditCardHolderName(chat_ctx=self.chat_ctx)):
            await self.session.generate_reply(instructions="Inform the user that you are unable to proceed and will end the call.")
            if not self.done():
                self.complete(ToolError(f"couldn't get all the data: failed to obtain cardholder name (reason: {holder_name})"))
            return
        
        await self.session.generate_reply(instructions="Inform the user we're going to move on to the card number.")
        if not (cc_number := await GetCreditCardNumber(chat_ctx=self.chat_ctx)):
            await self.session.generate_reply(instructions="Inform the user that you are unable to proceed and will end the call.")
            if not self.done():
                self.complete(ToolError(f"couldn't get all the data: failed to obtain card number (reason: {cc_number})"))
            return
        
        await self.session.generate_reply(instructions="Inform the user we're going to move on to the card expiration date.")
        if not (expiration_date := await GetCreditCardExpirationDate(chat_ctx=self.chat_ctx)):
            await self.session.generate_reply(instructions="Inform the user that you are unable to proceed and will end the call.")
            if not self.done():
                self.complete(ToolError(f"couldn't get all the data: failed to obtain card expiration date (reason: {expiration_date})"))
            return
        
        await self.session.generate_reply(instructions="Inform the user we're going to move on to the security code.")
        if not (security_code := await GetCreditCardSecurityCode(chat_ctx=self.chat_ctx)):
            await self.session.generate_reply(instructions="Inform the user that you are unable to proceed and will end the call.")
            if not self.done():
                self.complete(ToolError(f"couldn't get all the data: failed to obtain card security code (reason: {security_code})"))
            return

        await self.session.generate_reply(instructions="Inform the user the card data was stored successfully.")
        if not self.done():
            self.complete(CreditCardData(holder_name=holder_name, number=cc_number, expiration_date=expiration_date, security_code=security_code))

    @function_tool()
    async def decline_data_capture(self, reason: str) -> None:
        """
        Handles the case when the user explicitly declines to provide the card's data. 
        Call this tool also when `collect_data` fails.

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


SINGLE_FIELD_TOOLS_PROMPT = """
You are only a single step in a broader system, responsible solely for capturing a {data_name}.
Handle input as noisy voice transcription. Expect that users will say the data aloud with formats like:
{example_formats}
Call `{update_function}` at the first opportunity whenever you form a new hypothesis about the data (before asking any questions or providing any answers.) 
Don't invent the content of any of the fields, stick strictly to what the user said. 
Call `{confirm_function}` after the user confirmed the data is correct. 
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.
"""

class GetCreditCardHolderName(AgentTask[CardHolderName]):
    """
    Task that collects credit card holder name
    """
    def __init__(
        self,
        extra_instructions: str = "",
        chat_ctx: NotGivenOr[llm.ChatContext] = NOT_GIVEN,
        turn_detection: NotGivenOr[TurnDetectionMode | None] = NOT_GIVEN,
        tools: NotGivenOr[list[llm.FunctionTool | llm.RawFunctionTool]] = 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,
    ) -> None:
        super().__init__(
            instructions=(SINGLE_FIELD_TOOLS_PROMPT.format(
                    data_name = "credit card's holder name",
                    example_formats = "\n".join(["- 'theo t h e o smith' (name followed by spelling)"]),
                    update_function = "update_name",
                    confirm_function = "confirm_name"
                ) + extra_instructions
            ),
            chat_ctx=chat_ctx,
            turn_detection=turn_detection,
            tools=tools,
            stt=stt,
            vad=vad,
            llm=llm,
            tts=tts,
            allow_interruptions=allow_interruptions,
        )
        self._name = ""
        self._name_update_speech_handle: SpeechHandle | None = None

    async def on_enter(self) -> None:
        await self.session.generate_reply(instructions="Ask the user to provide the name of the credit card holder, as spelled on the card.")

    @function_tool
    async def update_name(self, name: str, ctx: RunContext) -> str:
        """Update the name provided by the user.

        Args:
            name: The name provided by the user
        """
        self._name_update_speech_handle = ctx.speech_handle
        self._name = name.strip()

        return (
            f"The card holder name has been updated to {self._name}\n"
            f"Repeat the full name, first as written and then spelling it character by character.\n"
            f"Prompt the user for confirmation, do not call `confirm_name` directly."
        )

    @function_tool()
    async def confirm_name(self, ctx: RunContext) -> None:
        """Validates/confirms the name provided by the user."""
        await ctx.wait_for_playout()

        if ctx.speech_handle == self._name_update_speech_handle:
            raise ToolError("error: the user must confirm the name explicitly")

        if not self._name.strip():
            raise ToolError(
                "error: no name was provided, `update_name` must be called before"
            )

        if not self.done():
            self.complete(CardHolderName(full_name=self._name))

    @function_tool()
    async def decline_name_capture(self, reason: str) -> None:
        """Handles the case when the user explicitly declines to provide the card holder's name.

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


class GetCreditCardNumber(AgentTask[CardNumber]):
    """
    Task that collects the credit card number
    """
    def __init__(
        self,
        extra_instructions: str = "",
        chat_ctx: NotGivenOr[llm.ChatContext] = NOT_GIVEN,
        turn_detection: NotGivenOr[TurnDetectionMode | None] = NOT_GIVEN,
        tools: NotGivenOr[list[llm.FunctionTool | llm.RawFunctionTool]] = 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,
    ) -> None:
        super().__init__(
            instructions=(SINGLE_FIELD_TOOLS_PROMPT.format(
                    data_name = "credit card's number",
                    example_formats = "\n".join(["- '1 3 6 8 3 ...' (digit by digit)", "- '13 68 33 ...' (in pairs)", "- '1368 3003 ...' (in groups of four)", "- '1 3 6 8 six zeros 3 3 ...' (grouping similar digits)", "- other arbitrary combinations"]),
                    update_function = "update_number",
                    confirm_function = "confirm_number"
                ) + extra_instructions
            ),
            chat_ctx=chat_ctx,
            turn_detection=turn_detection,
            tools=tools,
            stt=stt,
            vad=vad,
            llm=llm,
            tts=tts,
            allow_interruptions=allow_interruptions,
        )
        self._number = ""
        self._number_update_speech_handle: SpeechHandle | None = None

    async def on_enter(self) -> None:
        await self.session.generate_reply(instructions="Ask the user to provide the number of the credit card, as shown on the card.")

    @function_tool
    async def update_number(self, number: str, ctx: RunContext) -> str:
        """Update the credit card number provided by the user.

        Args:
            number: The credit card number provided by the user
        """
        self._number_update_speech_handle = ctx.speech_handle
        number = number.replace(" ", "")
        if len(number) != 16:
            raise ToolError(f"Invalid credit card number provided: {number}")

        self._number = number
        return (
            f"The credit card number has been updated to {self._number}\n"
            f"Repeat the full number, digit by digit.\n"
            f"Prompt the user for confirmation, do not call `confirm_number` directly."
        )

    @function_tool()
    async def confirm_number(self, ctx: RunContext) -> None:
        """Validates/confirms the number provided by the user."""
        await ctx.wait_for_playout()

        if ctx.speech_handle == self._number_update_speech_handle:
            raise ToolError("error: the user must confirm the number explicitly")

        if not self._number.strip():
            raise ToolError("error: no number was provided, `update_number` must be called before")

        if not self.done():
            self.complete(CardNumber(number=self._number))

    @function_tool()
    async def decline_number_capture(self, reason: str) -> None:
        """Handles the case when the user explicitly declines to provide the card number.

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


class GetCreditCardExpirationDate(AgentTask[CardExpirationDate]):
    """
    Task that collects the credit card expiration date
    """
    def __init__(
        self,
        extra_instructions: str = "",
        chat_ctx: NotGivenOr[llm.ChatContext] = NOT_GIVEN,
        turn_detection: NotGivenOr[TurnDetectionMode | None] = NOT_GIVEN,
        tools: NotGivenOr[list[llm.FunctionTool | llm.RawFunctionTool]] = 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,
    ) -> None:
        super().__init__(
            instructions=(SINGLE_FIELD_TOOLS_PROMPT.format(
                    data_name = "credit card's expiration date",
                    example_formats = "\n".join([
                        "- 'november two thousand twenty six' ( which means 'month: 11, year: 2026', using the month's name)", 
                        "- 'november twenty six' ( which means 'month: 11, year: 2026', using the month's name and short year)", 
                        "- 'zero nine slash two six' (which means 'month: 9, year: 2026', reading the string)", 
                        "- 'zero nine two six' (which means 'month: 9, year: 2026', forgetting the slash)"
                    ]),
                    update_function = "update_date",
                    confirm_function = "confirm_date"
                ) + extra_instructions
            ),
            chat_ctx=chat_ctx,
            turn_detection=turn_detection,
            tools=tools,
            stt=stt,
            vad=vad,
            llm=llm,
            tts=tts,
            allow_interruptions=allow_interruptions,
        )
        self._date = None
        self._date_update_speech_handle: SpeechHandle | None = None

    async def on_enter(self) -> None:
        await self.session.generate_reply(instructions="Ask the user to provide the expiration date of the credit card, as shown on the card.")

    @function_tool
    async def update_expiration_date(self, month: int, year: int, ctx: RunContext) -> str:
        """Update the credit card expiration date provided by the user.

        Args:
            month: The credit card expiration month provided by the user
            year: The credit card expiration year provided by the user
        """
        self._date_update_speech_handle = ctx.speech_handle

        current_year = datetime.datetime.now().year
        current_month = datetime.datetime.now().month

        if year < 100: # the LLM returned only two digits, like "25" for "2025"
            year += 2000

        if month > 12:
            raise ToolError(f"Invalid credit card expiration date provided: {month}/{year}")
        
        if year < current_year or (year == current_year and month < current_month):
            raise ToolError(f"Invalid credit card expiration date provided: {month}/{year}")
        
        self._date = datetime.datetime(year=year, month=month, day=1)
        return (
            f"The credit card expiration date has been updated to {self._date}\n"
            f"Repeat the month (by name) and then the year (full length).\n"
            f"Prompt the user for confirmation, do not call `confirm_date` directly."
        )

    @function_tool()
    async def confirm_date(self, ctx: RunContext) -> None:
        """Validates/confirms the expiration date provided by the user."""
        await ctx.wait_for_playout()

        if ctx.speech_handle == self._date_update_speech_handle:
            raise ToolError("error: the user must confirm the expiration date explicitly")

        if not self._date:
            raise ToolError("error: no expiration date was provided, `update_date` must be called before")

        if not self.done():
            self.complete(CardExpirationDate(month=self._date.month, year=self._date.year))

    @function_tool()
    async def decline_date_capture(self, reason: str) -> None:
        """Handles the case when the user explicitly declines to provide the card expiration date.

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


class GetCreditCardSecurityCode(AgentTask[CardSecurityCode]):
    """
    Task that collects the credit card security code
    """
    def __init__(
        self,
        extra_instructions: str = "",
        chat_ctx: NotGivenOr[llm.ChatContext] = NOT_GIVEN,
        turn_detection: NotGivenOr[TurnDetectionMode | None] = NOT_GIVEN,
        tools: NotGivenOr[list[llm.FunctionTool | llm.RawFunctionTool]] = 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,
    ) -> None:
        super().__init__(
            instructions=(SINGLE_FIELD_TOOLS_PROMPT.format(
                    data_name = "credit card's security code",
                    example_formats = "\n".join([
                        "- '1 3 6' (digit by digit)", 
                        "- '136' (as a single number)"
                    ]),
                    update_function = "update_code",
                    confirm_function = "confirm_code"
                ) + extra_instructions
            ),
            chat_ctx=chat_ctx,
            turn_detection=turn_detection,
            tools=tools,
            stt=stt,
            vad=vad,
            llm=llm,
            tts=tts,
            allow_interruptions=allow_interruptions,
        )
        self._code = ""
        self._code_update_speech_handle: SpeechHandle | None = None

    async def on_enter(self) -> None:
        await self.session.generate_reply(instructions="Ask the user to provide the security code of the credit card, as shown on the card.")

    @function_tool
    async def update_code(self, code: str, ctx: RunContext) -> str:
        """Update the credit card security code provided by the user.

        Args:
            code: The credit card security code provided by the user
        """
        self._code_update_speech_handle = ctx.speech_handle
        code = code.strip()
        if len(code) != 3:
            raise ToolError(f"Invalid credit card security code provided: {code}")

        self._code = code
        return (
            f"The credit card security code has been updated to {self._code}\n"
            f"Repeat the full security code, digit by digit.\n"
            f"Prompt the user for confirmation, do not call `confirm_code` directly."
        )

    @function_tool()
    async def confirm_code(self, ctx: RunContext) -> None:
        """Validates/confirms the security code provided by the user."""
        await ctx.wait_for_playout()

        if ctx.speech_handle == self._code_update_speech_handle:
            raise ToolError("error: the user must confirm the security code explicitly")

        if not self._code.strip():
            raise ToolError("error: no security code was provided, `update_code` must be called before")

        if not self.done():
            self.complete(CardSecurityCode(code=self._code))

    @function_tool()
    async def decline_code_capture(self, reason: str) -> None:
        """Handles the case when the user explicitly declines to provide the card security code.

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


class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="""You are a helpful voice AI assistant. The user is interacting with you via voice, even if you perceive the conversation as text.
            You eagerly assist users with their questions by providing information from your extensive knowledge.
            Your responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols.
            You are curious, friendly, and have a sense of humor. Do not switch language during the call unless the user explicitly asks for it.
            Right after greeting the user, call the `collect_cc_data` tool.
            """,
        )

    async def on_enter(self) -> None:
        await self.session.generate_reply(instructions="Greet the user and introduce yourself. Ask them if they're ready to begin.")

    @function_tool
    async def collect_cc_data(self, context: RunContext):
        """
        Use this tool immediately after introducing yourself.
        """
        try:
            if (card_data := await GetCreditCardData(chat_ctx=self.chat_ctx)):
                # Do what you need with the data
                await self.session.generate_reply(instructions="Offer your assistance to the user.")
            else:
                await self.terminate_call()
        except ToolError as e:
            await self.terminate_call()


    async def terminate_call(self):
        await self.session.generate_reply(instructions="Inform the user that you are unable to proceed and will end the call.")
        job_ctx = get_job_context()
        await job_ctx.api.room.delete_room(api.DeleteRoomRequest(room=job_ctx.room.name))


def prewarm(proc: JobProcess):
    proc.userdata["vad"] = silero.VAD.load()


async def entrypoint(ctx: JobContext):
    ctx.log_context_fields = {
        "room": ctx.room.name,
    }

    session = AgentSession(
        llm=openai.realtime.RealtimeModel(voice="marin")
    )
    usage_collector = metrics.UsageCollector()

    @session.on("metrics_collected")
    def _on_metrics_collected(ev: MetricsCollectedEvent):
        metrics.log_metrics(ev.metrics)
        usage_collector.collect(ev.metrics)

    async def log_usage():
        summary = usage_collector.get_summary()
        logger.info(f"Usage: {summary}")

    ctx.add_shutdown_callback(log_usage)

    await session.start(
        agent=Assistant(),
        room=ctx.room,
        room_input_options=RoomInputOptions(
            noise_cancellation=noise_cancellation.BVC(),
        ),
    )
    await ctx.connect()


if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, prewarm_fnc=prewarm))