Skip to main content
LangSmith 可以通过 OpenTelemetry 插桩来捕获由 Pipecat 生成的追踪。本指南将向您展示如何自动捕获来自 Pipecat 语音 AI 管道的追踪，并将其发送到 LangSmith 进行监控和分析。 有关完整实现，请参阅 演示仓库。

​
安装

安装所需的包： 
pip install langsmith "pipecat-ai[whisper,openai,local]" opentelemetry-exporter-otlp python-dotenv
如果您计划使用高级音频录制功能，还需安装：pip install scipy numpy

​
快速入门教程

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

​
步骤 1：设置您的环境

在项目目录中创建一个 .env 文件：
.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：下载 span 处理器

添加启用 LangSmith 追踪的 自定义 span 处理器文件。将其保存为项目目录中的 langsmith_processor.py。
span 处理器使用与 LangSmith 兼容的属性来丰富 Pipecat 的 OpenTelemetry span，以便您的追踪能在 LangSmith 中正确显示。主要功能：
  • 将 Pipecat span 类型（stt、llm、tts、turn、conversation）转换为 LangSmith 格式。
  • 添加 gen_ai.prompt.* 和 gen_ai.completion.* 属性以实现消息可视化。
  • 跨轮次跟踪和聚合对话消息。
  • 处理音频文件附件（用于高级用法）。
当您在代码中导入该处理器时，它会自动激活。

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

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

​
第 1 部分：导入依赖项

import asyncio
import uuid
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 导入 Pipecat 组件
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

# 导入 span 处理器以启用 LangSmith 追踪
from langsmith_processor import span_processor

​
第 2 部分：定义主函数

async def main():
    # 为 LangSmith 生成唯一的对话 ID
    conversation_id = str(uuid.uuid4())
    print(f"Starting conversation: {conversation_id}")

    # 配置带有语音活动检测的音频输入/输出
    transport = LocalAudioTransport(
        LocalAudioTransportParams(
            audio_in_enabled=True,
            audio_out_enabled=True,
            vad_analyzer=SileroVADAnalyzer(),
        )
    )

    # 初始化 AI 服务
    stt = WhisperSTTService()
    llm = OpenAILLMService(model="gpt-5.4-mini")
    tts = OpenAITTSService(voice="alloy")

    # 设置带有系统提示的对话上下文
    context = OpenAILLMContext(
        messages=[
            {
                "role": "system",
                "content": "You are a helpful voice assistant. Keep responses concise and conversational."
            }
        ]
    )
    context_aggregator = llm.create_context_aggregator(context)

    # 构建处理管道
    pipeline = Pipeline([
        transport.input(),           # 捕获麦克风输入
        stt,                         # 将语音转换为文本
        context_aggregator.user(),   # 将用户消息添加到上下文
        llm,                         # 生成 AI 响应
        tts,                         # 将响应转换为语音
        transport.output(),          # 通过扬声器播放
        context_aggregator.assistant(),  # 将助手响应添加到上下文
    ])

    # 创建启用了追踪的任务
    task = PipelineTask(
        pipeline,
        params=PipelineParams(enable_metrics=True),
        enable_tracing=True,
        enable_turn_tracking=True,
        conversation_id=conversation_id,
    )

    # 运行代理
    runner = PipelineRunner()
    await runner.run(task)

​
第 3 部分：添加入口点

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

​
步骤 4：运行您的代理

运行您的语音代理： 
python agent.py
通过麦克风与代理交谈。所有追踪将自动出现在 LangSmith 中。以下是 LangSmith 中的一个追踪示例：使用 Pipecat 的 LangSmith 追踪。 查看完整的 agent.py 代码。

​
高级用法

​
自定义元数据和标签

您可以使用 span 属性为追踪添加自定义元数据： 
from opentelemetry import trace

tracer = trace.get_tracer(__name__)

async def run_voice_session():
    with tracer.start_as_current_span("voice_conversation") as span:
        # 添加自定义元数据
        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")

        # 您的 Pipecat 管道代码在此处
        task = PipelineTask(pipeline, enable_tracing=True)
        await task.queue_frames([TextFrame("Hello")])

​
录制音频并将其附加到追踪

您可以从语音对话中捕获音频，并将其附加到 LangSmith 中的追踪。这允许您在转录和 AI 响应的同时收听实际音频。

​
完整对话录制

请参阅 AudioRecorder 实现，它处理输入（麦克风）和输出（TTS）音频之间的采样率不匹配问题。 从开始到结束捕获所有音频并将其附加到对话 span： 
from pathlib import Path
from datetime import datetime
from audio_recorder import AudioRecorder

