can i add lang smith to trace the costs of my call if i am using a realtime model?

Thanks for reaching out, Youssef. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

Langfuse Tracing

Thanks — I’ll walk through how you can use Langfuse tracing to track costs when using a realtime model with LiveKit, and what I need from you if you want concrete code for your setup. Acknowledging your request • You asked: “Can I add Langfuse tracing to trace the costs of my call if I am using a realtime model?” • Short answer: Yes. You can export OpenTelemetry traces from LiveKit Agents (including realtime model usage) to Langfuse and emit the usage metrics needed to compute costs per call or per session. What you can do (high level) • Enable OpenTelemetry tracing in your Agent/AgentSession and configure an OTLP exporter pointed to Langfuse (cloud or self-hosted). • Capture the model usage metrics LiveKit emits (input_tokens, output_tokens for LLMs; transcription duration for STT; character counts for TTS) from the

metrics_collected

events. • Compute cost per call using the model/service pricing (e.g., token or per-minute/per-character rates) and attach that cost as attributes on the relevant spans or as separate trace events. • Add

user_id

and

session_id

attributes to spans so traces in Langfuse can be correlated with your LiveKit sessions/users. Concrete pieces you’ll typically implement 1. Configure an OTLP exporter that sends traces to Langfuse. ◦ Use the Langfuse traces endpoint and credentials when you create your OTLPSpanExporter/OTLP exporter in your language of choice. 1. Instrument your session to consume LiveKit’s metrics events and compute cost: ◦ Listen for

metrics_collected

(or similar) events from the AgentSession. ◦ Extract

input_tokens

,

output_tokens

,

transcription_duration

,

character_count

, etc. ◦ Apply your pricing formula (tokens → $), then set attributes on the current span: ▪︎ e.g.,

llm.input_cost

,

llm.output_cost

,

llm.total_cost

1. Add identifying attributes to spans: ◦ Inject

user_id

and

session_id

into spans so Langfuse dashboards and queries can group by user/session. ◦ You can add a custom span processor that sets these attributes on span start. Example (summary of approach) • OTLP exporter -> Langfuse endpoint • On `metrics_collected`: ◦ input_tokens, output_tokens -> cost = input_tokens * input_price/1e6 + output_tokens * output_price/1e6 ◦ span.set_attribute("llm.total_cost", cost) ◦ span.set_attribute("user_id", "<your-user-id>"); span.set_attribute("session_id", "<session-id>") Notes and supported flows • LiveKit emits the token/character/duration metrics you need; these are available in the agent metrics events and are suitable for cost computation. • Any realtime model supported by LiveKit (OpenAI Realtime, Azure Realtime, etc.) can be traced this way because tracing is done via OpenTelemetry instrumentation in the Agents framework. • This approach lets you track cost per turn, per call, or aggregated per session in Langfuse. What I need from you to give exact setup code Please tell me: 1. Which language/Agent runtime are you using (Python, Node.js, etc.)? 2. Are you using a specific realtime model/provider (OpenAI Realtime, Azure, Gemini, etc.)? 3. Do you want costs reported per-turn, per-call, or aggregated per session/user? 4. Do you use a specific pricing table (if you want me to include exact cost calculations), or should I show a placeholder calculation you can fill with your prices? If you confirm those, I’ll provide a minimal, copy-paste-ready example (OTLP exporter config, metric handler, and span attribute injection) tailored to your stack. Sources: Using the Testing Framework with Realtime Models | LiveKit | Agents (Node.js) | livekit/livekit_composite | DeepWiki | Setting custom user_id and session_id in Langfuse traces

Python, OpenAI Realtime, per-call,

Thanks — great details. Below is a focused, copy-paste-ready Python example that shows how to: • Configure an OTLP exporter to send traces to Langfuse. • Instrument a LiveKit AgentSession using an OpenAI realtime model (text modality). • Listen for LiveKit's

metrics_collected

events, compute per-call cost using a placeholder pricing table, and attach the cost to the current span with

user_id

and

session_id

attributes. Replace placeholders (keys, endpoints, and pricing) with your real values.

# Requirements:
# pip install opentelemetry-sdk opentelemetry-exporter-otlp opentelemetry-api
# plus your LiveKit / Agents client libs and OpenAI realtime client

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, SpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

# --- 1) Configure OTLP exporter to Langfuse ---
# Replace with your Langfuse endpoint and credentials.
langfuse_otlp_endpoint = "<https://cloud.langfuse.com/api/traces>"  # or self-hosted endpoint
langfuse_public_key = "LF_PUBLIC_KEY"
langfuse_secret_key = "LF_SECRET_KEY"

otlp_exporter = OTLPSpanExporter(
    endpoint=langfuse_otlp_endpoint,
    # If your exporter supports basic auth via username/password:
    username=langfuse_public_key,
    password=langfuse_secret_key,
    # Some OTLP exporters accept headers instead; adapt as needed for your library/version.
)

tracer_provider = TracerProvider()
tracer_provider.add_span_processor(BatchSpanProcessor(otlp_exporter))
trace.set_tracer_provider(tracer_provider)
tracer = trace.get_tracer(__name__)

