追踪 Pipecat 应用

NOTE

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

使用 LangSmith Pipecat 集成将 Pipecat 语音 agent 追踪到 LangSmith。有关高层约定，请参阅语音追踪基础。

NOTE

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

该集成会接入 Pipecat 已经发出的 span，并将它们映射到 LangSmith 的追踪格式，因此每个对话都会成为一条 LangSmith 追踪，每个流水线阶段（STT、LLM、TTS）对应一个 span。这同时涵盖 STT/LLM/TTS 级联和语音到语音（实时）模型。实时模型（例如 OpenAIRealtimeLLMService）需要额外调用一次来捕获用户转录。请参阅在使用实时模型时配合 Pipecat。

安装

安装该集成以及你的流水线所使用的 Pipecat 服务附加组件：

pip

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

uv

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

设置环境变量

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

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

# Install the tracer and the LangSmith span processor.
configure_pipecat()

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

NOTE

设置 enable_tracing=True、enable_turn_tracking=True 和 enable_metrics=True。追踪需要开启轮次跟踪，指标会为每个 span 提供延迟和 token 数据。

使用 LangGraph 或 LangChain agent 作为 LLM

如果你的 LLM 阶段是进程内的 LangGraph 或 LangChain agent，其模型和工具运行应嵌套在 Pipecat 的 llm span 内，而不是形成单独的追踪。为此：

• 传入 configure_pipecat(llm_span_kind="chain")。这样可以避免嵌套并不真正代表推理请求的 llm span。
• 在环境中设置 LANGSMITH_TRACING_MODE=otel。否则，这些运行会直接发布到 LangSmith，并形成单独的追踪，而不是嵌套。

使用你自己的 tracer provider

configure_pipecat() 会构建一个 TracerProvider，注册 LangSmith span processor，并将其接入 Pipecat。要通过你已管理的 TracerProvider（例如同时导出到另一个 OpenTelemetry 后端的 provider）发送 span，请跳过 configure_pipecat，直接将 processor 添加到你的 provider：

from langsmith.integrations.pipecat import PipecatLangSmithSpanProcessor

provider.add_span_processor(PipecatLangSmithSpanProcessor())

将对话分组到线程

要将对话的运行分组到 LangSmith 线程中以获得线程级视图以及 token 和成本聚合，请在其 span 发出之前，为每个对话调用一次 set_thread_id：

from langsmith.integrations.pipecat import configure_pipecat, set_thread_id

configure_pipecat()
set_thread_id(conversation_id)

在使用实时模型时配合 Pipecat

使用语音到语音（实时）模型时，没有单独的语音转文本阶段，因此用户的转录永远不会作为 OTel span 发出。相反，它会通过用户上下文聚合器的 on_user_turn_message_added 回调到达，Pipecat 在获得最终用户文本后会触发该回调。如果不接入它，追踪只会显示 assistant 一侧。

在构建上下文聚合器之后立即调用一次 instrument_user_aggregator，这样 SDK 会为你订阅该回调，并将每个转录与其轮次配对。它通过你传给 set_thread_id 的 id 进行关联，因此请先设置该 id，并传入相同的 id：

NOTE

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 = "..."  # any id that identifies the conversation
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(
            # Enable input-audio transcription: OpenAI Realtime sends the model
            # raw audio and, by default, produces no user-side text. Without this
            # the context aggregator never fires, so the user's turns never reach
            # the trace.
            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)  # capture the user transcript

对于 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()

# Stereo: user on the left channel, agent on the right.
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,                     # after output(): records what was heard
    context_aggregator.assistant(),
])

await audiobuffer.start_recording()

对话结束时，集成会将录音附加到对话根节点。有关底层附件 API，请参阅使用追踪上传文件。

后续步骤

•Voice fundamentals

追踪语音 agent 的核心约定。

•Upload files with traces

将对话音频录音附加到你的追踪。

---