dev

dev

uv run agent.py dev

pnpm dev

agent_name

@server.rtc_session(agent_name="...")

agent_name

@server.rtc_session()

RoomConfiguration

agents=[RoomAgentDispatch(...)]

LIVEKIT_API_KEY

LIVEKIT_API_SECRET

LIVEKIT_URL

agent_name

from livekit.api import AccessToken, RoomConfiguration, RoomAgentDispatch

   token = (
       AccessToken()
       .with_identity("user123")
       .with_grants(VideoGrants(room_join=True, room="your-room-name"))
       .with_room_config(
           RoomConfiguration(
               agents=[RoomAgentDispatch(agent_name="test-agent", metadata='{"user":"123"}')]
           )
       )
       .to_jwt()
   )

lk dispatch create --room your-room-name --agent-name test-agent

rtc_session

agent_name

@server.rtc_session(agent_name="...")

RoomConfiguration

agents=[RoomAgentDispatch(...)]

test-agent

lk token create

lk token create

@server.rtc_session(agent_name="test-agent")

RoomConfiguration

agents=[RoomAgentDispatch(...)]

from livekit import api
from livekit.api import RoomAgentDispatch, RoomConfiguration

token = (
    api.AccessToken(api_key="YOUR_KEY", api_secret="YOUR_SECRET", identity="user123")
    .with_grants(api.VideoGrants(room_join=True, room="my-room"))
    .with_room_config(
        RoomConfiguration(
            agents=[RoomAgentDispatch(agent_name="test-agent", metadata='{"user":"123"}')]
        )
    )
    .to_jwt()
)
print(token)

lk dispatch create --agent-name test-agent --room my-room --metadata '{"user_id":"12345"}'

lk

dev

connect

start

start

start

console

dev

agent_name = "test-agent"

console

uv run agent.py console

uv run agent.py dev

text_output

RoomOptions(text_output=True)

@server.rtc_session(agent_name="test-agent")

RoomConfiguration

lk dispatch create --agent-name test-agent --room my-room --metadata '{"user_id":"12345"}'

dev

MetricsCollected

dev

LIVEKIT_API_KEY

LIVEKIT_API_SECRET

LIVEKIT_URL

console

console

dev

start

@server.rtc_session(...)

dev

console

dev

MetricsCollected

Bunları loglayacağım basit kodu yazar mısın?

session

AgentSession

dev

console

import asyncio
from datetime import datetime

# Örnek async handler: MetricsCollected olayını loglar
async def handle_metrics_collected(event):
    m = event.metrics
    now = datetime.utcnow().isoformat()
    print(f"[MetricsCollected] {now}")
    # Örnek alanlar; agent/session kitaplığına göre isimler küçük farklılık gösterebilir
    print(f"  Source: {getattr(event, 'source', 'unknown')}")
    print(f"  TTFT: {getattr(m, 'ttft', 'N/A')} s")
    print(f"  TTFB: {getattr(m, 'ttfb', 'N/A')} s")
    print(f"  Duration: {getattr(m, 'duration', 'N/A')} s")
    print(f"  Prompt tokens: {getattr(m, 'prompt_tokens', 'N/A')}")
    print(f"  Completion tokens: {getattr(m, 'completion_tokens', 'N/A')}")
    print(f"  Timestamp: {getattr(m, 'timestamp', 'N/A')}")
    print("-" * 40)

# session: AgentSession örneği (kendi kodunda zaten oluşturulmuş olmalı)
# session.on ile event adını ve handler'ı kaydet
# Event adlandırması kullandığın SDK sürümüne göre değişebilir; örnek: "metrics_collected" veya AgentSessionEventTypes.MetricsCollected
def attach_metrics_logger(session):
    try:
        # Eğer SDK string event isimleri kullanıyorsa:
        session.on("metrics_collected", handle_metrics_collected)

        # Veya enum/const kullanılıyorsa (örnek kullanım, SDK'na uyarlayın):
        # from livekit.agents import AgentSessionEventTypes
        # session.on(AgentSessionEventTypes.MetricsCollected, handle_metrics_collected)

        print("MetricsCollected handler eklendi.")
    except Exception as e:
        print("MetricsCollected handler eklenirken hata:", e)

session

attach_metrics_logger(session)

dev

console

livekit.agents

record=True

record=True

session = AgentSession(..., record=True)

record=False

dev

