livekit-evals 0.2.13

Track and evaluate LiveKit agent sessions with automatic metrics, transcripts, and usage analytics

LiveKit Evals

Track and evaluate your LiveKit voice AI agents with just 3 lines of code.

Automatically capture transcripts, usage metrics, latency data, and session analytics from your LiveKit agents. Perfect for monitoring, debugging, and optimizing your voice AI applications.

✨ Features

• 🎯 3-Line Integration - Add to any LiveKit agent in seconds
• 📝 Precise Transcripts - Accurate timing using VAD state change events
• 📊 Usage Metrics - Track LLM tokens, STT duration, TTS characters
• ⚡ Latency Tracking - Monitor LLM, STT, and TTS performance
• 🔍 Auto-Detection - Automatically extracts models, providers, and configuration
• 📞 SIP Support - Detects SIP trunking and phone numbers
• 🎥 Recording URLs - Captures egress recording links
• 🎙️ Call Recordings - Automatic call recording to S3 (MP3 format, enabled by default, no S3 config needed)
• 🔊 Stereo Recording - Dual-channel recording with agent on left, caller on right (one param)
• 🗂️ Custom Data - Attach arbitrary JSON to every webhook payload via custom_data
• 🔐 Secure - API key authentication; temporary S3 credentials fetched per-session

🚀 Quick Start

Prerequisites

1. Get your API key from https://app.superbryn.com/api-keys
2. Set environment variable:

   export SUPERBRYN_API_KEY=your_api_key_here
   

Installation

pip install livekit-evals

Integration (3 Lines)

Add these lines to your LiveKit agent:

from livekit_evals import create_webhook_handler

async def entrypoint(ctx: JobContext):
    # ... your existing setup code ...
    
    # 1. Create webhook handler (recording enabled by default)
    webhook_handler = create_webhook_handler(
        room=ctx.room,
        is_deployed_on_lk_cloud=True,  # Set to False if self-hosting
        # disable_recording=True  # Uncomment to disable call recording
    )
    
    # ... create your session ...
    session = AgentSession(
        llm=openai.LLM(model="gpt-4o-mini"),
        stt=deepgram.STT(model="nova-3"),
        tts=cartesia.TTS(voice="..."),
    )
    
    # ... your session setup ...
    await session.start(agent=YourAgent(), room=ctx.room)
    
    # 2. Attach to session (MUST be after session.start)
    if webhook_handler:
        webhook_handler.attach_to_session(session)
        # 3. Send webhook on shutdown
        ctx.add_shutdown_callback(webhook_handler.send_webhook)
    
    await ctx.connect()

That's it! 🎉 Your agent will now automatically track all session data and send it to your webhook endpoint.

📖 Full Example

Here's a complete working example:

import logging
from dotenv import load_dotenv
from livekit.agents import (
    Agent,
    AgentSession,
    JobContext,
    WorkerOptions,
    cli,
)
from livekit.plugins import cartesia, deepgram, openai, silero

# Import livekit-evals
from livekit_evals import create_webhook_handler

logger = logging.getLogger("agent")
load_dotenv()


class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="""You are a helpful voice AI assistant.
            You eagerly assist users with their questions.""",
        )


async def entrypoint(ctx: JobContext):
    # Logging setup
    ctx.log_context_fields = {"room": ctx.room.name}

    # Initialize webhook handler (auto-detects all metadata)
    webhook_handler = create_webhook_handler(
        room=ctx.room,
        is_deployed_on_lk_cloud=True  # Set to False if self-hosting
    )

    # Set up voice AI pipeline
    session = AgentSession(
        llm=openai.LLM(model="gpt-4o-mini"),
        stt=deepgram.STT(model="nova-3", language="en"),
        tts=cartesia.TTS(voice="your-voice-id"),
        vad=silero.VAD.load(),
    )

    # Start the session
    await session.start(agent=Assistant(), room=ctx.room)

    # Attach webhook handler to capture events
    # IMPORTANT: Must be after session.start()
    if webhook_handler:
        webhook_handler.attach_to_session(session)
        ctx.add_shutdown_callback(webhook_handler.send_webhook)

    # Connect to room
    await ctx.connect()


if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))

