Langfuse — открытая платформа наблюдаемости для приложений на языковых моделях: она пишет трассировку каждого вызова, хранит датасеты и считает оценки прогонов. Нужна она ради одного вопроса, который встаёт на второй неделе любого LLM-проекта: поправил промпт — стало лучше или хуже? Пока приложение живёт на трёх тестовых вопросах, ответ даётся на глаз. Когда вопросов сотня, а моделей в проекте три, «на глаз» превращается в спор без аргументов, а на проде — в жалобы пользователей, которые доходят до вас через неделю. Ниже — как выглядит тестирование LLM с трассировкой, датасетом и моделью-судьёй, с кодом из учебного стенда и из рабочего RAG-сервиса.
Если наблюдаемость нужно поднять в своём проекте, а не просто понять её устройство, — это часть программы курса «ИИ для 1С: разработка ИИ-агентов, RAG и MCP-серверов».
Почему оценка «на глаз» не работает?
Проблема в двух свойствах языковых моделей сразу. Первое: один и тот же вход даёт разные ответы, поэтому «проверил три раза, вроде нормально» ничего не доказывает. Второе, и оно хуже: правка промпта чинит тот случай, на котором вы её придумали, и ломает два соседних, которых вы в этот момент не смотрели. Приложение при этом молчит — оно не падает с ошибкой, оно просто начинает отвечать чуть хуже.
Второй слепой пятно — что реально ушло в модель. В коде вы видите шаблон промпта. В модель уходит результат: шаблон плюс подставленная история (уже обрезанная), плюс найденный RAG-контекст (возможно, пустой), плюс результаты инструментов. Отладка «по ответу приложения» здесь бесполезна: вы разглядываете выход, не видя входа. Половина странных ответов объясняется не моделью, а тем, что в контекст уехало не то — про это отдельный разбор, как управлять контекстным окном агента.
Отсюда два инструмента, которые закрывают обе дыры: трассировка (что ушло и что вернулось) и датасет с оценками (стало ли в среднем лучше).
Из чего состоит Langfuse?
Четыре сущности, и понимать их полезно раздельно.
Трассировка. Каждый вызов модели пишется как observation: полный вход после всех подстановок, полный выход, имя модели, число входных и выходных токенов, стоимость, задержка. Вызовы вкладываются друг в друга: у RAG-запроса внутри трассы отдельными шагами лежат поиск и генерация, у агента — каждая итерация цикла. Это и есть базовый мониторинг LLM-приложения: не дашборд с графиком аптайма, а возможность открыть конкретный плохой ответ и увидеть, из чего он собрался.
Датасет. Список пар «вход — эталонный ответ». Живёт на сервере, а не в репозитории, потому что пополнять его будут и не-программисты.
Эксперимент (dataset run). Прогон датасета под именем. Прогоны сравниваются между собой — в этом весь смысл.
Оценки (scores). Числа, привязанные к трассе или к элементу прогона. Откуда число взялось — дело ваше: точное совпадение, регулярка, метрика, вторая модель, живой человек в интерфейсе.
Ставится Langfuse self-hosted: docker compose поднимает сам сервис плюс PostgreSQL, ClickHouse, Redis и S3-совместимое хранилище; для кластера есть helm-чарт. Клиентская библиотека — пакет langfuse из PyPI, в проект добавляется одной строкой. Для приложения из всего этого нужны три параметра: host, public_key, secret_key. Разворачивать у себя — не блажь: в трассы попадает всё, что видит модель, включая персональные данные клиентов и внутренние документы, поэтому для многих проектов вариант «облачный сервис» отпадает на этапе согласования.
Как встроить Langfuse в приложение на LangChain?
Минимальная обвязка — один класс на два объекта:
from langfuse import Langfuse
from langfuse.callback import CallbackHandler
from src.config import settings
class LangFuseService:
def __init__(self):
self.langfuse_handler = CallbackHandler(
public_key=settings.langfuse.public_key,
secret_key=settings.langfuse.secret_key,
host=settings.langfuse.host,
)
self.langfuse = Langfuse(
public_key=settings.langfuse.public_key,
secret_key=settings.langfuse.secret_key,
host=settings.langfuse.host,
)
langfuse_handler — колбэк для LangChain, он и пишет трассы. langfuse — клиент для всего остального: датасеты, прогоны, оценки. Разделение важное: трассировка включается на уровне вызова, а датасеты и оценки — это отдельная работа со стороны кода тестов.
Дальше в коде приложения не меняется ничего, кроме одного параметра вызова:
completion = structured_llm.invoke(
messages, config={"callbacks": [callback_handler]}
)
Это работает одинаково для голого invoke, для цепочки и для графа LangGraph — колбэк протекает вниз по всей конструкции, и вложенные шаги сами разложатся по трассе. На агенте с инструментами эффект заметнее всего: видно, на какой итерации модель решила полезть в инструмент, что она туда передала и что получила обратно.
В свежих версиях SDK ключи читаются из переменных окружения, и обвязка становится короче:
from langfuse import get_client
from langfuse.langchain import CallbackHandler
self.langfuse = get_client()
self.langfuse_handler = CallbackHandler()
if self.langfuse.auth_check():
logger.info("Langfuse client is authenticated and ready!")
auth_check() при старте стоит вызывать всегда. Langfuse спроектирован так, чтобы не ронять приложение: если сервер недоступен или ключи неверны, трассы просто не пишутся, молча. Удобно в проде и крайне неудобно, когда вы полдня отлаживаете «почему пусто в интерфейсе».
Ещё одна деталь из боевого сервиса: колбэк там отдаётся не всегда, а по флагу настроек — метод _get_callback_handler() возвращает либо handler, либо None. Трассировать в проде каждый вызов дорого и шумно; обычно включают на dev-контуре, на отдельном проценте запросов или на время разбора инцидента.
Почему датасет нужен свой, а не чужой бенчмарк?
Публичные лидерборды отвечают на вопрос «какая модель умнее вообще». Ваш вопрос другой: какая модель лучше справляется с вашей задачей, на ваших формулировках, с вашим контекстом. Это принципиально разные измерения. Модель, проигрывающая по общим бенчмаркам, спокойно выигрывает на короткой русскоязычной классификации запросов — потому что задача узкая, а промпт под неё вылизан. Обратное тоже верно.
Датасет — это пары «вход — эталон», залитые на сервер:
langfuse.create_dataset(name=dataset_name)
for index, row in df.iterrows():
langfuse.create_dataset_item(
dataset_name=dataset_name,
input=row[1],
expected_output=row[2],
)
Откуда брать вопросы. Лучший источник — логи: реальные запросы пользователей, особенно те, на которых система уже опозорилась. Второй — руками, из головы предметника. Третий — сгенерировать. В учебном стенде датасет генерится из базы знаний (файл вопросов-ответов службы поддержки университета) через giskard: текст превращается в KnowledgeBase, дальше generate_testset собирает 200 вопросов пятью генераторами — простые, усложнённые перефразированием, двойные, с отвлекающим элементом и ситуационные. Разнообразие тут не для красоты: агент, который отвечает на «какая столица Франции», может рассыпаться на «Италия прекрасна, но какова столица Франции?».
Синтетика — стартовый набор, чтобы было с чего начать. Заменить реальные вопросы пользователей она не может.
Прогон нескольких моделей на одном датасете
Дальше начинается собственно эксперимент. Задача в стенде прикладная: классифицировать вопрос пользователя в одну из трёх категорий — заказы и доставка, остатки и наличие, всё остальное. Это роутер бота, и от его точности зависит вообще всё, что происходит дальше.
Модели заводятся словарём, прогон — циклом по нему:
models = {
"GigaChat-2-Lite": {"model": giga_llm_lite, "use_structured_output": True},
"GigaChat-2-Max": {"model": giga_llm_max, "use_structured_output": True},
"ChatGPT-4o-mini": {"model": openai_llm_4_mini, "use_structured_output": True},
"ChatGPT-4o": {"model": openai_llm_4, "use_structured_output": True},
"Qwen-2.5-14B": {"model": qwen25_14b, "use_structured_output": False},
}
for model_name, llm in models.items():
experiment = TestClassification(
lfs=lfs,
dataset_name="classification_orders",
model_name=model_name,
llm=llm["model"],
use_structured_output=llm["use_structured_output"],
)
experiment.run_experiment()
Имя прогона собирается как f"{model_name}_{current_datetime}" — в интерфейсе Langfuse все запуски встают рядом, и средние оценки сравниваются в одной таблице.
Флаг use_structured_output — не украшение, а первая честная находка такого эксперимента. Структурированный вывод заставляет модель отвечать строго по схеме:
class QuestionClassification(BaseModel):
"""Классификация вопросов пользователей"""
category_id: int = Field(description="Категория запроса пользователя (1, 2 или 3)")
Облачные модели это поддерживают, локальные через Ollama — далеко не все. Поэтому в коде две ветки: с with_structured_output(schema=...) и обычный invoke с разбором ответа регуляркой, где заодно вырезается блок <think>...</think> у рассуждающих моделей. То есть часть кандидатов отваливается не по качеству ответов, а по формату — и узнаёте вы это на прогоне, а не на демонстрации заказчику.
Оценка в задаче классификации простейшая, судья тут не нужен:
def evaluation_function(self, question, llm_response, reference_response, metadata, callback_handler):
"""Оценивает совпадение ответа модели с эталонным."""
score = 1.0 if llm_response == reference_response else 0.0
return {"score": score}
Правило общее: если ответ можно сравнить строкой, числом или категорией — сравнивайте, не привлекайте вторую модель. Детерминированная метрика дешевле, быстрее и не врёт.
LLM-as-a-judge: когда эталон — свободный текст
Со связным ответом такой номер не проходит: «Оплата производится после заключения договора» и «Сначала заключается договор, потом выставляется квитанция» — один и тот же ответ и ноль совпадения по строке. Тогда оценивает вторая модель по инструкции — это и называется LLM as a judge.
В стенде заведены две схемы оценки, и разделение здесь важнее самих цифр:
class EvaluationQuestion(BaseModel):
"""Оценка ответа на вопрос"""
evaluation: int = Field(description="Оценка ответа числом от 1 до 10")
class EvaluationEmbQuestion(BaseModel):
"""Оценка наличия релевантного контекста в тексте"""
evaluation: int = Field(description="Оценка контекста числом от 1 до 3")
Ответ RAG-системы портится в двух разных местах. Либо поиск не нашёл нужный документ — и модель физически не могла ответить правильно. Либо нашёл, а модель ответила плохо. Одна общая оценка эти случаи склеивает, и вы не понимаете, что чинить: ретривер или промпт. Поэтому шкала контекста отдельная и короткая: 1 — контекст пустой, 2 — что-то нашлось, но не по делу, 3 — нашлось нужное. А качество ответа против эталона судится по десятибалльной.
Обе оценки уходят в трассу вместе с нормализованной суммой:
normalized_score = (embedding_score + llm_score) / 13.0
return {
"score": normalized_score,
"embedding_score": embedding_score / 3,
"llm_score": llm_score / 10,
}
Каждая метрика пишется отдельным вызовом trace.score(name=key, value=value), так что в интерфейсе видно и общий балл прогона, и его составляющие по колонкам.
Промпт судьи — обычный текстовый файл рядом с кодом, и пишется он как инструкция строгому проверяющему: вернуть целое число без пояснений, снижать оценку не выше 5 при фактических ошибках, не выше 7 за обобщённый ответ, минус два балла за лишнюю информацию. Судью держат на низкой температуре (0.1) и на модели не слабее оцениваемой — судить сильную модель слабой бессмысленно.
Где судья врёт
Про это в статьях про LLM-as-a-judge пишут неохотно, а знать надо до того, как вы построите на этих числах решение.
- Судья любит длинные и уверенные ответы. Многословный ответ с неточностью регулярно получает больше баллов, чем короткий и правильный. Стиль подменяет содержание.
- Судья завышает ответы собственной модели. Если и генерирует, и судит одна и та же модель, оценка систематически сдвинута вверх. Разводите модели.
- Шкала не абсолютна. «7» сегодня и «7» через месяц с другим промптом судьи — разные семёрки. Сравнивать можно только прогоны, снятые одним судьёй с одним промптом. Меняете промпт судьи — старые прогоны обнуляются как база сравнения.
- В узкой предметке судья не видит фактическую ошибку. Перепутанный регламент, неверная ставка НДС, несуществующий реквизит конфигурации — у судьи нет источника истины, кроме эталонного ответа, и всё, что эталон не покрывает, он оценивает по правдоподобию.
Отсюда единственное рабочее правило: человеческая разметка хотя бы на части выборки. Берёте 30–50 элементов датасета, размечаете руками, сравниваете со свежими оценками судьи. Совпадает по порядку — судье можно доверить остальные несколько сотен. Не совпадает — чините промпт и шкалу судьи, а не радуетесь среднему баллу. Судья масштабирует разметку, а не отменяет её.
Как подбирать шкалы под конкретную задачу, чем измерять согласие судьи с человеком и как из этого собрать регламент выкатки промптов — разбираем на практике в курсе «ChatGPT и нейросети в 1С», на живых проектах.
Как это выглядит в зрелом сервисе
Учебный стенд — отдельные скрипты, которые запускаются руками. В рабочем сервисе eval-контур встроен внутрь приложения. Разберу на RAG-сервисе: FastAPI, индексация документов, поиск по векторной базе, чат — и рядом эндпоинты /eval/datasets, /eval/run, /eval/retrieval-run.
Ключевое решение там архитектурное, а не про Langfuse. Есть модуль rag_pipeline.py с общей логикой — сборка контекста из найденных чанков, сборка сообщений, константа CONTEXT_TOP_K. И его докстринг прямо фиксирует зачем: «Используется в chat_service и eval_service, чтобы не дублировать логику».
Звучит как обычный вынос общего кода, но смысл глубже. Если оценочный прогон собирает промпт своим кодом, вы измеряете не то приложение, которое стоит у пользователя. Другой top_k, другой формат заголовка источника, чуть иначе сформулированный системный промпт — и цифра перестаёт что-либо значить ровно в тот момент, когда на неё нужно опереться. Один общий пайплайн для чата и для оценки — техническое условие того, что измерение вообще про ваш прод.
Трасса в eval-контуре собирается вручную, а не колбэком, потому что нужны именованные шаги:
with langfuse.start_as_current_observation(
name=f"rag_eval_{run_name}", as_type="span",
input={"question": question},
metadata={"dataset": dataset_name, "run_name": run_name},
) as span:
results = await search_documents(query=question, top_k=CONTEXT_TOP_K)
with span.start_as_current_observation(
name="retriever", as_type="retriever",
input={"query": question},
output={"chunks": sources_payload(results), "count": len(results)},
):
pass
# ... генерация ответа во вложенном span "generation" ...
raw_score, norm_score = await _judge_answer(llm, question, expected, full_response)
span.score(name="answer_quality", value=norm_score)
span.score(name="answer_score_raw", value=float(raw_score))
Типы наблюдений (retriever, generation) — не косметика: Langfuse по ним раскладывает трассу и считает токены и стоимость отдельно по генерациям. Плюс каждый span привязывается к элементу датасета через dataset_run_items.create — в интерфейсе прогон виден строкой напротив исходного вопроса.
Режимов оценки там два, и это прямое следствие двух схем судьи. run_dataset_eval гоняет полный цикл: поиск, генерация, судья ответа. run_retrieval_eval останавливается на поиске и судит только релевантность найденного. Второй в разы дешевле и отвечает на вопрос «виноват ли ретривер» вообще без генерации — а этот вопрос при разборе плохих ответов встаёт первым.
С чего начинать наблюдаемость?
Порядок, который у меня работает на проектах.
1. Сначала трассировка, только на dev. Один колбэк, ноль изменений в логике приложения. На этом шаге обычно находится половина сюрпризов: в промпт уезжает пустой контекст, история режется не там, модель зовётся дважды на один запрос, в системный промпт подставился шаблон вместо значения. До всяких метрик.
2. Решите, что логировать. Обязательный минимум: полный вход модели после всех подстановок, полный выход, имя модели, версия промпта, идентификатор запроса приложения, чтобы связать трассу с записью в своих логах. Персональные данные в трассе будут — это ещё один аргумент за self-hosted.
3. Минимальный полезный датасет — 20–30 вопросов. Не двести. Тридцать вопросов, из которых десять — реальные провалы, вытащенные из логов, уже ловят регрессию промпта. Дальше датасет растёт инцидентами: сломалось у пользователя — вопрос едет в датасет, и больше эта поломка мимо вас не пройдёт.
4. Оценка — простейшая из возможных. Категория, число, наличие подстроки, попадание в справочник. Судья подключается только там, где без свободного текста не обойтись, и в паре с человеческой разметкой.
5. Одно изменение за прогон. Поменяли промпт — прогнали. Поменяли модель — прогнали. Меняете и то и другое сразу — узнаете, что стало лучше, но не узнаете от чего.
Ни один из этих шагов не требует переписывать приложение. Это ровно тот случай, когда порог входа низкий, а отдача начинается с первого же дня: вы перестаёте спорить о качестве и начинаете его показывать.
Что в итоге
Langfuse закрывает разрыв между «мне кажется, стало лучше» и проверяемым утверждением. Трассировка отвечает, что реально ушло в модель и вернулось из неё. Датасет с прогонами — стало ли в среднем лучше после правки. Судья масштабирует оценку свободного текста, но требует калибровки по живому человеку. А общий пайплайн для чата и для оценки — условие, без которого все эти цифры описывают несуществующее приложение.
Полный eval-контур в проде — это ещё подбор метрик под задачу, разбор режимов отказа, версионирование промптов и регламент выкатки. Собираем это вместе с обвязкой на LangChain и LangGraph, на реальных проектах, в курсе «ChatGPT и нейросети в 1С». Если у вас уже крутится LLM-сервис без наблюдаемости — начните с пункта 1 списка выше: один колбэк, вечер работы, и вы впервые увидите, что происходит внутри.