LiveKit voice agents

1
import asyncio
2
import logging
3
import os
4
from livekit.agents import AutoSubscribe, JobContext, llm, WorkerOptions, cli
5
from livekit.agents.voice_assistant import VoiceAssistant
6
from livekit.plugins import openai, silero
7
from zep_cloud.client import AsyncZep
8
from zep_livekit import ZepUserAgent

1
async def entrypoint(ctx: JobContext):
2
    # Initialize Zep client
3
    zep_client = AsyncZep(api_key=os.environ.get("ZEP_API_KEY"))
4
    
5
    # Create unique user and thread IDs
6
    participant_name = ctx.room.remote_participants[0].name or "User"
7
    user_id = f"livekit_{participant_name}_{ctx.room.name}"
8
    thread_id = f"thread_{ctx.room.name}"
9
    
10
    # Create user in Zep (if not exists)
11
    try:
12
        await zep_client.user.add(
13
            user_id=user_id,
14
            first_name=participant_name
15
        )
16
    except Exception as e:
17
        logging.info(f"User might already exist: {e}")
18
    
19
    # Create thread for conversation memory
20
    try:
21
        await zep_client.thread.create(thread_id=thread_id, user_id=user_id)
22
    except Exception as e:
23
        logging.info(f"Thread might already exist: {e}")

1
    # Create Zep memory agent with enhanced configuration
2
    zep_agent = ZepUserAgent(
3
        zep_client=zep_client,
4
        user_id=user_id,
5
        thread_id=thread_id,
6
        context_mode="basic",  # or "summary"
7
        user_message_name=participant_name,
8
        assistant_message_name="Assistant",
9
        instructions="You are a helpful voice assistant with persistent memory. "
10
                    "Remember details from previous conversations and reference them naturally."
11
    )
12
    
13
    # Create voice assistant with persistent memory
14
    assistant = VoiceAssistant(
15
        vad=silero.VAD.load(),
16
        stt=openai.STT(),
17
        llm=openai.LLM(),
18
        tts=openai.TTS(),
19
        chat_ctx=[
20
            llm.ChatMessage(
21
                role="system",
22
                content="You are a helpful voice assistant with persistent memory. "
23
                       "Remember details from previous conversations and reference them naturally."
24
            )
25
        ],
26
    )
27
    
28
    # Connect Zep memory to voice assistant
29
    assistant.start(ctx.room)
30
    await zep_agent.start(assistant)

1
    # Voice assistant will now have persistent memory
2
    logging.info("Voice assistant with Zep memory is running")
3
    
4
    # Keep the agent running
5
    await assistant.aclose()

1
from zep_cloud.external_clients.ontology import EntityModel, EntityText
2
from pydantic import Field
3
4
class Person(EntityModel):
5
    """A person entity for voice interactions."""
6
    role: EntityText = Field(
7
        description="person's role or profession",
8
        default=None
9
    )
10
    interests: EntityText = Field(
11
        description="topics the person is interested in",
12
        default=None
13
    )
14
15
class Topic(EntityModel):
16
    """A conversation topic or subject."""
17
    category: EntityText = Field(
18
        description="category of the topic",
19
        default=None
20
    )
21
    importance: EntityText = Field(
22
        description="importance level of this topic to the user",
23
        default=None
24
    )

1
from zep_livekit import ZepGraphAgent
2
3
async def setup_graph_agent(ctx: JobContext):
4
    zep_client = AsyncZep(api_key=os.environ.get("ZEP_API_KEY"))
5
    
6
    # Set ontology for structured knowledge
7
    await zep_client.graph.set_ontology(
8
        entities={
9
            "Person": Person,
10
            "Topic": Topic,
11
        }
12
    )
13
    
14
    # Create knowledge graph
15
    graph_id = f"livekit_graph_{ctx.room.name}"
16
    try:
17
        await zep_client.graph.create(
18
            graph_id=graph_id,
19
            name="LiveKit Voice Knowledge Graph"
20
        )
21
    except Exception as e:
22
        logging.info(f"Graph might already exist: {e}")

1
    # Create Zep graph agent
2
    zep_agent = ZepGraphAgent(
3
        zep_client=zep_client,
4
        graph_id=graph_id,
5
        facts_limit=15,  # Max facts in context
6
        entity_limit=8   # Max entities in context
7
    )
8
    
9
    # Create voice assistant with knowledge graph memory
10
    assistant = VoiceAssistant(
11
        vad=silero.VAD.load(),
12
        stt=openai.STT(),
13
        llm=openai.LLM(),
14
        tts=openai.TTS(),
15
        chat_ctx=[
16
            llm.ChatMessage(
17
                role="system",
18
                content="You are a knowledgeable voice assistant. Use the provided "
19
                       "context about entities and facts to give informed responses."
20
            )
21
        ],
22
    )
23
    
24
    assistant.start(ctx.room)
25
    await zep_agent.start(assistant)