LiveKit Plugin

pip install deepslate-livekit

from livekit import agents
from livekit.agents import AgentServer, AgentSession, Agent, room_io

import deepslate.livekit.realtime
from deepslate.livekit.realtime import ElevenLabsTtsConfig

class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(instructions="You are a helpful voice AI assistant.")

server = AgentServer()

@server.rtc_session()
async def my_agent(ctx: agents.JobContext):
    session = AgentSession(
        llm=deepslate.livekit.realtime.RealtimeModel(
            tts_config=ElevenLabsTtsConfig.from_env()
        ),
    )

    await session.start(
        room=ctx.room,
        agent=Assistant(),
        room_options=room_io.RoomOptions(),
    )

    await session.generate_reply(
        instructions="Greet the user and offer your assistance."
    )

if __name__ == "__main__":
    agents.cli.run_app(server)

import asyncio
from livekit import agents
from livekit.agents import AgentSession, Agent
import deepslate.livekit.realtime
from deepslate.livekit.realtime import ElevenLabsTtsConfig

class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(instructions="You are a helpful voice AI assistant.")

@server.rtc_session()
async def my_agent(ctx: agents.JobContext):
    model = deepslate.livekit.realtime.RealtimeModel(
        tts_config=ElevenLabsTtsConfig.from_env()
    )
    session = AgentSession(llm=model)

    deepslate_session = model.session()
    deepslate_session.on("session_initialized", lambda _: asyncio.create_task(
        deepslate_session.speak_direct("Hello! How can I help you today?")
    ))

    await session.start(room=ctx.room, agent=Assistant())

from livekit.agents import function_tool

@function_tool()
async def lookup_weather(location: str) -> str:
    """Get the current weather for a location.

    Args:
        location: The city or location to get weather for
    """
    # Your implementation here
    return f"The weather in {location} is sunny and 22°C"

class WeatherAssistant(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="You are a helpful weather assistant.",
            tools=[lookup_weather],
        )

# Inside a session or agent callback
await session.llm.session().speak_direct(
    "Welcome back! How can I help you today?",
    include_in_history=True,  # Record as an assistant turn (default: True)
)

summary = await session.llm.session().query_conversation(
    prompt="Summarize the conversation so far in one sentence.",
)
print(summary)  # e.g., "The user asked about weather in Berlin."

from livekit.agents import EventTypes

# Listen for the export result
@session.llm.session().on("chat_history_exported")
def on_history(messages):
    for msg in messages:
        print(msg["role"], msg["content"])

# Request the export
await session.llm.session().export_chat_history(
    await_pending=False,  # Set True to wait for any in-flight operations first
)

await session.llm.update_options(
    system_prompt="You are now a concise assistant. Keep replies under two sentences.",
    temperature=0.8,
)