LiveKit voice agents

1
import asyncio
2
import logging
3
import os
4
from livekit import agents
5
from livekit.agents import AutoSubscribe, JobContext, WorkerOptions, cli
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 agent session with components
2
    session = agents.AgentSession(
3
        stt=openai.STT(),
4
        llm=openai.LLM(model="gpt-4o-mini"),
5
        tts=openai.TTS(),
6
        vad=silero.VAD.load(),
7
    )
8
    
9
    # Create Zep memory agent with enhanced configuration
10
    zep_agent = ZepUserAgent(
11
        zep_client=zep_client,
12
        user_id=user_id,
13
        thread_id=thread_id,
14
        user_message_name=participant_name,
15
        assistant_message_name="Assistant",
16
        instructions="You are a helpful voice assistant with persistent memory. "
17
                    "Remember details from previous conversations and reference them naturally."
18
    )
19
    
20
    # Start the session with the agent
21
    await session.start(agent=zep_agent, room=ctx.room)

1
    # Voice assistant will now have persistent memory
2
    logging.info("Voice assistant with Zep memory is running")
3
    
4
    # Keep the session running
5
    await session.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 agent session with components
2
    session = agents.AgentSession(
3
        stt=openai.STT(),
4
        llm=openai.LLM(model="gpt-4o-mini"),
5
        tts=openai.TTS(),
6
        vad=silero.VAD.load(),
7
    )
8
    
9
    # Create Zep graph agent
10
    zep_agent = ZepGraphAgent(
11
        zep_client=zep_client,
12
        graph_id=graph_id,
13
        facts_limit=15,  # Max facts in context
14
        entity_limit=8,   # Max entities in context
15
        search_filters={
16
            "node_labels": ["Person"]  # Filter to Person entities only
17
        },
18
        instructions="You are a knowledgeable voice assistant. Use the provided "
19
                    "context about entities and facts to give informed responses."
20
    )
21
    
22
    # Start the session with the graph agent
23
    await session.start(agent=zep_agent, room=ctx.room)