This message was deleted.

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

Usage summary: AgentSessionUsage(model_usage=[LLMModelUsage(type='llm_usage', provider='livekit', model='openai/gpt-4.1', input_tokens=18337, input_cached_tokens=16896, output_tokens=76), TTSModelUsage(type='tts_usage', provider='livekit', model='elevenlabs/eleven_turbo_v2_5', characters_count=227, audio_duration=12.492375000000003), STTModelUsage(type='stt_usage', provider='livekit', model='deepgram/flux', audio_duration=42.75000000000017)]) using python this is now what i have but i want the tokens iteration wise like what part is consuming more tokens wether it is tool call or something like that

Thanks — I’ll help you get per-iteration token breakdowns in Python and correlate them to conversation parts (user text, system prompt, tool calls, etc.). A reliable approach: • Log each conversation item when it’s added (use

conversation_item_added

) and assign a unique turn ID/timestamp. • Listen for

metrics_collected

events (which include

RealtimeModelMetrics

) and attach the same turn ID/timestamp from the request/trace metadata so you can map metrics to the exact turn. • Store both logs (item content + metrics) and then compute token deltas or attribute tokens to the corresponding item types (text, audio, tool). Minimal Python sketch showing the pattern (adapt to your AgentSession APIs and data model):

import uuid
import time
import asyncio

# in-memory store for demo; replace with DB if needed
turn_store = {}

def now_ts():
    return time.time()

# conversation_item_added handler
@session.on("conversation_item_added")
def on_item_added(ev):
    item = ev.item
    turn_id = str(uuid.uuid4())
    ts = now_ts()

    # join text pieces if content is list-like
    text = "".join(c for c in getattr(item, "content", []) if isinstance(c, str))

    # save minimal metadata to correlate later
    turn_store[turn_id] = {
        "ts": ts,
        "role": getattr(item, "role", None),
        "type": getattr(item, "type", None),
        "content": text,
        "metrics": None,
    }

    # If the SDK provides a request/trace id on the event, save it too:
    trace_id = getattr(ev, "trace_id", None)
    if trace_id:
        turn_store[turn_id]["trace_id"] = trace_id

    print(f"[{turn_id}] item_added role={item.role} len={len(text)}")

# metrics_collected handler
@session.on("metrics_collected")
def on_metrics_collected(ev):
    metrics = ev.metrics  # RealtimeModelMetrics
    ts = now_ts()

    # metrics often include a trace_id or turn id you can use to correlate
    trace_id = getattr(metrics, "trace_id", None)

    # Attempt 1: correlate by trace_id if present
    if trace_id:
        for tid, rec in turn_store.items():
            if rec.get("trace_id") == trace_id:
                rec["metrics"] = metrics
                rec["metrics_ts"] = ts
                print_metrics_for_turn(tid, rec)
                return

    # Attempt 2: correlate by nearest timestamp (fallback)
    # find the most recent turn before this metrics timestamp without metrics
    candidates = [(tid, r) for tid, r in turn_store.items() if r.get("metrics") is None and r["ts"] <= ts]
    if candidates:
        # choose the most recent
        tid, rec = max(candidates, key=lambda x: x[1]["ts"])
        rec["metrics"] = metrics
        rec["metrics_ts"] = ts
        print_metrics_for_turn(tid, rec)
        return

    print("Received metrics but could not correlate to a turn:", metrics)

def print_metrics_for_turn(turn_id, rec):
    m = rec["metrics"]
    if not m:
        print(f"[{turn_id}] no metrics")
        return

    # Example fields from RealtimeModelMetrics
    input_tokens = getattr(m, "input_tokens", None)
    output_tokens = getattr(m, "output_tokens", None)
    details = getattr(m, "input_token_details", None)

    print(f"--- Turn {turn_id} ---")
    print(f"role={rec['role']} type={rec.get('type')} ts={rec['ts']}")
    print(f"content (len={len(rec['content'])}): {rec['content'][:200]}")
    print(f"input_tokens={input_tokens} output_tokens={output_tokens}")
    if details:
        print(f" text_tokens={getattr(details,'text_tokens',None)}")
        print(f" audio_tokens={getattr(details,'audio_tokens',None)}")
        print(f" cached_tokens={getattr(details,'cached_tokens',None)}")
    print("--------------------")

Notes and tips: • If

metrics

includes a

trace_id

or request id, use it — that's the most accurate correlation. •

input_token_details

gives modality breakdown (text vs audio vs cached). Use that to see whether tool calls (usually text/tool descriptions) or audio inputs consume more tokens. • For tool calls: capture the tool call event or mark tool outputs in the

conversation_item_added

content/type so you can separate tool text from user/system text when summarizing tokens. • Persist logs (DB) for easier analytics and to compute per-iteration summaries and aggregates. If you want, tell me: • whether

metrics

in your events contains a

trace_id

or similar, and • an example of a

conversation_item_added

payload you receive — and I’ll adapt the code to use those exact fields and produce a small per-iteration report. , with interactive elements