This is a Jupyter notebook

Trace ElevenLabs Agents with Langfuse

This notebook shows how to forward OpenTelemetry traces of ElevenLabs Agents conversations to Langfuse. You get each conversation as a full trace next to your other LLM application traces: transcript turns, LLM generations with token usage and cost, tool calls, and RAG retrievals. The notebook is self-contained: it creates a demo agent with a tool and a knowledge base, runs a text-only conversation, and forwards the resulting trace.

What is ElevenLabs Agents? ElevenLabs Agents is a platform for building voice-based conversational AI agents that handle calls, answer questions, and execute tools on behalf of users.

What is Langfuse? Langfuse is an open-source AI engineering platform that helps teams trace, debug, and evaluate their LLM applications.

Step 1: Install dependencies

%pip install elevenlabs requests -U

Step 2: Set up environment variables

Get your Langfuse keys from the project settings in Langfuse Cloud or set up self-hosting.

import os

# Get keys for your project from the project settings page: https://langfuse.com/cloud
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..."
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..."
os.environ["LANGFUSE_BASE_URL"] = "https://cloud.langfuse.com" # 🇪🇺 EU region (API host)
# os.environ["LANGFUSE_BASE_URL"] = "https://us.cloud.langfuse.com" # 🇺🇸 US region (API host)

os.environ["ELEVENLABS_API_KEY"] = "sk_..."  # ElevenLabs API key with Eleven Agents read + write permissions

Step 3: Create a demo agent

We create a small agent with two data sources, so the trace contains a real tool call and a real RAG flow:

a webhook tool that looks up currency exchange rates via the free Frankfurter API
a knowledge base document with a fee schedule, indexed for RAG

If you already have an agent, keep the client initialization (the next step needs it) and replace everything from document = ... onward with AGENT_ID = "agent_...".

import time

from elevenlabs import ElevenLabs

client = ElevenLabs(api_key=os.environ["ELEVENLABS_API_KEY"])

# Knowledge base document the agent answers fee questions from (via RAG)
document = client.conversational_ai.knowledge_base.documents.create_from_text(
    name="Demo Bank fee schedule",
    text=(
        "Demo Bank fee schedule.\n\n"
        "International wire transfers cost 25 EUR per transfer for standard delivery "
        "(2-3 business days) and 45 EUR for express delivery (same day).\n\n"
        "Domestic transfers are free of charge.\n\n"
        "Card replacement costs 10 EUR. Express card replacement costs 30 EUR."
    ),
)

# RAG indexing runs asynchronously, poll until the index is ready
while True:
    index = client.conversational_ai.knowledge_base.document.compute_rag_index(
        document.id, model="multilingual_e5_large_instruct"
    )
    if index.status == "succeeded":
        break
    if index.status == "failed":
        raise RuntimeError("RAG index computation failed")
    time.sleep(5)

agent = client.conversational_ai.agents.create(
    name="Langfuse Cookbook Demo Agent",
    conversation_config={
        "conversation": {"text_only": True},  # skip the audio pipeline so full agent text is stored
        "agent": {
            "first_message": "Hi! I can help you with exchange rates and Demo Bank fees. What would you like to know?",
            "prompt": {
                "prompt": (
                    "You are a helpful assistant for Demo Bank. Use the get_exchange_rate tool "
                    "for currency exchange rates and the knowledge base for questions about fees. "
                    "Keep answers short."
                ),
                "knowledge_base": [
                    {"type": "text", "id": document.id, "name": "Demo Bank fee schedule", "usage_mode": "auto"}
                ],
                "rag": {"enabled": True, "embedding_model": "multilingual_e5_large_instruct"},
                "tools": [
                    {
                        "type": "webhook",
                        "name": "get_exchange_rate",
                        "description": "Get the latest currency exchange rate between two currencies",
                        "api_schema": {
                            "url": "https://api.frankfurter.dev/v1/latest",
                            "method": "GET",
                            "query_params_schema": {
                                "properties": {
                                    "base": {"type": "string", "description": "Base currency code, e.g. USD"},
                                    "symbols": {"type": "string", "description": "Target currency code, e.g. EUR"},
                                },
                                "required": ["base", "symbols"],
                            },
                        },
                    }
                ],
            },
        }
    },
)
AGENT_ID = agent.agent_id
print(f"Created agent: {AGENT_ID}")

Step 4: Run a text-only conversation

ElevenLabs agents support chat mode: with text_only enabled on the agent, the SDK's Conversation runs without microphone or speakers. This creates a real conversation in your call history, exactly like a voice call.

