import json import logging from datetime import datetime from pathlib import Path from typing import Any from livekit.agents import AgentSession, JobContext from livekit.agents.llm.tool_context import ( get_function_info, is_function_tool, is_raw_function_tool, ) from livekit.agents.llm.utils import function_arguments_to_pydantic_model from livekit.agents.metrics import LLMMetrics logger = logging.getLogger("token_metrics") logger.setLevel(logging.DEBUG) class TokenMetricsService: def __init__(self, ctx: JobContext, session: AgentSession) -> None: self.ctx = ctx self.session = session stamp = datetime.now().strftime("%Y%m%d_%H%M%S") self.dump_dir = Path("./llm_dumps") / f"{ctx.room.name}_{stamp}" self.dump_dir.mkdir(parents=True, exist_ok=True) self._call_idx = 0 self._totals = {"prompt": 0, "cached": 0, "completion": 0} def register(self) -> None: self.session.on("metrics_collected", self._on_metrics) self.ctx.add_shutdown_callback(self._log_final_summary) def dump_llm_request(self, chat_ctx: Any, tools: list[Any]) -> None: self._call_idx += 1 idx = self._call_idx messages = chat_ctx.to_dict(exclude_image=True, exclude_audio=True).get( "items", [] ) tool_dicts = [] for t in tools: info = get_function_info(t) if is_function_tool(t): schema = function_arguments_to_pydantic_model(t).model_json_schema() elif is_raw_function_tool(t): schema = getattr(info, "raw_schema", {}) else: schema = {} tool_dicts.append( { "type": "function", "function": { "name": info.name, "description": info.description or "", "parameters": schema, }, } ) payload = {"messages": messages, "tools": tool_dicts} out = self.dump_dir / f"llm_call_{idx:03d}.json" out.write_text(json.dumps(payload, indent=2, default=str, ensure_ascii=False)) logger.debug(f"[LLM #{idx}] dumped -> {out}") def _on_metrics(self, ev: Any) -> None: m = ev.metrics if not isinstance(m, LLMMetrics): return prompt = m.prompt_tokens or 0 cached = getattr(m, "prompt_cached_tokens", 0) or 0 completion = m.completion_tokens or 0 self._totals["prompt"] += prompt self._totals["cached"] += cached self._totals["completion"] += completion logger.info( f"[LLM #{self._call_idx}] prompt={prompt} (cached={cached}, fresh={prompt - cached}) " f"completion={completion}" ) async def _log_final_summary(self) -> None: t = self._totals logger.info( f"=== TOTAL === prompt={t['prompt']} (cached={t['cached']}, " f"fresh={t['prompt'] - t['cached']}) completion={t['completion']}" ) i want to analyze the per call tokens used so basically i want to print the number of tokens per llm call

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.

i am doing this in order to get to know how many tokens does my one call require and which tool or function is actually using more and more tokens so i can edit that function or tool call

Thanks — that goal makes sense. You already have the pieces needed to identify which tools/functions consume the most tokens; here’s a concise plan and small, concrete tweaks to make correlation easy. Acknowledgement • You want per-call token counts and a reliable way to map each LLM call to the set of tools/functions used so you can see which tool uses the most tokens. What you already have •

dump_llm_request()

increments

self._call_idx

and writes a JSON file per call with the

messages

and

tools

used. •

_on_metrics()

receives

LLMMetrics

and logs prompt, cached, and completion tokens per call using

self._call_idx

. • Correlation is possible by matching the call index in the logs to the corresponding

llm_call_{idx:03d}.json

file. Recommended small changes (make analysis explicit and robust) 1. Include

call_idx

in the dumped JSON and in the metrics log line so correlation is explicit:

# in dump_llm_request
payload = {"call_idx": idx, "messages": messages, "tools": tool_dicts}
# in _on_metrics
logger.info(f"[LLM #{self._call_idx}] call_idx={self._call_idx} prompt={prompt} (cached={cached}) completion={completion}")

