Мониторинг и Observability AI-агентов

В отличие от классического софта, AI-агенты — это недетерминированные цепочки вызовов. Один пользовательский запрос может породить 3, 7 или 15 последовательных LLM-вызовов, каждый со своими параметрами, промптами и ответами. Без observability вы буквально слепы: невозможно понять, почему агент ответил так, а не иначе — и сколько это стоило.

Ключевые проблемы, которые решает observability:

• Отладка в проде. Агент дал странный ответ пользователю. Без трассировки вам придётся гадать, на каком шаге произошла ошибка. Трейс показывает весь путь: промпт → LLM-ответ → парсинг → вызов тула → результат тула → следующий промпт.
• Контроль затрат. Каждый вызов GPT-4o стоит денег. Без метрик вы не знаете, что 30% бюджета уходит на повторные вызовы из-за ошибок парсинга JSON-ответов агента.
• Оценка качества. Continuous Evaluation позволяет автоматически прогонять агента на тестовых сценариях после каждого изменения промпта и сравнивать метрики (точность, релевантность, латентность).
• Версионирование промптов. Когда вы меняете системный промпт, важно видеть A/B-сравнение: улучшила ли новая версия ответы или наоборот.

LangSmith vs LangFuse: что выбрать?

Для стартапов и проектов с ограниченным бюджетом рекомендуем LangFuse в self-hosted режиме. Для enterprise-команд, глубоко интегрированных в экосистему LangChain — LangSmith. Ниже разберём настройку обоих.

LangSmith — SaaS-платформа от создателей LangChain. Главное преимущество: нулевая конфигурация для LangChain-проектов. Достаточно установить переменные окружения — и все вызовы автоматически трассируются.

Шаг 1: Регистрация и API-ключ

Зарегистрируйтесь на smith.langchain.com. После регистрации создайте API-ключ в разделе Settings → API Keys.

Шаг 2: Установка и базовая настройка

Установка LangSmith SDK

pip install langsmith langchain langchain-openai

Базовый трейсинг для агента на LangGraph

import os
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langsmith import Client, traceable

# ▸ Установите переменные окружения
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"
os.environ["LANGCHAIN_API_KEY"] = "lsv2_pt_xxxx...ваш_ключ"
os.environ["LANGCHAIN_PROJECT"] = "qantcore-agent-prod"

# ▸ Этих 4 строк достаточно — всё автоматически трассируется!
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# ▸ Определяем инструменты агента
def search_knowledge_base(query: str) -> str:
    """Поиск по внутренней базе знаний QantCore."""
    return f"Результаты поиска по: {query}"

tools = [search_knowledge_base]

# ▸ Создаём ReAct-агента — каждый шаг попадёт в трейс
agent = create_react_agent(llm, tools)

# ▸ Тестовый запуск — смотрим трейс в LangSmith UI
result = agent.invoke({
    "messages": [("user", "Какие модели LLM поддерживает QantCore?")]
})
print(f"Ответ агента: {result["messages"][-1].content}")

Шаг 3: Кастомная трассировка с @traceable

Для более гранулярного контроля используйте декоратор @traceable — он позволяет отслеживать не только LangChain-вызовы, но и любые функции вашего кода:

from langsmith import traceable
from langsmith.run_helpers import get_current_run_tree

@traceable(run_type="tool", name="validate_agent_output")
def validate_output(raw_response: str, expected_schema: dict) -> dict:
    """Валидирует JSON-ответ агента и логирует ошибки в трейс."""
    import json
    run = get_current_run_tree()
    try:
        parsed = json.loads(raw_response)
        # Логируем метаданные в текущий трейс
        if run:
            run.add_metadata({"output_keys": list(parsed.keys())})
        return {"valid": True, "data": parsed}
    except json.JSONDecodeError as e:
        if run:
            run.add_metadata({"validation_error": str(e)})
        return {"valid": False, "error": str(e)}

# ▸ Используем в пайплайне агента
validation_result = validate_output(
    raw_response='{"action": "search", "query": "LLM models"}',
    expected_schema={"action": str, "query": str}
)
print(f"Валидация: {validation_result}")

Шаг 4: Оценка (Evaluation) в LangSmith

LangSmith предоставляет LLM-as-Judge — автоматическую оценку качества ответов с помощью другой LLM:

from langsmith import Client
from langsmith.evaluation import evaluate, LangChainStringEvaluator

client = Client()

# ▸ Создаём датасет с тестовыми запросами
dataset = client.create_dataset(
    dataset_name="QantCore Agent Eval v1",
    description="Тестовые сценарии для оценки агента поддержки"
)

test_cases = [
    {"input": "Как подключить API?", "expected_output": "API ключ в разделе настроек"},
    {"input": "Какие лимиты у бесплатного тарифа?", "expected_output": "100 запросов в день"},
    {"input": "Что делать при ошибке 429?", "expected_output": "Превышен rate limit"},
]

for tc in test_cases:
    client.create_example(
        inputs={"input": tc["input"]},
        outputs={"output": tc["expected_output"]},
        dataset_id=dataset.id
    )

# ▸ Запускаем оценку — LLM-as-Judge
def target_function(inputs: dict) -> dict:
    """Функция, которую оцениваем (наш агент)."""
    result = agent.invoke({"messages": [("user", inputs["input"])]})
    return {"output": result["messages"][-1].content}

# ▸ Критерии оценки
evaluators = [
    LangChainStringEvaluator("cot_qa"),       # Chain-of-Thought QA оценка
    LangChainStringEvaluator("labeled_criteria", config={
        "criteria": "correctness"
    }),
]

results = evaluate(
    target_function,
    data="QantCore Agent Eval v1",
    evaluators=evaluators,
    experiment_prefix="qantcore-eval-run-001",
)
print(f"Результаты оценки: {results}")

LangFuse — это open-source (MIT) платформа observability для LLM-приложений. Ключевое преимущество: полный self-hosting на вашей инфраструктуре. Это критически важно для проектов с требованиями к конфиденциальности данных (GDPR, HIPAA, банковский сектор).

Установка LangFuse через Docker

docker-compose для self-hosted LangFuse

# docker-compose.yml
version: "3.8"
services:
  langfuse-server:
    image: ghcr.io/langfuse/langfuse:latest
    ports:
      - "3000:3000"
    environment:
      - DATABASE_URL=postgresql://postgres:postgres@postgres:5432/postgres
      - NEXTAUTH_SECRET=mysecret
      - SALT=mysalt
      - ENCRYPTION_KEY=0000000000000000000000000000000000000000000000000000000000000000
    depends_on:
      - postgres

  postgres:
    image: postgres:15
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=postgres
    volumes:
      - langfuse_pgdata:/var/lib/postgresql/data

volumes:
  langfuse_pgdata:

Запуск

docker-compose up -d
✔ Container langfuse-postgres-1  Started
✔ Container langfuse-server-1    Started
LangFuse доступен на http://localhost:3000

Интеграция LangFuse с Python-агентом

В отличие от LangSmith, LangFuse не привязан к LangChain — он работает с любым LLM-фреймворком через универсальный SDK:

pip install langfuse openai

from langfuse import Langfuse
from openai import OpenAI
import time

# ▸ Инициализация LangFuse
langfuse = Langfuse(
    secret_key="sk-lf-...",
    public_key="pk-lf-...",
    host="http://localhost:3000"  # self-hosted URL
)

openai_client = OpenAI(api_key="sk-...")

# ▸ Создаём трейс для агента
trace = langfuse.trace(
    name="qantcore-agent-query",
    user_id="user_42",
    session_id="session_abc123",
    metadata={"agent_version": "v2.1.0", "env": "production"}
)

# ▸ Шаг 1: Первый LLM-вызов (reasoning)
span_reasoning = trace.span(
    name="agent-reasoning",
    input={"query": "Какие тарифы QantCore?"},
)
t0 = time.time()
response_1 = openai_client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "Ты агент поддержки QantCore"},
        {"role": "user", "content": "Какие тарифы?"}
    ],
    temperature=0.1
)
latency_1 = (time.time() - t0) * 1000  # ms

# ▸ Завершаем спан с метриками
span_reasoning.end(
    output={"decision": response_1.choices[0].message.content},
    metadata={
        "model": "gpt-4o",
        "latency_ms": latency_1,
        "prompt_tokens": response_1.usage.prompt_tokens,
        "completion_tokens": response_1.usage.completion_tokens,
        "total_tokens": response_1.usage.total_tokens,
        "cost_usd": round(
            response_1.usage.prompt_tokens * 0.000005 +
            response_1.usage.completion_tokens * 0.000015, 6
        )
    }
)