# --- 2) (Optional) Span processor to inject user_id/session_id on span start ---
class UserSessionSpanProcessor(SpanProcessor):
    def __init__(self, user_id_getter, session_id_getter):
        # user_id_getter and session_id_getter are callables returning current ids
        self.user_id_getter = user_id_getter
        self.session_id_getter = session_id_getter

    def on_start(self, span, parent_context):
        try:
            user_id = self.user_id_getter()
            session_id = self.session_id_getter()
            if user_id:
                span.set_attribute("user_id", user_id)
            if session_id:
                span.set_attribute("session_id", session_id)
        except Exception:
            pass

    def on_end(self, span):  # noqa: N802
        pass

    def shutdown(self):
        pass

    def force_flush(self, timeout_millis: int = 30000) -> bool:
        return True

# Example getters (adapt to your auth/session management)
_current_user_id = None
_current_session_id = None
def get_user_id():
    return _current_user_id

def get_session_id():
    return _current_session_id

tracer_provider.add_span_processor(UserSessionSpanProcessor(get_user_id, get_session_id))

# --- 3) Pricing (placeholder) - set to your actual prices ---
# Example prices (fill with real values):
PRICING = {
    "input_per_million": 2.50,   # $ per 1M input tokens
    "output_per_million": 10.00, # $ per 1M output tokens
    # add STT/TTS pricing if you use those modalities
}

def compute_llm_cost(input_tokens: int, output_tokens: int) -> float:
    input_cost = input_tokens * PRICING["input_per_million"] / 1_000_000
    output_cost = output_tokens * PRICING["output_per_million"] / 1_000_000
    return input_cost + output_cost

# --- 4) LiveKit AgentSession & metrics handling (pseudocode adapted for typical LiveKit Agents API) ---
# Import and setup your LiveKit Agents session and OpenAI realtime model as you normally would.
# This pseudocode assumes the AgentSession emits a "metrics_collected" event with usage metrics.

# Example: create a session configured for OpenAI realtime text modality
# from livekit_agents import AgentSession
# from openai import realtime as openai_realtime
#
# session = AgentSession(
#     llm=openai_realtime.RealtimeModel(modalities=["text"]),
#     ...
# )

# For demonstration, here's a minimal handler that would run when metrics are emitted.
def on_metrics_collected(metrics):
    """
    Expected `metrics` structure (fields used below):
      - metrics.input_tokens (int)
      - metrics.output_tokens (int)
      - metrics.call_id or metrics.turn_id (identifier for the call)
      - metrics.user_id and metrics.session_id (optional, may be set elsewhere)
    Adapt field names to the exact structure your Agents runtime provides.
    """
    input_tokens = getattr(metrics, "input_tokens", 0) or 0
    output_tokens = getattr(metrics, "output_tokens", 0) or 0
    call_id = getattr(metrics, "call_id", None) or getattr(metrics, "turn_id", None)

    # Compute cost
    total_cost = compute_llm_cost(int(input_tokens), int(output_tokens))

    # Record cost on a span associated with this call
    # If you have an active span for the call, use that; otherwise create a span here.
    with tracer.start_as_current_span(f"llm.call.{call_id or 'unknown'}") as span:
        # ensure user/session attributes are present for correlation
        if getattr(metrics, "user_id", None):
            span.set_attribute("user_id", metrics.user_id)
        if getattr(metrics, "session_id", None):
            span.set_attribute("session_id", metrics.session_id)

        span.set_attribute("llm.input_tokens", int(input_tokens))
        span.set_attribute("llm.output_tokens", int(output_tokens))
        span.set_attribute("llm.total_cost", float(total_cost))
        # Optionally log a structured event
        span.add_event("llm.cost_computed", {
            "call_id": call_id,
            "input_tokens": int(input_tokens),
            "output_tokens": int(output_tokens),
            "total_cost": float(total_cost),
        })

# --- 5) Wiring the handler into your AgentSession ---
# Example: session.on("metrics_collected", on_metrics_collected)
# The exact subscription API depends on your LiveKit Agents Python client.

# --- 6) Example flow for a single call (high level) ---
def handle_incoming_call(call_info):
    """
    1) Set current user/session for span processor
    2) Start call/span
    3) Let the agent run -> LiveKit will emit metrics_collected at end of call (or per turn)
    4) on_metrics_collected computes & attaches cost to span (sent to Langfuse)
    """
    global _current_user_id, _current_session_id
    _current_user_id = call_info.get("user_id")
    _current_session_id = call_info.get("session_id")

    with tracer.start_as_current_span(f"call.{call_info.get('call_id')}"):
        # run your agent logic that uses OpenAI Realtime model
        # ...
        # metrics_collected will be invoked by the AgentSession lifecycle

        pass  # placeholder

# Note: Make sure to flush/ shutdown tracer provider on application exit to export remaining spans:
# tracer_provider.shutdown()

Key points and tips • LiveKit Agents expose token/character/duration metrics you need; capture them from the

metrics_collected

event. • Compute cost per-call in your metrics handler and attach it to the span (attribute or event). Langfuse will ingest these attributes/events for filtering and reporting. • Use

user_id

and

session_id

attributes on spans so you can group and query costs per user or session in Langfuse. • Ensure your OTLP exporter supports authentication for Langfuse; some versions accept username/password, others require headers — adapt the exporter instantiation accordingly. • Flush/shutdown the tracer provider at process exit to ensure spans are delivered. If you’d like, I can: • Provide a Node/Python exact example that matches the exact LiveKit Agents Python client API you’re using (if you paste the snippet you use to create AgentSession). • Show how to aggregate per-call costs into daily reports in Langfuse using span queries. Which would you prefer next?