1. Write a structured per-call token log (JSONL) for easier programmatic analysis:

# append per-call metrics to a JSONL file (open in append mode)
record = {
  "call_idx": self._call_idx,
  "prompt": prompt, "prompt_cached": cached, "completion": completion,
  "timestamp": datetime.utcnow().isoformat()
}
(self.dump_dir / "llm_token_log.jsonl").write_text(json.dumps(record, ensure_ascii=False) + "\n", append=True)

(Use your preferred file-append method.) 1. Optionally annotate each tool entry with an explicit

tool_id

or

name

in

llm_call_{idx}.json

so you can immediately see which functions were present for that call (you already add

name

in `tool_dicts`; ensure it’s always populated). 2. Verify logging and ordering: • Ensure

register()

is called and logger level is INFO/DEBUG. • If metrics can arrive out of order, include a unique request id (call_idx or UUID) in both the dumped request and the LLM request/metadata so metrics can be reliably matched. How to analyze after collecting logs • Parse

llm_token_log.jsonl

and for each call, open

llm_call_{call_idx}.json

to get the tool list. • Aggregate token totals per tool name (e.g., attribute tokens to all tools present in that call or refine by instrumenting tools to report their own token deltas). If you’d like, I can: • Provide a ready-to-use code snippet that appends JSONL correctly (with safe file append). • Show a short Python script to aggregate token usage by tool name from the dumped files. Which of those would you like next? , with interactive elements

ok

Thanks — below is a compact, ready-to-use code snippet that appends structured per-call token metrics to a JSONL file and ensures each metric record includes the same call index you dump with

llm_call_{idx}.json

. This makes it trivial to correlate tokens -> tools. Add these changes to your class: 1) Ensure

dump_llm_request

writes the call index into the dumped request:

# in dump_llm_request, after computing idx and payload
payload = {"call_idx": idx, "messages": messages, "tools": tool_dicts}
out = self.dump_dir / f"llm_call_{idx:03d}.json"
out.write_text(json.dumps(payload, indent=2, default=str, ensure_ascii=False))

2) Append structured per-call metrics in

_on_metrics

(use atomic append):

import json
from datetime import datetime
from pathlib import Path

def _append_jsonl(path: Path, record: dict) -> None:
    # atomic append: open in append mode and write one JSON line
    with path.open("a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

def _on_metrics(self, ev: Any) -> None:
    m = ev.metrics
    if not isinstance(m, LLMMetrics):
        return

    prompt = m.prompt_tokens or 0
    cached = getattr(m, "prompt_cached_tokens", 0) or 0
    completion = m.completion_tokens or 0

    self._totals["prompt"] += prompt
    self._totals["cached"] += cached
    self._totals["completion"] += completion

    # include the same call index used when dumping the request
    call_idx = self._call_idx

    logger.info(
        f"[LLM #{call_idx}] call_idx={call_idx} prompt={prompt} (cached={cached}, fresh={prompt - cached}) completion={completion}"
    )

    # write structured record for later analysis
    record = {
        "call_idx": call_idx,
        "prompt": prompt,
        "prompt_cached": cached,
        "completion": completion,
        "fresh": prompt - cached,
        "timestamp": datetime.utcnow().isoformat() + "Z",
    }
    _append_jsonl(self.dump_dir / "llm_token_log.jsonl", record)

How to analyze afterwards • For each JSONL record, open

llm_call_{call_idx:03d}.json

to see the

tools

list and map tokens to those tool names. • Aggregate token totals per tool name (either attribute the full call tokens to all tools present, or refine by instrumenting functions to report per-tool deltas if you need per-tool precision). If you want, I can: • Provide a short Python script to parse

llm_token_log.jsonl

and aggregate tokens by tool name based on the dumped

llm_call_*.json

files. • Show how to attach a UUID per request (if metrics arrive out of order) so correlation is robust. Which would you like next? , with interactive elements