Observability for Pipecat with Opik

1
import os
2
import getpass
3
4
if "OPENAI_API_KEY" not in os.environ:
5
    os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")
6
7
if "DEEPGRAM_API_KEY" not in os.environ:
8
    os.environ["DEEPGRAM_API_KEY"] = getpass.getpass("Enter your Deepgram API key: ")
9
10
# Set project name for organization
11
os.environ["OPIK_PROJECT_NAME"] = "pipecat-voice-agent-demo"

1
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
2
from opentelemetry import trace
3
from opentelemetry.sdk.trace import TracerProvider
4
from opentelemetry.sdk.trace.export import BatchSpanProcessor
5
from opentelemetry.sdk.resources import Resource
6
from pipecat.pipeline.pipeline import Pipeline
7
from pipecat.pipeline.runner import PipelineRunner
8
from pipecat.pipeline.task import PipelineTask, PipelineParams
9
from pipecat.services.openai import OpenAILLMService
10
from pipecat.services.deepgram import DeepgramSTTService
11
from pipecat.services.cartesi import CartesiTTSService
12
from pipecat.transports.websocket import WebSocketTransport
13
from pipecat.frames.frames import TextFrame
14
import os
15
16
# Initialize OpenTelemetry with the OTLP exporter
17
def setup_tracing(service_name="pipecat-demo"):
18
    # Create resource
19
    resource = Resource.create({
20
        "service.name": service_name,
21
        "service.version": "1.0.0",
22
    })
23
    
24
    # Create tracer provider
25
    tracer_provider = TracerProvider(resource=resource)
26
    trace.set_tracer_provider(tracer_provider)
27
    
28
    # Create OTLP exporter
29
    exporter = OTLPSpanExporter(
30
        endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT"),
31
        headers=os.getenv("OTEL_EXPORTER_OTLP_HEADERS")
32
    )
33
    
34
    # Add span processor
35
    span_processor = BatchSpanProcessor(exporter)
36
    tracer_provider.add_span_processor(span_processor)
37
    
38
    return tracer_provider
39
40
# Configure tracing
41
setup_tracing("pipecat-voice-agent")
42
43
# Create your Pipecat pipeline
44
async def main():
45
    # Create transport
46
    transport = WebSocketTransport("ws://localhost:8765")
47
    
48
    # Create services
49
    llm = OpenAILLMService(
50
        api_key=os.getenv("OPENAI_API_KEY"),
51
        model="gpt-4o-mini"
52
    )
53
    
54
    stt = DeepgramSTTService(
55
        api_key=os.getenv("DEEPGRAM_API_KEY"),
56
        model="nova-2"
57
    )
58
    
59
    tts = CartesiTTSService(
60
        api_key=os.getenv("CARTESI_API_KEY"),
61
        voice="samantha"
62
    )
63
    
64
    # Create pipeline
65
    pipeline = Pipeline([
66
        stt,
67
        llm,
68
        tts,
69
        transport.output()
70
    ])
71
    
72
    # Enable tracing in your PipelineTask
73
    task = PipelineTask(
74
        pipeline,
75
        params=PipelineParams(
76
            allow_interruptions=True,
77
            enable_metrics=True,  # Required for some service metrics
78
        ),
79
        enable_tracing=True,  # Enables both turn and conversation tracing
80
        conversation_id="customer-123",  # Optional - will auto-generate if not provided
81
    )
82
    
83
    # Run the pipeline
84
    runner = PipelineRunner()
85
    await runner.run(task)
86
87
if __name__ == "__main__":
88
    import asyncio
89
    asyncio.run(main())

1
from opik import track
2
3
@track
4
def process_voice_input(transcription: str, conversation_id: str):
5
    """Process voice input and generate response."""
6
    
7
    # Your custom logic here
8
    response = f"Hello! I heard you say: {transcription}"
9
    
10
    return {
11
        "transcription": transcription,
12
        "response": response,
13
        "conversation_id": conversation_id,
14
        "timestamp": time.time()
15
    }
16
17
# Use in your Pipecat pipeline
18
async def handle_voice_input(frame):
19
    if isinstance(frame, TextFrame):
20
        result = process_voice_input(frame.text, conversation_id)
21
        # Process the result...

1
from opik.evaluation import evaluate
2
from opik.evaluation.metrics import Hallucination
3
4
# Define your evaluation task
5
def evaluation_task(x):
6
    return {
7
        "message": x["transcription"],
8
        "output": x["response"],
9
        "reference": x["expected_response"]
10
    }
11
12
# Create the Hallucination metric
13
hallucination_metric = Hallucination()
14
15
# Run the evaluation
16
evaluation_results = evaluate(
17
    experiment_name="pipecat-voice-agent-evaluation",
18
    dataset=conversation_data,
19
    task=evaluation_task,
20
    scoring_metrics=[hallucination_metric],
21
)

1
from opentelemetry import trace
2
from opentelemetry.trace import Status, StatusCode
3
4
tracer = trace.get_tracer(__name__)
5
6
class CustomVoiceService:
7
    @tracer.start_as_current_span("custom_voice_processing")
8
    async def process_audio(self, audio_data):
9
        with tracer.start_as_current_span("audio_analysis") as span:
10
            # Your custom processing logic
11
            span.set_attribute("audio.duration", len(audio_data))
12
            span.set_attribute("audio.format", "wav")
13
            
14
            result = await self.analyze_audio(audio_data)
15
            
16
            span.set_attribute("analysis.confidence", result.confidence)
17
            span.set_status(Status(StatusCode.OK))
18
            
19
            return result

1
from opik import track
2
3
@track
4
def maintain_conversation_context(transcription: str, context: dict):
5
    """Maintain conversation context across turns."""
6
    
7
    # Update context with new information
8
    context["last_user_input"] = transcription
9
    context["turn_count"] = context.get("turn_count", 0) + 1
10
    
11
    # Generate contextual response
12
    response = generate_contextual_response(transcription, context)
13
    
14
    return {
15
        "response": response,
16
        "updated_context": context,
17
        "turn_number": context["turn_count"]
18
    }