追踪 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(
        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."))

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

from langsmith.integrations.livekit import LiveKitLangSmithSpanProcessor

provider = TracerProvider()  # 您自己的提供者
provider.add_span_processor(LiveKitLangSmithSpanProcessor())
telemetry.set_tracer_provider(provider)

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  # 或任何标识对话的 ID
    set_thread_id(thread_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  # 或任何标识对话的 ID
    set_thread_id(thread_id)

    session = AgentSession(llm=lk_openai.realtime.RealtimeModel(voice="marin"))
    processor.instrument_session(session, thread_id)  # 捕获用户转录文本

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

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

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  # 每个会话唯一；控制台模式下 ctx.room.name 为 "console"
    set_thread_id(thread_id)  # 将此对话的跨度分组到一个线程中
    key = f"recordings/{thread_id}.ogg"

    # 启动一个仅音频的房间组合 Egress 到您的存储。
    lkapi = api.LiveKitAPI()  # 读取 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"],
                    ),
                )
            ],
        )
    )
    # 保持追踪打开，直到录音准备就绪。
    processor.expect_recording(thread_id)

    async def attach_recording():
        try:
            await wait_for_egress(lkapi, egress.egress_id)  # 轮询直到 EGRESS_COMPLETE
            audio = download_from_storage(RECORDING_BUCKET, key)  # 您的存储客户端
            processor.complete_recording(thread_id, audio, name="call.ogg")
        except Exception:
            processor.complete_recording(thread_id, None)  # 无音频释放

    ctx.add_shutdown_callback(attach_recording)

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