Langfuse: пошаговый туториал LLM observability

LLM-приложение в production без observability — чёрный ящик. Пользователь получил плохой ответ? Непонятно, какой промпт отработал, сколько токенов ушло, была ли ошибка на промежуточном шаге. HTTP 200 OK ничего не говорит о качестве — модель может вернуть корректный JSON с бессмысленным содержимым.

Langfuse — open-source платформа (MIT) для LLM observability: трейсинг, prompt management, evaluations, cost tracking. Self-hosted без лимитов, 22 000+ звёзд на GitHub. Python SDK v4 построен на OpenTelemetry — совместимость с любой OTEL-инструментацией из коробки.

В этом туториале — полный путь от установки до production: каждый компонент с примерами кода на Python.

Что такое Langfuse

Langfuse решает четыре задачи, которые стандартные APM-инструменты (Datadog, Grafana) не покрывают для LLM-приложений:

Трейсинг. Каждый запрос пользователя раскладывается на дерево операций: LLM-вызовы, retrieval, preprocessing, postprocessing. Видно input/output каждого шага, latency, модель, количество токенов.

Prompt Management. Промпты живут не в коде, а в Langfuse. Версионирование, labels (production/staging), откат в один клик. Изменение промпта применяется без деплоя.

Evaluations. Автоматическая оценка качества ответов: LLM-as-a-Judge, human annotations, custom scores через API. Числа вместо «вроде работает».

Cost Tracking. Автоматический расчёт стоимости каждого LLM-вызова по модели и токенам. Разбивка по фичам, пользователям, моделям.

┌─────────────────────────────────────────────────────┐
│                  LLM Observability                   │
├──────────┬──────────┬──────────────┬────────────────┤
│ Tracing  │   Cost   │   Prompt     │  Evaluation    │
│          │ Tracking │  Management  │                │
├──────────┼──────────┼──────────────┼────────────────┤
│ Что      │ Сколько  │ Какой промпт │ Насколько      │
│ произошло│ стоило   │ в продакшене │ хороший ответ  │
└──────────┴──────────┴──────────────┴────────────────┘

Зачем нужен Langfuse

Без observability три вещи остаются в слепой зоне.

Дебаг

Пользователь сообщает: «AI дал неправильный ответ». Без трейсинга вы открываете код, ставите print-ы, пытаетесь воспроизвести. С Langfuse — открываете трейс по user_id, видите точный промпт, ответ модели, все промежуточные шаги. Проблема локализована за минуту.

Стоимость

Один неоптимальный промпт может стоить сотни долларов в месяц. GPT-5.5 с контекстом на 10 000 токенов — 3-5 центов за вызов. При 1 000 пользователей в день разница между хорошим и плохим промптом ощутима в счёте. Без трекинга вы узнаёте об этом post factum — из счёта от OpenAI.

Качество

«Вроде стало лучше» — не метрика. Изменили промпт — качество выросло или упало? Без evaluations это вопрос веры, с evaluations — числа: средний score по relevance до и после изменения.

Langfuse vs альтернативы

	Langfuse	LangSmith	Helicone
Лицензия	MIT	Проприетарный	Apache 2.0
Self-hosted	Бесплатно, без лимитов	Enterprise only	Да
Framework lock-in	Нет	LangChain-first	OpenAI-first
Prompt management	Да + MCP-сервер	Да	Да (beta)
Evaluations	LLM-as-Judge + custom	LLM-as-Judge + custom	Базовые
Free tier (cloud)	50k units/мес (Hobby)	5k traces/мес	100k req/мес

Langfuse не привязан к фреймворку. OpenAI, Anthropic, open-source модели, LangChain, LlamaIndex, LiteLLM — всё через один SDK.

Установка

Два варианта: self-hosted (Docker) или cloud (langfuse.com). Для туториала начнём с self-hosted — полный контроль, без лимитов.

Self-hosted через Docker Compose

Минимальные требования: Docker, 4 ядра, 16 ГБ RAM (рекомендация Langfuse). Langfuse v3 — два контейнера приложения (web + worker) плюс четыре storage-сервиса: PostgreSQL, ClickHouse, Redis и S3-совместимое хранилище (MinIO для local). Обзор более простой v2-архитектуры и мониторинга самого Langfuse — в обзорной статье.

Dev vs production. Приведённый ниже compose — для локального запуска и знакомства с платформой. Он не предназначен для production: отсутствуют high-availability, автоматические бэкапы и scaling. Для production Langfuse рекомендует Kubernetes. Если всё же используете compose в production — обязательно смените все дефолтные секреты (NEXTAUTH_SECRET, SALT, пароли PostgreSQL и MinIO) на случайные значения (openssl rand -base64 32). Актуальный официальный compose всегда доступен по curl -LO https://langfuse.com/docker-compose.yml.

