Skip to main content
Test your Pipecat-powered voice agents by connecting them to UserTrace. Our simulated users will interact with your Pipecat-based agents through real-time voice conversations.

Getting Started

1. Connect Your Pipecat Agent

In the UserTrace dashboard, navigate to Agent Setup and select Pipecat Integration. Required Information:
  • Pipecat agent endpoint URL
  • Authentication credentials (if required)
  • Transport method (WebRTC, WebSocket, or SIP)
Example:
Agent Endpoint: wss://your-agent.com/pipecat
Transport: WebRTC
Authentication: API Key

2. Set Evaluation Context

Define your pass and fail criteria for voice interactions: Example Pass/Fail Criteria:
Pass Criteria:
- Establishes connection within 3 seconds
- Maintains real-time conversation flow
- Processes interruptions naturally
- Provides contextually relevant responses
- Handles audio quality gracefully

Fail Criteria:
- Connection timeouts or failures
- Unnatural or robotic responses
- Poor handling of interruptions
- Off-topic or irrelevant responses
- Audio artifacts or quality issues

Testing Process

Real-Time Pipeline Testing

Pipecat testing focuses on the real-time processing pipeline:
  1. UserTrace connects to your Pipecat agent endpoint
  2. Real-time audio processing pipeline is established
  3. Simulated user begins conversation based on scenario
  4. Pipecat processes audio through its pipeline stages
  5. Agent responses are generated and delivered
  6. Conversation quality and pipeline performance are monitored
Pipeline Flow:
  1. Audio Input: User speech captured and processed
  2. Speech-to-Text: Real-time transcription
  3. LLM Processing: Context understanding and response generation
  4. Text-to-Speech: Natural voice synthesis
  5. Audio Output: Delivered to user
  6. Quality Metrics: Latency and accuracy tracking

Pipecat-Specific Features

Pipeline Architecture

Modular Components:
  • Audio input/output processors
  • Speech recognition services
  • LLM integrations
  • Text-to-speech engines
  • Transport mechanisms
Real-Time Processing:
  • Streaming audio processing
  • Low-latency pipeline execution
  • Interrupt handling
  • Context preservation

Transport Options

WebRTC Transport:
  • Browser-based connections
  • Ultra-low latency
  • Built-in echo cancellation
  • Network adaptation
WebSocket Transport:
  • Simple integration
  • Custom audio protocols
  • Server-to-server communication
  • Scalable architecture
SIP Integration:
  • Traditional telephony systems
  • PBX compatibility
  • Carrier-grade reliability
  • Standards compliance

Best Practices

Pipeline Optimization

Performance Tuning• Minimize processing latency • Optimize buffer sizes • Use appropriate audio codecs • Monitor pipeline bottlenecks

Audio Quality

Sound Processing• Configure noise suppression • Implement echo cancellation • Handle variable audio quality • Test with different microphones

Implementation Examples

Basic Pipecat Agent

Python Implementation:
import asyncio
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.task import PipelineTask
from pipecat.services.openai import OpenAILLMService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.services.deepgram import DeepgramSTTService
from pipecat.transports.network.websocket import WebsocketTransport

async def run_agent():
    # Configure services
    stt_service = DeepgramSTTService(api_key="your-deepgram-key")
    llm_service = OpenAILLMService(api_key="your-openai-key")
    tts_service = ElevenLabsTTSService(api_key="your-elevenlabs-key")
    
    # Set up transport
    transport = WebsocketTransport(
        host="localhost",
        port=8765,
        params={"access_token": "your-token"}
    )
    
    # Create pipeline
    pipeline = Pipeline([
        transport.input_processor(),
        stt_service,
        llm_service, 
        tts_service,
        transport.output_processor()
    ])
    
    # Run the task
    task = PipelineTask(pipeline)
    await task.run()

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

Advanced Configuration

Custom Pipeline:
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.audio.audio_buffer import AudioBuffer

# Configure advanced pipeline with custom processors
pipeline = Pipeline([
    # Audio preprocessing
    AudioBuffer(sample_rate=16000),
    
    # Speech recognition
    DeepgramSTTService(
        api_key="your-key",
        model="nova-2",
        language="en-US"
    ),
    
    # Context management
    OpenAILLMContext(
        messages=[
            {"role": "system", "content": "You are a helpful assistant."}
        ]
    ),
    
    # LLM processing
    OpenAILLMService(
        api_key="your-key",
        model="gpt-4",
        max_tokens=150
    ),
    
    # Voice synthesis
    ElevenLabsTTSService(
        api_key="your-key",
        voice_id="your-voice-id"
    )
])

Common Scenarios

Conversational AI:
  • Personal assistants
  • Customer service bots
  • Educational tutors
  • Healthcare assistants
Real-Time Applications:
  • Live translation services
  • Meeting assistants
  • Voice-controlled systems
  • Interactive voice response (IVR)
Multi-Modal Experiences:
  • Video conferencing bots
  • Smart home interfaces
  • Automotive assistants
  • Gaming characters

Advanced Features

Interrupt Handling

Barge-in Support:
from pipecat.processors.interruption import InterruptionHandler

interruption_handler = InterruptionHandler(
    enable_barge_in=True,
    sensitivity=0.7,
    timeout_seconds=1.0
)

Context Management

Conversation Memory:
from pipecat.processors.context import ConversationContext

context = ConversationContext(
    max_turns=10,
    context_window=4000,
    preserve_system_message=True
)

Custom Processors

Audio Processing:
from pipecat.processors.base import AudioProcessor

class CustomAudioProcessor(AudioProcessor):
    async def process_audio(self, audio_data):
        # Custom audio processing logic
        processed_audio = self.apply_filters(audio_data)
        return processed_audio

Troubleshooting

Common Issues: Pipeline Problems:
  • High latency: Optimize processor order and buffer sizes
  • Audio dropouts: Check network stability and audio codec settings
  • Memory issues: Monitor processor memory usage and cleanup
  • Context loss: Verify context management configuration
Service Integration:
  • API errors: Validate service credentials and rate limits
  • Model failures: Test with different AI models and configurations
  • Transport issues: Check network connectivity and protocol settings
  • Audio quality: Verify codec compatibility and audio processing

Development Setup

Local Development

Docker Compose:
version: '3.8'
services:
  pipecat-agent:
    build: .
    ports:
      - "8765:8765"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - DEEPGRAM_API_KEY=${DEEPGRAM_API_KEY}
      - ELEVENLABS_API_KEY=${ELEVENLABS_API_KEY}
    volumes:
      - ./config:/app/config
Environment Configuration:
export OPENAI_API_KEY="your-openai-key"
export DEEPGRAM_API_KEY="your-deepgram-key" 
export ELEVENLABS_API_KEY="your-elevenlabs-key"
export TRANSPORT_HOST="0.0.0.0"
export TRANSPORT_PORT="8765"

Testing Pipeline

Unit Testing:
import pytest
from pipecat.pipeline.pipeline import Pipeline

@pytest.mark.asyncio
async def test_pipeline_latency():
    pipeline = create_test_pipeline()
    start_time = time.time()
    
    # Test audio processing
    result = await pipeline.process(test_audio)
    
    latency = time.time() - start_time
    assert latency < 0.5  # Target < 500ms
Need help with Pipecat setup? Check the Pipecat documentation or contact [email protected].