uv run agent.py dev

MetricsCollected

record

MetricsCollected

MetricsCollected

# metrics_exporter.py
from prometheus_client import start_http_server, Gauge, Counter
import asyncio
import time
from datetime import datetime

# Prometheus metric tanımları
TTFT = Gauge('agent_ttft_seconds', 'Time to first token', ['agent_name', 'room_id', 'model'])
TTFB = Gauge('agent_ttfb_seconds', 'Time to first byte (TTS)', ['agent_name', 'room_id', 'model'])
DURATION = Gauge('agent_duration_seconds', 'Processing duration', ['agent_name', 'room_id', 'model'])
INPUT_TOKENS = Gauge('agent_input_tokens', 'Input tokens', ['agent_name', 'room_id', 'model'])
OUTPUT_TOKENS = Gauge('agent_output_tokens', 'Output tokens', ['agent_name', 'room_id', 'model'])
TOTAL_TOKENS = Gauge('agent_total_tokens', 'Total tokens', ['agent_name', 'room_id', 'model'])
REQUESTS = Counter('agent_requests_total', 'Total inference requests', ['agent_name', 'model', 'status'])

# Basit metrics logger / handler
async def handle_metrics_collected(event):
    m = event.metrics
    labels = {
        'agent_name': getattr(event, 'agent_name', 'unknown'),
        'room_id': getattr(event, 'room_id', 'unknown'),
        'model': getattr(event, 'model_name', getattr(m, 'model_name', 'unknown'))
    }
    # Safely al (bazı alanlar eksik olabilir)
    try:
        if getattr(m, 'ttft', None) is not None and m.ttft >= 0:
            TTFT.labels(**labels).set(m.ttft)
        if getattr(m, 'ttfb', None) is not None and m.ttfb >= 0:
            TTFB.labels(**labels).set(m.ttfb)
        if getattr(m, 'duration', None) is not None:
            DURATION.labels(**labels).set(m.duration)
        if getattr(m, 'input_tokens', None) is not None:
            INPUT_TOKENS.labels(**labels).set(m.input_tokens)
        if getattr(m, 'output_tokens', None) is not None:
            OUTPUT_TOKENS.labels(**labels).set(m.output_tokens)
        if getattr(m, 'total_tokens', None) is not None:
            TOTAL_TOKENS.labels(**labels).set(m.total_tokens)

        # Basit request sayacı (hata durumu da varsa 'status' label'ı ver)
        REQUESTS.labels(agent_name=labels['agent_name'], model=labels['model'], status='ok').inc()
    except Exception as e:
        # Hata sayacı
        REQUESTS.labels(agent_name=labels['agent_name'], model=labels['model'], status='error').inc()
        print("Metrics handler error:", e)

def attach_metrics_exporter(session, port=8000):
    # Prometheus exporter'ı başlat
    start_http_server(port)
    print(f"Prometheus exporter started on :{port}")
    # SDK event kaydı. SDK event ismi sürüme göre değişir: "metrics_collected" veya enum kullan
    session.on("metrics_collected", lambda ev: asyncio.create_task(handle_metrics_collected(ev)))
    print("MetricsCollected handler attached.")

attach_metrics_exporter(session, port=8000)

scrape_configs:
  - job_name: 'livekit_agents'
    static_configs:
      - targets: ['<agent-host>:8000']

AgentSession

.on(...)

.on

asyncio.create_task(...)

MetricsCollected

handle_metrics

asyncio.create_task

import asyncio
from datetime import datetime

# Bu örnekte event nesnesinin .metrics alanı olduğunu varsayıyoruz.
# SDK sürümüne göre event isimleri/alanlar değişebilir; gerekirse uyarlayalım.

async def handle_metrics(event):
    """Asenkron iş yapan gerçek handler"""
    m = event.metrics
    now = datetime.utcnow().isoformat()
    # Güvenli erişim için getattr kullanıyoruz (bazı alanlar olmayabilir)
    ttft = getattr(m, "ttft", None)
    ttfb = getattr(m, "ttfb", None)
    duration = getattr(m, "duration", None)
    input_tokens = getattr(m, "input_tokens", None)
    output_tokens = getattr(m, "output_tokens", None)

    # Burada istediğin işlemi yap: DB'ye yaz, Prometheus'a gönder, dosyaya logla vb.
    print(f"[Metrics] {now} room={getattr(event, 'room_id', 'unknown')} model={getattr(event, 'model_name', 'unknown')}")
    if ttft is not None:
        print(f"  TTFT: {ttft}s")
    if ttfb is not None:
        print(f"  TTFB: {ttfb}s")
    if duration is not None:
        print(f"  Duration: {duration}s")
    if input_tokens is not None or output_tokens is not None:
        print(f"  Tokens - in: {input_tokens} out: {output_tokens}")
    # Örnek: bekleyip async I/O yapma
    await asyncio.sleep(0)  # placeholder, gerçek async iş burada olabilir