# docker-compose.yml (для локальной разработки и знакомства)
# Актуальная версия: curl -LO https://langfuse.com/docker-compose.yml

services:
  langfuse-web:
    image: langfuse/langfuse:3
    ports:
      - "3000:3000"
    environment:
      DATABASE_URL: postgresql://postgres:postgres@db:5432/langfuse
      NEXTAUTH_SECRET: your-secret-key-change-me  # openssl rand -base64 32 — ОБЯЗАТЕЛЬНО сменить в production
      SALT: your-salt-change-me                    # openssl rand -base64 32 — ОБЯЗАТЕЛЬНО сменить в production
      NEXTAUTH_URL: http://localhost:3000
      CLICKHOUSE_URL: http://clickhouse:8123
      REDIS_HOST: redis
      LANGFUSE_S3_EVENT_UPLOAD_BUCKET: langfuse
      LANGFUSE_S3_EVENT_UPLOAD_ENDPOINT: http://minio:9000
      LANGFUSE_S3_EVENT_UPLOAD_ACCESS_KEY_ID: minio
      LANGFUSE_S3_EVENT_UPLOAD_SECRET_ACCESS_KEY: miniopassword  # сменить в production
      LANGFUSE_S3_EVENT_UPLOAD_FORCE_PATH_STYLE: "true"
    depends_on:
      - db
      - clickhouse
      - redis
      - minio

  langfuse-worker:
    image: langfuse/langfuse:3
    environment:
      DATABASE_URL: postgresql://postgres:postgres@db:5432/langfuse
      NEXTAUTH_SECRET: your-secret-key-change-me
      SALT: your-salt-change-me
      CLICKHOUSE_URL: http://clickhouse:8123
      REDIS_HOST: redis
      LANGFUSE_S3_EVENT_UPLOAD_BUCKET: langfuse
      LANGFUSE_S3_EVENT_UPLOAD_ENDPOINT: http://minio:9000
      LANGFUSE_S3_EVENT_UPLOAD_ACCESS_KEY_ID: minio
      LANGFUSE_S3_EVENT_UPLOAD_SECRET_ACCESS_KEY: miniopassword
      LANGFUSE_S3_EVENT_UPLOAD_FORCE_PATH_STYLE: "true"
    command: ["node", "packages/worker/dist/index.js"]
    depends_on:
      - db
      - clickhouse
      - redis
      - minio

  db:
    image: postgres:16
    environment:
      POSTGRES_PASSWORD: postgres  # сменить в production
      POSTGRES_DB: langfuse
    volumes:
      - langfuse_pg:/var/lib/postgresql/data

  clickhouse:
    image: clickhouse/clickhouse-server:24
    volumes:
      - langfuse_ch:/var/lib/clickhouse

  redis:
    image: redis:7
    volumes:
      - langfuse_redis:/data

  minio:
    image: minio/minio
    command: server /data --console-address ":9001"
    environment:
      MINIO_ROOT_USER: minio
      MINIO_ROOT_PASSWORD: miniopassword  # сменить в production
    volumes:
      - langfuse_minio:/data

volumes:
  langfuse_pg:
  langfuse_ch:
  langfuse_redis:
  langfuse_minio:

docker compose up -d

Через 2-3 минуты Langfuse доступен на localhost:3000. Создайте аккаунт, организацию, проект. В настройках проекта скопируйте Public Key и Secret Key — они понадобятся для SDK.

Cloud (langfuse.com)

Регистрация на cloud.langfuse.com. Создайте проект, скопируйте ключи. Hobby план — 50 000 units в месяц бесплатно. Достаточно для разработки и небольших проектов.

Установка Python SDK

pip install langfuse

Версии SDK. Статья написана под SDK v4 (OpenTelemetry-based, from langfuse import get_client). SDK v4 вышел в марте 2026 и с тех пор является стандартом — pip install langfuse без pin устанавливает v4. Основные breaking changes относительно v3: update_current_trace() разложен на propagate_attributes(), set_current_trace_io() и set_current_trace_as_public(); start_span() / start_generation() → start_observation(as_type=...); DatasetItemClient.run() удалён, замена — dataset.run_experiment(); Pydantic v1 больше не поддерживается. Базовые паттерны (@observe, get_client, update_current_span) работают без изменений. Полный список — миграционный гайд v3→v4.

SDK автоматически определяет конфигурацию через переменные окружения:

export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_BASE_URL="http://localhost:3000"  # self-hosted
# export LANGFUSE_BASE_URL="https://cloud.langfuse.com"  # cloud EU
# export LANGFUSE_BASE_URL="https://us.cloud.langfuse.com"  # cloud US

Проверка подключения:

from langfuse import get_client

langfuse = get_client()
# Если переменные окружения заданы — клиент готов к работе

Трейсинг

