```
## Set up tracing
Call `configure_livekit` once before creating your `AgentServer`.
```python theme={null}
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."),
record={"audio": True},
)
```
`configure_livekit()` uses LiveKit session recordings by default. The `record={"audio": True}` option tells LiveKit to create the recording.
This setup works for both STT/LLM/TTS cascades and speech-to-speech (realtime) models. Realtime models need one extra call to capture the user's transcript. For more information, see [Use a realtime model](#use-a-realtime-model).
### Use your own tracer provider
If your application already manages an OpenTelemetry `TracerProvider`, add the LangSmith processor to that provider and register it with LiveKit:
```python theme={null}
from langsmith.integrations.livekit import LiveKitLangSmithSpanProcessor
from livekit.agents import telemetry
from opentelemetry.sdk.trace import TracerProvider
provider = TracerProvider() # your own provider
processor = LiveKitLangSmithSpanProcessor()
provider.add_span_processor(processor)
telemetry.set_tracer_provider(provider)
```
## Group a conversation into a thread
To group a conversation's runs into a LangSmith [thread](/langsmith/threads), call `set_thread_id` inside the session handler. Use a unique ID for each active session:
```python theme={null}
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)
...
```
## Use a realtime model
For a speech-to-speech (realtime) model, call `instrument_session` after creating the `AgentSession` to capture the user's transcript. Pass the same thread ID to `set_thread_id` and `instrument_session`:
```python theme={null}
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."),
record={"audio": True},
)
```
Only call `instrument_session` for realtime models. STT/LLM/TTS cascades already capture the user's transcript, so calling it there records each user turn twice.
## Record the conversation audio
By default, the integration uses LiveKit's session recording. Use Egress mode when you record to external storage instead.
### Record with LiveKit's session recording
Turn on LiveKit's session recording.
```python theme={null}
from langsmith.integrations.livekit import configure_livekit
from livekit import agents
from livekit.agents import Agent, AgentServer, AgentSession
configure_livekit()
server = AgentServer()
@server.rtc_session()
async def my_agent(ctx: agents.JobContext):
session = AgentSession(...)
await session.start(
room=ctx.room,
agent=Agent(instructions="You are a helpful assistant."),
record={"audio": True},
)
```
By default, the LiveKit integration will capture the recording from LiveKit if present.
In console mode, also pass `--record` on the command line (`python agent.py console --record`). Without it LiveKit creates the recorder but never starts it, so there is no file to attach. The recording reflects what was played to the client, so a barge-in shows up truncated.
### Record with Egress
Use [LiveKit Egress](https://docs.livekit.io/home/egress/overview/) when you want the recording in your own object storage or need video. Egress recording delivery requires a thread ID. Configure the integration for Egress, then call `complete_recording` after the Egress file is available:
```python theme={null}
import asyncio
import os
import time
from langsmith.integrations.livekit import configure_livekit, set_thread_id
from livekit import agents, api
from livekit.agents import Agent, AgentServer, AgentSession
RECORDING_BUCKET = os.environ["RECORDING_BUCKET"]
processor = configure_livekit(
recording_mode="egress",
recording_timeout_seconds=180,
)
server = AgentServer()
async def wait_for_egress(
lkapi: api.LiveKitAPI,
egress_id: str,
timeout_seconds: float = 120,
) -> api.EgressInfo:
deadline = time.monotonic() + timeout_seconds
failed_statuses = {
api.EgressStatus.EGRESS_FAILED,
api.EgressStatus.EGRESS_ABORTED,
api.EgressStatus.EGRESS_LIMIT_REACHED,
}
while time.monotonic() < deadline:
response = await lkapi.egress.list_egress(
api.ListEgressRequest(egress_id=egress_id)
)
if response.items:
info = response.items[0]
if info.status == api.EgressStatus.EGRESS_COMPLETE:
return info
if info.status in failed_statuses:
raise RuntimeError(f"Egress failed with status {info.status}")
await asyncio.sleep(1)
raise TimeoutError(f"Egress {egress_id} did not complete in time")
@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) # routes the Egress recording to this trace
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"],
),
)
],
)
)
async def attach_recording():
try:
info = 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,
data=audio,
# EgressInfo.started_at is a Unix timestamp in nanoseconds. This is
# when the egress worker began recording, which is what aligns the
# audio start with the trace.
started_at=info.started_at / 1e9,
)
except Exception:
processor.complete_recording(thread_id, data=None)
ctx.add_shutdown_callback(attach_recording)
session = AgentSession(...)
await session.start(room=ctx.room, agent=Agent(instructions="..."))
```
`download_from_storage` represents your storage client's download operation. The default attachment name and MIME type are `recording.ogg` and `audio/ogg`. Set `name` or `mime_type` in `complete_recording` if your Egress output uses another format.
Always call `complete_recording`, including on failure with `data=None`. Otherwise, the integration waits for `recording_timeout_seconds` (30 seconds by default) before exporting the trace without audio. Using `complete_recording` to capture Egress recordings requires setting a `thread_id`.
## 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).