def metrics_wrapper(event):
    """.on(...) için senkron wrapper — asenkron handler'ı arka planda çalıştırır."""
    try:
        asyncio.create_task(handle_metrics(event))
    except RuntimeError:
        # Eğer event loop yoksa (ör. module import sırasında), loop başlat veya logla
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        loop.create_task(handle_metrics(event))

def attach_metrics_listener(session):
    """
    session: SDK'daki AgentSession veya ilgili nesne.
    .on event kayıt yöntemi SDK'ya göre farklı olabilir (string veya enum).
    Örnek: session.on("metrics_collected", metrics_wrapper)
    veya component bazında: session.llm.on("metrics_collected", metrics_wrapper)
    """
    # Örnek: genel oturum event'i
    try:
        session.on("metrics_collected", metrics_wrapper)
        print("MetricsCollected handler eklendi (wrapper ile).")
    except Exception as e:
        print("Handler eklenirken hata:", e)

attach_metrics_listener(session)

dev

console

handle_metrics

.on

session.llm.on(...)

session.tts.on(...)

session.stt.on(...)

asyncio.create_task

RuntimeError

handle_metrics

handle_metrics

attach_metrics_listener(session)

attach_metrics_listener(session)

session

asyncio.create_task

attach_metrics_listener(session)

metrics_collected

asyncio.create_task

metrics_collected

import asyncio
from datetime import datetime

# --- Basit metrics handler'ları ---
async def handle_llm_metrics(event):
    m = event.metrics
    now = datetime.utcnow().isoformat()
    print(f"[LLM Metrics] {now} room={getattr(event, 'room_id', 'unknown')} model={getattr(event, 'model_name', 'unknown')}")
    print(f"  TTFT: {getattr(m, 'ttft', 'N/A')}")
    print(f"  Duration: {getattr(m, 'duration', 'N/A')}")
    print(f"  Prompt tokens: {getattr(m, 'prompt_tokens', 'N/A')}")
    print(f"  Completion tokens: {getattr(m, 'completion_tokens', 'N/A')}")
    print("-" * 40)
    await asyncio.sleep(0)  # placeholder for async I/O (DB, exporter, vb.)

async def handle_tts_metrics(event):
    m = event.metrics
    now = datetime.utcnow().isoformat()
    print(f"[TTS Metrics] {now} room={getattr(event, 'room_id', 'unknown')}")
    print(f"  TTFB: {getattr(m, 'ttfb', 'N/A')}")
    print(f"  Duration: {getattr(m, 'duration', 'N/A')}")
    print("-" * 40)
    await asyncio.sleep(0)

async def handle_stt_metrics(event):
    m = event.metrics
    now = datetime.utcnow().isoformat()
    print(f"[STT Metrics] {now} room={getattr(event, 'room_id', 'unknown')}")
    print(f"  Duration: {getattr(m, 'duration', 'N/A')}")
    print("-" * 40)
    await asyncio.sleep(0)

# --- Wrapper'lar: .on(...) sync callable beklediği için asyncio.create_task kullanıyoruz ---
def attach_metrics_listener(session):
    """
    session: AgentSession nesnesi (kendi SDK'na göre isim farklı olabilir).
    Bu fonksiyon session üzerinde metrics event'lerini dinler.
    """
    # Eğer SDK component bazlı event veriyorsa session.llm/tts/stt kullan:
    try:
        # LLM metrics
        session.llm.on("metrics_collected", lambda ev: asyncio.create_task(handle_llm_metrics(ev)))
    except Exception:
        # Eğer session.llm yoksa genel event üzerinden dinle
        try:
            session.on("llm_metrics_collected", lambda ev: asyncio.create_task(handle_llm_metrics(ev)))
        except Exception:
            pass

    # TTS metrics
    try:
        session.tts.on("metrics_collected", lambda ev: asyncio.create_task(handle_tts_metrics(ev)))
    except Exception:
        try:
            session.on("tts_metrics_collected", lambda ev: asyncio.create_task(handle_tts_metrics(ev)))
        except Exception:
            pass

    # STT metrics
    try:
        session.stt.on("metrics_collected", lambda ev: asyncio.create_task(handle_stt_metrics(ev)))
    except Exception:
        try:
            session.on("stt_metrics_collected", lambda ev: asyncio.create_task(handle_stt_metrics(ev)))
        except Exception:
            pass

    print("attach_metrics_listener: handler'lar eklendi.")

