追踪 Pipecat 应用 - 以编程方式使用文档

LangSmith 可以捕获由 Pipecat 使用 OpenTelemetry 检测生成的追踪。本指南向您展示如何自动从 Pipecat 语音 AI 流水线中捕获追踪，并将其发送到 LangSmith 进行监控和分析。

关于追踪语音代理的高级指导原则，请参阅语音追踪基础.

完整的实现示例，请参阅语音演示仓库.

安装

安装所需的包：

pip install langsmith "pipecat-ai[whisper,openai,local]" opentelemetry-exporter-otlp python-dotenv

uv add langsmith "pipecat-ai[whisper,openai,local]" opentelemetry-exporter-otlp python-dotenv

快速入门教程

按照此分步教程创建带有 Pipecat 和 LangSmith 追踪的语音 AI 代理。您将通过复制粘贴代码片段来构建一个完整的可工作示例。

第 1 步：设置您的环境

在您的项目目录中创建一个 .env 文件：

OTEL_EXPORTER_OTLP_ENDPOINT=https://api.smith.langchain.com/otel
OTEL_EXPORTER_OTLP_HEADERS=x-api-key=<your-langsmith-api-key>, Langsmith-Project=pipecat-voice
OPENAI_API_KEY=<your-openai-api-key>

第 2 步：下载跨度处理器

Pipecat 发出 OpenTelemetry 跨度，但其属性名称不是 LangSmith 默认识别的属性。自定义跨度处理器会转换这些属性，以便您的追踪在 LangSmith 中正确呈现。

添加自定义跨度处理器文件并将其保存为 langsmith_processor.py 在您的项目目录中。

What does the span processor do?

跨度处理器使用与 LangSmith 兼容的属性丰富了 Pipecat 的 OpenTelemetry 跨度，以便您的追踪在 LangSmith 中正确显示。

主要功能：

- 将 Pipecat 跨度类型（stt、llm、tts、turn、conversation）转换为 LangSmith 格式。
- 添加 gen_ai.prompt.* 和 gen_ai.completion.* 属性用于消息可视化。
- 将整个对话记录渲染到根运行上。
- 处理音频文件附件（用于高级用法）。

它只会重塑其识别的 Pipecat 跨度；任何其他跨度（例如，嵌套的 LangChain 或 LangGraph 运行）都会原样通过。处理器在您在代码中导入它时激活。

第 3 步：创建您的语音代理文件

创建一个名为 agent.py 的新文件并添加以下代码。我们将分段构建，以便您可以复制粘贴每个部分。

第 1 部分：导入依赖

from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Import Pipecat components
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.whisper.stt import WhisperSTTService
from pipecat.services.openai import OpenAILLMService, OpenAITTSService
from pipecat.transports.local.audio import LocalAudioTransport, LocalAudioTransportParams

# Import the span processor setup to enable LangSmith tracing
from langsmith_processor import setup_langsmith_tracing

第 2 部分：定义主函数

async def main():
    # Generate unique thread ID for LangSmith
    thread_id = str(uuid.uuid4())
    print(f"Starting conversation: {thread_id}")

    # Configure OpenTelemetry export to LangSmith and register the span processor.
    # This reads OTEL_EXPORTER_OTLP_ENDPOINT / OTEL_EXPORTER_OTLP_HEADERS from your
    # environment and returns the processor so you can register a recording later.
    span_processor = setup_langsmith_tracing()

    # Configure audio input/output with voice activity detection
    transport = LocalAudioTransport(
        LocalAudioTransportParams(
            audio_in_enabled=True,
            audio_out_enabled=True,
            vad_analyzer=SileroVADAnalyzer(),
        )
    )

    # Initialize AI services
    stt = WhisperSTTService()
    llm = OpenAILLMService(model="gpt-5.4-mini")
    tts = OpenAITTSService(voice="alloy")

    # Set up conversation context with system prompt
    context = OpenAILLMContext(
        messages=[
            {
                "role": "system",
                "content": "You are a helpful voice assistant. Keep responses concise and conversational."
            }
        ]
    )
    context_aggregator = llm.create_context_aggregator(context)

    # Build the processing pipeline
    pipeline = Pipeline([
        transport.input(),           # Capture microphone input
        stt,                         # Convert speech to text
        context_aggregator.user(),   # Add user message to context
        llm,                         # Generate AI response
        tts,                         # Convert response to speech
        transport.output(),          # Play through speakers
        context_aggregator.assistant(),  # Add assistant response to context
    ])

    # Create task with tracing enabled
    task = PipelineTask(
        pipeline,
        params=PipelineParams(enable_metrics=True),
        enable_tracing=True,
        enable_turn_tracking=True,
        conversation_id=thread_id,
    )

    # Run the agent
    runner = PipelineRunner()
    await runner.run(task)