# ▸ Шаг 2: Вызов инструмента (поиск по БЗ)
span_tool = trace.span(
    name="tool-search-kb",
    input={"query": "тарифы qantcore"},
)
# Здесь реальный вызов к вашей БЗ или API
kb_result = "Тарифы: Free — 100 запросов/день, Pro — 1000 запросов/день, Enterprise — ∞"
span_tool.end(
    output={"results": kb_result},
    metadata={"source": "internal_kb", "confidence": 0.95}
)

# ▸ Шаг 3: Финальный LLM-вызов (генерация ответа)
span_final = trace.span(
    name="agent-final-response",
    input={"kb_results": kb_result},
)
t2 = time.time()
response_2 = openai_client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "Сформулируй дружелюбный ответ на основе данных"},
        {"role": "user", "content": f"Результаты: {kb_result}"}
    ]
)
latency_2 = (time.time() - t2) * 1000
final_answer = response_2.choices[0].message.content
span_final.end(
    output={"answer": final_answer},
    metadata={
        "model": "gpt-4o-mini",
        "latency_ms": latency_2,
        "total_tokens": response_2.usage.total_tokens,
    }
)

# ▸ Обновляем общий трейс с итоговыми метриками
trace.update(
    output={"final_answer": final_answer},
    metadata={
        "total_latency_ms": latency_1 + latency_2,
        "total_tokens": response_1.usage.total_tokens + response_2.usage.total_tokens,
        "tool_calls_count": 1,
    }
)

# ▸ Обязательный flush для гарантированной отправки
langfuse.flush()
print(f"Трейс отправлен в LangFuse. Ответ: {final_answer}")

LangFuse с декораторами (современный подход)

Начиная с версии 2.x, LangFuse поддерживает декораторы, похожие на LangSmith @traceable:

from langfuse.decorators import observe, langfuse_context

@observe(as_type="generation")
def call_llm(prompt: str, model: str = "gpt-4o") -> str:
    """LLM-вызов с автоматической трассировкой."""
    # langfuse_context даёт доступ к текущему трейсу
    langfuse_context.update_current_trace(
        tags=["llm-call", model],
        metadata={"caller": "agent_orchestrator"}
    )
    client = OpenAI()
    resp = client.chat.completions.create(
        model=model, messages=[{"role": "user", "content": prompt}]
    )
    return resp.choices[0].message.content

@observe()
def agent_pipeline(user_query: str) -> dict:
    """Основной пайплайн агента — каждый вызов call_llm создаст вложенный спан."""
    reasoning = call_llm(f"Проанализируй запрос: {user_query}")
    final_answer = call_llm(f"Сгенерируй ответ: {reasoning}")
    return {"reasoning": reasoning, "answer": final_answer}

# ▸ Вызов — LangFuse автоматически создаст трейс
result = agent_pipeline("Как сменить тарифный план QantCore?")
langfuse_context.flush()
print(f"Ответ: {result['answer']}")

Observability без метрик — это просто красивые графики без смысла. Нужны конкретные числовые показатели, которые позволяют принимать решения. Разберём ключевые метрики для AI-агентов.

1. Латентность (Latency)

Для агентов критично измерять латентность на разных уровнях:

• Time-to-First-Token (TTFT) — время от запроса до первого токена первого LLM-вызова. Важно для UX: пользователь не должен ждать > 2 секунд перед тем, как увидит, что агент «думает».
• End-to-End Latency — полное время от запроса до финального ответа. Для агентов, которые делают несколько LLM-вызовов, это может быть 10–30+ секунд.
• Per-Step Latency — время каждого отдельного LLM-вызова внутри агента. Помогает найти узкие места (например, вызов GPT-4o для тривиальной классификации — там достаточно GPT-4o-mini).

Расчёт перцентилей латентности через LangSmith SDK

from langsmith import Client
import statistics

client = Client()

# ▸ Получаем все трейсы за последние 24 часа
traces = client.list_runs(
    project_name="qantcore-agent-prod",
    start_time="2026-07-03T06:00:00Z",
)

latencies = []
for run in traces:
    if run.end_time and run.start_time:
        lat_ms = (run.end_time - run.start_time).total_seconds() * 1000
        latencies.append(lat_ms)