Why text_only matters here: ElevenLabs stores what the agent actually said, not what the LLM generated. In voice mode, sending the next message while the agent is speaking interrupts the TTS, and the stored transcript (and trace) only contains the spoken part, e.g. "Hi!...". In production voice traces, a trailing ... on an agent message means the caller interrupted the agent.

import threading

from elevenlabs.conversational_ai.conversation import Conversation

agent_responded = threading.Event()


def on_agent_response(text: str) -> None:
    print(f"Agent: {text}")
    agent_responded.set()


conversation = Conversation(
    client=client,
    agent_id=AGENT_ID,
    requires_auth=True,
    callback_agent_response=on_agent_response,
)

conversation.start_session()
agent_responded.wait(timeout=15)  # wait for the agent's greeting

for message in [
    "What is the current exchange rate from US dollars to euros?",
    "How much does an international wire transfer cost?",
    "Great, thank you. Goodbye!",
]:
    agent_responded.clear()
    print(f"User: {message}")
    conversation.send_user_message(message)
    agent_responded.wait(timeout=45)

conversation.end_session()
conversation_id = conversation.wait_for_session_end()
print(f"Conversation ID: {conversation_id}")

Step 5: Fetch the conversation as an OpenTelemetry trace

ElevenLabs does not push traces to Langfuse. It hands you a complete trace as OTLP JSON (a resourceSpans object) on three surfaces, and you forward it:

Surface	When you get data	Best for
GET conversation API	On demand, after the conversation exists	Backfill, debugging (used in this notebook)
Post-call webhook	After the conversation ends	Production pipelines (see Step 9)
Monitoring WebSocket	Live, during the conversation	Live dashboards (Enterprise feature)

All three surfaces share the same trace ID per conversation and the same elevenlabs.* span attributes. We poll until ElevenLabs has processed the conversation, then request it in OpenTelemetry format.

import time

import requests

headers = {"xi-api-key": os.environ["ELEVENLABS_API_KEY"]}
url = f"https://api.elevenlabs.io/v1/convai/conversations/{conversation_id}"

# Wait until ElevenLabs has finished processing the conversation.
# Keep the regular conversation JSON, Step 7 needs its transcript.
for _ in range(30):
    conversation = requests.get(url, headers=headers).json()
    if conversation.get("status") == "done":
        break
    time.sleep(5)

response = requests.get(url, params={"format": "opentelemetry"}, headers=headers)
response.raise_for_status()
otlp_traces = response.json()["otlp_traces"]

span_names = [
    span["name"]
    for resource_span in otlp_traces["resourceSpans"]
    for scope_span in resource_span["scopeSpans"]
    for span in scope_span["spans"]
]
print(f"Fetched {len(span_names)} spans:")
print("\n".join(f"- {name}" for name in span_names))

Step 6: Enrich the spans with Langfuse attributes

ElevenLabs spans only carry elevenlabs.* attributes. Langfuse would ingest them as generic spans with empty input/output and all details buried in metadata. This transformation adds langfuse.* mapped attributes so the trace becomes readable and filterable:

The root conversation span becomes an agent observation with the first user message as input, the last agent response as output, and the call summary in the trace metadata.
The conversation ID becomes the Langfuse session ID, so application-side traces that know the conversation ID group together with the voice trace.
Agent turns backed by an LLM call (elevenlabs.producing_llm) become generation observations with the user message they respond to as input, plus model name, token usage, and cost parsed from elevenlabs.llm_usage.*. Evaluators see the full exchange, and cost dashboards work out of the box.
Tool call spans become tool observations with parameters as input and results as output.
RAG retrievals become retriever observations nested under the generation that used them, with the rewritten retrieval query as input plus embedding model and latency. ElevenLabs reports RAG as elevenlabs.rag.* attributes on the turn rather than a separate span, so the transformation synthesizes a child span with a deterministic ID. Step 7 adds the retrieved chunks as output.
Transcript turns get the user/agent text as observation input/output.
The ElevenLabs environment maps to the Langfuse environment, keeping test and production traces separate.
Empty agent turns (no text, no LLM call, no children) are dropped as noise.
Turn order is restored: ElevenLabs exports timestamps with one-second granularity, so spans within the same second would render in arbitrary order. Each span is nudged by its elevenlabs.turn.index in milliseconds.

All original elevenlabs.* attributes (latency metrics, analysis results, dynamic variables) stay available in each observation's metadata. The mapping follows the trace best practices. If your application knows the end user, also map a user identifier to langfuse.user.id (see the commented line in the code).

import ast
import hashlib
import json


def _get(attributes, key):
    for attribute in attributes or []:
        if attribute["key"] == key:
            return next(iter(attribute["value"].values()), None)
    return None


