该 Gemini Live API 支持通过持久 WebSocket 连接与 Gemini 模型进行低延迟、双向的语音交互。本指南展示了如何将使用 Google Agent Development Kit (ADK) 构建的 Gemini Live 语音代理追踪到 LangSmith。

Gemini Live is a speech-to-speech model: it processes audio natively and exchanges a continuous stream of 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 principles on getting the most out of your voice agent traces, see 语音追踪基础.

ADK Live 事件模型

对话运行时，ADK 会向您的应用程序流式发送一系列事件。每个事件报告对话中发生的事情：音频片段、转录片段、工具调用、轮次边界或中断。每个事件具有相同的结构，且大多数字段是可选的，因此您需要根据 **哪些字段被填充**:

事件如何映射到 LangSmith 运行

为了充分利用追踪效果，请在单个对话追踪中捕获每个有意义的事件及其包含的数据，每个事件一个 span：

conversation                           ← root run (combined audio recording; ls_modality="audio")
│   metadata: thread_id, model, event_count, duration_s
│
├─ input_transcription                 ← a fragment of the user's speech transcript
├─ output_transcription                ← a fragment of the agent's speech transcript
├─ function_call: get_weather          ← the model requested the tool
├─ function_response: get_weather      ← ADK ran the tool; result heading back
├─ turn_complete                       ← turn boundary
└─ interrupted                         ← barge-in

安装

pip install "google-adk>=2.0" google-genai "langsmith>=0.4"

安装 sounddevice 以及 numpy ，如果您想捕获本地音频并附加对话录音。

设置您的环境

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

快速入门

步骤 1：构建 RunConfig

from google.adk.agents.run_config import RunConfig, StreamingMode
from google.genai import types as genai_types

run_config = RunConfig(
    response_modalities=["AUDIO"],
    streaming_mode=StreamingMode.BIDI,
    input_audio_transcription=genai_types.AudioTranscriptionConfig(),
    output_audio_transcription=genai_types.AudioTranscriptionConfig(),
)

步骤 2：打开对话根运行

为整个对话打开一个运行，并使用 ls_modality="audio"将其标记为语音追踪，遵循 单追踪约定。在会话的整个生命周期内保持此运行处于打开状态，并在会话结束时最终化它。

from langsmith import RunTree

session = RunTree(
    name="conversation",
    run_type="chain",
    extra={"metadata": {"thread_id": thread_id, "model": MODEL, "ls_modality": "audio"}},
)
session.post()

第3步：追踪每个事件

定义一个小辅助函数，为一个事件打开一个子运行，记录其清理后的负载，并在代码块退出时关闭它。 scrub 传递会用占位符替换原始音频字节，使跨度保持较小：

from contextlib import contextmanager


def scrub(obj):
    """Replace raw audio bytes with a placeholder so spans stay small."""
    if isinstance(obj, bytes):
        return f"<{len(obj)} bytes>"
    if isinstance(obj, dict):
        return {k: scrub(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [scrub(v) for v in obj]
    return obj


@contextmanager
def event_span(parent, event, *, name, inbound):
    """Trace one event as a child run under the conversation root.

    User-to-model events land in `inputs`; model-to-user events land in
    `outputs`, so the trace reads in the natural direction of flow.
    """
    payload = scrub(event.raw.model_dump())
    child = parent.create_child(
        name=name,
        run_type="chain",
        inputs=payload if inbound else {},
    )
    child.post()
    try:
        yield child
    finally:
        child.end(outputs={} if inbound else payload)
        child.patch()

然后遍历您应用中 run_live 流的数据，跳过仅音频的数据块并为其余数据创建跨度。 runner, adk_session和 queue 来自您的 ADK Live 应用（参见 演示代理); LiveEvent 是下面注释中定义的包装器：

async for raw_event in runner.run_live(
    user_id=USER_ID,
    session_id=adk_session.id,
    live_request_queue=queue,
    run_config=run_config,
):
    event = LiveEvent(raw_event)
    if event.is_audio_only:
        continue  # tracing audio will make your traces very noisy

    with event_span(session, event, name=event.label, inbound=event.is_inbound):
        ...  # handle the event: capture the transcript, run a tool, and so on

附加音频

要边听对话边查看其转录文本，请将整个对话的单个合并录音附加到根运行。将双方录制到一个立体声 WAV 文件中（用户的麦克风在左声道，代理的音频在右声道），这样中断会显示为两个声道之间的重叠。在发送时写入用户的麦克风帧，并在扬声器上窃听代理的音频，这样在打断时刷新到扬声器的音频永远不会进入录音，文件反映的是用户实际听到的内容。

有关底层附件 API，请参阅 上传文件并附加到追踪。有关跨提供商的原理，请参阅 录制单个合并音频文件.

在会话结束时最终化根运行。将事件循环包装在 try/finally 中，这样运行总是会关闭，即使发生错误：

try:
    ...  # the run_live event loop from Step 3
except Exception as exc:
    session.error = f"{type(exc).__name__}: {exc}"  # surface failures on the root run
finally:
    session.end()
    session.patch()

演示代码仓库为每个框架封装了完整的录制流程，包括麦克风重采样、扬声器窃听捕获和立体声 WAV 重建。对于 Gemini Live，请参阅 ADK 代理 和共享的 录制辅助函数.

故障排除

• 没有转录配置意味着追踪看起来为空。 这是最常见的失败模式。 input_audio_transcription 和 output_audio_transcription 都必须设置在 RunConfig.
• 不要累积转录片段。 使用 finished=True 事件的完整文本；片段仅用于实时 UI 显示。
• 不要为仅音频的事件创建跨度。 几分钟的对话会产生成千上万个这样的事件。
• 字段会同时出现。 按优先级分类，不要假设每个事件只有一个字段。
• 工具运行在 ADK 内部。 不要合成您自己的工具运行。这样做会重复计算 function_call 和 function_response 已经记录的内容。
• 重新采样麦克风 如果您的捕获不是 16 kHz（ADK 输入为 16 kHz，输出为 24 kHz）。
• 静音 ADK 的启动噪音 用于控制台 UI： logging.getLogger("google_adk").setLevel(logging.ERROR) 抑制实验性功能警告 run_live 以及 MCP 未安装的提示行。

后续步骤