if latencies:
    print(f"📊 Статистика латентности (n={len(latencies)}):")
    print(f"   P50: {statistics.median(latencies):.0f} ms")
    print(f"   P95: {sorted(latencies)[int(len(latencies) * 0.95)]:.0f} ms")
    print(f"   P99: {sorted(latencies)[int(len(latencies) * 0.99)]:.0f} ms")
    print(f"   Max: {max(latencies):.0f} ms")

2. Потребление токенов и стоимость

Цены на токены различаются для разных моделей. GPT-4o: $5/1M input токенов, $15/1M output токенов. GPT-4o-mini: $0.15/$0.60. Разница в 33 раза!

Автоматический расчёт стоимости в трейсе

MODEL_PRICING = {
    "gpt-4o":       {"input": 5.00,  "output": 15.00},  # $/1M tokens
    "gpt-4o-mini":  {"input": 0.15,  "output": 0.60},
    "claude-3-5-sonnet": {"input": 3.00, "output": 15.00},
}

def calc_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    """Рассчитывает стоимость LLM-вызова в USD."""
    pricing = MODEL_PRICING.get(model)
    if not pricing:
        return 0.0
    cost = (input_tokens / 1_000_000) * pricing["input"] + \
           (output_tokens / 1_000_000) * pricing["output"]
    return round(cost, 6)

# ▸ Пример: один вызов GPT-4o с 2000 input и 500 output токенов
cost = calc_cost("gpt-4o", 2000, 500)
print(f"Стоимость вызова: ${cost}")
# → Стоимость вызова: $0.0175

3. Процент успешных завершений (Success Rate)

Измеряйте, какой процент запусков агента завершается успешно (без исключений, без превышения лимита итераций, с валидным JSON на выходе):

from langfuse import Langfuse

langfuse = Langfuse(secret_key="sk-lf-...", public_key="pk-lf-...")

# ▸ LangFuse позволяет добавлять score к трейсам
def score_agent_run(trace_id: str, success: bool, note: str = ""):
    """Выставляет оценку (score) для трейса агента."""
    langfuse.score(
        trace_id=trace_id,
        name="success",
        value=1.0 if success else 0.0,
        comment=note
    )

# ▸ После каждого запуска агента:
try:
    result = agent_pipeline(user_query)
    score_agent_run(trace_id, True, "OK")
except Exception as e:
    score_agent_run(trace_id, False, str(e))
    raise

Промпт — это код. Как и любой код, его нужно версионировать, тестировать и развёртывать контролируемо. И LangSmith, и LangFuse предоставляют для этого инструменты.

LangSmith Prompt Hub

LangSmith позволяет хранить промпты централизованно и подтягивать их в код по имени и версии:

from langsmith import Client

client = Client()

# ▸ Публикуем промпт в Prompt Hub (один раз, через UI или API)
client.pull_prompt("qantcore-agent-system")
# Возвращает актуальную production-версию промпта

# ▸ Использование промпта с привязкой к конкретному коммиту
prompt = client.pull_prompt(
    "qantcore-agent-system",
    include_model=True  # промпт может содержать настройки модели
)
messages = prompt.invoke({
    "user_name": "Анна",
    "kb_context": "QantCore — платформа для AI-агентов",
})
print(f"Системный промпт: {messages[0].content[:100]}...")

LangFuse Prompt Management

LangFuse использует похожий подход, но с явным указанием версий:

from langfuse import Langfuse

langfuse = Langfuse(secret_key="sk-lf-...", public_key="pk-lf-...")

# ▸ Получаем production-версию промпта
prompt = langfuse.get_prompt("qantcore-agent")

# ▸ Компилируем промпт с переменными
compiled = prompt.compile(
    user_name="Анна",
    company="QantCore",
    context="Платформа для создания и мониторинга AI-агентов"
)

# ▸ Логируем использование конкретной версии промпта в трейсе
trace = langfuse.trace(name="agent-query")
generation = trace.generation(
    name="agent-response",
    model="gpt-4o",
    prompt=prompt,  # LangFuse автоматически свяжет генерацию с версией промпта
)
print(f"Версия промпта: {prompt.version}, лейбл: {prompt.name}")

A/B-тестирование промптов

