Trace LiveKit applications

This integration is in beta, so its API may change.

Trace your LiveKit Agents voice agents to LangSmith with the LangSmith LiveKit integration. For high-level conventions, see Voice tracing fundamentals.

The LiveKit integration requires langsmith[livekit]>=0.9.7.

The integration hooks into the spans LiveKit already emits and maps them onto LangSmith’s tracing format, so each conversation becomes a single LangSmith trace: a span per pipeline event, plus LiveKit’s latency and token metrics.

Install

Install the integration along with the LiveKit plugins your agent uses:

pip

pip install "langsmith[livekit]" "livekit-agents[openai,silero,turn-detector]"

uv add "langsmith[livekit]" "livekit-agents[openai,silero,turn-detector]"

Set environment variables

The integration reads your LangSmith credentials from the environment and exports to LangSmith for you via OpenTelemetry:

.env

LANGSMITH_API_KEY=<your-langsmith-api-key>
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=<your-desired-langsmith-project>
LIVEKIT_URL=<your-livekit-url>
LIVEKIT_API_KEY=<your-livekit-api-key>
LIVEKIT_API_SECRET=<your-livekit-api-secret>
OPENAI_API_KEY=<your-openai-api-key>

Set up tracing

Import configure_livekit and call it once before creating your AgentServer. It builds the tracer provider, registers the LangSmith span processor, and wires it into LiveKit:

from langsmith.integrations.livekit import configure_livekit
from livekit import agents
from livekit.agents import Agent, AgentServer, AgentSession

# Enable tracing before creating agents.
configure_livekit()

server = AgentServer()

@server.rtc_session()
async def my_agent(ctx: agents.JobContext):
    session = AgentSession(
        stt="openai/gpt-4o-mini-transcribe",
        llm="openai/gpt-4o-mini",
        tts="openai/tts-1:alloy",
    )
    await session.start(room=ctx.room, agent=Agent(instructions="You are a helpful assistant."))

This works for both the STT/LLM/TTS cascade and speech-to-speech (realtime) models. Realtime models (for example, lk_openai.realtime.RealtimeModel(...)) need one extra call to capture the user’s transcript. Refer to When using LiveKit with a realtime model.

Use your own tracer provider

configure_livekit() builds a TracerProvider, registers the LangSmith span processor, and wires it into LiveKit. To use a TracerProvider you already manage, construct the processor yourself, add it to your provider, and register that provider with LiveKit’s tracer hook. LiveKit only emits spans through the provider its tracer is bound to:

from livekit.agents import telemetry
from opentelemetry.sdk.trace import TracerProvider

from langsmith.integrations.livekit import LiveKitLangSmithSpanProcessor

provider = TracerProvider()  # your own provider
provider.add_span_processor(LiveKitLangSmithSpanProcessor())
telemetry.set_tracer_provider(provider)

Group a conversation into a thread

To group a conversation’s runs into a LangSmith thread, for thread-level views and token and cost aggregation, call set_thread_id once per conversation, inside its @server.rtc_session() handler before its spans are emitted:

from langsmith.integrations.livekit import configure_livekit, set_thread_id

configure_livekit()

@server.rtc_session()
async def my_agent(ctx: agents.JobContext):
    thread_id = ctx.job.id  # or any id that identifies the conversation
    set_thread_id(thread_id)
    ...

When using LiveKit with a realtime model

With a speech-to-speech (realtime) model there is no separate speech-to-text step, so LiveKit transcribes the user’s audio asynchronously and delivers the transcript through the session’s user_input_transcribed event rather than on the OTel traces it emits.

instrument_session requires langsmith[livekit]>=0.10.4.

Call instrument_session once, right after creating the AgentSession, so the SDK subscribes to that event for you and pairs each transcript with its turn. It correlates by thread id, so set that first with set_thread_id, and pass the same id:

from langsmith.integrations.livekit import configure_livekit, set_thread_id
from livekit.plugins import openai as lk_openai

processor = configure_livekit()

@server.rtc_session()
async def my_agent(ctx: agents.JobContext):
    thread_id = ctx.job.id  # or any id that identifies the conversation
    set_thread_id(thread_id)

    session = AgentSession(llm=lk_openai.realtime.RealtimeModel(voice="marin"))
    processor.instrument_session(session, thread_id)  # capture the user transcript

    await session.start(room=ctx.room, agent=Agent(instructions="You are a helpful assistant."))