def _set(attributes, key, value):
    if value is not None:
        attributes.append({"key": key, "value": {"stringValue": str(value)}})


def _llm_usage(attributes):
    """Parse elevenlabs.llm_usage.<model>.<kind> attributes into usage/cost dicts."""
    usage, cost = {}, {}
    for attribute in attributes:
        key = attribute["key"]
        if not key.startswith("elevenlabs.llm_usage."):
            continue
        kind = key.rsplit(".", 1)[-1]  # model names contain dots, kind is the last segment
        kind = {"output_total": "output"}.get(kind, kind)
        entry = ast.literal_eval(next(iter(attribute["value"].values())))
        if entry.get("tokens"):
            usage[kind] = entry["tokens"]
        if entry.get("price"):
            cost[kind] = entry["price"]
    if cost:
        cost["total"] = round(sum(cost.values()), 10)
    return usage, cost


def _rag_retrieval_span(parent):
    """Synthesize a retriever child span from the elevenlabs.rag.* attributes
    that ElevenLabs sets on the turn that used retrieval. The span ID is derived
    deterministically from the parent, so re-sending the same payload stays idempotent."""
    attributes = parent["attributes"]
    start = int(parent["startTimeUnixNano"])
    latency_ms = float(_get(attributes, "elevenlabs.rag.latency_ms") or 0)
    span = {
        "traceId": parent["traceId"],
        "spanId": hashlib.sha256(f"rag-{parent['spanId']}".encode()).hexdigest()[:16],
        "parentSpanId": parent["spanId"],
        "name": "elevenlabs.rag.retrieve",
        "kind": parent.get("kind", 1),
        "startTimeUnixNano": str(start),
        "endTimeUnixNano": str(start + int(latency_ms * 1_000_000)),
        "attributes": [],
    }
    _set(span["attributes"], "langfuse.observation.type", "retriever")
    _set(span["attributes"], "langfuse.observation.input", _get(attributes, "elevenlabs.rag.retrieval_query"))
    for key in ("elevenlabs.rag.embedding_model", "elevenlabs.rag.latency_ms"):
        _set(span["attributes"], key, _get(attributes, key))
    return span


def enrich_for_langfuse(otlp_traces: dict) -> dict:
    """Add langfuse.* attributes to ElevenLabs spans so they map to typed,
    readable observations in Langfuse. Modifies the payload in place."""
    for resource_span in otlp_traces.get("resourceSpans", []):
        resource_attributes = resource_span.get("resource", {}).get("attributes", [])
        conversation_id = _get(resource_attributes, "elevenlabs.conversation_id")

        for scope_span in resource_span.get("scopeSpans", []):
            spans = scope_span.get("spans", [])

            # ElevenLabs exports timestamps with one-second granularity, so spans of the
            # same second would render in arbitrary order. Nudge each span by its turn
            # index in milliseconds to restore the true conversational order.
            offsets = {
                span["spanId"]: int(_get(span.get("attributes"), "elevenlabs.turn.index")) * 1_000_000
                for span in spans
                if _get(span.get("attributes"), "elevenlabs.turn.index") is not None
            }
            for span in spans:
                offset = offsets.get(span["spanId"], offsets.get(span.get("parentSpanId"), 0))
                span["startTimeUnixNano"] = str(int(span["startTimeUnixNano"]) + offset)
                span["endTimeUnixNano"] = str(int(span["endTimeUnixNano"]) + offset)

            spans = sorted(spans, key=lambda span: int(span.get("startTimeUnixNano", 0)))
            root = next((s for s in spans if s.get("name") == "elevenlabs.conversation"), None)
            environment = _get(root.get("attributes") if root else [], "elevenlabs.environment")
            parent_ids = {span.get("parentSpanId") for span in spans}

            user_texts, agent_texts, kept = [], [], []
            for span in spans:
                attributes = span.setdefault("attributes", [])
                name = span.get("name", "")
                _set(attributes, "langfuse.session.id", conversation_id)
                _set(attributes, "langfuse.environment", environment)

                if name.startswith("elevenlabs.tool."):
                    _set(attributes, "langfuse.observation.type", "tool")
                    _set(attributes, "langfuse.observation.input", _get(attributes, "elevenlabs.tool.params"))
                    _set(attributes, "langfuse.observation.output", _get(attributes, "elevenlabs.tool.result"))
                elif name.endswith(".user_transcript"):
                    text = _get(attributes, "elevenlabs.user.text")
                    _set(attributes, "langfuse.observation.input", text)
                    if text:
                        user_texts.append(text)
                elif name.endswith(".agent_response"):
                    text = _get(attributes, "elevenlabs.agent.text")
                    model = _get(attributes, "elevenlabs.producing_llm")
                    # Drop empty agent turns (no text, no LLM call, no children)
                    if not text and not model and span.get("spanId") not in parent_ids:
                        continue
                    _set(attributes, "langfuse.observation.output", text)
                    if text:
                        agent_texts.append(text)
                    if model:
                        usage, cost = _llm_usage(attributes)
                        _set(attributes, "langfuse.observation.type", "generation")
                        _set(attributes, "langfuse.observation.model.name", model)
                        # The user message this turn responds to, so evaluators see the full exchange
                        if user_texts:
                            _set(attributes, "langfuse.observation.input", user_texts[-1])
                        _set(attributes, "langfuse.observation.usage_details", json.dumps(usage) if usage else None)
                        _set(attributes, "langfuse.observation.cost_details", json.dumps(cost) if cost else None)
                        if _get(attributes, "elevenlabs.rag.retrieval_query") is not None:
                            rag_span = _rag_retrieval_span(span)
                            _set(rag_span["attributes"], "langfuse.session.id", conversation_id)
                            _set(rag_span["attributes"], "langfuse.environment", environment)
                            kept.append(rag_span)
                kept.append(span)

            scope_span["spans"] = kept

            if root is not None:
                attributes = root["attributes"]
                _set(attributes, "langfuse.observation.type", "agent")
                _set(attributes, "langfuse.trace.metadata.elevenlabs_agent_id", _get(attributes, "elevenlabs.agent_id"))
                _set(attributes, "langfuse.trace.metadata.call_summary", _get(attributes, "elevenlabs.analysis.transcript_summary"))
                # If you pass a user identifier to your agent (e.g. a `user_id` dynamic
                # variable), map it so Langfuse can group traces per user:
                # _set(attributes, "langfuse.user.id", _get(attributes, "elevenlabs.dynamic_variable.user_id"))
                if user_texts:
                    _set(attributes, "langfuse.observation.input", user_texts[0])
                if agent_texts:
                    _set(attributes, "langfuse.observation.output", agent_texts[-1])

    return otlp_traces

