import asyncio
import uuid
import contextlib
from dotenv import load_dotenv
from fi.simulate import AgentDefinition, Scenario, Persona, TestRunner, evaluate_report
from livekit import rtc
from livekit.api import AccessToken, VideoGrants
from livekit.agents import Agent, AgentSession, function_tool
from livekit.plugins import openai, silero
from livekit.agents.voice.room_io import RoomInputOptions, RoomOutputOptions
import logging
logging.basicConfig(level=logging.INFO)
class SupportAgent(Agent):
def __init__(self, *, room: rtc.Room, **kwargs):
super().__init__(**kwargs)
self._room = room
@function_tool()
async def end_call(self) -> None:
self.session.say("I'm glad I could help. Have a great day! Goodbye.")
await asyncio.sleep(0.2)
self.session.shutdown()
# Disconnect room if still connected
try:
if getattr(self._room, "isconnected", False):
if callable(self._room.isconnected):
if self._room.isconnected():
await self._room.disconnect()
elif self._room.isconnected:
await self._room.disconnect()
except Exception:
pass
async def run_support_agent(lk_url: str, lk_api_key: str, lk_api_secret: str, room_name: str):
token = (
AccessToken(lk_api_key, lk_api_secret)
.with_identity("support-agent")
.with_grants(VideoGrants(room_join=True, room=room_name))
.to_jwt()
)
room = rtc.Room()
await room.connect(lk_url, token)
agent = SupportAgent(
room=room,
stt=openai.STT(),
llm=openai.LLM(model="gpt-4o-mini", temperature=0.7),
tts=openai.TTS(voice="alloy"),
vad=silero.VAD.load(),
allow_interruptions=True,
min_endpointing_delay=0.4,
max_endpointing_delay=2.2,
instructions=(
"You are a helpful support agent. Be friendly and proactive. "
"Ask clarifying questions and provide step-by-step guidance. "
"Keep the conversation going for at least 6 turns unless the issue is resolved. "
"When the customer confirms their issue is resolved or they say they're done, "
"call the `end_call` tool to gracefully end the call."
),
)
session = AgentSession(
stt=agent.stt,
llm=agent.llm,
tts=agent.tts,
vad=None,
turn_detection="stt",
allow_interruptions=True,
discard_audio_if_uninterruptible=True,
min_interruption_duration=0.25,
min_endpointing_delay=0.35,
max_endpointing_delay=2.0,
preemptive_generation=True,
)
await session.start(
agent,
room=room,
room_input_options=RoomInputOptions(
delete_room_on_close=False,
# ensure the agent hears both simulator and other agents
participant_kinds=[rtc.ParticipantKind.PARTICIPANT_KIND_STANDARD,
rtc.ParticipantKind.PARTICIPANT_KIND_AGENT],
),
room_output_options=RoomOutputOptions(transcription_enabled=False),
)
# small delay so tracks publish before the greeting
await asyncio.sleep(0.6)
session.say("Hello! How can I help you today?")
# Wait until session closes
closed = asyncio.Event()
session.on("close", lambda ev: closed.set())
await closed.wait()
# Ensure disconnect
try:
if getattr(room, "isconnected", False):
if callable(room.isconnected):
if room.isconnected():
await room.disconnect()
elif room.isconnected:
await room.disconnect()
except Exception:
pass
