import ast import json import logging import os import uuid from database.Aiden import AidenDB from dotenv import load_dotenv from livekit import api, rtc from livekit.agents import ( Agent, AgentSession, JobContext, JobProcess, MetricsCollectedEvent, RoomInputOptions, WorkerOptions, cli, inference, metrics, RunContext, get_job_context, ) from livekit.plugins import noise_cancellation, silero from livekit.plugins.turn_detector.multilingual import MultilingualModel from livekit.plugins import openai from livekit.agents.llm import ToolError, function_tool from openai.types.beta.realtime.session import TurnDetection from database.connection import get_db_transients, get_db_meta, get_db_main, get_db_aiden from dataclasses import dataclass, field from typing import Optional from database.Transients import TransientsDB from prompts.default_prompt import DEFAULT_PROMPT logger = logging.getLogger("agent") load_dotenv(".env.local") @dataclass class ContextData: # Shared data that's used by the editor agent. # This structure is passed as a parameter to function calls. facility: Optional[int] = None dialed_number: Optional[str] = None # ensure the following variables/env vars are set SIP_TRUNK_ID = os.getenv("SIP_TRUNK_ID") # "ST_abcxyz" SUPERVISOR_PHONE_NUMBER = os.getenv("SUPERVISOR_PHONE_NUMBER") # "+12003004000" # SIP_OUTBOUND_CALLER_ID = os.getenv("SIP_OUTBOUND_CALLER_ID") # Twilio-verified caller ID class Assistant(Agent): def __init__(self) -> None: super().__init__( instructions=DEFAULT_PROMPT ) # def _normalize_destination(self, destination: str) -> str: # normalized = destination.strip().replace(" ", "") # if not normalized: # raise ToolError("A valid phone number or SIP URI is required.") # if not ( # normalized.startswith(("sip:", "+")) # or normalized.isdigit() # ): # raise ToolError( # "Destination must be an E.164 phone number or a SIP URI, for example +15551234567 or sip:user@example.com." # ) # return normalized # def _resolve_sip_participant( # self, # job_ctx: JobContext, # participant_identity: Optional[str] = None, # ) -> rtc.RemoteParticipant: # sip_callers = [ # remote # for remote in job_ctx.room.remote_participants.values() # if remote.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP # ] # if not sip_callers: # raise ToolError("No SIP caller is connected, so the transfer cannot be performed.") # if participant_identity: # participant = job_ctx.room.remote_participants.get(participant_identity) # if participant and participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP: # return participant # logger.warning( # "Requested participant identity '%s' was not found; defaulting to first SIP caller.", # participant_identity, # ) # return sip_callers[0] async def on_enter(self): # trigger an initial response as soon as the agent joins the room self.session.generate_reply() @function_tool async def facility_data( self, context: RunContext[ContextData], ): """ When the customer requests contact information about the storage, location, phone number and other relevant facility details, this tool is consulted. Args: facility: The ID of the facility """ db = TransientsDB() result = db.get_facility_data(context.userdata["facility_id"]) return result.model_dump_json(indent=2) @function_tool async def units_data( self, context: RunContext[ContextData], ): """ Tool used to check the units available at the location, as well as their characteristics such as price, sizes, discounts, etc. Args: facility: The ID of the facility """ db = TransientsDB() result = db.get_units_data(context.userdata["facility_id"]) return json.dumps(result) # @function_tool # async def transfer_call( # self, # context: RunContext[ContextData], # destination_number: str, # play_dialtone: bool = True, # participant_identity: Optional[str] = None, # ): # """Transfer the current SIP caller to another phone number or SIP URI. # Args: # destination_number: Target E.164 number (e.g. +15551234567) or SIP URI (sip:user@example.com). # play_dialtone: Whether LiveKit should play a dialtone while the transfer takes place. # participant_identity: Optional explicit participant identity to transfer; defaults to the first SIP caller. # """ # normalized_destination = self._normalize_destination(destination_number) # job_ctx = get_job_context() # participant = self._resolve_sip_participant(job_ctx, participant_identity) # try: # await job_ctx.transfer_sip_participant( # participant=participant, # transfer_to=normalized_destination, # play_dialtone=play_dialtone, # ) # except Exception as exc: # logger.exception("Failed to transfer SIP participant") # raise ToolError(f"Unable to transfer the call right now: {exc}") from exc # return { # "status": "success", # "transfer_to": normalized_destination, # "participant_identity": participant.identity, # } # @function_tool # async def warm_transfer_to_operator( # self, # context: RunContext[ContextData], # operator_number: Optional[str] = None, # caller_identity: Optional[str] = None, # operator_name: str = "Human Operator", # wait_for_answer: bool = True, # play_dialtone: bool = True, # caller_id_number: Optional[str] = None, # ): # """Initiate a warm transfer to a human operator following https://docs.livekit.io/sip/transfer-warm/. # Args: # operator_number: E.164 number or SIP URI for the human operator. Falls back to SUPERVISOR_PHONE_NUMBER. # caller_identity: Explicit SIP participant identity to hand off. Defaults to the first SIP participant in the room. # operator_name: Friendly name shown in LiveKit for the operator leg. # wait_for_answer: Whether to wait for the operator to pick up before returning (recommended for warm transfers). # play_dialtone: Whether LiveKit should play a dial tone while the outbound leg is ringing. # caller_id_number: Optional verified caller ID (E.164 or SIP URI). Falls back to SIP_OUTBOUND_CALLER_ID. # """ # destination = operator_number or SUPERVISOR_PHONE_NUMBER # if not destination: # raise ToolError( # "Provide 'operator_number' or set the SUPERVISOR_PHONE_NUMBER environment variable to use warm transfer." # ) # sip_trunk_id = SIP_TRUNK_ID # if not sip_trunk_id: # raise ToolError("SIP_TRUNK_ID must be configured to dial an operator.") # normalized_destination = self._normalize_destination(destination) # normalized_caller_id = None # configured_caller_id = caller_id_number or SIP_OUTBOUND_CALLER_ID # if configured_caller_id: # normalized_caller_id = self._normalize_destination(configured_caller_id) # else: # logger.warning( # "No outbound caller ID configured. Some providers (e.g. Twilio) " # "require SIP_OUTBOUND_CALLER_ID or caller_id_number to be set." # ) # job_ctx = get_job_context() # caller = self._resolve_sip_participant(job_ctx, caller_identity) # operator_identity = f"operator-{caller.identity}-{uuid.uuid4().hex[:6]}" # try: # operator_info = await job_ctx.api.sip.create_sip_participant( # api.CreateSIPParticipantRequest( # room_name=job_ctx.room.name, # sip_trunk_id=sip_trunk_id, # sip_call_to=normalized_destination, # sip_number=normalized_caller_id, # participant_identity=operator_identity, # participant_name=operator_name, # wait_until_answered=wait_for_answer, # play_dialtone=play_dialtone, # # caller_id_number= SIP_OUTBOUND_CALLER_ID, # ) # ) # except Exception as exc: # logger.exception("Failed to start warm transfer to operator") # raise ToolError(f"Unable to reach the operator right now: {exc}") from exc # status = "operator_connected" if wait_for_answer else "dialing_operator" # return { # "status": status, # "operator_identity": operator_identity, # "operator_number": normalized_destination, # "caller_identity": caller.identity, # "sip_call_id": getattr(operator_info, "sip_call_id", None), # "operator_room": getattr(operator_info, "room_name", job_ctx.room.name), # } def prewarm(proc: JobProcess): proc.userdata["vad"] = silero.VAD.load() async def entrypoint(ctx: JobContext): # Logging setup # Add any other context you want in all log entries here metadata = {} raw_metadata = ctx.job.metadata if raw_metadata: if isinstance(raw_metadata, dict): metadata = raw_metadata else: try: metadata = json.loads(raw_metadata) except json.JSONDecodeError as json_err: try: metadata = ast.literal_eval(raw_metadata) except (ValueError, SyntaxError) as ast_err: logger.warning( "Unable to parse job metadata. json_error=%s ast_error=%s", json_err, ast_err, ) aiden_db= AidenDB() dialed_number= metadata.get("dialed_number", None) facility_id = aiden_db.get_facility_id(dialed_number) ctx.log_context_fields = { "room": ctx.room.name, "facility_id": facility_id, "dialed_number": dialed_number, } # Set up a voice AI pipeline using OpenAI, Cartesia, AssemblyAI, and the LiveKit turn detector session = AgentSession( # Speech-to-text (STT) is your agent's ears, turning the user's speech into text that the LLM can understand # See all available models at https://docs.livekit.io/agents/models/stt/ stt=inference.STT(model="assemblyai/universal-streaming", language="en"), # A Large Language Model (LLM) is your agent's brain, processing user input and generating a response # See all available models at https://docs.livekit.io/agents/models/llm/ llm=inference.LLM(model="openai/gpt-4.1-mini"), # Text-to-speech (TTS) is your agent's voice, turning the LLM's text into speech that the user can hear # See all available models as well as voice selections at https://docs.livekit.io/agents/models/tts/ tts=inference.TTS( model="cartesia/sonic-3", voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc" ), # VAD and turn detection are used to determine when the user is speaking and when the agent should respond # See more at https://docs.livekit.io/agents/build/turns # llm= openai.realtime.RealtimeModel( # voice="marin", # model="gpt-realtime", # turn_detection=TurnDetection( # type="server_vad", # threshold=0.5, # prefix_padding_ms=300, # silence_duration_ms=500, # create_response=True, # interrupt_response=True, # )), turn_detection=MultilingualModel(), vad=ctx.proc.userdata["vad"], userdata={ "facility_id": facility_id, "dialed_number": dialed_number, }, # allow the LLM to generate a response while waiting for the end of turn # See more at https://docs.livekit.io/agents/build/audio/#preemptive-generation # preemptive_generation=True, ) # To use a realtime model instead of a voice pipeline, use the following session setup instead. # (Note: This is for the OpenAI Realtime API. For other providers, see https://docs.livekit.io/agents/models/realtime/)) # 1. Install livekit-agents[openai] # 2. Set OPENAI_API_KEY in .env.local # 3. Add `from livekit.plugins import openai` to the top of this file # 4. Use the following session setup instead of the version above # session = AgentSession( # llm=openai.realtime.RealtimeModel(voice="marin") # ) # Metrics collection, to measure pipeline performance # For more information, see https://docs.livekit.io/agents/build/metrics/ 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) # # Add a virtual avatar to the session, if desired # # For other providers, see https://docs.livekit.io/agents/models/avatar/ # avatar = hedra.AvatarSession( # avatar_id="...", # See https://docs.livekit.io/agents/models/avatar/plugins/hedra # ) # # Start the avatar and wait for it to join # await avatar.start(session, room=ctx.room) # Start the session, which initializes the voice pipeline and warms up the models await session.start( agent=Assistant(), room=ctx.room, room_input_options=RoomInputOptions( # For telephony applications, use `BVCTelephony` for best results noise_cancellation=noise_cancellation.BVCTelephony(), ), ) # Join the room and connect to the user await ctx.connect() if __name__ == "__main__": cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, prewarm_fnc=prewarm))