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

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

Langfuse покрывает трейсинг, prompt management, evaluations и cost tracking. Ядро распространяется под MIT, а enterprise-модули в ee/ имеют отдельную коммерческую лицензию. Актуальный Python SDK v4 построен на OpenTelemetry.

Гайд начинается с поддерживаемых способов установки, а затем проводит один Python setup через tracing, prompts, evaluation, cost, privacy и failure handling.

Если сначала нужно решить, что измерять команде, начните с production-модели LLM observability. Здесь остаётся практика: установка Langfuse и ежедневная работа с ним.

Что такое Langfuse

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

Трейсинг. Представьте пользовательскую задачу root observation, а model, retrieval, tool и validation steps - дочерними observations. Видны только поля, которые реально записал SDK или integration.

Prompt Management. Храните versioned prompts и labels отдельно от application release, но оставьте local fallback для пути, который не может зависеть от network fetch.

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

Cost Tracking. Группируйте usage и рассчитанный cost по feature, user или model, когда generation содержит необходимые model и usage fields.

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

Зачем нужен Langfuse

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

Дебаг

Пользователь сообщает: «AI дал неправильный ответ». Trace может показать captured prompt, model response, tool calls и validation steps этого request. Это сужает область поиска, если sensitive fields были записаны намеренно.

Стоимость

Длинный контекст и повторные tool calls превращают небольшую цену одного запроса в заметный месячный счёт. Trace связывает токены и model usage с фичей и задачей, которые их породили: invoice провайдера перестаёт быть первым сигналом проблемы.

Качество

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

Почему в туториале используется Langfuse

Здесь важна не очередная сравнительная таблица: тарифы и feature lists быстро устаревают. У Langfuse есть OpenTelemetry-based SDK, self-hostable MIT-ядро и интеграции с распространёнными model и agent frameworks. Перед закупкой всё равно нужно сверить актуальную документацию каждого вендора.

Установка

Есть два пути: Langfuse Cloud или self-hosted deployment. Для локальной проверки используйте поддерживаемый Docker Compose из репозитория Langfuse.

Self-hosted через Docker Compose

В актуальной документации Docker Compose назван самым простым вариантом для local или single-VM запуска. Это не high-availability production topology: scaling, backups и failover остаются вашей задачей. Для high availability и высокой нагрузки Langfuse рекомендует Kubernetes.

Не копируйте статичный compose из туториала. Клонируйте upstream-репозиторий, чтобы application images, storage services, environment variables и migrations относились к одному релизу:

git clone https://github.com/langfuse/langfuse.git
cd langfuse

# Review every CHANGEME value before the first start.
docker compose up -d

Официальный файл сейчас запускает web и worker вместе с PostgreSQL, ClickHouse, Redis и MinIO. Замените все placeholder secrets, не публикуйте internal storage ports в интернет и настройте backups до загрузки реальных трейсов. Поддерживаемые шаги и границы production-сценария описаны в Docker Compose deployment guide.

Когда services станут healthy, откройте http://localhost:3000, создайте organization и project, затем скопируйте project-scoped Public Key и Secret Key для SDK.

Cloud (langfuse.com)

Зарегистрируйтесь на cloud.langfuse.com, создайте project и скопируйте его keys. На 24 августа 2026 года официальный pricing page указывает для бесплатного Hobby 50 000 units в месяц, 30 дней доступа к данным и двух пользователей. Перед расчётом лимитов проверьте страницу ещё раз.

Установка Python SDK

pip install langfuse

Версии SDK. Примеры рассчитаны на актуальный Python SDK v4 и Python 3.10+ с from langfuse import get_client. После проверки зафиксируйте версию в lockfile проекта. V4 по умолчанию использует актуальные Observations и Metrics APIs, которым нужен self-hosted server v4. Перед обновлением существующей instrumentation прочитайте миграционный гайд 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()
if not langfuse.auth_check():
    raise RuntimeError("Langfuse authentication failed")

Трейсинг

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

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

В Langfuse v4 trace - это набор observations с общим trace ID. Root observation представляет пользовательскую задачу; дочерние observations - validation, retrieval, tools и model calls. Общие input и output принадлежат root observation.

Root span: "generate-travel-plan"
│
├── Span: "validate-input" (8ms)
│
├── Span: "retrieve-context" (200ms)
│   └── Span: "vector-search" (180ms)
│
├── Generation: "plan-draft" (model-a, 1800 tokens)
│
├── Generation: "plan-review" (model-b, 600 tokens)
│
└── Span: "format-output" (3ms)

Generation может хранить model, input/output, usage, cost, latency и metadata. SDK не создаёт поля, которых wrapper не записал. Native integrations могут заполнить их сами; manual wrapper должен обновить generation явно.

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

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

