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

> Полное руководство по Langfuse: трейсинг LLM-запросов, prompt management, evaluations, cost tracking. Self-hosted и cloud. Примеры на Python.
> Author: Roman Belov · Published: 2026-07-03 · Source: https://futurecraft.pro/ru/blog/langfuse-step-by-step/

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 — в [обзорной статье](/ru/blog/llm-observability-langfuse/).

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

```yaml
# 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:
```

```bash
docker compose up -d
```

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

### Cloud (langfuse.com)

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

### Установка Python SDK

```bash
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](https://langfuse.com/docs/observability/sdk/upgrade-path/python-v3-to-v4).

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

```bash
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
```

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

```python
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.

```python
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:

```python
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):

```python
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:
```json
[
  {"role": "system", "content": "Ты — travel-ассистент для {{destination}}."},
  {"role": "user", "content": "{{user_query}}"}
]
```

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

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

```python
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** — последняя созданная версия.

```python
# 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 тестирование промптов

```python
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 — внешняя зависимость. Если он недоступен, промпт всё равно нужен.

```python
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:

```python
# Кешировать промпт на 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 и комментарии.

```json
{
  "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

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

```python
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

```python
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 с предыдущей версией.

```python
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

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

```python
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` покажет, сколько стоит генерация маршрутов. Отдельно — чат, отдельно — рекомендации.

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

```python
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 вручную:

```python
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

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

```python
# Было:
# 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`-трейс:

```python
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:

```bash
pip install openinference-instrumentation-anthropic
```

```python
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

```python
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 инструментация:

```bash
pip install openinference-instrumentation-llama-index
```

```python
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:

```python
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 инструментатор:

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

```python
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 — цепочка запросов одного пользователя за одну сессию:

```python
@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).

```bash
# Через переменную окружения (рекомендуется):
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.

```python
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 МБ в месяц. Настройте удаление старых данных:

```sql
-- Для 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()` перед завершением:

```python
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:

```bash
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](https://github.com/langfuse/langfuse) — исходный код и self-hosted установка
- [Langfuse Docs](https://langfuse.com/docs) — документация
- [Python SDK](https://langfuse.com/docs/observability/sdk/python/overview) — reference
- [Миграция Python SDK v3 → v4](https://langfuse.com/docs/observability/sdk/upgrade-path/python-v3-to-v4) — breaking changes
- [Интеграции](https://langfuse.com/docs/observability/sdk/instrumentation) — OpenAI, Anthropic, LangChain, LlamaIndex
- [Prompt Management](https://langfuse.com/docs/prompt-management/get-started) — управление промптами
- [Evaluations](https://langfuse.com/docs/evaluation/overview) — оценка качества

---

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