Трейсинг стоит подключить первым. Без него остальные компоненты Langfuse не работают: evaluations привязываются к трейсам, cost tracking считается по generations внутри трейсов.

Структура трейса

Трейс в Langfuse — дерево. На верхнем уровне — trace (один запрос пользователя). Внутри — spans (промежуточные шаги) и generations (LLM-вызовы).

Trace: "generate-travel-plan"
│
├── Span: "validate-input" (8ms)
│
├── Span: "retrieve-context" (200ms)
│   └── Span: "vector-search" (180ms)
│
├── Generation: "plan-draft" (GPT-5.5, 1800 tokens, $0.015)
│
├── Generation: "plan-review" (GPT-5.4-Mini, 600 tokens, $0.001)
│
└── Span: "format-output" (3ms)

Для каждой generation видно: input/output (полный промпт и ответ), модель, токены (input/output/total), стоимость в долларах, latency, metadata.

@observe — основной способ трейсинга

Декоратор @observe — самый простой способ инструментировать код. Создаёт span для функции, захватывает аргументы, возвращаемое значение и время выполнения. Вложенность — автоматическая, через OpenTelemetry context propagation.

from langfuse import get_client, observe
from openai import OpenAI

openai_client = OpenAI()
langfuse = get_client()


@observe()
def validate_input(user_query: str) -> str:
    """Валидация и нормализация пользовательского запроса."""
    cleaned = user_query.strip().lower()
    if len(cleaned) < 3:
        raise ValueError("Запрос слишком короткий")
    return cleaned


@observe()
def search_context(query: str) -> list[str]:
    """Поиск релевантного контекста в базе знаний."""
    # Ваша логика поиска: vector DB, Elasticsearch, etc.
    results = vector_db.search(query, top_k=5)
    return [doc.text for doc in results]


@observe(as_type="generation")
def generate_response(query: str, context: list[str]) -> str:
    """LLM-вызов для генерации ответа."""
    context_text = "\n".join(context)
    response = openai_client.chat.completions.create(
        model="gpt-5.5",
        messages=[
            {"role": "system", "content": f"Контекст:\n{context_text}"},
            {"role": "user", "content": query},
        ],
    )
    return response.choices[0].message.content


@observe()
def handle_request(user_query: str, user_id: str) -> str:
    """Главная функция обработки запроса. Создаёт корневой trace."""
    query = validate_input(user_query)
    context = search_context(query)
    response = generate_response(query, context)
    return response


# Вызов
result = handle_request("Порекомендуй кафе в Москве", user_id="user-123")
langfuse.flush()

Что произойдёт:

• handle_request создаст корневой trace
• validate_input, search_context, generate_response — вложенные spans
• generate_response отмечен как as_type="generation" — будет отображаться как LLM-вызов с моделью и токенами
• Вложенность автоматическая — OpenTelemetry propagation

Обновление текущего span

Внутри @observe-функции можно добавить метаданные через update_current_span(). В SDK v4 user_id и session_id передаются через propagate_attributes() — они автоматически распространяются на все дочерние observations:

from langfuse import get_client, observe, propagate_attributes

langfuse = get_client()


@observe()
def process_with_metadata(query: str, user_id: str) -> str:
    # user_id и session_id — через propagate_attributes (SDK v4)
    with propagate_attributes(
        user_id=user_id,
        metadata={"source": "api", "version": "2.1"},
        tags=["production", "feature:search"],
    ):
        result = do_something(query)

        # Обновляем output, если нужно кастомное значение
        langfuse.update_current_span(output={"processed_result": result})
        return result

Низкоуровневый API

Для случаев, когда декоратор неудобен (динамическое создание spans, интеграция с event loop):

from langfuse import get_client

langfuse = get_client()

# Context manager — автоматическое вложение
with langfuse.start_as_current_observation(name="process-request") as span:
    span.update(input={"query": "Кафе в Москве"})

    # Вложенный span
    with langfuse.start_as_current_observation(name="search-places") as search_span:
        results = search_places("Кафе в Москве")
        search_span.update(output={"count": len(results)})

    # Generation
    with langfuse.start_as_current_observation(
        as_type="generation",
        name="llm-response",
        model="gpt-5.5",
        input=[{"role": "user", "content": "Кафе в Москве"}],
    ) as generation:
        response = call_llm("Кафе в Москве")
        generation.update(
            output=response,
            usage_details={"input_tokens": 42, "output_tokens": 128},
        )

    span.update(output={"response": response})

langfuse.flush()

Что трейсинг показывает сразу

Три паттерна, которые обнаруживаются в первый же день:

Скрытые retry. Retry-логика вызывает модель 2-3 раза на один запрос. Без трейсинга видно только финальный ответ. С трейсингом — каждый вызов со стоимостью.

Model mismatch. Один эндпоинт должен использовать GPT-5.4-Mini, но тянет GPT-5.5. Фильтр по полю model в Langfuse сразу покажет несоответствие.