from langfuse import get_client, observe, propagate_attributes
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]:
    """Поиск релевантного контекста в базе знаний."""
    # Your search logic: 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)
    model = "your-openai-model-id"
    messages = [
        {"role": "system", "content": f"Контекст:\n{context_text}"},
        {"role": "user", "content": query},
    ]
    response = openai_client.chat.completions.create(
        model=model,
        messages=messages,
    )
    output = response.choices[0].message.content or ""
    langfuse.update_current_generation(
        model=model,
        input=messages,
        output=output,
        usage_details={
            "input": response.usage.prompt_tokens,
            "output": response.usage.completion_tokens,
        } if response.usage else None,
    )
    return output


@observe()
def handle_request(user_query: str, user_id: str) -> str:
    """Обработать одну задачу в root observation."""
    with propagate_attributes(user_id=user_id):
        query = validate_input(user_query)
        context = search_context(query)
        response = generate_response(query, context)
        return response


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

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

• handle_request создаст root observation и trace context
• validate_input и search_context станут вложенными spans
• generate_response станет generation и запишет model и usage
• Вложенность автоматическая - 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:
    # Set user_id and session_id through 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)

        # Update output when a custom value is needed
        langfuse.update_current_span(output={"processed_result": result})
        return result

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

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

from langfuse import get_client

langfuse = get_client()

# The context manager handles nesting automatically
with langfuse.start_as_current_observation(name="process-request") as span:
    span.update(input={"query": "Кафе в Москве"})

    # Nested 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="your-openai-model-id",
        input=[{"role": "user", "content": "Кафе в Москве"}],
    ) as generation:
        response = call_llm("Кафе в Москве")
        generation.update(
            output=response,
            usage_details={"input": 42, "output": 128},
        )

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

langfuse.flush()

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

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

Скрытые retry. Retry logic может вызвать модель несколько раз. Отдельные generation observations показывают каждую записанную попытку и её usage.

Model mismatch. Endpoint должен вызывать model-b, но всё ещё вызывает model-a. Фильтр по полю model показывает ошибочный route.

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

Prompt Management

Remote prompt registry отделяет prompt release от application release. Rollback может стать короче, но появляется fetch и cache policy, которую нужно проверить.

Создание промпта в 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()

# Load the prompt (defaults to the version labeled "production")
prompt = langfuse.get_prompt("travel-assistant", type="chat")

# Compile with variables
compiled = prompt.compile(
    destination="Москва",
    user_query="Порекомендуй кафе в центре",
)

# compiled is a string for a text prompt or a messages list for a chat prompt
# Use it in the LLM call
response = openai_client.chat.completions.create(
    model="your-openai-model-id",
    messages=compiled,
)

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

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

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

# Production version (default)
prod_prompt = langfuse.get_prompt("travel-assistant", type="chat")

# Staging version for testing
staging_prompt = langfuse.get_prompt(
    "travel-assistant", type="chat", label="staging"
)

# Specific version number
v3_prompt = langfuse.get_prompt("travel-assistant", type="chat", version=3)

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

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

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

import random
from langfuse import get_client, observe

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", type="chat", label="production"
        )
    else:
        prompt = langfuse.get_prompt(
            "travel-assistant", type="chat", label="staging"
        )

    compiled = prompt.compile(user_query=query)

    # Tag for analytics filters
    langfuse.update_current_span(
        tags=[f"ab-test:prompt-{variant}"],
        metadata={"prompt_variant": variant},
    )

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

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

Fallback-паттерн

Langfuse - внешняя зависимость. У свежего процесса ещё нет local prompt cache, поэтому на случай первого неудачного fetch задайте SDK fallback.

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


prompt = langfuse.get_prompt(
    "travel-assistant",
    type="chat",
    label="production",
    fallback=FALLBACK_MESSAGES,
)
compiled = prompt.compile(user_query="Порекомендуй два кафе")

Поле prompt.is_fallback позволяет записать метрику fallback. Другой вариант - pre-fetch важных промптов при startup. Выберите одну политику и проверьте cold start при недоступном Langfuse API.

Кеширование

SDK хранит промпты в памяти с default TTL 60 секунд. После его истечения SDK может вернуть stale value и обновить cache в фоне. Более длинный TTL подходит лишь тогда, когда такая задержка обновления допустима:

# Cache the prompt for five minutes
prompt = langfuse.get_prompt(
    "travel-assistant", type="chat", cache_ttl_seconds=300
)

В development задайте cache_ttl_seconds=0, если каждый вызов должен получать текущую версию. Этот cache не связан с provider-side prompt caching.