Step 7: Add retrieved chunks to the retriever observations

The OTLP export records that retrieval happened (query, embedding model, latency) but not what was retrieved. The conversation JSON from Step 5 has the missing piece: each agent turn carries rag_retrieval_info with the retrieved chunk IDs and vector distances, and the knowledge base API returns the chunk text.

This step joins them into the retriever observations, so the retrieved text becomes the observation output. You can then check the agent's answer against the exact text it retrieved, e.g. with an LLM-as-a-judge evaluator for faithfulness.

def attach_rag_chunks(otlp_traces: dict, conversation: dict, client: ElevenLabs) -> dict:
    """Set the retrieved chunks as output on the synthesized retriever spans.
    Chunk IDs and distances come from the transcript's rag_retrieval_info,
    chunk text from the knowledge base API. Run after enrich_for_langfuse."""
    rag_by_turn = {
        index: turn["rag_retrieval_info"]
        for index, turn in enumerate(conversation.get("transcript", []))
        if turn.get("rag_retrieval_info")
    }
    chunk_texts = {}  # (document_id, chunk_id) -> content, fetched once

    for resource_span in otlp_traces.get("resourceSpans", []):
        for scope_span in resource_span.get("scopeSpans", []):
            spans = scope_span.get("spans", [])
            turn_by_span_id = {
                span["spanId"]: _get(span.get("attributes"), "elevenlabs.turn.index")
                for span in spans
            }
            for span in spans:
                if span.get("name") != "elevenlabs.rag.retrieve":
                    continue
                turn_index = turn_by_span_id.get(span.get("parentSpanId"))
                info = rag_by_turn.get(int(turn_index)) if turn_index is not None else None
                if not info:
                    continue
                documents = []
                for chunk in info["chunks"]:
                    key = (chunk["document_id"], chunk["chunk_id"])
                    if key not in chunk_texts:
                        # Chunks are stored per RAG index, so request the one the query used
                        chunk_texts[key] = client.conversational_ai.knowledge_base.documents.chunk.get(
                            *key, embedding_model=info["embedding_model"]
                        ).content
                    documents.append({**chunk, "content": chunk_texts[key]})
                _set(span["attributes"], "langfuse.observation.output", json.dumps(documents))

    return otlp_traces

Step 8: Forward the trace to Langfuse

Langfuse accepts OTLP over HTTP in JSON format, so the payload can be sent as-is with a single POST request. No OpenTelemetry SDK or protobuf encoding needed. The x-langfuse-ingestion-version: 4 header enables real-time ingestion on the v4 data model.

