追踪 LiveKit 应用

NOTE

此集成处于 beta 阶段，因此其 API 可能会发生变化。

使用 LangSmith LiveKit 集成追踪 LiveKit Agents 语音 agent，包括其转录和音频录制。有关高层约定，请参阅语音追踪基础。

NOTE

此设置需要 langsmith[livekit]>=0.11.2 和 livekit-agents>=1.6。LiveKit Agents 1.7 及更高版本需要 langsmith[livekit]>=0.12.4。

每个对话都会显示为一条 LangSmith 追踪，包含其流水线事件、延迟和 token 指标。

安装

安装该集成以及你的 agent 使用的 LiveKit 插件：

pip

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

uv

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

设置环境变量

该集成会从环境中读取你的 LangSmith 凭据，并通过 OpenTelemetry 为你导出到 LangSmith：

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>

设置追踪

在创建 AgentServer 之前调用一次 configure_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."),
        record={"audio": True},
    )

configure_livekit() 默认使用 LiveKit 会话录制。record=""audio": True" 选项告诉 LiveKit 创建录制。

此设置适用于 STT/LLM/TTS 级联和语音到语音（实时）模型。实时模型需要额外调用一次来捕获用户转录。有关更多信息，请参阅使用实时模型。

使用你自己的 tracer provider

如果你的应用已经管理 OpenTelemetry TracerProvider，请将 LangSmith processor 添加到该 provider，并将其注册到 LiveKit：

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)

将对话分组到线程

要将对话的运行分组到 LangSmith 线程，请在会话处理程序中调用 set_thread_id。为每个活动会话使用唯一 ID：

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)
    ...

使用实时模型

对于语音到语音（实时）模型，在创建 AgentSession 后调用 instrument_session 以捕获用户转录。将相同的线程 ID 传给 set_thread_id 和 instrument_session：

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},
    )

仅对实时模型调用 instrument_session。STT/LLM/TTS 级联已经捕获用户转录，因此在那里调用它会把每个用户轮次记录两次。

录制对话音频

默认情况下，集成使用 LiveKit 的会话录制。如果你改为录制到外部存储，请使用 Egress 模式。

使用 LiveKit 的会话录制进行录制

开启 LiveKit 的会话录制。

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},
    )

默认情况下，如果存在 LiveKit 录制，LangSmith 集成会捕获该录制。

NOTE

在 console 模式下，还要在命令行传入 --record（python agent.py console --record）。否则 LiveKit 会创建录制器但从不启动它，因此没有可附加的文件。录制反映的是播放给客户端的内容，因此打断会显示为截断。

使用 Egress 进行录制

当你希望将录制保存在自己的对象存储中或需要视频时，请使用 LiveKit Egress。Egress 录制交付需要线程 ID。为 Egress 配置集成，然后在 Egress 文件可用后调用 complete_recording：

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 代表你的存储客户端的下载操作。默认附件名和 MIME 类型为 recording.ogg 和 audio/ogg。如果你的 Egress 输出使用其他格式，请在 complete_recording 中设置 name 或 mime_type。

NOTE

始终调用 complete_recording，包括失败时使用 data=None。否则，集成会等待 recording_timeout_seconds（默认 30 秒）后导出不带音频的追踪。使用 complete_recording 捕获 Egress 录制需要设置 thread_id。

后续步骤

•Voice fundamentals

追踪语音 agent 的核心约定。

•Upload files with traces

将对话音频录音附加到你的追踪。

---