Authenticated MCP-сервер

Langfuse предоставляет project data через Streamable HTTP MCP endpoint. Актуальная документация рекомендует CLI или agent skill, если утверждённый агент умеет выполнять shell-команды; MCP нужен инструментам без такого доступа. Endpoint по умолчанию содержит read и write tools.

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

Создайте project-scoped key для этой интеграции, не храните закодированное значение в репозитории и подключайте только утверждённый client. Для read-only сценария добавьте в allow-list только lookup tools и исключите write tools. Перед настройкой сверьтесь с каноническим MCP server reference: список tools и client snippets меняется вместе с платформой.

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 - точная evaluator model и её версия
• Target - какие трейсы оценивать (фильтр по тегам, имени, дате)
• Score name - имя метрики (например, relevance)
• Score type - Numeric (0-1), Categorical, Boolean

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

Evaluator’ы можно настраивать через UI, API и MCP-сервер. Можно оценивать 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)

    # Automatic check: the answer is not empty
    langfuse.score_current_trace(
        name="not-empty",
        value=bool(response.strip()),
        data_type="BOOLEAN",
    )

    # Automatic check: the length stays within the expected range
    langfuse.score_current_trace(
        name="length-ok",
        value=50 < len(response) < 5000,
        data_type="BOOLEAN",
    )

    # Validate the format when JSON is expected
    try:
        import json
        json.loads(response)
        langfuse.score_current_trace(
            name="valid-json", value=True, data_type="BOOLEAN"
        )
    except json.JSONDecodeError:
        langfuse.score_current_trace(
            name="valid-json", value=False, data_type="BOOLEAN"
        )

    return response

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

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

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

Dataset - набор пар input/expected_output. Изменили промпт → прогнали dataset → сравнили scores с предыдущей версией. Схема самого эксперимента разобрана отдельно в гайде по A/B-тестированию промптов.

from langfuse import get_client

langfuse = get_client()

# Create a dataset
langfuse.create_dataset(name="travel-queries-v1")

# Add 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"],
    )

# Run the experiment with the 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 creates the run, traces each item, and links the results
# Manual item.link() calls are not needed in v4
dataset.run_experiment(
    name="prompt-v3-model-b",
    task=my_task,
)

langfuse.flush()

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

Cost Tracking

Langfuse может рассчитать cost, когда generation содержит usage и model совпадает с настроенным price definition. Проверяйте результат для каждого model release. Неизвестной или self-hosted model нужна явная definition либо cost_details; не предполагайте, что цена новой модели уже настроена правильно.

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

• Total cost за период (день, неделя, месяц)
• Cost per trace - средняя стоимость одного запроса пользователя
• Cost per user - расходы конкретного пользователя
• Cost per model - распределение между model IDs, которые реально отправляет application
• 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 calls
        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 calls
        return result

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

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

Не копируйте старые примеры с fetch_traces(): в v4 это не aggregate data path. Запрашивайте cost и usage через Metrics API v2, группируйте по стабильному user или feature dimension из trace attributes и записывайте aggregate в billing database. Billing job должен быть idempotent и сверяться с invoices провайдера. Актуальная request schema находится в Metrics API documentation.

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

Для custom или self-hosted models записывайте usage и рассчитанный cost явно:

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": count_tokens(query),
            "output": count_tokens(response),
        },
        cost_details={
            "input": estimate_input_cost(query),
            "output": estimate_output_cost(response),
        },
    )

Интеграции с frameworks

Framework integrations выпускаются по собственным графикам. Выберите одну integration на уже существующей границе model calls, зафиксируйте packages и проверьте success, stream, tool call и provider error до широкого включения.

OpenAI Python wrapper

Поддерживаемая OpenAI integration оборачивает client, сохраняя его interface:

from langfuse.openai import OpenAI

client = OpenAI()
response = client.chat.completions.create(
    model="your-openai-model-id",
    messages=[{"role": "user", "content": "Порекомендуй два кафе"}],
)

Wrapper может записывать поля model call, которые вернул provider. Это не делает export raw prompts безопасным. Примените ту же capture и masking policy, что и для manual observations, затем проверьте фактический field set в Langfuse.

Другие stacks

Не смешивайте snippets из разных поколений SDK. Начинайте с актуального Langfuse integrations index и записывайте точные package versions в deployment manifest.

Dashboards и аналитика

Langfuse dashboards и Metrics API показывают данные, дошедшие до платформы. Freshness зависит от SDK, server version и ingestion path; проверьте её до использования dashboard в alerts.

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

Latency. Распределение времени ответа по traces. Сравнивайте percentiles по model, feature или release, а не один average.

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 session показывает в хронологическом порядке captured traces с общим session_id. Пропущенные или unsampled traces там не появятся. Сама session - не audit log.

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

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