🔧 Configuration

Environment Variables

Variable	Required	Description	Default
SUPERBRYN_API_KEY	✅ Yes	API key for webhook authentication and call recording	-
LIVEKIT_PROJECT_ID	⚪ Optional	LiveKit project ID	Auto-detected from LIVEKIT_URL
AGENT_ID	⚪ Optional	Unique agent identifier	"livekit-agent"
VERSION_ID	⚪ Optional	Agent version identifier	"v1"

Note: Call recording is enabled by default. Temporary S3 credentials are fetched automatically using your SUPERBRYN_API_KEY -- no S3 configuration needed.

Setting Environment Variables

Linux/Mac:

export SUPERBRYN_API_KEY=your_api_key_here

Windows (CMD):

set SUPERBRYN_API_KEY=your_api_key_here

Windows (PowerShell):

$env:SUPERBRYN_API_KEY="your_api_key_here"

Docker:

docker run -e SUPERBRYN_API_KEY=your_api_key_here ...

.env file:

SUPERBRYN_API_KEY=your_api_key_here
LIVEKIT_PROJECT_ID=my-project-id
AGENT_ID=my-agent
VERSION_ID=v1.0.0

📊 What Gets Tracked

Transcript Data

• Precise timing using VAD state change events
• Speaker turns (user/assistant)
• Start/end timestamps (ISO 8601)
• Start/end times in milliseconds (relative to call start)
• Response delays between turns
• Interruption detection
• Confidence scores (when available)
• Language detection
• Speaker IDs

Tool Calls

• Function name and raw JSON arguments for every tool invoked by the agent
• Tool result (output string) paired with each call
• Error flag (is_error) when the tool execution raised an exception
• Timing — start_ms (call dispatched) and end_ms (result received), both as ms offsets from call start

Usage Metrics

• LLM: Input tokens, output tokens, total tokens, model, provider
• STT: Audio duration, model, provider
• TTS: Character count, audio duration, model, provider, voice ID

Latency Metrics

• LLM: Time to first token (TTFT), total duration
• STT: Processing duration
• TTS: Time to first byte (TTFB), total duration
• Aggregated: Average latencies per component

Session Metadata

• Agent ID and version
• LiveKit project ID
• System prompt
• Call duration
• Phone number (if SIP call)
• SIP trunking detection
• Egress recording URLs
• LiveKit Cloud deployment status

🔍 How It Works

1. Event Listening: Attaches to LiveKit session events (user_state_changed, agent_state_changed, conversation_item_added) and to the per-plugin metrics_collected events on STT/LLM/TTS (the non-deprecated metrics surface), with session_usage_updated as a fallback for realtime models
2. Data Aggregation: Collects and processes events during the session
3. Auto-Detection: Extracts configuration from session objects
4. Webhook Delivery: Sends comprehensive payload to webhook endpoint when session ends

Webhook Payload Format

