Trace Pipecat applications ↗

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

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-4.1-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)

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

python agent.py

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

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

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