LiveKit Integration

pip install livekit-plugins-camb

pip install livekit-plugins-camb 'livekit-agents[silero]' python-dotenv

# Camb.ai API key
CAMB_API_KEY=your_camb_api_key

# LiveKit credentials
LIVEKIT_URL=wss://your-project.livekit.cloud
LIVEKIT_API_KEY=your_livekit_api_key
LIVEKIT_API_SECRET=your_livekit_api_secret

# LLM provider (e.g., OpenAI)
OPENAI_API_KEY=your_openai_api_key

# STT provider (e.g., Deepgram)
DEEPGRAM_API_KEY=your_deepgram_api_key

import logging

from dotenv import load_dotenv
from livekit.agents import (
    Agent,
    AgentServer,
    AgentSession,
    JobContext,
    JobProcess,
    cli,
    room_io,
)
from livekit.plugins import camb, silero

load_dotenv()

logger = logging.getLogger("voice-agent")


class MyAgent(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions=(
                "You are a helpful voice assistant. "
                "Keep your responses concise and conversational. "
                "Do not use emojis or special formatting."
            ),
        )

    async def on_enter(self):
        # Generate initial greeting
        self.session.generate_reply(allow_interruptions=False)


server = AgentServer()


def prewarm(proc: JobProcess):
    # Pre-load VAD model for faster startup
    proc.userdata["vad"] = silero.VAD.load()


server.setup_fnc = prewarm


@server.rtc_session()
async def entrypoint(ctx: JobContext):
    session = AgentSession(
        # Speech-to-text
        stt="deepgram/nova-3",
        # Language model
        llm="openai/gpt-4.1-mini",
        # Camb.ai TTS
        tts=camb.TTS(),
        # Voice activity detection
        vad=ctx.proc.userdata["vad"],
    )

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


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

python your_agent.py dev

python your_agent.py start

import asyncio
from livekit.plugins import camb

async def main():
    voices = await camb.list_voices()
    for voice in voices:
        print(f"ID: {voice['id']}, Name: {voice['name']}, Gender: {voice['gender']}")

asyncio.run(main())

tts=camb.TTS(voice_id=147320)

# English (US)
tts=camb.TTS(language="en-us")

# French
tts=camb.TTS(language="fr-fr")

# Spanish
tts=camb.TTS(language="es-es")

# German
tts=camb.TTS(language="de-de")

# Japanese
tts=camb.TTS(language="ja-jp")

# Change voice mid-conversation
session.tts.update_options(voice_id=1234)

# Switch model
session.tts.update_options(model="mars-pro")

# Switch language
session.tts.update_options(language="es-es")

from livekit.agents.llm import function_tool
from livekit.agents import RunContext

class MyAgent(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="You are a helpful assistant that can look up weather.",
        )

    @function_tool
    async def lookup_weather(
        self,
        context: RunContext,
        location: str,
        latitude: str,
        longitude: str
    ):
        """Called when the user asks for weather related information.
        Ensure the user's location (city or region) is provided.
        When given a location, please estimate the latitude and longitude of the location and
        do not ask the user for them.

        Args:
            location: The location they are asking for
            latitude: The latitude of the location, do not ask user for it
            longitude: The longitude of the location, do not ask user for it
        """
        # Your weather lookup logic here
        return "sunny with a temperature of 70 degrees."

import aiohttp

async with aiohttp.ClientSession() as session:
    tts = camb.TTS(http_session=session)
    # Use TTS...

from livekit.agents import metrics, MetricsCollectedEvent

usage_collector = metrics.UsageCollector()

@session.on("metrics_collected")
def on_metrics(ev: MetricsCollectedEvent):
    metrics.log_metrics(ev.metrics)
    usage_collector.collect(ev.metrics)

# Get usage summary after session
summary = usage_collector.get_summary()
logger.info(f"Session usage: {summary}")

pip install livekit-plugins-camb 'livekit-agents[silero]' python-dotenv

import logging

from dotenv import load_dotenv

from livekit.agents import (
    Agent,
    AgentServer,
    AgentSession,
    JobContext,
    JobProcess,
    MetricsCollectedEvent,
    RunContext,
    cli,
    metrics,
    room_io,
)
from livekit.agents.llm import function_tool
from livekit.plugins import camb, silero
from livekit.plugins.turn_detector.multilingual import MultilingualModel

logger = logging.getLogger("basic-agent")

load_dotenv()


class MyAgent(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="Your name is Kelly. You would interact with users via voice."
            "with that in mind keep your responses concise and to the point."
            "do not use emojis, asterisks, markdown, or other special characters in your responses."
            "You are curious and friendly, and have a sense of humor."
            "you will speak english to the user",
        )

    async def on_enter(self):
        # when the agent is added to the session, it'll generate a reply
        # according to its instructions
        # Keep it uninterruptible so the client has time to calibrate AEC
        self.session.generate_reply(allow_interruptions=False)

    @function_tool
    async def lookup_weather(
        self, context: RunContext, location: str, latitude: str, longitude: str
    ):
        """Called when the user asks for weather related information.
        Ensure the user's location (city or region) is provided.
        When given a location, please estimate the latitude and longitude of the location and
        do not ask the user for them.

        Args:
            location: The location they are asking for
            latitude: The latitude of the location, do not ask user for it
            longitude: The longitude of the location, do not ask user for it
        """
        logger.info(f"Looking up weather for {location}")
        return "sunny with a temperature of 70 degrees."


server = AgentServer()


def prewarm(proc: JobProcess):
    proc.userdata["vad"] = silero.VAD.load()


server.setup_fnc = prewarm


@server.rtc_session()
async def entrypoint(ctx: JobContext):
    ctx.log_context_fields = {"room": ctx.room.name}

    session = AgentSession(
        stt="deepgram/nova-3",
        llm="openai/gpt-4.1-mini",
        tts=camb.TTS(),
        turn_detection=MultilingualModel(),
        vad=ctx.proc.userdata["vad"],
        preemptive_generation=True,
        resume_false_interruption=True,
        false_interruption_timeout=1.0,
    )

    usage_collector = metrics.UsageCollector()

    @session.on("metrics_collected")
    def _on_metrics_collected(ev: MetricsCollectedEvent):
        metrics.log_metrics(ev.metrics)
        usage_collector.collect(ev.metrics)

    async def log_usage():
        summary = usage_collector.get_summary()
        logger.info(f"Usage: {summary}")

    ctx.add_shutdown_callback(log_usage)

    await session.start(
        agent=MyAgent(),
        room=ctx.room,
        room_options=room_io.RoomOptions(
            audio_input=room_io.AudioInputOptions(),
        ),
    )


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

python your_agent.py dev

python your_agent.py start

import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("livekit.plugins.camb")
logger.setLevel(logging.DEBUG)