{
  "event": "call.ended",
  "call": {
    "id": "room-name",
    "room_name": "room-name",
    "participant_identity": "user-123",
    "started_at": "2025-10-19T12:00:00.000Z",
    "ended_at": "2025-10-19T12:05:30.000Z",
    "duration_seconds": 330,
    "transcript": {
      "turns": [
        {
          "speaker": "user",
          "text": "Hello, how are you?",
          "timestamp": "2025-10-19T12:00:05.000Z",
          "start_timestamp": "2025-10-19T12:00:05.000Z",
          "end_timestamp": "2025-10-19T12:00:07.000Z",
          "start_time_ms": 5000,
          "end_time_ms": 7000,
          "interrupted": false,
          "confidence_score": 0.98,
          "language": "en"
        },
        {
          "speaker": "assistant",
          "text": "I'm doing great, thanks for asking!",
          "timestamp": "2025-10-19T12:00:08.000Z",
          "start_timestamp": "2025-10-19T12:00:08.000Z",
          "end_timestamp": "2025-10-19T12:00:11.000Z",
          "start_time_ms": 8000,
          "end_time_ms": 11000,
          "response_delay_ms": 1000,
          "interrupted": false
        }
      ]
    },
    "tool_calls": [
      {
        "id": "call_abc123",
        "function_name": "book_appointment",
        "arguments": "{\"date\": \"2026-06-20\", \"time\": \"10:00\"}",
        "result": "{\"confirmation_id\": \"APT-9001\", \"status\": \"confirmed\"}",
        "is_error": false,
        "start_ms": 12500,
        "end_ms": 13800,
        "timestamp_ms": 12500
      }
    ],
    "recording_url": "https://...",
    "stereo_recording_url": "https://...",
    "metadata": {
      "agent_id": "my-agent",
      "livekit_project_id": "my-project",
      "llm_model": "gpt-4o-mini",
      "llm_provider": "openai",
      "stt_model": "nova-3",
      "stt_provider": "deepgram",
      "tts_model": "sonic-english",
      "tts_provider": "cartesia",
      "tts_voice_id": "...",
      "system_prompt": "You are a helpful assistant...",
      "sip_trunking_enabled": false,
      "egress_enabled": true,
      "lk_agent_enabled": true,
      "phone_number": null
    },
    "usage": {
      "llm_model": "gpt-4o-mini",
      "llm_provider": "openai",
      "llm_input_tokens": 1250,
      "llm_output_tokens": 850,
      "llm_total_tokens": 2100,
      "stt_provider": "deepgram",
      "stt_model": "nova-3",
      "stt_duration_seconds": 45.2,
      "audio_duration_seconds": 45.2,
      "tts_provider": "cartesia",
      "tts_model": "sonic-english",
      "tts_characters": 1200,
      "tts_audio_duration_seconds": 42.5
    },
    "latency": {
      "llm_ms": 450.5,
      "stt_ms": 120.3,
      "tts_ms": 180.7,
      "total_ms": 751.5
    },
    "custom_data": {
      "ticket_id": "TKT-9001",
      "customer_tier": "enterprise"
    }
  }
}

🛠️ Advanced Usage

Custom Data

Attach any JSON-serializable fields to the webhook payload using custom_data. These are forwarded verbatim in payload["call"]["custom_data"] and are never interpreted by the package — they're purely for your own downstream use.

At creation time (data known at session startup):

webhook_handler = create_webhook_handler(
    room=ctx.room,
    is_deployed_on_lk_cloud=True,
    custom_data={
        "ticket_id": "TKT-9001",
        "customer_tier": "enterprise",
        "lead_source": "website",
    },
)

During the session (data discovered at runtime, e.g. after a tool call):

# Replace the entire dict
webhook_handler.set_custom_data({"resolved": True, "resolution_code": "answered"})

# Or merge additional keys while keeping existing ones
webhook_handler.update_custom_data({"appointment_booked": True, "slot": "2026-06-05T10:00"})

Both methods can be called at any point before send_webhook() fires on shutdown.

Custom API Key

Pass API key directly instead of using environment variable:

webhook_handler = create_webhook_handler(
    room=ctx.room,
    is_deployed_on_lk_cloud=True,
    api_key="your_api_key_here"
)

Custom LiveKit Project ID

webhook_handler = create_webhook_handler(
    room=ctx.room,
    is_deployed_on_lk_cloud=True,
    livekit_project_id="my-custom-project-id"
)

Self-Hosted Agents

If you're self-hosting your LiveKit agents (not using LiveKit Cloud):

webhook_handler = create_webhook_handler(
    room=ctx.room,
    is_deployed_on_lk_cloud=False  # Important for cost calculation
)

Custom Telephony Rates

If you're using custom telephony providers (Twilio, Vonage, etc.) with specific per-minute rates:

webhook_handler = create_webhook_handler(
    room=ctx.room,
    is_deployed_on_lk_cloud=True,
    call_rate_usd=0.015  # Your custom rate per minute ($/min)
)

This overrides default provider costs and ensures accurate cost tracking for your telephony usage.

Call Recording (Enabled by Default)

Call recording is automatically enabled. Recordings are:

• ✅ MP3 format (universal compatibility)
• ✅ Publicly accessible via direct URL
• ✅ Secured with short-lived credentials (30-minute expiry, scoped per session)
• ✅ Automatically included in webhook payload

No S3 keys, buckets, or regions need to be configured -- the package fetches temporary upload credentials from SuperBryn's credentials service using your SUPERBRYN_API_KEY.

