跟踪 Pipecat 应用

此集成为测试版，其 API 可能会发生变化。

使用 LangSmith Pipecat 集成，将您的 Pipecat 语音代理跟踪到 LangSmith。有关高级约定，请参阅语音跟踪基础。

Pipecat 集成需要 langsmith[pipecat]>=0.9.7。

该集成会挂钩到 Pipecat 已发出的跨度，并将其映射到 LangSmith 的跟踪格式，因此每次对话都成为单个 LangSmith 跟踪，每个管道阶段（STT、LLM、TTS）一个跨度。这涵盖了 STT/LLM/TTS 级联和语音到语音（实时）模型。实时模型（例如，OpenAIRealtimeLLMService）需要一次额外调用来捕获用户的转录文本。请参阅当将 Pipecat 与实时模型结合使用时。

安装

安装集成以及您的管道使用的 Pipecat 服务附加组件：

pip install "langsmith[pipecat]" "pipecat-ai[openai,local,tracing]"

uv add "langsmith[pipecat]" "pipecat-ai[openai,local,tracing]"

设置环境变量

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

.env

LANGSMITH_API_KEY=<your-langsmith-api-key>
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=pipecat-voice
OPENAI_API_KEY=<your-openai-api-key>

设置跟踪

导入 configure_pipecat 并在构建管道之前调用一次。在 PipelineTask 上启用跟踪：

from langsmith.integrations.pipecat import configure_pipecat
from pipecat.pipeline.task import PipelineParams, PipelineTask

# 安装跟踪器和 LangSmith 跨度处理器。
configure_pipecat()

task = PipelineTask(
    pipeline,
    params=PipelineParams(enable_metrics=True),
    enable_tracing=True,
    enable_turn_tracking=True,
    conversation_id=conversation_id,
)

设置 enable_tracing=True、enable_turn_tracking=True 和 enable_metrics=True。轮次跟踪是跟踪所必需的，指标驱动每个跨度的延迟和令牌数据。

将 LangGraph 或 LangChain 代理用作 LLM

如果您的 LLM 阶段是一个进程内 LangGraph 或 LangChain 代理，其模型和工具运行应该嵌套在 Pipecat 的 llm 跨度内，而不是形成单独的跟踪。要实现这一点：

传递 configure_pipecat(llm_span_kind="chain")。这可以避免实际上并不代表推理请求的嵌套 llm 跨度。
在环境中设置 LANGSMITH_TRACING_MODE=otel。如果没有它，这些运行将直接发布到 LangSmith，并形成单独的跟踪，而不是嵌套。

使用您自己的跟踪器提供程序

configure_pipecat() 会构建一个 TracerProvider，注册 LangSmith 跨度处理器，并将其连接到 Pipecat。要通过您已经管理的 TracerProvider 发送跨度（例如，也导出到另一个 OpenTelemetry 后端的提供程序），请跳过 configure_pipecat，直接将处理器添加到您的提供程序：

from langsmith.integrations.pipecat import PipecatLangSmithSpanProcessor

provider.add_span_processor(PipecatLangSmithSpanProcessor())

将对话分组到线程中

要将对话的运行分组到 LangSmith 线程中，以便进行线程级视图和令牌与成本聚合，请在发出跨度之前，为每个对话调用一次 set_thread_id：

from langsmith.integrations.pipecat import configure_pipecat, set_thread_id

configure_pipecat()
set_thread_id(conversation_id)

当将 Pipecat 与实时模型结合使用时

使用语音到语音（实时）模型时，没有单独的语音到文本阶段，因此用户的转录文本永远不会作为 OTel 跨度发出。相反，它通过用户上下文聚合器的 on_user_turn_message_added 回调到达，Pipecat 在获得最终确定的用户文本后会触发此回调。如果不将其连接起来，跟踪将只显示助手端的内容。在构建上下文聚合器后立即调用一次 instrument_user_aggregator，这样 SDK 就会为您订阅该回调，并将每个转录文本与其轮次配对。它通过您传递给 set_thread_id 的 id 进行关联，因此请先设置该 id 并传递相同的 id：