Latency bottleneck. В цепочке из четырёх шагов один занимает 80% времени. Без spans — только общее время. С ними — точное место для оптимизации.

Prompt Management

Стандартный подход: промпт зашит в код. Изменить — PR, code review, merge, deploy. A/B тест — два деплоя. Langfuse выносит промпты из кода.

Создание промпта в UI

В Langfuse UI: Prompts → New Prompt. Два типа:

Text prompt — одна строка с переменными:

Ты — travel-ассистент. Порекомендуй места в {{destination}}.
Учитывай предпочтения: {{preferences}}.

Chat prompt — массив messages:

[
  {"role": "system", "content": "Ты — travel-ассистент для {{destination}}."},
  {"role": "user", "content": "{{user_query}}"}
]

Переменные в двойных фигурных скобках {{variable}} заменяются при компиляции.

Загрузка и компиляция промпта

from langfuse import get_client

langfuse = get_client()

# Загрузка промпта (по умолчанию — версия с label "production")
prompt = langfuse.get_prompt("travel-assistant")

# Компиляция с переменными
compiled = prompt.compile(
    destination="Москва",
    preferences="вегетарианская кухня",
    user_query="Порекомендуй кафе в центре",
)

# compiled — строка (для text prompt) или список messages (для chat prompt)
# Используем в LLM-вызове
response = openai_client.chat.completions.create(
    model="gpt-5.5",
    messages=compiled,
)

Версионирование и labels

Каждое сохранение промпта — новая версия (1, 2, 3…). Labels назначаются на конкретные версии:

• production — текущая рабочая версия. get_prompt("name") возвращает её по умолчанию.
• staging — версия для тестирования.
• latest — последняя созданная версия.

# Production-версия (по умолчанию)
prod_prompt = langfuse.get_prompt("travel-assistant")

# Staging-версия для тестирования
staging_prompt = langfuse.get_prompt("travel-assistant", label="staging")

# Конкретная версия по номеру
v3_prompt = langfuse.get_prompt("travel-assistant", version=3)

Рабочий процесс:

1. Создали новую версию промпта в UI
2. Назначили label staging
3. Протестировали на staging-окружении
4. Результаты устроили → переназначили label production на эту версию
5. Следующий запрос в production подтянет новый промпт — без деплоя

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

import random
from langfuse import get_client, observe, propagate_attributes

langfuse = get_client()


@observe()
def generate_with_ab_test(query: str) -> str:
    """A/B тест двух версий промпта."""
    variant = "A" if random.random() < 0.5 else "B"

    if variant == "A":
        prompt = langfuse.get_prompt("travel-assistant", label="production")
    else:
        prompt = langfuse.get_prompt("travel-assistant", label="staging")

    compiled = prompt.compile(user_query=query)

    # Тег для фильтрации в аналитике
    langfuse.update_current_span(
        tags=[f"ab-test:prompt-{variant}"],
        metadata={"prompt_variant": variant},
    )

    response = openai_client.chat.completions.create(
        model="gpt-5.5",
        messages=compiled,
    )
    return response.choices[0].message.content

В Langfuse UI фильтруете по тегу ab-test:prompt-A vs ab-test:prompt-B, сравниваете scores и cost.

Fallback-паттерн

Langfuse — внешняя зависимость. Если он недоступен, промпт всё равно нужен.

FALLBACK_MESSAGES = [
    {"role": "system", "content": "Ты — travel-ассистент. Помогай с рекомендациями."},
    {"role": "user", "content": "{user_query}"},
]


def get_prompt_safe(name: str, **variables) -> list[dict]:
    """Загрузка промпта с fallback на хардкод."""
    try:
        prompt = langfuse.get_prompt(name, label="production")
        return prompt.compile(**variables)
    except Exception:
        # Langfuse недоступен — используем fallback
        return [
            {**msg, "content": msg["content"].format(**variables)}
            for msg in FALLBACK_MESSAGES
        ]

В production этот паттерн обязателен. Langfuse ускоряет итерацию, но не должен быть single point of failure.

Кеширование

SDK кеширует промпты клиент-стороне. По умолчанию TTL не задан — промпт запрашивается при каждом вызове get_prompt(). Для production задавайте TTL:

# Кешировать промпт на 5 минут
prompt = langfuse.get_prompt("travel-assistant", cache_ttl_seconds=300)

cache_ttl_seconds управляет клиентским кешем — сколько времени SDK использует локальную копию вместо запроса к серверу. Это не влияет на provider-side caching (Anthropic prompt caching, OpenAI caching) — два механизма независимы.

MCP-сервер: Langfuse прямо в IDE

У Langfuse есть MCP-сервер — управляйте платформой прямо из Claude Code, Cursor и других AI-ассистентов. С мая 2026 MCP охватывает не только промпты: observations, scores, datasets, annotation queues, metrics и комментарии.