# --- Tam bir örnek giriş noktası (entrypoint) ---
async def main():
    """
    Burada session oluşturma yerini kendi agent koduna göre uyarla.
    Örneğin:
      - session = AgentSession(...) 
      - veya agent = MyAgent(...) ve agent.session
    """
    # --- Placeholder: kendi SDK ile değiştir ---
    # from livekit.agents import AgentSession
    # session = AgentSession(...) 
    session = create_fake_session_for_example()  # sadece örnek, kendi session'ını kullan

    # Metrics listener'ı ekle
    attach_metrics_listener(session)

    # Start / connect agent session; kendi kodunda bu farklı olabilir
    # Ör: await session.start(...) veya await agent.run(...)
    try:
        await session.start()  # kendi SDK çağrınla değiştir
    except AttributeError:
        # Eğer örnek session fake ise, sadece bekle
        print("Session.start() çağrısı örnek için atlandı.")

    # Örnek: sonsuz bekle (gerçek agent çalışırken event'ler tetiklenecek)
    while True:
        await asyncio.sleep(60)

# --- Örnek placeholder session (sadece demo, gerçek kodunda kaldır) ---
class FakeEvent:
    def __init__(self, metrics, room_id="RM_demo", model_name="demo-model"):
        self.metrics = metrics
        self.room_id = room_id
        self.model_name = model_name

class FakeMetrics:
    def __init__(self, ttft=None, ttfb=None, duration=None, prompt_tokens=None, completion_tokens=None):
        self.ttft = ttft
        self.ttfb = ttfb
        self.duration = duration
        self.prompt_tokens = prompt_tokens
        self.completion_tokens = completion_tokens

class FakeComponent:
    def __init__(self):
        self._handlers = {}

    def on(self, event_name, handler):
        self._handlers.setdefault(event_name, []).append(handler)

    # test için event tetikle
    def emit(self, event_name, ev):
        for h in self._handlers.get(event_name, []):
            h(ev)

class FakeSession:
    def __init__(self):
        self.llm = FakeComponent()
        self.tts = FakeComponent()
        self.stt = FakeComponent()

    async def start(self):
        # örnek: çalıştıktan sonra periyodik test event'i gönder
        await asyncio.sleep(1)
        # test metrics
        m1 = FakeMetrics(ttft=0.12, duration=1.3, prompt_tokens=120, completion_tokens=30)
        self.llm.emit("metrics_collected", FakeEvent(m1))
        await asyncio.sleep(1)
        m2 = FakeMetrics(ttfb=0.25, duration=0.8)
        self.tts.emit("metrics_collected", FakeEvent(m2))
        await asyncio.sleep(1)
        m3 = FakeMetrics(duration=0.6)
        self.stt.emit("metrics_collected", FakeEvent(m3))

def create_fake_session_for_example():
    return FakeSession()

# --- Çalıştır ---
if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("Çıkış yapılıyor.")

session

attach_metrics_listener(session)

session.llm

session.tts

session.stt

session.on("metrics_collected", ...)

async def

asyncio.create_task(...)

handle_*

prometheus_client.start_http_server(8000)

room

model

event.room_id

event.context.room_id

event.job.room_id

event.model_name

event.metrics.model_name

def inspect_event(event):
    print("=== EVENT INSPECT ===")
    # En yaygın alanlar
    for attr in ["room_id", "room", "model_name", "model", "agent_name", "context", "job", "metadata"]:
        if hasattr(event, attr):
            print(f"{attr}: {getattr(event, attr)}")
    # Tüm __dict__ (varsa)
    try:
        print("__dict__:", event.__dict__)
    except Exception:
        pass
    # metrics alt yapısı
    try:
        m = event.metrics
        print("metrics.__dict__:", getattr(m, "__dict__", str(m)))
    except Exception:
        pass
    print("=====================")

