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 the span processor setup to enable LangSmith tracing
from langsmith_processor import setup_langsmith_tracing

async def main():
    # Generate unique conversation ID for LangSmith
    conversation_id = str(uuid.uuid4())
    print(f"Starting conversation: {conversation_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=conversation_id,
    )

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

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

python agent.py

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

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