Настройка мониторинга ИИ с помощью Langfuse

Настройка мониторинга ИИ с помощью Langfuse

Меня часто спрашивают, как отличить начинающего ИИ-инженера от опытного. Мой ответ прост: посмотри на его мониторинг. Если промпты, логи и расходы невидимы, команда летит вслепую. В этом посте разберём, почему мониторинг должен быть в первой версии твоего ИИ-сервиса, и как быстро развернуть Langfuse.

Почему мониторинг нужен раньше файнтюнинга

Ты не доберёшься до файнтюнинга или навороченных пайплайнов, если не видишь, что модель делает в проде. Начни отслеживать эти метрики с первого дня:

• стоимость запроса (токены промпта + токены ответа)
• задержка ответа и доля таймаутов
• качество генерации и примеры неудачных ответов
• контекст пользователя: кто сделал запрос и в каком сценарии

Без этих данных ты не сможешь объяснить внезапный скачок расходов на API или почему пользователи жалуются на «молчаливого» ассистента. Частая ошибка — оставлять историю разговора включённой, из-за чего контекстное окно раздувается, а расходы растут экспоненциально.

Чем помогает Langfuse

Langfuse — это платформа observability для LLM-приложений. Она доступна и как бесплатный облачный тариф, и как self-hosted вариант, так что вписать её в политики компании легко. Ключевые преимущества:

• фиксирует трейсы запросов, цепочки промптов и вложенные вызовы инструментов
• хранит сырые входы и выходы модели для разбора инцидентов
• считает токены и расходы по пользователю или фиче
• поддерживает OpenAI, Hugging Face, Anthropic и кастомные модели
• интегрируется с OpenTelemetry, так что можно объединять данные с другими метриками

Что отслеживать

Относись к мониторингу как к продуктовой аналитике:

1. Экономика: куда уходит бюджет и сколько стоит запуск разных фич.
2. Качество: на какие промпты приходится больше всего жалоб и сколько времени занимают ответы.
3. Эксперименты: помечай A/B-тесты промптов или моделей, чтобы сравнивать результаты.
4. Операционная готовность: всплески токенов и ошибок — ранние сигналы, которые позволяют исправить деградацию до того, как клиент заведёт тикет.

Подключение Langfuse к Python-проекту

Разберём базовую интеграцию с OpenAI.

nodejs

// tracedOpenAi.ts

import { observeOpenAI } from '@langfuse/openai';
import { NodeSDK } from '@opentelemetry/sdk-node';
import { LangfuseSpanProcessor } from '@langfuse/otel';
import Config from '@app/config';
import { OpenAI } from 'openai/client';

const sdk = new NodeSDK({
  spanProcessors: [
    new LangfuseSpanProcessor({
      publicKey: Config.langfusePublicKey,
      secretKey: Config.langfuseSecretKey,
      baseUrl: Config.langfuseHost,
    }),
  ],
});

sdk.start();

// Wrap the OpenAI client with Langfuse tracing
export const tracedOpenAI = observeOpenAI(new OpenAI({ apiKey: Config.openAiKey }), {
  // Configure trace-level attributes for all API calls
  traceName: 'my-openai-trace', // Name for the trace
  sessionId: 'user-session-123', // Track user session
  userId: 'user-abc', // Track user identity
  tags: ['openai-integration'], // Add searchable tags
});

python

# GptClient.py

from typing import Union

from langfuse.openai import OpenAI  # type: ignore

from openai.types import ChatModel
from openai.types.chat import (
    ChatCompletionSystemMessageParam,
    ChatCompletionUserMessageParam,
)

from src.config import Config


class GptClient:
    def __init__(self, api_key: str):
        self.api_key = api_key

        self.client = OpenAI(
            api_key=Config.gpt_key(),
        )

    def exec_request(
        self,
        setup_message: str,
        request_text: str,
        temperature: float = 0.1,
        model: Union[str, ChatModel] | None = None,
    ) -> str:
        final_model = model if model else Config.gpt_model()

        messages = [
            ChatCompletionSystemMessageParam(role="system", content=setup_message),
            ChatCompletionUserMessageParam(role="user", content=request_text),
        ]

        completion = self.client.chat.completions.create(
            max_tokens=40000,
            model=final_model,
            messages=messages,
            temperature=temperature,
            stream=True,
        )

        answer_parts = []
        for chunk in completion:
            part = None
            if hasattr(chunk, "choices"):
                delta = getattr(chunk.choices[0], "delta", None)
                if delta and hasattr(delta, "content"):
                    part = delta.content
            elif isinstance(chunk, dict) and "choices" in chunk:
                part = chunk["choices"][0]["delta"].get("content", "")
            if part:
                answer_parts.append(part)

        answer = "".join(answer_parts)
        return answer

Что смотреть в дашбордах

Как только интеграция заработает, открой Langfuse и отслеживай:

• распределение токенов по пользователям и сценариям
• топ-5 самых дорогих промптов
• долю ошибочных ответов (5xx, таймауты, валидация)
• примеры генераций, помеченных как «плохие» (при необходимости добавляй метки вручную)

Эти данные помогают расставлять приоритеты: ты будешь знать, где оптимизировать, где делать файнтюнинг, а где пересмотреть UX.

Дальнейшие шаги

После базовой настройки изучи продвинутые функции Langfuse: кастомные метрики, автоматические повторы цепочек и сбор обратной связи от пользователей прямо в UI. Делись находками — чем шире наш общий инструментарий, тем быстрее мы вместе построим надёжные ИИ-продукты.