import asyncio
import os
from dotenv import load_d...
# ask-aib
busy-breakfast-32623
09/18/2025, 5:19 AMimport asyncio
import os
from dotenv import load_dotenv
from livekit import agents
from livekit.agents import AgentSession, Agent, RoomInputOptions
from livekit.plugins import (
openai,
noise_cancellation,
google
)
import prompts
from tools import lookup_weather,send_message_to_aman
from constants import get_constant
load_dotenv()
import time
async def wait_for_file(file_path: str, timeout: int = 5):
start = time.time()
while not os.path.exists(file_path):
await asyncio.sleep(0.2)
if time.time() - start > timeout:
return False
return True
class Assistant(Agent):
def __init__(self,prompt="you are a helpful assistant") -> None:
super().__init__(instructions=prompt,
tools=[lookup_weather,send_message_to_aman]
)
async def entrypoint(ctx: agents.JobContext):
system_prompt="your name is ravina tandon"
session = AgentSession(
llm=google.beta.realtime.RealtimeModel(
model="gemini-2.0-flash-exp",
voice="Aoede",
temperature=0.8,
instructions="start with a greeting",
),
)
print("session created")
await session.start(
room=ctx.room,
agent=Assistant(prompt=system_prompt),
room_input_options=RoomInputOptions(
noise_cancellation=noise_cancellation.BVC(),
),
)
print("Room created:", ctx.room)
await ctx.connect()
await session.generate_reply(
instructions="start with a greeting ",
)
if name == "__main__":
agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint,num_idle_processes=2))
i want to print the total token consumed for each session how to implement this here
t
tall-airline-16756
09/18/2025, 5:19 AMThanks for reaching out, Aman. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
tall-airline-16756
09/18/2025, 5:20 AMI'd love to help you track and print the total token consumption for each session! You can do this easily using the
UsageCollectorlivekit.agents.metricsUsageCollectormetrics_collectedAgentSessionAgentMetricsusage_collector.collect()ctx.add_shutdown_callback() Copy code
from livekit.agents.metrics import UsageCollector
from livekit.agents import MetricsCollectedEvent
# Initialize UsageCollector
usage_collector = UsageCollector()
# Subscribe to metrics_collected event
@session.on("metrics_collected")
def _on_metrics_collected(ev: MetricsCollectedEvent):
usage_collector.collect(ev.metrics)
# Log usage summary at session end
async def log_usage_summary():
summary = usage_collector.get_summary()
print(f"Total LLM Tokens: {summary.llm_prompt_tokens + summary.llm_completion_tokens}")
print(f"Total TTS Characters: {summary.tts_characters_count}")
print(f"Total STT Audio Duration: {summary.stt_audio_duration:.2f}s")
ctx.add_shutdown_callback(log_usage_summary)