• 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).

# Use an environment variable (recommended):
export LANGFUSE_SAMPLE_RATE=0.1  # 10% трейсов

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

Универсальной таблицы по числу запросов здесь нет. Сначала определите debugging и evaluation cases, которые нельзя потерять, оцените их объём и проверьте, не скрывает ли random sampling редкие failures. Изменение environment variable требует перезагрузки конфигурации процесса; не рассчитывайте, что уже запущенный client подхватит её динамически.

PII Masking

LLM traces могут содержать prompts, responses, tool arguments и персональные данные. Сначала определите, что разрешено отправлять из application, затем маскируйте или исключайте всё остальное до export.

import re
from langfuse import observe


def mask_pii(text: str) -> str:
    """Маскирование персональных данных."""
    # Email
    text = re.sub(r'[\w.-]+@[\w.-]+\.\w+', '[EMAIL]', text)
    # Phone number in Russian format
    text = re.sub(r'\+?7[\s-]?\(?\d{3}\)?[\s-]?\d{3}[\s-]?\d{2}[\s-]?\d{2}', '[PHONE]', text)
    # Payment card number
    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)

    # Record masked data manually
    langfuse.update_current_span(
        input=mask_pii(query),
        output=mask_pii(result),
    )
    return result

Ручной pattern подходит, когда одна функция владеет всем sensitive input. Он не проверяет attributes, созданные сторонней OpenTelemetry instrumentation. Для новых Python v4 setups Langfuse рекомендует export-stage hook mask_otel_spans. Функция должна быть быстрой и детерминированной: некорректный результат может удалить span или целый export batch.

Выбирайте минимальный data path:

1. отключите capture через capture_input=False, capture_output=False, если raw IO не нужен;
2. записывайте только намеренно очищенное подмножество в observations, которыми управляете;
3. применяйте mask_otel_spans к attributes, уходящим через Langfuse client, и отдельно настройте redaction для каждого другого exporter.

Перед реализацией сверьте актуальный masking contract. Regex из примера не заменяет полноценную PII policy.

Retention

Объём зависит от payload, media, metadata и sampling. Измеряйте собственный trace mix, а не переносите оценку размера из чужой нагрузки.

Не удаляйте данные прямым SQL по скопированным именам ClickHouse tables. Langfuse поддерживает удаление traces через UI и API, а retention feature вместе удаляет старые traces, observations, scores и media. Доступность зависит от Cloud plan или self-hosted license. Без policy self-hosted event data по умолчанию хранится бессрочно. Сверьте актуальную документацию по retention и deletion, сначала проверьте процесс на non-production data, а после удаления повторно запросите данные.

Async и flush

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

langfuse = get_client()

# ... application logic ...

# Wait for pending exports before shutdown
langfuse.flush()

В short-lived Python jobs и serverless Python handlers вызывайте flush до заморозки или завершения runtime. Ограничьте операцию оставшимся временем handler: outage telemetry не должен превращаться в бесконечный retry.

Team Access

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

• Organization - команда или компания
• Project - отдельное приложение или окружение
• У каждого project свои API keys; они не привязаны к конкретному user
• Organization roles доступны обычно, а fine-grained project roles зависят от Cloud plan или self-hosted Enterprise license

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

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

Разные keys и projects снижают риск смешать окружения. User access всё равно зависит от organization и project roles; проверьте его least-privilege аккаунтом. Актуальная матрица доступности находится в RBAC documentation.

Мониторинг Langfuse

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

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

Добавьте endpoint в мониторинг. Если application startup или request получает prompts из Langfuse, проверьте cached или local fallback в outage test.

Self-hosted или Cloud

Решение определяется стоимостью эксплуатации, а не выдуманной границей по числу traces в месяц.

Self-hosting оправдан, когда зафиксированная data boundary или network requirement важнее работы по эксплуатации PostgreSQL, ClickHouse, Redis, object storage, web и worker services. Cloud подходит, когда managed operations важнее контроля этого stack. Одной цены недостаточно: учитывайте backups, upgrades, incident response и security features, которые действительно нужны команде.

Смена SDK endpoint переводит новую telemetry. Перенос исторических traces, prompts и datasets - отдельный migration project: составьте inventory объектов, используйте поддерживаемые exports и APIs, проверьте связи и сохраните rollback copy. Не обещайте прозрачную миграцию истории без фактического прогона.

Итог

Начните с малого:

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

Начните с одного endpoint. До расширения instrumentation проверьте репрезентативные success, failure, retry и sensitive-data cases.

Ссылки:

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

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