With LiveKit

1
import asyncio
2
import logging
3
4
from dotenv import load_dotenv
5
6
from livekit.agents import (
7
    Agent,
8
    AgentServer,
9
    AgentSession,
10
    JobContext,
11
    cli,
12
    function_tool,
13
)
14
from livekit.plugins.phonic.realtime import RealtimeModel
15
16
logger = logging.getLogger("phonic-agent")
17
18
load_dotenv()
19
20
21
class MyAgent(Agent):
22
    def __init__(self) -> None:
23
        super().__init__(
24
            instructions="You are a helpful voice AI assistant named Sabrina.",
25
            llm=RealtimeModel(
26
                voice="sabrina",
27
                audio_speed=1.2,
28
            ),
29
        )
30
31
    @function_tool(
32
        description="Toggle a light on or off. Available lights are A05, A06, A07, and A08."
33
    )
34
    async def toggle_light(self, light_id: str, state: str) -> str:
35
        """Called when the user asks to toggle a light on or off.
36
37
        Args:
38
            light_id: The ID of the light to toggle
39
            state: Whether to turn the light on or off, e.g., 'on', 'off'
40
        """
41
        logger.info(f"Turning {state} light {light_id}")
42
        await asyncio.sleep(1.0)
43
        return f"Light {light_id} turned {state}"
44
45
46
server = AgentServer()
47
48
49
@server.rtc_session()
50
async def entrypoint(ctx: JobContext):
51
    session = AgentSession()
52
    await session.start(agent=MyAgent(), room=ctx.room)
53
    await session.generate_reply(
54
        instructions="Greet the user, asking about their day.",
55
    )
56
57
58
if __name__ == "__main__":
59
    cli.run_app(server)