Only call instrument_session for realtime models. In the STT/LLM/TTS cascade the transcript is already captured (from the speech-to-text step), so calling it there would record the user’s turns a second time.

Record the conversation audio

The integration attaches the call recording to the conversation root span. How you capture that recording differs between local development and production.

Development: embed a local file

In console and local development, enable LiveKit’s session recording and point audio_path_provider at the audio.ogg LiveKit writes under ctx.session_directory. The integration reads that file and embeds the bytes in the trace.

from pathlib import Path

_audio_path: Path | None = None
configure_livekit(audio_path_provider=lambda: _audio_path)

@server.rtc_session()
async def my_agent(ctx: agents.JobContext):
    global _audio_path
    _audio_path = ctx.session_directory / "audio.ogg"
    await session.start(
        room=ctx.room,
        agent=Agent(instructions="You are a helpful assistant."),
        record={"audio": True},
    )

In console mode, also pass --record on the command line. The recording reflects what was played to the client, so a barge-in shows up truncated.

Do not use audio_path_provider in production. In a deployed worker, ctx.session_directory is an ephemeral temporary directory that LiveKit deletes when the session ends, so there is no durable file to embed.

Production: record with Egress and attach the file

In production, record the room with LiveKit Egress into your own object storage, then attach the finished recording to the trace as a real audio attachment. Egress finishes uploading after the call ends, so the integration holds the conversation’s root span open until you supply the bytes:

1. Call processor.expect_recording(thread_id) when you start egress.
2. After the call, wait for egress to complete, download the file from your storage, and call processor.complete_recording(thread_id, audio_bytes). The integration embeds the bytes and exports the trace.

You must use the same thread_id for expect_recording and complete_recording, so the recording is matched to the right conversation.

import os

from livekit import agents, api
from livekit.agents import Agent, AgentServer, AgentSession
from langsmith.integrations.livekit import configure_livekit, set_thread_id

RECORDING_BUCKET = os.environ["RECORDING_BUCKET"]

processor = configure_livekit()
server = AgentServer()

@server.rtc_session()
async def my_agent(ctx: agents.JobContext):
    thread_id = ctx.job.id  # unique per session; ctx.room.name is "console" in console mode
    set_thread_id(thread_id)  # groups this conversation's spans into a thread
    key = f"recordings/{thread_id}.ogg"

    # Start an audio-only room-composite egress to your storage.
    lkapi = api.LiveKitAPI()  # reads LIVEKIT_URL / LIVEKIT_API_KEY / LIVEKIT_API_SECRET
    egress = await lkapi.egress.start_room_composite_egress(
        api.RoomCompositeEgressRequest(
            room_name=ctx.room.name,
            audio_only=True,
            file_outputs=[
                api.EncodedFileOutput(
                    file_type=api.EncodedFileType.OGG,
                    filepath=key,
                    s3=api.S3Upload(
                        bucket=RECORDING_BUCKET,
                        region=os.environ["AWS_REGION"],
                        access_key=os.environ["AWS_ACCESS_KEY_ID"],
                        secret=os.environ["AWS_SECRET_ACCESS_KEY"],
                    ),
                )
            ],
        )
    )
    # Hold the trace open until the recording is ready.
    processor.expect_recording(thread_id)

    async def attach_recording():
        try:
            await wait_for_egress(lkapi, egress.egress_id)  # poll until EGRESS_COMPLETE
            audio = download_from_storage(RECORDING_BUCKET, key)  # your storage client
            processor.complete_recording(thread_id, audio, name="call.ogg")
        except Exception:
            processor.complete_recording(thread_id, None)  # release without audio

    ctx.add_shutdown_callback(attach_recording)

    session = AgentSession(...)
    await session.start(room=ctx.room, agent=Agent(instructions="..."))

wait_for_egress polls list_egress until the status is EGRESS_COMPLETE (or subscribe to the egress_ended webhook), and download_from_storage reads the object with your cloud provider’s client. LiveKit Egress also writes to Google Cloud Storage and Azure: swap s3= for gcp=api.GCPUpload(...) or azure=api.AzureBlobUpload(...).

Always call complete_recording because the trace’s root span is held until it runs, including on failure with data=None. If the worker stops first, the integration will flush the trace without audio.

Next steps

Voice fundamentals

Core conventions for tracing voice agents.

Upload files with traces

Attach the conversation audio recording to your trace.

---

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.