Recording URLs are included in the webhook payload:

{
  "call": {
    "recording_url": "https://superbryn-call-recordings.s3.ap-south-1.amazonaws.com/call_recordings/[phone]/20250106-153045/call.mp3"
  }
}

To disable recording:

webhook_handler = create_webhook_handler(
    room=ctx.room,
    is_deployed_on_lk_cloud=True,
    disable_recording=True  # Disable call recording
)

Bring Your Own Egress (External Recording URL)

If you already run your own egress, you can disable SuperBryn's recording and supply your own recording URL instead. Disable our egress with disable_recording=True, then call set_external_recording_url() once your recording is ready (any time before the webhook fires on shutdown):

webhook_handler = create_webhook_handler(
    room=ctx.room,
    is_deployed_on_lk_cloud=True,
    disable_recording=True,  # don't start SuperBryn's egress
)

# ... later, once your own egress has produced a file ...
reachable = await webhook_handler.set_external_recording_url(
    "https://my-bucket.s3.amazonaws.com/calls/room-123/call.mp3?X-Amz-Signature=...",
    # probe=False,  # skip the call-time reachability check
)

Only public or pre-signed URLs are supported. When you call this method the URL is validated at call time with a lightweight ranged GET (Range: bytes=0-0):

• 200/206 → reachable; the URL is marked usable for mirroring.
• 401/403 → private object or bad/expired signature — a clear error is logged and mirroring should be skipped.
• 404 → object not uploaded yet or wrong path (a timing/path issue).

A ranged GET is used instead of HEAD because S3 pre-signed URLs are signed for a single HTTP method — a HEAD on a GET-signed URL returns 403 and would falsely look private. Note the check reflects reachability at call time; a pre-signed URL can still expire before a later (e.g. server-side) mirror runs, so sign for a long-enough TTL or mirror promptly.

The webhook payload gains two fields so the consumer can decide whether to mirror:

{
  "call": {
    "recording_url": "https://my-bucket.s3.amazonaws.com/.../call.mp3?...",
    "recording_url_source": "external",     // "superbryn_s3" for the managed flow
    "recording_url_reachable": true          // call-time probe result (null if not probed)
  }
}

For the default managed flow these are "superbryn_s3" / true — the file is already in SuperBryn's bucket and needs no mirroring.

Stereo Recording (Dual-Channel)

Record in dual-channel stereo where the agent is on the left channel and all other participants (caller/SIP) are on the right channel. This is useful for separate-speaker transcription and analysis.

Stereo is opt-in — the default is mono. Enabling it is a one-line change: add stereo_recording=True. There is nothing else to configure or install — it reuses the exact same recording pipeline (egress + S3 credentials), just switching the egress to DUAL_CHANNEL_AGENT mixing instead of a mono mix.

webhook_handler = create_webhook_handler(
    room=ctx.room,
    is_deployed_on_lk_cloud=True,
    stereo_recording=True  # L=agent, R=caller
)

Prerequisites: the only requirement is that recording already works for you (i.e. you currently get a recording_url in your webhook payload). That requires:

• SUPERBRYN_API_KEY set (used to fetch temporary S3 credentials — no S3 config needed)
• LiveKit Egress available on your deployment (enabled by default on LiveKit Cloud; the egress service must be running if self-hosted)
• Standard LiveKit env vars (LIVEKIT_URL, LIVEKIT_API_KEY, LIVEKIT_API_SECRET)

If recording works today, stereo_recording=True is the only change needed — no version bump, no new dependency. stereo_recording=True also implies recording is enabled (it overrides disable_recording).

When stereo is enabled, both recording_url and stereo_recording_url are populated in the webhook payload. Note: a single MP3 file is written — the stereo separation lives in the file's two channels, so both fields point to the same URL:

{
  "call": {
    "recording_url": "https://...call.mp3",
    "stereo_recording_url": "https://...call.mp3"
  }
}

If you need to stop the recording before deleting the room (e.g. in a graceful shutdown), call stop_egress() to ensure the file is finalized on S3:

await webhook_handler.stop_egress()  # finalize recording before room deletion

Semantic Call End Reasons