{
  "mcpServers": {
    "langfuse": {
      "type": "http",
      "url": "https://your-langfuse.com/api/public/mcp",
      "headers": {
        "Authorization": "Basic base64(publicKey:secretKey)"
      }
    }
  }
}

AI-ассистент видит текущие промпты, трейсы и scores, предлагает изменения, применяет их через MCP — без переключения между IDE и браузером.

Evaluations

Трейсинг показывает что произошло. Evaluations — насколько хорошо.

Три подхода к оценке

LLM-as-a-Judge. Одна модель оценивает ответы другой. Масштабируется, но дорого и не всегда точно.

Human Annotations. Ручная оценка через UI Langfuse. Точно, но не масштабируется. Подходит для калибровки LLM-as-Judge.

Custom Scores via SDK. Программная оценка через API: regex-проверки, метрики из кода, пользовательский фидбек. Быстро, дёшево, закрывает механические проверки.

LLM-as-a-Judge

Настраивается в UI Langfuse: Evaluation → New Evaluator.

Параметры:

• Template — промпт для judge-модели (relevance, helpfulness, toxicity, correctness)
• Model — какая модель оценивает (GPT-5.5, Claude Sonnet)
• Target — какие трейсы оценивать (фильтр по тегам, имени, дате)
• Score name — имя метрики (например, relevance)
• Score type — Numeric (0-1), Categorical, Boolean

Langfuse прогоняет judge-модель по каждому подходящему трейсу и записывает score. В дашборде видно распределение scores, тренд по времени, корреляцию с другими метриками.

Evaluator’ы можно настраивать и через API / MCP-сервер — не только в UI. Можно оценивать не только trace целиком, но и отдельные observations (generation, span): в pipeline из четырёх шагов — отдельный evaluator для каждого шага. Типы scores: Numeric (0–1), Categorical, Boolean — выбираются при создании evaluator’а.

Custom Scores через SDK

Для автоматических проверок в коде:

from langfuse import get_client, observe

langfuse = get_client()


@observe()
def generate_and_evaluate(query: str) -> str:
    """Генерация ответа с автоматической оценкой."""
    response = call_llm(query)

    # Автоматическая проверка: ответ не пустой
    langfuse.score(
        name="not-empty",
        value=1 if len(response.strip()) > 0 else 0,
        data_type="NUMERIC",
    )

    # Автоматическая проверка: длина в разумных пределах
    langfuse.score(
        name="length-ok",
        value=1 if 50 < len(response) < 5000 else 0,
        data_type="NUMERIC",
    )

    # Проверка формата (если ожидаем JSON)
    try:
        import json
        json.loads(response)
        langfuse.score(name="valid-json", value=1, data_type="NUMERIC")
    except json.JSONDecodeError:
        langfuse.score(name="valid-json", value=0, data_type="NUMERIC")

    return response

Пользовательский фидбек как score

def record_user_feedback(trace_id: str, thumbs_up: bool, comment: str = ""):
    """Записать фидбек пользователя как score в Langfuse."""
    langfuse.score(
        trace_id=trace_id,
        name="user-feedback",
        value=1 if thumbs_up else 0,
        data_type="NUMERIC",
        comment=comment,
    )
    langfuse.flush()

Datasets: регрессионное тестирование промптов

Dataset — набор пар input/expected_output. Изменили промпт → прогнали dataset → сравнили scores с предыдущей версией.

from langfuse import get_client

langfuse = get_client()

# Создание dataset
langfuse.create_dataset(name="travel-queries-v1")

# Добавление test cases
test_cases = [
    {
        "input": {"query": "Кафе в центре Москвы"},
        "expected_output": "Список из 5+ кафе с адресами и рейтингами",
    },
    {
        "input": {"query": "Бюджетные отели в Казани"},
        "expected_output": "Список отелей до 3000 руб/ночь с описанием",
    },
    {
        "input": {"query": "Маршрут по Золотому кольцу на 3 дня"},
        "expected_output": "Подробный маршрут с остановками и логистикой",
    },
]

for case in test_cases:
    langfuse.create_dataset_item(
        dataset_name="travel-queries-v1",
        input=case["input"],
        expected_output=case["expected_output"],
    )

# Запуск эксперимента (SDK v4: run_experiment API)
dataset = langfuse.get_dataset("travel-queries-v1")


def my_task(*, item, **kwargs):
    """Функция, которую run_experiment выполнит для каждого item."""
    return run_pipeline(item.input["query"])


# run_experiment автоматически создаёт run, трейсирует каждый item
# и связывает результаты — ручной item.link() в v4 не нужен
dataset.run_experiment(
    name="prompt-v3-gpt5.5",
    task=my_task,
)

langfuse.flush()

