```
### 步骤 2：下载 span 处理器
添加启用 LangSmith 追踪的 [自定义 span 处理器文件](https://github.com/langchain-ai/voice-agents-tracing/blob/main/pipecat/langsmith_processor.py)。将其保存为项目目录中的 `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 部分：导入依赖项
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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 部分：定义主函数
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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 部分：添加入口点
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
if __name__ == "__main__":
asyncio.run(main())
```
### 步骤 4：运行您的代理
运行您的语音代理：
```bash theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
python agent.py
```
通过麦克风与代理交谈。所有追踪将自动出现在 LangSmith 中。以下是 LangSmith 中的一个追踪示例：[使用 Pipecat 的 LangSmith 追踪](https://smith.langchain.com/public/07721f41-cd27-413e-bc79-90bd23b6807d/r)。
查看完整的 [agent.py 代码](https://github.com/langchain-ai/voice-agents-tracing/blob/main/pipecat/agent.py)。
## 高级用法
### 自定义元数据和标签
您可以使用 span 属性为追踪添加自定义元数据：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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 实现](https://github.com/langchain-ai/voice-agents-tracing/blob/main/pipecat/audio_recorder.py)，它处理输入（麦克风）和输出（TTS）音频之间的采样率不匹配问题。
从开始到结束捕获所有音频并将其附加到对话 span：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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 实现](https://github.com/langchain-ai/voice-agents-tracing/blob/main/pipecat/turn_audio_recorder.py)，它为每个轮次分别捕获用户语音和 AI 响应。
为每个对话轮次捕获单独的音频片段，用户语音和 AI 响应保存为单独的文件：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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 的服务。
### 高级：音频录制故障排除
有关高级音频录制功能的问题，请参阅 [完整演示文档](https://github.com/langchain-ai/voice-agents-tracing)。
***
[将这些文档](/use-these-docs)通过 MCP 连接到 Claude、VSCode 等，以获取实时答案。
[在 GitHub 上编辑此页面](https://github.com/langchain-ai/docs/edit/main/src/langsmith/trace-with-pipecat.mdx) 或 [提交问题](https://github.com/langchain-ai/docs/issues/new/choose)。