# 设置录制目录
recordings_dir = Path(__file__).parent / "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"

# 创建音频录制器
audio_recorder = AudioRecorder(str(recording_path))

# 注册到 span 处理器以附加到对话 span
span_processor.register_recording(
    conversation_id,
    str(recording_path),
    audio_recorder=audio_recorder
)

# 添加到您的管道
pipeline = Pipeline([
    transport.input(),
    stt,
    context_aggregator.user(),
    llm,
    tts,
    audio_recorder,              # 将音频录制器添加到管道
    transport.output(),
    context_aggregator.assistant(),
])

# 运行管道
runner = PipelineRunner()
try:
    await runner.run(task)
finally:
    # 重要：在对话 span 完成之前保存录制
    audio_recorder.save_recording()

​
按轮次录制

请参阅 TurnAudioRecorder 实现，它为每个轮次分别捕获用户语音和 AI 响应。 为每个对话轮次捕获单独的音频片段，用户语音和 AI 响应保存为单独的文件： 
from turn_audio_recorder import TurnAudioRecorder

# 创建轮次音频录制器
turn_audio_recorder = TurnAudioRecorder(
    span_processor=span_processor,
    conversation_id=conversation_id,
    recordings_dir=recordings_dir,
    turn_tracker=None,  # 将在任务创建后设置
)

# 注册到 span 处理器
span_processor.register_turn_audio_recorder(conversation_id, turn_audio_recorder)

# 添加到您的管道
pipeline = Pipeline([
    transport.input(),
    stt,
    context_aggregator.user(),
    llm,
    tts,
    audio_recorder,              # 完整对话录制
    turn_audio_recorder,         # 按轮次音频片段
    transport.output(),
    context_aggregator.assistant(),
])

# 创建任务
task = PipelineTask(
    pipeline,
    params=PipelineParams(enable_metrics=True),
    enable_tracing=True,
    enable_turn_tracking=True,  # 按轮次音频录制所需
    conversation_id=conversation_id,
)

# 在任务创建后连接轮次跟踪器
if task.turn_tracking_observer:
    turn_audio_recorder.connect_to_turn_tracker(task.turn_tracking_observer)

# 运行管道
runner = PipelineRunner()
try:
    await runner.run(task)
finally:
    audio_recorder.save_recording()

​
故障排除

​
Span 未出现在 LangSmith 中

如果追踪未显示在 LangSmith 中：
  1. 验证环境变量：确保 OTEL_EXPORTER_OTLP_ENDPOINT 和 OTEL_EXPORTER_OTLP_HEADERS 在您的 .env 文件中设置正确。
  2. 检查 API 密钥：确认您的 LangSmith API 密钥具有写入权限。
  3. 验证导入：确保您从 langsmith_processor.py 导入了 span_processor。
  4. 检查 .env 加载：确保在导入 Pipecat 组件之前调用了 load_dotenv()。

​
消息显示不正确

如果对话消息显示不正确：
  1. 检查 span 处理器：验证 langsmith_processor.py 在您的项目目录中并已正确导入。
  2. 验证对话 ID：确保您在 PipelineTask 中设置了唯一的 conversation_id。
  3. 启用轮次跟踪：确保在 PipelineTask 中设置了 enable_turn_tracking=True。

​
音频不工作

如果您的麦克风或扬声器不工作：
  1. 检查权限：确保您的终端/IDE 具有麦克风访问权限。
  2. 测试音频设备：验证您的麦克风和扬声器在其他应用程序中正常工作。
  3. VAD 设置：如果未检测到语音，请尝试调整 SileroVADAnalyzer() 设置。
  4. 检查服务：确保 OpenAI API 密钥有效并有权访问 Whisper 和 TTS。

​
导入错误

如果您遇到导入错误：
  1. 安装依赖项：运行 pip install langsmith "pipecat-ai[whisper,openai,local]" opentelemetry-exporter-otlp python-dotenv。
  2. 检查 Python 版本：确保您使用的是 Python 3.9 或更高版本。
  3. 验证 langsmith_processor：确保 langsmith_processor.py 已下载并与您的 agent.py 位于同一目录中。

​
性能问题

如果响应缓慢：
  1. 使用更快的模型：为 LLM 切换到 gpt-5.4-mini（教程中已使用）。
  2. 检查网络：确保 API 调用的互联网连接稳定。
  3. 本地 STT：考虑使用本地 Whisper 而不是基于 API 的服务。

​
高级：音频录制故障排除

有关高级音频录制功能的问题，请参阅 完整演示文档。
将这些文档通过 MCP 连接到 Claude、VSCode 等，以获取实时答案。
在 GitHub 上编辑此页面 或 提交问题。