Классический сценарий: вы хотите проверить, даёт ли новый промпт лучшие результаты. Стратегия:

1. Публикуете промпт v2 с лейблом experiment в Prompt Hub.
2. 10% трафика направляете на v2, 90% — на стабильный v1.
3. Сравниваете метрики: success rate, user satisfaction score, latency, cost.
4. Если v2 лучше — продвигаете его в production.

import random
from langfuse import Langfuse

langfuse = Langfuse(secret_key="sk-lf-...", public_key="pk-lf-...")

def get_prompt_for_request(user_id: str) -> str:
    """A/B-тест: 10% пользователей получают экспериментальный промпт."""
    if hash(user_id) % 10 == 0:
        # Экспериментальная ветка (10%)
        return langfuse.get_prompt("qantcore-agent", label="experiment")
    else:
        # Стабильная ветка (90%)
        return langfuse.get_prompt("qantcore-agent", label="production")

# ▸ В коде агента:
def handle_user_query(user_id: str, query: str) -> str:
    prompt = get_prompt_for_request(user_id)
    trace = langfuse.trace(
        name="agent-query",
        user_id=user_id,
        metadata={"prompt_label": prompt.label}
    )
    # ... запуск агента с этим промптом ...
    return "ответ агента"

Трассировка и метрики бесполезны, если вы не узнаёте о проблемах до того, как их заметят пользователи. Настройте автоматические алерты и непрерывную оценку качества.

Алертинг через LangFuse Webhooks + Slack

LangFuse поддерживает вебхуки, которые срабатывают при определённых событиях. Интегрируйте их с вашей системой алертинга:

import requests
import os
from datetime import datetime, timedelta
from langfuse import Langfuse

langfuse = Langfuse(secret_key="sk-lf-...", public_key="pk-lf-...")
SLACK_WEBHOOK = os.environ["SLACK_WEBHOOK_URL"]

def send_slack_alert(title: str, message: str, color: str = "#ff0000"):
    """Отправляет алерт в Slack через webhook."""
    payload = {
        "attachments": [{
            "color": color,
            "title": title,
            "text": message,
            "footer": "QantCore Agent Monitor",
            "ts": int(datetime.now().timestamp())
        }]
    }
    requests.post(SLACK_WEBHOOK, json=payload)

def check_agent_health():
    """Проверяет здоровье агента за последние 5 минут."""
    cutoff = datetime.now() - timedelta(minutes=5)

    # ▸ LangFuse API: получаем метрики за период
    import langfuse.api as lf_api
    metrics = lf_api.get_metrics(
        name="success_rate",
        from_timestamp=cutoff.isoformat(),
    )

    success_rate = metrics.get("value", 1.0)

    if success_rate < 0.95:
        send_slack_alert(
            title="🚨 Агент QantCore: низкий Success Rate!",
            message=f"Success Rate за 5 мин: {success_rate * 100:.1f}% (порог: 95%)\n"
                      f"Проверьте дашборд: https://langfuse.qantcore.space",
            color="#ff0000"
        )

    if success_rate < 0.98:
        send_slack_alert(
            title="⚠️ Агент QantCore: повышенное количество ошибок",
            message=f"Success Rate за 5 мин: {success_rate * 100:.1f}%",
            color="#ffa500"
        )

# ▸ Запускать по cron каждые 5 минут
if __name__ == "__main__":
    check_agent_health()

Continuous Evaluation (непрерывная оценка)

Continuous Evaluation — это автоматический прогон агента на наборе тестовых сценариев при каждом изменении промпта или кода. Интеграция в CI/CD:

# .github/workflows/eval.yml
name: Agent Evaluation
on:
  push:
    paths: ['agent/**', 'prompts/**']
  pull_request:
    paths: ['agent/**', 'prompts/**']

jobs:
  eval:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: {python-version: '3.11'}
      - run: pip install -r requirements.txt
      - name: Run LangSmith Evals
        env:
          LANGCHAIN_API_KEY: ${{ secrets.LANGCHAIN_API_KEY }}
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: python scripts/run_evals.py --dataset "prod-eval"

Скрипт run_evals.py

import sys
from langsmith import Client
from langsmith.evaluation import evaluate

