rituearth/Restaurant_VoiceAgent

1import os
2from dotenv import load_dotenv
3from livekit.agents import (
4    Agent,
5    AgentSession,
6    AutoSubscribe,
7    JobContext,
8    JobProcess,
9    WorkerOptions,
10    cli,
11    metrics,
12    RoomInputOptions,
13)
14from livekit.plugins import (
15    cartesia,
16    openai,
17    deepgram,
18    noise_cancellation,
19    silero,
20)
21from livekit.plugins.turn_detector.multilingual import MultilingualModel
22
23# Load environment variables
24load_dotenv()
25
26class Assistant(Agent):
27    def __init__(self) -> None:
28        # Get API keys from environment variables
29        openai_api_key = os.getenv("OPENAI_API_KEY")
30        deepgram_api_key = os.getenv("DEEPGRAM_API_KEY")
31        cartesia_api_key = os.getenv("CARTESIA_API_KEY")
32
33        if not all([openai_api_key, deepgram_api_key, cartesia_api_key]):
34            raise ValueError("Missing required API keys in environment variables")
35
36        super().__init__(
37            instructions="You are a friendly and efficient restaurant order-taking assistant. Your goal is to help customers place their orders accurately. "
38            "Greet the customer warmly, ask them what they would like to order, and you can suggest popular items if they ask. "
39            "Repeat the order back to the customer for confirmation before finalizing. "
40            "Keep your responses concise and clear.",
41            stt=deepgram.STT(api_key=deepgram_api_key),
42            llm=openai.LLM(model="o3-mini", api_key=openai_api_key),
43            tts=cartesia.TTS(api_key=cartesia_api_key),
44            turn_detection=MultilingualModel(),
45        )
46
47    async def on_enter(self):
48        self.session.generate_reply(
49            instructions="Hey, how can I help you today?", allow_interruptions=True
50        )
51
52def prewarm(proc: JobProcess):
53    proc.userdata["vad"] = silero.VAD.load()
54
55async def entrypoint(ctx: JobContext):
56    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
57    participant = await ctx.wait_for_participant()
58    
59    usage_collector = metrics.UsageCollector()
60    
61    def on_metrics_collected(agent_metrics: metrics.AgentMetrics):
62        metrics.log_metrics(agent_metrics)
63        usage_collector.collect(agent_metrics)
64    
65    session = AgentSession(
66        vad=ctx.proc.userdata["vad"],
67        min_endpointing_delay=0.5,
68        max_endpointing_delay=5.0,
69    )
70    
71    session.on("metrics_collected", on_metrics_collected)
72    
73    await session.start(
74        room=ctx.room,
75        agent=Assistant(),
76        room_input_options=RoomInputOptions(
77            noise_cancellation=noise_cancellation.BVC(),
78        ),
79    )
80
81if __name__ == "__main__":
82    cli.run_app(
83        WorkerOptions(
84            entrypoint_fnc=entrypoint,
85            prewarm_fnc=prewarm,
86        ),
87    )