multi-human call transfer in a livekit application

human_transfer.py

import typing
from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import ChatPromptTemplate
from loguru import logger
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from livekit.agents import JobContext

from livekit.agents import function_tool
from livekit import api
from livekit.protocol.sip import TransferSIPParticipantRequest
from livekit import rtc


class CallTransferConfig(BaseModel):
    description: str
    to_number: str
    type: typing.Literal["human", "agent"] = "human"


class ConciergeOutputSchema(CallTransferConfig):
    score: int = Field(
        default=0,
        description="how sure are you that the agent is the right one",
        gt=-1,
        le=10,
    )


async def determine_agent(
    agents: list[CallTransferConfig], reason: str
) -> ConciergeOutputSchema | None:
    """Determines the type of agent based on the reason"""

    sys_tmpl = """
You are a concierge agent responsible for selecting the appropriate agent based on a provided reason and a list of agents with their corresponding trigger conditions. 
Analyze the reason below and determine which agent should be used for the conversation.
You will be penalized if the agent is not the right one or doesnt strictly follow the transfer conditions/descriptions - if no agent provided, return the `score` as 0

{format_instructions}

[Reason]:
{reason}

[Agents]
{agents}
"""

    logger.debug(f"Determining agent to transfer to: {reason}")

    model = ChatOpenAI(model="gpt-4o")
    parser = PydanticOutputParser(pydantic_object=ConciergeOutputSchema)
    prompt = ChatPromptTemplate(
        messages=[("system", sys_tmpl)],
        partial_variables={"format_instructions": parser.get_format_instructions()},
    )

    chain = prompt | model | parser

    response = await chain.ainvoke(
        {
            "agents": "\n".join([f"{i + 1}. {agent}" for i, agent in enumerate(agents)]),
            "reason": reason,
        }
    )
    response = typing.cast(ConciergeOutputSchema, response)

    if response and (response.to_number is None) or (response.score < 7):
        logger.warning(f"agent could not be determined: {response}")
        return None

    return ConciergeOutputSchema(**response.model_dump())


class SystemTools:
    def __init__(
        self,
        participant: rtc.RemoteParticipant,
        ctx: JobContext,
    ):
        self.participant = participant
        self.ctx = ctx
        self.room = self.ctx.room
        self.api = self.ctx.api


    @function_tool
    async def call_transfer(
        self,
        reason: typing.Annotated[
            str,
            "The reason for the call transfer. eg. user wants to speak with to the billings department",
        ],
    ) -> str:
        """Use this tool when you want to transfer the call to a human."""
        try:
            # TODO: This should come from your dynamic agent factory
            # Typically this would be a list of agents from your db or schema
            agents = [
                CallTransferConfig(
                    description="user wants to speak to the billings department",
                    to_number="+1234567890",
                ),
                CallTransferConfig(
                    description="user wants to speak to the sales department",
                    to_number="+9876543210",
                ),
                CallTransferConfig(
                    description="user has technical issues and needs technical support",
                    to_number="+1122334455",
                ),
                CallTransferConfig(
                    description="user is asking about a recent order or shipment",
                    to_number="+2233445566",
                ),
                CallTransferConfig(
                    description="user needs assistance with account login or password reset",
                    to_number="+3344556677",
                ),
                CallTransferConfig(
                    description="user wants to file a complaint or speak with customer service",
                    to_number="+4455667788",
                ),
            ]
            human_agent = await determine_agent(
                agents=agents,
                reason=reason,
            )

            if human_agent is None or human_agent.score == 0:
                return "No human agent available to transfer the call to."

            async with api.LiveKitAPI() as livekit_api:
                transfer_to = f"tel:{human_agent.to_number}"

                transfer_request = TransferSIPParticipantRequest(
                    participant_identity=self.participant.identity,
                    room_name=self.room.name,
                    transfer_to=transfer_to,
                    play_dialtone=True,
                )
                logger.debug(f"Transfer request: {transfer_request}")

                await livekit_api.sip.transfer_sip_participant(transfer_request)
                logger.info(
                    f"Successfully transferred participant {self.participant.identity} to {transfer_to}"
                )
                return "Call successfully transferred"
        except Exception as e:
            logger.error(f"Error while transferring call: {e}")
            return "There's a problem with the call transfer"

    @function_tool
    async def hangup(self):
        """Ends the call."""
        try:
            await self.api.room.remove_participant(
                api.RoomParticipantIdentity(
                    room=self.room.name, identity=self.participant.identity
                )
            )
            await self.api.room.delete_room(api.DeleteRoomRequest(room=self.room.name))
            self.ctx.shutdown()

        except Exception as e:
            logger.info(f"Error while ending call: {e}")
            return f"Error while ending call: {e}"

pyproject.toml

[project]
name = "demos"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "langchain>=0.3.24",
    "langchain-openai>=0.3.14",
    "livekit-agents[cartesia,deepgram,openai,silero,turn-detector]~=1.0",
    "loguru>=0.7.3",
]