import base64

LANGFUSE_AUTH = base64.b64encode(
    f"{os.environ['LANGFUSE_PUBLIC_KEY']}:{os.environ['LANGFUSE_SECRET_KEY']}".encode()
).decode()


def send_to_langfuse(otlp_traces: dict) -> None:
    response = requests.post(
        f"{os.environ['LANGFUSE_BASE_URL']}/api/public/otel/v1/traces",
        headers={
            "Authorization": f"Basic {LANGFUSE_AUTH}",
            "Content-Type": "application/json",
            "x-langfuse-ingestion-version": "4",
        },
        data=json.dumps(otlp_traces),
    )
    response.raise_for_status()


enriched = attach_rag_chunks(enrich_for_langfuse(otlp_traces), conversation, client)
send_to_langfuse(enriched)
print("Forwarded conversation to Langfuse")

Step 9: Production setup with a post-call webhook

In production, don't poll the GET API. Configure an ElevenLabs post-call webhook with the Transcript event and OpenTelemetry transcript payloads enabled (in Agents settings). ElevenLabs then POSTs a post_call_transcription_otel event with the same otlp_traces object after every completed call.

A minimal receiver that verifies the webhook signature and reuses enrich_for_langfuse and send_to_langfuse from this notebook:

import hashlib, hmac, json, os, time

from fastapi import FastAPI, Header, HTTPException, Request

app = FastAPI()


@app.post("/elevenlabs/post-call")
async def post_call(request: Request, elevenlabs_signature: str = Header(...)):
    body = await request.body()

    # Verify the HMAC signature: "t={unix},v0={hmac}" over "{timestamp}.{body}"
    parts = dict(part.split("=", 1) for part in elevenlabs_signature.split(","))
    if int(parts["t"]) < time.time() - 30 * 60:  # reject replayed deliveries
        raise HTTPException(status_code=403, detail="Request expired")
    expected = hmac.new(
        os.environ["ELEVENLABS_WEBHOOK_SECRET"].encode(),
        f"{parts['t']}.{body.decode()}".encode(),
        hashlib.sha256,
    ).hexdigest()
    if not hmac.compare_digest(parts["v0"], expected):
        raise HTTPException(status_code=401, detail="Invalid signature")

    payload = json.loads(body)
    if payload.get("type") == "post_call_transcription_otel":
        send_to_langfuse(enrich_for_langfuse(payload["data"]["otlp_traces"]))
    return {"ok": True}

To include the retrieved chunks (Step 7) in production, fetch the transcript once via the GET conversation API when the event arrives and run attach_rag_chunks before send_to_langfuse. (The webhook's regular post_call_transcription event also carries the transcript with rag_retrieval_info, but it is a separate delivery, so joining it with the OTel event requires buffering one of the two.)

Forward each conversation exactly once. The trace ID is stable per conversation, but ElevenLabs generates fresh span IDs on every export. Re-exporting a conversation via the GET API and forwarding it again duplicates all observations inside the same trace. Re-sending the same payload (e.g. your own retry around a failed POST) is idempotent. If you need retries, persist the exported payload and re-send that exact payload.

For live use cases (supervisor dashboards, real-time alerting), the monitoring WebSocket streams the same OTLP JSON span batches during the call. Forward each batch with the same send_to_langfuse function. Requires an ElevenLabs Enterprise workspace.

Limitations

Not a live push integration: traces arrive after the call ends (or during the call via the Enterprise-only monitoring WebSocket).
No trace context propagation: ElevenLabs mints its own trace IDs, so the conversation cannot be nested inside an existing application trace. Use the conversation ID (mapped to the Langfuse session ID above) to correlate application-side traces with the voice conversation.
LLM cost only: per-turn token usage and cost cover the LLM calls. The total conversation cost including voice (TTS/ASR) is in the elevenlabs.cost_fiat attribute in the root observation's metadata.
4 KB truncation: ElevenLabs truncates long tool parameters and results at 4 KB per span attribute.
RAG chunk contents require a join: the OTLP trace only carries the retrieval query, embedding model, and latency. Step 7 joins the retrieved chunk IDs and distances from the conversation transcript and the chunk text from the knowledge base API.

Step 10: View the trace in Langfuse

Open Langfuse Cloud to see the full conversation trace: the conversation as the root agent observation, transcript turns, LLM generations with token usage and cost, the tool call with its parameters and result, and the RAG retrievals with their queries and retrieved chunks. The conversation ID is the Langfuse session ID, so all traces of the same conversation group together in the Sessions view.

Example trace in Langfuse

Was this page helpful?

On this page