В UI Langfuse: Datasets → travel-queries-v1 → Runs. Сравнение prompt-v2-gpt5.5 vs prompt-v3-gpt5.5 по каждому item с scores.

Cost Tracking

Langfuse считает стоимость каждого LLM-вызова автоматически по модели и токенам. Встроенная таблица pricing обновляется с каждым релизом — новые модели появляются в день выхода.

Что видно в дашборде

• Total cost за период (день, неделя, месяц)
• Cost per trace — средняя стоимость одного запроса пользователя
• Cost per user — расходы конкретного пользователя
• Cost per model — распределение между GPT-5.5, GPT-5.4-Mini, Claude Sonnet и т.д.
• Cost trend — динамика расходов по дням

Per-feature cost tracking

Добавляя теги к трейсам, вы группируете расходы по фичам:

from langfuse import get_client, observe, propagate_attributes

langfuse = get_client()


@observe()
def generate_itinerary(destination: str, user_id: str) -> str:
    with propagate_attributes(user_id=user_id, tags=["feature:itinerary", "tier:premium"]):
        # ... LLM-вызовы
        return result


@observe()
def chat_response(message: str, user_id: str) -> str:
    with propagate_attributes(user_id=user_id, tags=["feature:chat", "tier:free"]):
        # ... LLM-вызовы
        return result

Фильтр по тегу feature:itinerary покажет, сколько стоит генерация маршрутов. Отдельно — чат, отдельно — рекомендации.

Синхронизация расходов в свою БД

def sync_user_spending(user_id: str) -> float:
    """Получить суммарные расходы пользователя из Langfuse."""
    traces = langfuse.fetch_traces(user_id=user_id)
    total_cost = sum(t.total_cost or 0 for t in traces.data)

    # Записать в свою БД для лимитов, биллинга, аналитики
    db.update_user_spending(user_id=user_id, amount=total_cost)
    return total_cost

Нужно для freemium с кредитами, per-user лимитов на LLM-расходы, отображения потребления пользователю.

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

Langfuse считает стоимость автоматически, если знает модель. Для кастомных моделей или self-hosted LLM задавайте usage вручную:

with langfuse.start_as_current_observation(
    as_type="generation",
    name="local-llama",
    model="llama-3.1-70b",
    input=[{"role": "user", "content": query}],
) as generation:
    response = call_local_llm(query)
    generation.update(
        output=response,
        usage_details={
            "input_tokens": count_tokens(query),
            "output_tokens": count_tokens(response),
        },
        # Стоимость для self-hosted: compute cost
        cost_details={
            "input": 0.001,   # $0.001 за вызов (ваша оценка)
            "output": 0.002,
        },
    )

Интеграция с фреймворками

Langfuse интегрируется с основными LLM-фреймворками — вместо ручного трейсинга каждого вызова автоматическая инструментация.

OpenAI SDK — drop-in replacement

Замена одного импорта — все вызовы автоматически трейсятся:

# Было:
# from openai import OpenAI

# Стало:
from langfuse.openai import OpenAI

client = OpenAI()

# Используете как обычный OpenAI SDK
# Каждый вызов автоматически попадает в Langfuse
response = client.chat.completions.create(
    model="gpt-5.5",
    messages=[{"role": "user", "content": "Порекомендуй кафе в Москве"}],
)

Захватывается автоматически: промпт и ответ, модель, токены (input/output/total), стоимость, latency, ошибки. Работает со streaming, async, function/tool calls.

Для вложения в @observe-трейс:

from langfuse import observe
from langfuse.openai import OpenAI

client = OpenAI()


@observe()
def handle_request(query: str) -> str:
    # OpenAI-вызов автоматически станет дочерним span
    response = client.chat.completions.create(
        model="gpt-5.5",
        messages=[{"role": "user", "content": query}],
    )
    return response.choices[0].message.content

Anthropic SDK

Для Anthropic используется OpenTelemetry instrumentor:

pip install openinference-instrumentation-anthropic

from anthropic import Anthropic
from openinference.instrumentation.anthropic import AnthropicInstrumentor

# Инструментация — один раз при старте приложения
AnthropicInstrumentor().instrument()

client = Anthropic()

# Все вызовы автоматически трейсятся в Langfuse
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Порекомендуй кафе в Москве"}],
)

OpenTelemetry spans автоматически вкладываются в @observe-трейсы.

LangChain

from langfuse.langchain import CallbackHandler

langfuse_handler = CallbackHandler()

# Передаётся в config при вызове chain
chain = prompt | llm | output_parser

result = chain.invoke(
    {"question": "Порекомендуй кафе"},
    config={"callbacks": [langfuse_handler]},
)

CallbackHandler захватывает все шаги цепочки: промпты, LLM-вызовы, tool calls, retrieval — с полной вложенностью.

LlamaIndex