第 3 部分：添加入口点

if __name__ == "__main__":
    asyncio.run(main())

第 4 步：运行您的代理

运行您的语音代理：

python agent.py

通过麦克风与代理对话。所有追踪将自动出现在 LangSmith 中。

查看完整的 agent.py 代码.

高级用法

追踪嵌套的 LangGraph 代理

您可以使用进程内的 LangChain 或 LangGraph agent 作为流水线的 LLM 阶段。稍作调整后，agent 的模型和工具运行会嵌套在 Pipecat 的 llm span 内，这样整个对话保持为单个 trace。

实现这一点需要三个要素：

**设置 LANGSMITH_TRACING_MODE=otel.** This makes the LangSmith SDK emit your LangChain/LangGraph runs as OpenTelemetry spans through the same provider, so they nest under Pipecat's llm span，而不是形成单独的顶层 traces。
**使用已追踪的 LLM 服务。** 裸 FrameProcessor 不产生 llm span 用于图的运行嵌套在其下。继承已追踪服务，例如 OpenAILLMService 并从追踪的上下文处理器运行你的图，这样其运行会落在 llm span.
**在 span 处理器上设置 llm_span_kind="chain" 。** 当图嵌套在其中时，Pipecat 的 llm span 不再自行执行推理：图自身的模型节点才是真正的 LLM 运行。将 wrapper 标记为 chain 可以避免在一个 LLM 运行中嵌套另一个 LLM 运行。

最终生成的 trace 如下：

conversation                        ← root: whole transcript + audio recording
└── turn × N
    ├── stt
    ├── llm                         ← chain (orchestrates the graph)
    │   ├── model                   ← ChatOpenAI (may emit tool calls)
    │   ├── tools: lookup_weather   ← tool run
    │   └── model                   ← final answer (spoken)
    └── tts

完整的可运行实现，请参阅语音演示仓库.

自定义元数据和标签

你可以使用 span 属性向 traces 添加自定义元数据：

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

async def run_voice_session():
    with tracer.start_as_current_span("voice_conversation") as span:
        # Add custom metadata
        span.set_attribute("langsmith.metadata.session_type", "voice_assistant")
        span.set_attribute("langsmith.metadata.user_id", "user_123")
        span.set_attribute("langsmith.span.tags", "pipecat,voice-ai,stt-llm-tts")

        # Your Pipecat pipeline code here
        task = PipelineTask(pipeline, enable_tracing=True)
        await task.queue_frames([TextFrame("Hello")])

录制音频并附加到 traces

录制对话并把音频附加到根运行，这样你就可以在转录文本旁边收听。对于底层附件 API，请参阅上传文件与 traces 关联.

录制 **听到的内容**，而不是生成的内容。简单的方法是在流水线上添加录制 FrameProcessor ，会截取上游 TTS 帧的 transport.output()，所以会过度捕获：当中断截断 agent 说到一半的句子时，它包含用户从未听到的音频。相反，应该在设备写入边界处截取，输出传输只有 *在* 中断截断之后

