Интеграция с LiveKit Agents

1
import os
2
import asyncio
3
4
from agentinbox import Agentinbox, AsyncAgentinbox, Subscribe, MessageReceivedEvent
5
from agentinbox_toolkit.livekit import AgentinboxToolkit

1
class EmailAssistant(Agent):
2
    inbox_id: str
3
    ws_task: asyncio.Task | None = None
4
5
    def __init__(self) -> None:
6
        client = Agentinbox()
7
8
        # By setting the client_id the inbox is created only once.
9
        username = os.getenv("AGENTINBOX_USERNAME")
10
        inbox = client.inboxes.create(username=username, client_id=f"{username}-inbox")
11
12
        self.inbox_id = inbox.inbox_id
13
14
        super().__init__(
15
            instructions=f"""
16
            Ты полезный голосовой и почтовый AI-ассистент. Твоё имя — Agent Inbox. Ты получаешь письма на {self.inbox_id}. Можешь отправлять письма и отвечать.
17
            При вызове почтовых инструментов указывай "{self.inbox_id}" как inbox_id. В подписи писем указывай «Agent Inbox».
18
            Говори по-русски.
19
            ВАЖНО: {self.inbox_id} — твой ящик, а не ящик пользователя.
20
            """,
21
            # The Agent Inbox Toolkit has ready-to-go tools for LiveKit agents.
22
            tools=AgentinboxToolkit(client=client).get_tools(
23
                [
24
                    "list_threads",
25
                    "get_thread",
26
                    "get_attachment",
27
                    "send_message",
28
                    "reply_to_message",
29
                ]
30
            ),
31
        )
32
33
    async def _websocket_task(self):
34
        # Open a websocket connection to Agent Inbox.
35
        async with AsyncAgentinbox().websockets.connect() as socket:
36
            # Subscribe to events from the inbox.
37
            await socket.send_subscribe(Subscribe(inbox_ids=[self.inbox_id]))
38
39
            while True:
40
                data = await socket.recv()
41
42
                # If a message is received by the inbox, interrupt the current conversation and generate a reply.
43
                if isinstance(data, MessageReceived):
44
                    self.session.interrupt()
45
46
                    await self.session.generate_reply(
47
                        instructions=f"""Скажи «Я получил письмо», затем прочитай письмо.""",
48
                        user_input=data.message.model_dump_json(),
49
                    )
50
51
    # Open the websocket connection and generate a greeting when the agent enters the call.
52
    async def on_enter(self):
53
        self.ws_task = asyncio.create_task(self._websocket_task())
54
55
        await self.session.generate_reply(
56
            instructions=f"""По-русски поздоровайся с пользователем, представься как Agent Inbox, скажи, что можешь получать письма на {self.inbox_id}, и предложи помощь.""",
57
            allow_interruptions=False,
58
        )
59
60
    # Close the websocket connection when the agent exits the call.
61
    async def on_exit(self):
62
        if self.ws_task:
63
            self.ws_task.cancel()

1
await session.start(
2
    room=ctx.room,
3
    agent=EmailAssistant(),  # Replace Assistant with EmailAssistant.
4
    room_input_options=RoomInputOptions(
5
        noise_cancellation=noise_cancellation.BVC()
6
    ),
7
)