Используется OpenTelemetry-based инструментация:

pip install openinference-instrumentation-llama-index

from openinference.instrumentation.llama_index import LlamaIndexInstrumentor

# Инструментация — один раз при старте
LlamaIndexInstrumentor().instrument()

# Все вызовы LlamaIndex автоматически трейсятся
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
response = query_engine.query("Кафе в Москве")

LiteLLM — универсальный прокси

LiteLLM — прокси к любому LLM-провайдеру. Одна строка для Langfuse:

import litellm

litellm.success_callback = ["langfuse"]

# Любой провайдер — автоматический трейсинг
response = litellm.completion(
    model="gpt-5.5",
    messages=[{"role": "user", "content": "Привет"}],
    metadata={
        "trace_name": "chat-response",
        "trace_user_id": "user-123",
        "generation_name": "greeting",
        "tags": ["production", "chat"],
    },
)

LiteLLM сам отправит в Langfuse: модель, токены, стоимость, latency, input/output.

Claude Agent SDK

Для трейсинга агентов на Claude Agent SDK используется OpenInference инструментатор:

pip install langfuse claude-agent-sdk openinference-instrumentation-claude-agent-sdk

from claude_agent_sdk import Agent, Runner
from openinference.instrumentation.claude_agent_sdk import ClaudeAgentSDKInstrumentor

# Инструментация — один раз при старте приложения
ClaudeAgentSDKInstrumentor().instrument()

agent = Agent(
    name="travel-assistant",
    instructions="Ты — travel-ассистент.",
)

# Трейсинг агентских шагов: tool calls, model responses, handoffs
result = Runner.run_sync(agent, "Порекомендуй маршрут по Москве")

В Langfuse видно каждый шаг агента: вызовы модели, использование инструментов, передачу между агентами.

Dashboards и аналитика

В Langfuse встроены дашборды для анализа LLM-приложений с обновлением в реальном времени.

Основные метрики

Latency. Распределение времени ответа по трейсам. Фильтр по модели, фиче, пользователю. Percentiles: p50, p90, p99. Если p99 latency растёт — видно сразу.

Cost. Расходы по дням, моделям, фичам, пользователям. Тренд за период. Аномалии (резкий рост) заметны на графике.

Quality. Средний score по evaluations. Тренд по времени — качество растёт или падает после изменения промпта. Разбивка по evaluator (relevance, helpfulness, toxicity).

Volume. Количество трейсов по дням. Пики нагрузки. Распределение по моделям и фичам.

Sessions

Langfuse группирует трейсы в sessions — цепочка запросов одного пользователя за одну сессию:

@observe()
def handle_message(message: str, session_id: str, user_id: str) -> str:
    with propagate_attributes(session_id=session_id, user_id=user_id):
        return generate_response(message)

В UI: Sessions → конкретная сессия. Видно все трейсы пользователя в хронологическом порядке — полная картина взаимодействия.

Фильтрация и поиск

Фильтрация по:

• Name — имя трейса или observation
• Tags — произвольные теги
• User ID — конкретный пользователь
• Model — модель LLM
• Score — трейсы с определённым score (например, relevance < 0.5)
• Time range — временной диапазон
• Metadata — произвольные поля

Типичные запросы:

• «Все трейсы пользователя X за последнюю неделю» — для дебага
• «Трейсы с relevance < 0.3» — для анализа плохих ответов
• «Трейсы с cost > $0.10» — для оптимизации дорогих запросов

Production Best Practices

Sampling

В production не всегда нужен трейсинг 100% запросов. При высокой нагрузке sampling снижает объём данных и стоимость (cloud) или нагрузку на инфраструктуру (self-hosted).

# Через переменную окружения (рекомендуется):
export LANGFUSE_SAMPLE_RATE=0.1  # 10% трейсов

Значение от 0 до 1. Sampling работает на уровне трейса — если трейс попал в выборку, все его observations отправляются, если нет — ни одно.

Рекомендации:

• Development: 100% (всё трейсим)
• Staging: 100%
• Production, < 1 000 req/день: 100%
• Production, 1 000-10 000 req/день: 50%
• Production, > 10 000 req/день: 10-20%

Можно менять sampling rate динамически через environment variable.

PII Masking

LLM-трейсы содержат полные промпты и ответы — в них нередко оказываются персональные данные. Для GDPR/compliance нужно маскировать PII до отправки в Langfuse.

import re
from langfuse import observe