event.room_id

event.context.room_id

event.job.room_id

event.metadata['room_id']

event.model_name

event.metrics.model_name

event.metadata['model_name']

import asyncio
from datetime import datetime

def _get_room_id_from_event(event):
    # Olası yerleri sırayla kontrol et
    candidates = []
    if hasattr(event, "room_id"):
        candidates.append(getattr(event, "room_id"))
    if hasattr(event, "room"):
        # bazen room objesi veya adı gelebilir
        candidates.append(getattr(event, "room"))
    # context veya job içinde olabilir
    if hasattr(event, "context") and event.context:
        try:
            candidates.append(getattr(event.context, "room_id", None))
            candidates.append(getattr(event.context, "room", None))
        except Exception:
            pass
    if hasattr(event, "job") and event.job:
        try:
            candidates.append(getattr(event.job, "room_id", None))
        except Exception:
            pass
    # metadata dict içinde olabilir
    try:
        meta = getattr(event, "metadata", None) or {}
        if isinstance(meta, dict):
            candidates.append(meta.get("room_id"))
            candidates.append(meta.get("room"))
    except Exception:
        pass

    # Döndürülebilir ilk dolu değer
    for c in candidates:
        if c not in (None, "", "unknown"):
            return str(c)
    return "unknown"

def _get_model_from_event(event):
    candidates = []
    if hasattr(event, "model_name"):
        candidates.append(getattr(event, "model_name"))
    if hasattr(event, "model"):
        candidates.append(getattr(event, "model"))
    # metrics içinde olabilir
    try:
        m = getattr(event, "metrics", None)
        if m is not None:
            candidates.append(getattr(m, "model_name", None))
            candidates.append(getattr(m, "model", None))
    except Exception:
        pass
    # metadata
    try:
        meta = getattr(event, "metadata", None) or {}
        if isinstance(meta, dict):
            candidates.append(meta.get("model_name"))
            candidates.append(meta.get("model"))
    except Exception:
        pass

    for c in candidates:
        if c not in (None, "", "unknown"):
            return str(c)
    return "unknown"

async def handle_metrics(event):
    m = getattr(event, "metrics", None)
    room = _get_room_id_from_event(event)
    model = _get_model_from_event(event)
    now = datetime.utcnow().isoformat()

    print(f"[Metrics] {now} room={room} model={model}")
    if m is not None:
        print(f"  TTFT: {getattr(m, 'ttft', 'N/A')}")
        print(f"  TTFB: {getattr(m, 'ttfb', 'N/A')}")
        print(f"  Duration: {getattr(m, 'duration', 'N/A')}")
        print(f"  Input tokens: {getattr(m, 'input_tokens', 'N/A')}")
        print(f"  Output tokens: {getattr(m, 'output_tokens', 'N/A')}")
    else:
        print("  metrics: None")
    print("-" * 40)
    await asyncio.sleep(0)

def attach_metrics_listener(session):
    # component bazlı bağlan (varsa)
    try:
        session.llm.on("metrics_collected", lambda ev: asyncio.create_task(handle_metrics(ev)))
    except Exception:
        try:
            session.on("llm_metrics_collected", lambda ev: asyncio.create_task(handle_metrics(ev)))
        except Exception:
            pass

    try:
        session.tts.on("metrics_collected", lambda ev: asyncio.create_task(handle_metrics(ev)))
    except Exception:
        try:
            session.on("tts_metrics_collected", lambda ev: asyncio.create_task(handle_metrics(ev)))
        except Exception:
            pass

    try:
        session.stt.on("metrics_collected", lambda ev: asyncio.create_task(handle_metrics(ev)))
    except Exception:
        try:
            session.on("stt_metrics_collected", lambda ev: asyncio.create_task(handle_metrics(ev)))
        except Exception:
            pass

    # fallback: genel metrics_collected
    try:
        session.on("metrics_collected", lambda ev: asyncio.create_task(handle_metrics(ev)))
    except Exception:
        pass

    print("attach_metrics_listener: handlers attached.")

inspect_event(event)

room_id

model_name

_get_room_id_from_event

_get_model_from_event

start

start

start

start

dev

console

start

room

start

lk

lk dispatch create --agent-name test-agent --room my-room --metadata '{"...": "..."}'

start

start

agent_name

start

lk dispatch create