LiveKit integration

1
import logging
2
import os
3
4
from livekit import agents
5
from livekit.agents import AutoSubscribe
6
from livekit.plugins import openai, silero
7
from zep_cloud.client import AsyncZep
8
from zep_livekit import ZepUserAgent
9
10
11
async def entrypoint(ctx: agents.JobContext):
12
    zep_client = AsyncZep(api_key=os.environ.get("ZEP_API_KEY"))
13
14
    # Stable user identity from your auth system; thread scoped to the room
15
    user_id = ctx.job.metadata or "user-123"
16
    thread_id = f"thread_{ctx.room.name}"
17
18
    # Ensure the user exists, then create the room-scoped thread
19
    try:
20
        await zep_client.user.get(user_id=user_id)
21
    except Exception:
22
        await zep_client.user.add(user_id=user_id, first_name="Alice")
23
24
    await zep_client.thread.create(thread_id=thread_id, user_id=user_id)
25
26
    # Subscribe to audio only — a voice agent has no use for video tracks
27
    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
28
29
    # AgentSession owns the audio pipeline (STT, VAD, turn detection, TTS)
30
    session = agents.AgentSession(
31
        stt=openai.STT(),
32
        llm=openai.LLM(model="gpt-4o-mini"),
33
        tts=openai.TTS(),
34
        vad=silero.VAD.load(),
35
    )
36
37
    # Drop-in Agent replacement that adds Zep memory
38
    agent = ZepUserAgent(
39
        zep_client=zep_client,
40
        user_id=user_id,
41
        thread_id=thread_id,
42
        user_message_name="Alice",
43
        assistant_message_name="Assistant",
44
        instructions="You are a helpful voice assistant with long-term memory. "
45
        "Reference details from previous conversations naturally.",
46
    )
47
48
    await session.start(agent=agent, room=ctx.room)
49
    logging.info("Voice assistant with Zep memory is running")
50
51
52
if __name__ == "__main__":
53
    agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint))

1
import os
2
3
from livekit import agents
4
from livekit.agents import AutoSubscribe
5
from livekit.plugins import openai, silero
6
from pydantic import Field
7
from zep_cloud.client import AsyncZep
8
from zep_cloud.external_clients.ontology import EntityModel, EntityText
9
from zep_livekit import ZepGraphAgent
10
11
12
class Person(EntityModel):
13
    """A person entity for voice interactions."""
14
15
    role: EntityText = Field(description="person's role or profession", default=None)
16
    interests: EntityText = Field(description="topics the person is interested in", default=None)
17
18
19
class Topic(EntityModel):
20
    """A conversation topic or subject."""
21
22
    category: EntityText = Field(description="category of the topic", default=None)
23
    importance: EntityText = Field(description="importance to the user", default=None)
24
25
26
async def entrypoint(ctx: agents.JobContext):
27
    zep_client = AsyncZep(api_key=os.environ.get("ZEP_API_KEY"))
28
29
    # Optional: define a custom ontology for structured extraction
30
    await zep_client.graph.set_ontology(entities={"Person": Person, "Topic": Topic})
31
32
    # Room-scoped graph
33
    graph_id = f"graph_{ctx.room.name}"
34
    try:
35
        await zep_client.graph.get(graph_id)
36
    except Exception:
37
        await zep_client.graph.create(graph_id=graph_id, name="LiveKit Voice Knowledge Graph")
38
39
    # Subscribe to audio only — a voice agent has no use for video tracks
40
    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
41
42
    session = agents.AgentSession(
43
        stt=openai.STT(),
44
        llm=openai.LLM(model="gpt-4o-mini"),
45
        tts=openai.TTS(),
46
        vad=silero.VAD.load(),
47
    )
48
49
    agent = ZepGraphAgent(
50
        zep_client=zep_client,
51
        graph_id=graph_id,
52
        facts_limit=15,  # Max facts (edges) to retrieve
53
        entity_limit=8,  # Max entities (nodes) to retrieve
54
        episode_limit=2,  # Max episodes to retrieve
55
        search_filters={"node_labels": ["Person"]},  # Constrain to Person entities
56
        instructions="You are a knowledgeable voice assistant. Use the provided "
57
        "context about entities and facts to give informed responses.",
58
    )
59
60
    await session.start(agent=agent, room=ctx.room)
61
62
63
if __name__ == "__main__":
64
    agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint))