Skip to content
LangSmith 文档版本 跟随上游 源快照 948a43f0 更新于 2026-10-18

追踪 OpenAI Realtime 应用

使用 LangSmith SDK 在 LangSmith 中追踪 OpenAI Realtime 语音 agent。

NOTE

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

OpenAI Realtime 是一种通过 WebSocket 流式传输类型化事件的语音到语音模型。无论你是使用原始连接还是 OpenAI Agents SDK 构建它,该集成都会将每个对话捕获为单条 LangSmith 追踪,并为每个有意义的事件(转录、模型响应和工具调用)提供一个按轮次分组的 span。

将你的 OpenAI Realtime API 语音 agent 追踪到 LangSmith。有关高层约定,请参阅语音追踪基础。

选择一种方法

使用 OpenAI Realtime API 构建有两种方式,LangSmith 为每种方式都提供了追踪集成:

  • 如果你是直接使用实时客户端连接,请使用 wrap_realtime。
  • 如果你是使用 OpenAI Agents SDK 构建,请改用 wrap_realtime_session。

安装

langsmith[openai-realtime]>=0.9.7 附加项提供这两个包装器:

bash
pip install "langsmith[openai-realtime]"
bash
uv add "langsmith[openai-realtime]"

设置环境变量

bash
LANGSMITH_API_KEY=<your-langsmith-api-key>
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=<your-desired-langsmith-project>
OPENAI_API_KEY=<your-openai-api-key>

使用 Realtime 客户端

当你使用 client.realtime.connect() 自己打开 WebSocket 并驱动事件循环时,请使用此方法。

设置追踪

wrap_realtime 返回连接的透明代理。你现有的 async for event in connection 循环、session.update 和工具处理保持不变:

python
from langsmith.integrations.openai_realtime import wrap_realtime
from openai import AsyncOpenAI

client = AsyncOpenAI()

async with client.realtime.connect(model="gpt-realtime-2") as raw, wrap_realtime(
    raw,
    project_name="openai-realtime-voice",
) as connection:
    await connection.session.update(session={...})  # your existing config

    async for event in connection:
        ...  # your existing handling: play audio, run tools, update UI

NOTE

每个对话都会捕获为单独的追踪。要将其与 LangSmith 线程中的相关交互分组(例如继续之前的会话或从文本聊天接着进行),请传入 thread_id。在多个追踪中重复使用相同 ID 会将它们的事件关联在一起。

你在处理事件时运行的任何 @traceable 工具都会自动嵌套在该事件下。

NOTE

在 session.update 中启用 input_audio_transcription 和 agent 转录,以便在追踪中查看转录。

录制对话音频

当你向代理提供麦克风和播放音频时,它会在追踪中附加一个立体声录音(用户在左,agent 在右)。要标记打断,请使用 is_agent_speaking:

python
import base64

async with client.realtime.connect(model="gpt-realtime-2") as raw, wrap_realtime(
    raw,
    is_agent_speaking=lambda: speaker.buffered_bytes() > 0,
) as connection:
    # Record the agent's audio from the speaker, so the recording reflects only
    # what was played — audio a barge-in discards before playback is never recorded.
    speaker.set_played_callback(connection.record_agent_audio)

    # Record the user's mic where you send it to the model.
    async def send_mic(mic_chunk):
        await connection.input_audio_buffer.append(
            audio=base64.b64encode(mic_chunk).decode("ascii")
        )
        connection.record_user_audio(mic_chunk)          # user mic PCM16 as sent

    async for event in connection:
        if event.type == "response.output_audio.delta":
            speaker.play(base64.b64decode(event.delta))

NOTE

你应该从扬声器录制 agent 的音频,以便录制只反映实际播放的内容。在打断期间,生成的音频会在播放前被丢弃,不应被录制。这样做可以使追踪与用户实际听到的内容保持一致。

使用 OpenAI Agents SDK

当你使用 OpenAI Agents SDK(RealtimeAgent / RealtimeRunner)构建 agent 时,请使用此方法,该 SDK 负责轮次和工具调用循环。

NOTE

Agents SDK 内置的实时追踪会上传到 OpenAI 自己的仪表板。调用 agents.set_tracing_disabled(True) 以避免第二条单独的上传路径。

设置追踪

wrap_realtime_session 包装 RealtimeSession 并为你进入它。像迭代原始会话一样迭代它;SDK 会运行工具并管理轮次:

python
from agents import set_tracing_disabled
from agents.realtime import RealtimeAgent, RealtimeRunner
from langsmith.integrations.openai_realtime import wrap_realtime_session

set_tracing_disabled(True)

runner = RealtimeRunner(
    starting_agent=RealtimeAgent(name="assistant", instructions="...", tools=[...]),
)
session = await runner.run()

async with wrap_realtime_session(
    session,
    project_name="openai-realtime-voice",
) as conn:
    async for event in conn:
        ...  # your handling: play audio, update UI

对话转录会根据会话的 history 快照重建,因此即使 SDK 以部分片段的形式流式传输消息,消息也会出现。

NOTE

每个对话都会捕获为单独的追踪。要将其与 LangSmith 线程中的相关交互分组(例如继续之前的会话或从文本聊天接着进行),请传入 thread_id。在多个追踪中重复使用相同 ID 会将它们的事件关联在一起。

录制对话音频

向代理提供麦克风和播放音频:

python
# Record the agent's audio from the speaker, so the recording reflects only
# what was played — audio a barge-in discards before playback is never recorded.
speaker.set_played_callback(conn.record_agent_audio)

# Record the user's mic where you send it to the session.
async def send_mic(mic_chunk):
    await conn.send_audio(mic_chunk)
    conn.record_user_audio(mic_chunk)               # user mic PCM16 as sent

async for event in conn:
    if event.type == "audio":
        speaker.play(event.audio.data)

NOTE

你应该从扬声器录制 agent 的音频,以便录制只反映实际播放的内容。在打断期间,生成的音频会在播放前被丢弃,不应被录制。这样做可以使追踪与用户实际听到的内容保持一致。

后续步骤


BytesRouter 是社区翻译站点,所有译文仅供参考,一切以官方英文原文为准。