def mask_pii(text: str) -> str:
    """Маскирование персональных данных."""
    # Email
    text = re.sub(r'[\w.-]+@[\w.-]+\.\w+', '[EMAIL]', text)
    # Телефон (российский формат)
    text = re.sub(r'\+?7[\s-]?\(?\d{3}\)?[\s-]?\d{3}[\s-]?\d{2}[\s-]?\d{2}', '[PHONE]', text)
    # Номер карты
    text = re.sub(r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', '[CARD]', text)
    return text


@observe(capture_input=False, capture_output=False)
def handle_sensitive_request(query: str) -> str:
    """Для чувствительных данных — отключаем авто-захват IO."""
    result = generate_response(query)

    # Вручную записываем маскированные данные
    langfuse.update_current_span(
        input=mask_pii(query),
        output=mask_pii(result),
    )
    return result

Два подхода:

1. capture_input=False, capture_output=False + ручная запись маскированных данных
2. LANGFUSE_OBSERVE_DECORATOR_IO_CAPTURE_ENABLED=false — глобальное отключение авто-захвата

Retention

Трейсы занимают место: один трейс с промптом на 1 000 токенов — 8-12 КБ, при 1 000 трейсов в день — 300-400 МБ в месяц. Настройте удаление старых данных:

-- Для self-hosted: данные трейсов хранятся в ClickHouse, не в PostgreSQL
-- Подключитесь к ClickHouse и выполните:
ALTER TABLE traces DELETE WHERE created_at < NOW() - INTERVAL 90 DAY;
ALTER TABLE observations DELETE WHERE created_at < NOW() - INTERVAL 90 DAY;
-- (синтаксис ClickHouse: INTERVAL 90 DAY без кавычек)

В Langfuse Cloud retention настраивается в UI.

Async и flush

SDK отправляет трейсы асинхронно — основной поток не блокируется. В short-lived окружениях (serverless, scripts) данные могут не успеть отправиться. Вызывайте flush() перед завершением:

langfuse = get_client()

# ... ваша логика ...

# Перед завершением — дождаться отправки всех данных
langfuse.flush()

Для serverless (AWS Lambda, Cloudflare Workers) — flush() в конце handler-а.

Team Access

Иерархия Organizations → Projects:

• Organization — команда или компания
• Project — отдельное приложение или окружение
• У каждого проекта свои API-ключи и данные
• RBAC — в Cloud Core, Pro и Enterprise

Паттерн для multi-environment:

• Project myapp-dev — development
• Project myapp-staging — staging
• Project myapp-prod — production

Разные ключи в разных окружениях — данные изолированы.

Мониторинг Langfuse

Langfuse — внешняя зависимость. Health check:

curl http://localhost:3000/api/public/health

Добавьте в мониторинг (Zabbix, Uptime Robot, Grafana). Если Langfuse недоступен — fallback промпты из кода.

Self-hosted vs Cloud

Критерий	Self-hosted	Cloud
Стоимость	Только инфраструктура (~$20-40/мес VPS)	Core $29/мес, Pro $199/мес
Данные	На вашей инфраструктуре	На серверах Langfuse (EU/US)
Обслуживание	Обновления, бекапы, мониторинг — ваши	Всё включено
Лимиты	Нет	50k obs/мес (free), далее по тарифу
RBAC	Базовый	Полный (Pro/Enterprise)
SLA	Зависит от вашего DevOps	99.9% (Enterprise)
Подходит для	Команды с DevOps, compliance	Быстрый старт, маленькие команды

Когда self-hosted

• Регуляторные требования — данные не покидают контур
• Объём > 100 000 units/мес (экономия vs Cloud Core $29/мес)
• Уже есть инфраструктура (Kubernetes, Docker)
• Нужен полный контроль над данными и retention

Когда Cloud

• Быстрый старт без DevOps
• Маленькая команда (1-5 человек)
• Объём < 50 000 units/мес — Hobby план бесплатный
• Нужен RBAC, SSO, audit logs

Миграция между вариантами

Langfuse экспортирует и импортирует данные через API. Начать с Cloud и переехать на self-hosted — реалистичный путь. Промпты и datasets переносятся через SDK.

Итог

Langfuse закрывает четыре задачи LLM observability одним инструментом. Минимальный путь к первым данным:

1. Установить — pip install langfuse + Docker Compose (или Cloud)
2. Трейсинг — добавить @observe() к основным функциям
3. Cost tracking — работает автоматически после трейсинга
4. Prompt management — вынести один промпт из кода в Langfuse
5. Evaluations — настроить LLM-as-Judge для критичного endpoint

Каждый шаг самостоятельный. Начните с трейсинга одного endpoint, посмотрите данные за неделю — этого достаточно, чтобы понять, нужен ли инструмент дальше.

Ссылки:

• Langfuse GitHub — исходный код и self-hosted установка
• Langfuse Docs — документация
• Python SDK — reference
• Миграция Python SDK v3 → v4 — breaking changes
• Интеграции — OpenAI, Anthropic, LangChain, LlamaIndex
• Prompt Management — управление промптами
• Evaluations — оценка качества

---

Нужна помощь с observability для LLM-приложений? Я помогаю стартапам внедрять AI-решения и строить продукты — belov.works.