演示的 RecordingLocalAudioTransport 就是这样做的：它以 write_audio_frame 录制播放的 agent 音频，从输入回调录制用户音频，然后使用共享的 build_stereo_session_wav 辅助函数写出一个立体声 WAV（左声道用户，右声道 agent）。以此作为参考实现，并适配到你的项目中。

交换 LocalAudioTransport 为录制传输，注册录制到 span 处理器以便在对话 span 结束时附加，并将其保存在 finally block:

from pathlib import Path
from datetime import datetime
from recording_transport import ConversationRecorder, RecordingLocalAudioTransport

# Write the conversation recording to a per-run path
recordings_dir = Path.cwd() / "pipecat-recordings"
recordings_dir.mkdir(exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
recording_path = recordings_dir / f"conversation_{timestamp}.wav"

# Tap played agent audio and the user's mic at the device boundary
recorder = ConversationRecorder(recording_path)
transport = RecordingLocalAudioTransport(
    LocalAudioTransportParams(
        audio_in_enabled=True,
        audio_out_enabled=True,
        vad_analyzer=SileroVADAnalyzer(),
    ),
    recorder,
)

# Attach the WAV to the conversation root span when the span ends
span_processor.register_recording(
    thread_id, str(recording_path), audio_recorder=recorder
)

# Build the pipeline with the recording transport
pipeline = Pipeline([
    transport.input(),               # mic in (taps user audio)
    stt,
    context_aggregator.user(),
    llm,
    tts,
    transport.output(),              # speaker out (taps played agent audio)
    context_aggregator.assistant(),
])

# Run the pipeline, saving the recording on the way out
runner = PipelineRunner()
try:
    await runner.run(task)
finally:
    recorder.save_recording()

故障排除

LangSmith 中未显示 Spans

如果 traces 没有在 LangSmith 中显示：

**验证环境变量**：确保 OTEL_EXPORTER_OTLP_ENDPOINT 和 OTEL_EXPORTER_OTLP_HEADERS 已在您的中正确设置 .env file.
**检查 API 密钥**：确认您的 LangSmith API 密钥具有写入权限。
**验证导入**：确保您正在导入 setup_langsmith_tracing 从 langsmith_processor.py 并在运行管道之前调用它。
**检查 .env 加载**：确保 load_dotenv() 在导入 Pipecat 组件之前被调用。

消息显示不正确

如果对话消息未正确显示：

**检查 span 处理器**：验证 langsmith_processor.py 在您的项目目录中且导入正确。
**验证线程 ID**：确保您设置了唯一的 conversation_id in PipelineTask.
**启用轮次跟踪**：确保 enable_turn_tracking=True 在 PipelineTask.

音频不工作

如果您的麦克风或扬声器不工作：

**检查权限**: Ensure your terminal/IDE has microphone access.
**测试音频设备**：验证您的麦克风和扬声器在其他应用程序中正常工作。
**VAD 设置**：尝试调整 SileroVADAnalyzer() 设置（如果未检测到语音）。
**检查服务**：确保 OpenAI API 密钥有效且可访问 Whisper 和 TTS。

导入错误

如果您遇到导入错误：

**安装依赖项**：运行 pip install langsmith "pipecat-ai[whisper,openai,local]" opentelemetry-exporter-otlp python-dotenv.
**检查 Python 版本**：确保您使用的是 Python 3.9 或更高版本。
**验证 langsmith_处理器**：确保 langsmith_processor.py 已下载且在您的 agent.py.

性能问题

如果响应缓慢：

**使用更快的模型**：切换到 gpt-5.4-mini 用于 LLM（已在教程中）。
**检查网络**：确保 API 调用的网络连接稳定。
**本地 STT**：考虑使用本地 Whisper 而非基于 API 的服务。

高级：音频录制故障排除

有关高级音频录制功能的问题，请参阅完整演示文档.