```
## 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:
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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](#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:
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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](/langsmith/threads), 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:
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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:
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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.
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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](https://docs.livekit.io/home/egress/overview/) 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.
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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`](https://docs.livekit.io/home/egress/api/) 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](https://docs.livekit.io/home/egress/overview/): 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
Core conventions for tracing voice agents.
Attach the conversation audio recording to your trace.
***
[Connect these docs](/use-these-docs) to Claude, VSCode, and more via MCP for real-time answers.
[Edit this page on GitHub](https://github.com/langchain-ai/docs/edit/main/src/langsmith/trace-with-livekit.mdx) or [file an issue](https://github.com/langchain-ai/docs/issues/new/choose).