该 OpenAI Realtime API 为低延迟语音对语音语音代理提供支持。本指南展示如何将 Realtime 应用追踪到 LangSmith。

OpenAI Realtime is a speech-to-speech model: it processes audio natively and exchanges a continuous stream of typed JSON events with your application over a persistent WebSocket connection, rather than making discrete request/response calls. The following sections show those events and how to turn them into a LangSmith trace. For our high-level philosophy on getting the most out of your voice agent traces, see 语音追踪基础.

如需完整实现，请参阅 语音演示仓库.

事件模型

每个事件都有一个可辨识的 type 字符串来指示其代表的类型：音频、工具调用等。

客户端 **发送** 事件以配置会话和流式传输音频。您不应将这些作为跨度追踪；它们是请求，其效果作为服务器事件返回：

服务器 **发回** events:

事件如何映射到 LangSmith 运行

我们建议将整个对话作为单个追踪进行追踪，每个被追踪的事件按到达顺序创建一个跨度：

realtime_session                                 ← root run (chain)
│   metadata: thread_id, model, event_count, duration_s, ls_modality=audio
│   attachments: conversation.wav (stereo: L=user, R=agent)
│
├─ input_audio_buffer.speech_started
├─ input_audio_buffer.speech_stopped
├─ conversation.item.input_audio_transcription.completed
├─ response.created
├─ response.function_call_arguments.done
├─ response.done
│   └─ lookup_weather × N                         ← tool runs, nested under the event that announced them
├─ response.done                                  ← the spoken follow-up after tools
└─ error                                          ← only if the server sent one

安装

pip install "langsmith>=0.4" "openai>=1.50"

该演示还使用 sounddevice 和 numpy for the mic/speaker and to build the WAV attachment.

设置您的环境

以下步骤演示如何使用 LangSmith SDK 进行追踪。您也可以直接使用 OpenTelemetry 进行追踪。请参阅 使用 OpenTelemetry 追踪.

快速入门

步骤 1：在连接时打开对话根节点

每个对话使用一个 RunTree ：

from langsmith import RunTree

root = RunTree(
    name="realtime_session",
    run_type="chain",
    inputs={},
    project_name="my-voice-app",
    extra={"metadata": {"thread_id": thread_id, "model": MODEL, "ls_modality": "audio"}},
)
root.post()

一个稳定的 thread_id 您为每个对话生成的（例如 UUID）将追踪分组到 LangSmith 线程中; ls_modality="audio" 并将其标记为语音对话。

步骤 2：为每个收到的事件创建跨度，跳过噪音

定义一个小助手，为一个事件打开一个子运行，将清理后的负载记录为运行的输入，并在块退出时关闭它：

from contextlib import contextmanager

@contextmanager
def event_span(parent, event, *, name):
    """Trace one event as a child run, with its payload as the run's input."""
    payload = event.model_dump(mode="json")
    child = parent.create_child(name=name, run_type="chain", inputs={"event": payload})
    child.post()
    try:
        yield child
    finally:
        child.end()
        child.patch()

然后循环遍历开放 Realtime 中的事件 connection，跳过 .delta 噪音并追踪其余部分：

async for event in connection:
    if event.type.endswith(".delta"):
        continue  # the matching .done event repeats the full payload

    with event_span(root, event, name=event.type) as event_run:
        ...  # your handling for this event type

步骤 3：在宣布事件下嵌套运行工具

from langsmith.run_helpers import tracing_context

if event.type == "response.done":
    calls = [i for i in (event.response.output or []) if i.type == "function_call"]
    for call in calls:
        with tracing_context(parent=event_run):
            result = await execute_tool(call.name, call.arguments)  # traced child
        await connection.conversation.item.create(item={
            "type": "function_call_output",
            "call_id": call.call_id,
            "output": json.dumps(result),
        })
    if calls:
        await connection.response.create()   # ask for the spoken follow-up

附加对话音频

要同时收听对话及其转录，请将整个对话的单一合并录音附加到根运行。录制用户和代理的声音到一个文件中（例如，立体声 WAV，用户麦克风在一个声道，代理在另一个声道），从播放给客户端的内容中捕获，这样录音就能反映实际听到的内容，包括被插话切断的语音。Realtime API 将代理音频作为流式传输 response.output_audio.delta 事件：解码这些字节并写入您的输出设备，同时分接该输出以构建录音。

有关底层附加 API，请参阅 使用追踪上传文件。有关跨提供者原理，请参阅 录制单一合并音频文件.

当对话结束时，完成根运行：

root.end()
root.patch()

后续步骤