def run_ci_eval(dataset_name: str):
    """Прогоняет оценку и падает с exit code 1 при плохих метриках."""
    from my_agent import agent_pipeline

    results = evaluate(
        lambda inputs: {"output": agent_pipeline(inputs["query"])},
        data=dataset_name,
        evaluators=["cot_qa"],
        experiment_prefix="ci-run",
    )

    # ▸ Пороговые значения для CI/CD
    avg_score = sum(r["score"] for r in results) / len(results)
    print(f"Средний score: {avg_score:.2f}")

    if avg_score < 0.80:
        print("❌ Оценка ниже порога 0.80 — деплой заблокирован!")
        sys.exit(1)
    else:
        print("✅ Оценка пройдена, деплой разрешён.")

if __name__ == "__main__":
    run_ci_eval(sys.argv[1] if len(sys.argv) > 1 else "prod-eval")

Сбор обратной связи от пользователей

Интегрируйте пользовательские оценки (👍/👎) в трейсы для построения датасетов дообучения:

from langfuse import Langfuse

langfuse = Langfuse(secret_key="sk-lf-...", public_key="pk-lf-...")

def record_user_feedback(trace_id: str, rating: int, comment: str = ""):
    """Записывает пользовательскую оценку (1-5) в трейс.
    
    Вызывается из вашего API при нажатии 👍/👎 пользователем.
    """
    langfuse.score(
        trace_id=trace_id,
        name="user_satisfaction",
        value=rating / 5.0,  # нормализуем в 0-1
        comment=comment,
        data_type="NUMERIC"
    )

# ▸ Пример использования в FastAPI-эндпоинте:
# @app.post("/feedback")
# async def submit_feedback(trace_id: str, rating: int, comment: str = ""):
#     record_user_feedback(trace_id, rating, comment)
#     return {"status": "ok"}

Когда базовых дашбордов LangSmith/LangFuse недостаточно, стройте собственные на основе их API. LangFuse предоставляет полный REST API для извлечения трейсов, спанов и метрик.

Экспорт трейсов в Pandas для анализа

import pandas as pd
from langfuse import Langfuse
from datetime import datetime, timedelta

langfuse = Langfuse(secret_key="sk-lf-...", public_key="pk-lf-...")

def export_traces_to_dataframe(hours: int = 24) -> pd.DataFrame:
    """Экспортирует все трейсы за последние N часов в DataFrame."""
    from langfuse.api.resources.trace import TraceListResponse

    # Получаем трейсы через LangFuse API
    traces_data = langfuse.fetch_traces(
        from_timestamp=datetime.now() - timedelta(hours=hours),
        limit=1000
    )

    rows = []
    for trace in traces_data.data:
        rows.append({
            "trace_id": trace.id,
            "user_id": trace.user_id,
            "timestamp": trace.timestamp,
            "latency_ms": (trace.latency or 0) / 1_000_000,  # ns → ms
            "total_tokens": trace.total_tokens or 0,
            "total_cost": trace.total_cost or 0.0,
            "observation_count": len(trace.observations) if trace.observations else 0,
        })

    return pd.DataFrame(rows)

# ▸ Анализируем данные
df = export_traces_to_dataframe(hours=24)
print(f"Трейсов за 24ч: {len(df)}")
print(f"Средняя латентность: {df['latency_ms'].median():.0f} ms")
print(f"Суммарные затраты: ${df['total_cost'].sum():.4f}")
print(f"Среднее токенов на трейс: {df['total_tokens'].median():.0f}")

# ▸ Топ-5 самых дорогих запросов
top_expensive = df.nlargest(5, "total_cost")
print("\n💸 Топ-5 дорогих запросов:")
print(top_expensive[["trace_id", "total_cost", "total_tokens"]].to_string(index=False))

Кастомный дашборд с Grafana + PostgreSQL

LangFuse хранит все данные в PostgreSQL — используйте это для прямых SQL-запросов и Grafana-дашбордов:

-- Запрос для Grafana: почасовая стоимость использования агента
SELECT
    date_trunc('hour', "timestamp") AS hour,
    COUNT(*) AS request_count,
    SUM(COALESCE(total_cost, 0)) AS total_cost_usd,
    AVG(latency) / 1000000 AS avg_latency_ms,
    PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY latency) / 1000000 AS p95_latency_ms
FROM traces
WHERE "timestamp" >= NOW() - INTERVAL '7 days'
GROUP BY 1
ORDER BY 1;