If your agent knows the business reason for ending a call, set it explicitly on the WebhookHandler before closing the room. This helps preserve reasons such as transfer_to_human, conversation_complete, caller_hung_up, or no_answer_timeout in the final webhook payload.

Without this, LiveKit may only emit a generic close reason like "participant left" or "session closed".

# Example: transfer to human
webhook_handler.set_call_end_reason("transfer_to_human")
await webhook_handler.stop_egress()
await ctx.api.room.delete_room(...)

You can use any short snake_case reason string that fits your application.

Common examples:

• conversation_complete
• purpose_achieved
• transfer_to_human
• caller_hung_up
• main_agent_hung_up
• no_answer_timeout
• silence_timeout
• duration_limit

set_call_end_reason() is most useful when your application logic decides why the call is ending. For example:

• an end_call tool is invoked by the agent
• your app triggers a transfer to a human
• you enforce a silence timeout or no-answer timeout
• you intentionally delete the room during graceful shutdown

🐛 Troubleshooting

Webhook Not Sending

Check API Key:

echo $SUPERBRYN_API_KEY

Enable Debug Logging:

import logging
logging.basicConfig(level=logging.DEBUG)

Look for these log messages:

• SUPERBRYN_WEBHOOK_HANDLER_CREATED - Handler initialized
• SUPERBRYN_WEBHOOK_SENT - Webhook delivered successfully
• SUPERBRYN_WEBHOOK_UNAUTHORIZED - Invalid API key
• SUPERBRYN_WEBHOOK_FAILED - Delivery failed
• SUPERBRYN_WEBHOOK_ERROR - Exception occurred

Common Errors

Error	Cause	Solution
SUPERBRYN_API_KEY not configured	Missing API key	Set SUPERBRYN_API_KEY environment variable
SUPERBRYN_WEBHOOK_UNAUTHORIZED	Invalid API key	Verify your API key is correct
SUPERBRYN_WEBHOOK_FORBIDDEN	Expired/disabled key	Generate a new API key
No empty turn found to fill	State change timing issue	Usually harmless, check logs for patterns

Missing Transcript Data

Ensure webhook_handler.attach_to_session(session) is called:

• ✅ After await session.start()
• ✅ At the end of your entrypoint (no early returns)

Provider Detection Issues

The package auto-detects providers from model names. Supported providers (25+):

LLM Providers:

• OpenAI (gpt, whisper, tts-1, o1, o3)
• Anthropic (claude)
• Google (gemini, palm, bard, gemma)
• Meta (llama, meta-llama)
• Mistral (mistral, mixtral)
• Cohere (cohere, command)
• Perplexity (perplexity, pplx)
• Groq
• Together AI (together, togethercomputer)
• Replicate
• Hugging Face (huggingface, hf-)

TTS Providers:

• ElevenLabs (eleven, elevenlabs)
• Cartesia (cartesia, sonic)
• PlayHT (playht, play.ht)
• Resemble AI (resemble, resembleai)
• Murf (murf, murf.ai)
• WellSaid Labs (wellsaid, wellsaidlabs)
• Speechify
• Sarvam (saarika, sarvam, bulbul)
• Azure/Microsoft (azure, microsoft)
• AWS Polly (aws, polly, amazon)
• Google Cloud (gcloud, google-cloud)

STT Providers:

• Deepgram (deepgram, nova, aura)
• AssemblyAI (assemblyai, assembly)
• Rev.ai (rev.ai, revai)
• Speechmatics
• Gladia

Realtime/Multi-modal:

• LiveKit
• Twilio
• Vonage

If your provider isn't detected, it will show as "unknown" but won't affect functionality.

📝 Migration Guide

If you're currently using the standalone webhook_handler.py:

Before:

from webhook_handler import create_webhook_handler

After:

from livekit_evals import create_webhook_handler

Everything else stays the same! The API is identical.

🤝 Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

🔗 Links

• LiveKit Agents Documentation
• GitHub Repository
• Issue Tracker
• Get API Key

💡 Support

• 📧 Email: [e-mail]
• 💬 GitHub Issues: Report a bug
• 📚 Documentation: README

---

Made with ❤️ by SuperBryn