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.
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_livekitfrom livekit import agentsfrom 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 models: if you choose to build the AgentSession with a realtime model (for example, lk_openai.realtime.RealtimeModel(...)), the tracing setup is unchanged.
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 telemetryfrom opentelemetry.sdk.trace import TracerProviderfrom langsmith.integrations.livekit import LiveKitLangSmithSpanProcessorprovider = TracerProvider() # your own providerprovider.add_span_processor(LiveKitLangSmithSpanProcessor())telemetry.set_tracer_provider(provider)
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_idconfigure_livekit()@server.rtc_session()async def my_agent(ctx: agents.JobContext): set_thread_id(ctx.job.id) # set to desired thread id for the session ...
The integration attaches the call recording to the conversation root span. How you capture that recording differs between local development and production.
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 = Noneconfigure_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:
Call processor.expect_recording(thread_id) when you start egress.
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 osfrom livekit import agents, apifrom livekit.agents import Agent, AgentServer, AgentSessionfrom langsmith.integrations.livekit import configure_livekit, set_thread_idRECORDING_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.
Engine