instrument_user_aggregator 需要 langsmith[pipecat]>=0.10.6。

from langsmith.integrations.pipecat import configure_pipecat, set_thread_id
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.services.openai.realtime.events import (
    AudioConfiguration,
    AudioInput,
    AudioOutput,
    InputAudioTranscription,
    SessionProperties,
)
from pipecat.services.openai.realtime.llm import OpenAIRealtimeLLMService

conversation_id = "..."  # 任何标识对话的 id
span_processor = configure_pipecat()
set_thread_id(conversation_id)

llm = OpenAIRealtimeLLMService(
    api_key=openai_api_key,
    settings=OpenAIRealtimeLLMService.Settings(
        model="gpt-realtime",
        session_properties=SessionProperties(
            # 启用输入音频转录：OpenAI Realtime 将原始音频发送给模型，
            # 默认情况下不产生用户端文本。如果没有这个，
            # 上下文聚合器永远不会触发，因此用户的轮次永远不会到达跟踪。
            audio=AudioConfiguration(
                input=AudioInput(transcription=InputAudioTranscription()),
                output=AudioOutput(voice="marin"),
            ),
        ),
    ),
)

context = LLMContext(messages=[{"role": "system", "content": "..."}])
context_aggregator = LLMContextAggregatorPair(context, realtime_service_mode=True)
span_processor.instrument_user_aggregator(context_aggregator, conversation_id)  # 捕获用户转录文本

对于 OpenAI Realtime，还需要在会话上启用输入音频转录（InputAudioTranscription）；否则模型接收原始音频，不产生用户端文本，因此聚合器永远不会触发。其他通过聚合器自身提供用户文本的实时服务（例如，Gemini Live）只需要 instrument_user_aggregator 调用，无需额外的会话配置。仅为实时模型调用 instrument_user_aggregator。在 STT/LLM/TTS 级联中，转录文本已经被捕获（来自语音到文本阶段），因此在那里调用它会将用户的轮次记录两次。

记录对话音频

使用 Pipecat 的 AudioBufferProcessor 将对话音频附加到跟踪。将其放在 transport.output() 之后，以便捕获实际播放的内容（在任何打断截断之后），将其交给集成，并在会话运行后启动它：

from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor

span_processor = configure_pipecat()

# 立体声：用户在左声道，代理在右声道。
audiobuffer = AudioBufferProcessor(num_channels=2, buffer_size=32_000)
span_processor.attach_audio_buffer(audiobuffer, conversation_id=conversation_id)

pipeline = Pipeline([
    transport.input(),
    stt,
    context_aggregator.user(),
    llm,
    tts,
    transport.output(),
    audiobuffer,                     # 在 output() 之后：记录听到的内容
    context_aggregator.assistant(),
])

await audiobuffer.start_recording()

集成在会话结束时将录音附加到对话根目录。有关底层附件 API，请参阅上传文件与跟踪。

后续步骤

语音基础

跟踪语音代理的核心约定。

上传文件与跟踪

将对话录音附加到您的跟踪中。

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

Tracing setup

Configuration & troubleshooting

安装

设置环境变量

设置跟踪

将 LangGraph 或 LangChain 代理用作 LLM

使用您自己的跟踪器提供程序

将对话分组到线程中

当将 Pipecat 与实时模型结合使用时

记录对话音频

后续步骤

语音基础

上传文件与跟踪

​安装

​设置环境变量

​设置跟踪

​将 LangGraph 或 LangChain 代理用作 LLM

​使用您自己的跟踪器提供程序

​将对话分组到线程中

​当将 Pipecat 与实时模型结合使用时

​记录对话音频

​后续步骤

语音基础

上传文件与跟踪

安装

设置环境变量

设置跟踪

将 LangGraph 或 LangChain 代理用作 LLM

使用您自己的跟踪器提供程序

将对话分组到线程中

当将 Pipecat 与实时模型结合使用时

记录对话音频

后续步骤