LiveKit Agents Integration via OpenTelemetry

1
import logging
2
3
from livekit.agents import (
4
    Agent,
5
    AgentSession,
6
    JobContext,
7
    RunContext,
8
    WorkerOptions,
9
    cli,
10
    metrics,
11
)
12
from livekit.agents.llm import function_tool
13
from livekit.agents.telemetry import set_tracer_provider
14
from livekit.agents.voice import MetricsCollectedEvent
15
from livekit.plugins import deepgram, openai, silero
16
from livekit.plugins.turn_detector.multilingual import MultilingualModel
17
from opentelemetry.util.types import AttributeValue
18
19
logger = logging.getLogger("basic-agent")
20
21
22
def setup_opik_tracing(metadata: dict[str, AttributeValue] | None = None):
23
    """Set up Opik tracing for LiveKit Agents"""
24
    from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
25
    from opentelemetry.sdk.trace import TracerProvider
26
    from opentelemetry.sdk.trace.export import BatchSpanProcessor
27
28
    # Set up the tracer provider
29
    trace_provider = TracerProvider()
30
    trace_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
31
    set_tracer_provider(trace_provider, metadata=metadata)
32
33
    return trace_provider
34
35
36
@function_tool
37
async def lookup_weather(context: RunContext, location: str) -> str:
38
    """Called when the user asks for weather related information.
39
40
    Args:
41
        location: The location they are asking for
42
    """
43
44
    logger.info(f"Looking up weather for {location}")
45
46
    return "sunny with a temperature of 70 degrees."
47
48
49
class Kelly(Agent):
50
    def __init__(self) -> None:
51
        super().__init__(
52
            instructions="Your name is Kelly.",
53
            llm=openai.LLM(model="gpt-4o-mini"),
54
            stt=deepgram.STT(model="nova-3", language="multi"),
55
            tts=openai.TTS(voice="ash"),
56
            turn_detection=MultilingualModel(),
57
            tools=[lookup_weather],
58
        )
59
60
    async def on_enter(self):
61
        logger.info("Kelly is entering the session")
62
        self.session.generate_reply()
63
64
    @function_tool
65
    async def transfer_to_alloy(self) -> Agent:
66
        """Transfer the call to Alloy."""
67
        logger.info("Transferring the call to Alloy")
68
        return Alloy()
69
70
71
class Alloy(Agent):
72
    def __init__(self) -> None:
73
        super().__init__(
74
            instructions="Your name is Alloy.",
75
            llm=openai.realtime.RealtimeModel(voice="alloy"),
76
            tools=[lookup_weather],
77
        )
78
79
    async def on_enter(self):
80
        logger.info("Alloy is entering the session")
81
        self.session.generate_reply()
82
83
    @function_tool
84
    async def transfer_to_kelly(self) -> Agent:
85
        """Transfer the call to Kelly."""
86
87
        logger.info("Transferring the call to Kelly")
88
        return Kelly()
89
90
91
async def entrypoint(ctx: JobContext):
92
    # set up the langfuse tracer
93
    trace_provider = setup_opik_tracing(
94
        # metadata will be set as attributes on all spans created by the tracer
95
        metadata={
96
            "livekit.session.id": ctx.room.name,
97
        }
98
    )
99
100
    # (optional) add a shutdown callback to flush the trace before process exit
101
    async def flush_trace():
102
        trace_provider.force_flush()
103
104
    ctx.add_shutdown_callback(flush_trace)
105
106
    session = AgentSession(vad=silero.VAD.load())
107
108
    @session.on("metrics_collected")
109
    def _on_metrics_collected(ev: MetricsCollectedEvent):
110
        metrics.log_metrics(ev.metrics)
111
112
    await session.start(agent=Kelly(), room=ctx.room)
113
114
115
if __name__ == "__main__":
116
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))