Trace with Pipecat

LangSmith can capture traces generated by Pipecat using OpenTelemetry instrumentation. This guide shows you how to automatically capture traces from your Pipecat voice AI pipelines and send them to LangSmith for monitoring and analysis.

For a complete implementation, see the demo repository.

Installation

Install the required packages:

pip

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

If you plan to use the advanced audio recording features, also install: pip install scipy numpy

Quickstart tutorial

Follow this step-by-step tutorial to create a voice AI agent with Pipecat and LangSmith tracing. You’ll build a complete working example by copying and pasting code snippets.

Step 1: Set up your environment

Create a .env file in your project directory:

.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>

Step 2: Download the span processor

Add the custom span processor file that enables LangSmith tracing. Save it as langsmith_processor.py in your project directory.

What does the span processor do?

The span processor enriches Pipecat’s OpenTelemetry spans with LangSmith-compatible attributes so your traces display properly in LangSmith.

Key functions:

• Converts Pipecat span types (stt, llm, tts, turn, conversation) to LangSmith format.
• Adds gen_ai.prompt.* and gen_ai.completion.* attributes for message visualization.
• Tracks and aggregates conversation messages across turns.
• Handles audio file attachments (for advanced usage).

The processor automatically activates when you import it in your code.

Step 3: Create your voice agent file

Create a new file called agent.py and add the following code. We’ll build it section by section so you can copy and paste each part.

Part 1: Import dependencies

import asyncio
import uuid
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 span processor to enable LangSmith tracing
from langsmith_processor import span_processor

Part 2: Define the main function

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

    # 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-4o-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=conversation_id,
    )

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

Part 3: Add the entry point

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

Step 4: Run your agent

Run your voice agent:

python agent.py

Speak to the agent through your microphone. All conversation traces will automatically appear in LangSmith. View the complete agent.py code.

Advanced usage

Custom metadata and tags

You can add custom metadata to your traces using span attributes:

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")])

Recording and attaching audio to traces

You can capture audio from your voice conversations and attach it to traces in LangSmith. This allows you to listen to the actual audio alongside the transcriptions and AI responses.

Full conversation recording

See the AudioRecorder implementation which handles sample rate mismatches between input (microphone) and output (TTS) audio.

Capture all audio from start to finish and attach it to the conversation span:

from pathlib import Path
from datetime import datetime
from audio_recorder import AudioRecorder

# Setup recording directory
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"

# Create audio recorder
audio_recorder = AudioRecorder(str(recording_path))

# Register with span processor for attachment to conversation span
span_processor.register_recording(
    conversation_id,
    str(recording_path),
    audio_recorder=audio_recorder
)

# Add to your pipeline
pipeline = Pipeline([
    transport.input(),
    stt,
    context_aggregator.user(),
    llm,
    tts,
    audio_recorder,              # Add audio recorder to pipeline
    transport.output(),
    context_aggregator.assistant(),
])

# Run pipeline
runner = PipelineRunner()
try:
    await runner.run(task)
finally:
    # IMPORTANT: Save recording BEFORE conversation span completes
    audio_recorder.save_recording()

Per-turn recording

See the TurnAudioRecorder implementation which captures user speech and AI responses separately for each turn.

Capture separate audio snippets for each conversational turn, with user speech and AI responses saved as individual files:

from turn_audio_recorder import TurnAudioRecorder

# Create turn audio recorder
turn_audio_recorder = TurnAudioRecorder(
    span_processor=span_processor,
    conversation_id=conversation_id,
    recordings_dir=recordings_dir,
    turn_tracker=None,  # Will be set after task creation
)

# Register with span processor
span_processor.register_turn_audio_recorder(conversation_id, turn_audio_recorder)

# Add to your pipeline
pipeline = Pipeline([
    transport.input(),
    stt,
    context_aggregator.user(),
    llm,
    tts,
    audio_recorder,              # Full conversation recording
    turn_audio_recorder,         # Per-turn audio snippets
    transport.output(),
    context_aggregator.assistant(),
])

# Create task
task = PipelineTask(
    pipeline,
    params=PipelineParams(enable_metrics=True),
    enable_tracing=True,
    enable_turn_tracking=True,  # Required for turn audio recording
    conversation_id=conversation_id,
)

# Connect turn tracker after task creation
if task.turn_tracking_observer:
    turn_audio_recorder.connect_to_turn_tracker(task.turn_tracking_observer)

# Run pipeline
runner = PipelineRunner()
try:
    await runner.run(task)
finally:
    audio_recorder.save_recording()

Troubleshooting

Spans not appearing in LangSmith

If traces aren’t showing up in LangSmith:

1. Verify environment variables: Ensure OTEL_EXPORTER_OTLP_ENDPOINT and OTEL_EXPORTER_OTLP_HEADERS are set correctly in your .env file
2. Check API key: Confirm your LangSmith API key has write permissions.
3. Verify import: Make sure you’re importing span_processor from langsmith_processor.py.
4. Check .env loading: Ensure load_dotenv() is called before importing Pipecat components.

Messages not showing correctly

If conversation messages aren’t displaying properly:

1. Check span processor: Verify langsmith_processor.py is in your project directory and imported correctly.
2. Verify conversation ID: Ensure you’re setting a unique conversation_id in PipelineTask.
3. Enable turn tracking: Make sure enable_turn_tracking=True is set in PipelineTask.

Audio not working

If your microphone or speakers aren’t working:

1. Check permissions: Ensure your terminal/IDE has microphone access.
2. Test audio devices: Verify your microphone and speakers work in other applications.
3. VAD settings: Try adjusting SileroVADAnalyzer() settings if speech isn’t being detected.
4. Check services: Ensure OpenAI API key is valid and has access to Whisper and TTS.

Import errors

If you’re getting import errors:

1. Install dependencies: Run pip install langsmith "pipecat-ai[whisper,openai,local]" opentelemetry-exporter-otlp python-dotenv.
2. Check Python version: Ensure you’re using Python 3.9 or higher.
3. Verify langsmith_processor: Make sure langsmith_processor.py is downloaded and in the same directory as your agent.py.

Performance issues

If responses are slow:

1. Use faster models: Switch to gpt-4o-mini for the LLM (already in the tutorial).
2. Check network: Ensure stable internet connection for API calls.
3. Local STT: Consider using local Whisper instead of API-based services.

Advanced: Audio recording troubleshooting

For issues with the advanced audio recording features, see the complete demo documentation.

---

Edit the source of this page on GitHub.

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