Отслеживание пайплайна LangGraph для блогов с использованием Langfuse: сегменты, оценки, предупреждения
Оснастить многоагентный пайплайн LangGraph инструментами Langfuse: один трейс на статью, вложенные сегменты, предупреждения, оценки и рутину отладки для слабых черновиков.
Источники: Building Multi-Agent System From Scratch — Part 6: Observability and Debugging
Механизмы защиты (проверка исходных данных, проверка цитат, ограничение количества попыток, передача задачи человеку) полезны лишь в том случае, если после неудачной попытки можно восстановить процесс. В такой ситуации у многоагентной системы для обработки блогов возникает множество вопросов: были ли источники информации ненадежными? Игнорировал ли автор предупреждения о необходимости цитирования? Было ли правильным решение рецензента отклонить работу? Завершилась ли операция поиска из-за истечения времени? Сколько ресурсов потребовалось на вызовы модели и использование токенов?
Без возможности наблюдения система представляет собой «черный ящик», окруженный еще более множеством «черных ящиков». Трассировка действий многоагентной системы — это особенно сложная часть работы. В данной конфигурации к системе LangGraph для обработки блогов добавляется Langfuse, который генерирует одну трассу для каждой обработки статьи, содержащую информацию о передаче задач, вызовах моделей и инструментов, времени выполнения, затратах, ошибках и результатах оценки.
Определение того, что такое трасса
Прежде чем приступать к коду, выберите единицу для анализа. В этой системе один трейс соответствует одной попытке создания одной статьи. Он начинается с момента подачи темы и заканчивается, когда система завершает работу, делает паузу в ожидании человека или терпит неудачу. Внутри него содержатся записи по каждому выполненному действию:
blog-pipeline
├── research-agent
│ ├── search-web
│ └── research-model
├── writer-agent
│ └── writer-model
├── citation-check
│ └── citation-review-model
└── reviewer-agent
└── reviewer-model
Langfuse работает с тремя типами сущностей. Трейс — это один полный запуск от начала до конца. Наблюдение — это отдельный шаг внутри него, который может представлять собой фрагмент кода (логику приложения), вызов модели большого языка или небольшой маркер события. Сессия объединяет несколько трейсов, например, все попытки работы с одной и той же темой блога.
Иерархия отражает процесс отладки: сначала рассматривается полный запуск, затем выясняется, какой конкретный агент, инструмент или вызов модели стал причиной проблемы.
Настройка Langfuse
Создайте проект Langfuse и укажите его учетные данные с помощью переменных окружения. Секретный ключ никогда не следует записывать в исходные файлы.
pip install -U langfuse
export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_BASE_URL="https://cloud.langfuse.com"
export LANGFUSE_TRACING_ENVIRONMENT="development"
Пакет langfuse представляет собой Python SDK. Два ключа используются для аутентификации приложения в проекте, а LANGFUSE_BASE_URL определяет место развертывания. Пример настроен для европейского региона облака, поэтому при необходимости замените его на другой регион или самостоятельно развернутую инстанцу.
Метка среды позволяет разделять следы разработки и производственные данные. Хорошо подходят краткие и стабильные названия, такие как development, staging и production. Без них эксперименты и реальные запуски статей смешиваются в одном виде, что затрудняет отладку.
Подключение обработчика возврата
Langfuse поставляется с обработчиком возврата LangChain. LangGraph использует механизмы выполнения и обработки возврата из LangChain, поэтому тот же обработчик фиксирует события, связанные с графом, цепочкой, инструментами и моделями:
from langfuse import get_client, propagate_attributes
from langfuse.langchain import CallbackHandler
langfuse = get_client()
langfuse_handler = CallbackHandler()
get_client() создает клиент на основе переменных окружения. Обработчик должен быть создан один раз при запуске и использоваться повторно при каждом выполнении. Он прослушивает события LangChain и передаёт их в качестве наблюдений, что устраняет необходимость вручную отслеживать каждый вызов модели.
Затем вступает в игру функция propagate_attributes. Она проставляет одинаковые метки ко всем наблюдениям в ходе одного выполнения, что позволяет фильтровать запросы к поиску, генерацию текста и процесс рецензирования как части одного цикла обработки статьи.
Вызов графа оборачивается в корневой спан, причём обработчик передаётся через конфигурацию LangGraph:
def run_blog_pipeline(topic: str, blog_id: str, user_id: str):
initial_state = {
"topic": topic,
"audience": "developers new to agent systems",
"research_brief": "",
"sources": [],
"open_questions": [],
"article_draft": "",
"review_feedback": "",
"citation_issues": [],
"approved": False,
"revision_count": 0,
"status": "researching",
}
with langfuse.start_as_current_observation(
as_type="span",
name="blog-pipeline",
input={"topic": topic, "audience": initial_state["audience"]},
) as pipeline_span:
with propagate_attributes(
trace_name="blog-pipeline",
session_id=blog_id,
user_id=user_id,
tags=["blog-pipeline", "langgraph"],
version="1.0.0",
metadata={"workflow": "research-write-review"},
):
trace_id = langfuse.get_current_trace_id()
result = graph.invoke(
initial_state,
config={"callbacks": [langfuse_handler]},
)
pipeline_span.update(
output={
"status": result["status"],
"approved": result["approved"],
"revision_count": result["revision_count"],
}
)
return result, trace_id
Некоторые детали этого обертывающего элемента имеют важное значение. Внешний вызов start_as_current_observation() создаёт корневое наблюдение, причём его входные параметры остаются максимально сжатыми — только тема и аудитория. Именно это сразу видно в списке трейсов, поэтому информация об внутреннем состоянии туда не должна попадать.
Функция propagate_attributes() передаёт все атрибуты каждому вложенному наблюдению. Параметр blog_id служит идентификатором сессии, объединяя попытки обработки одной статьи. Значение version указывает на тип рабочего процесса и версию запроса, поэтому при изменении поведения трейсы можно фильтровать по версии, а не опираться на догадки.
Обработчик проходит по стандартному параметру config LangGraph и автоматически собирает детали на уровне фреймворка: информацию о каждом вызове узла, генерации LLM, вызовах инструментов, входных/выходных данных, задержке, количестве токенов и стоимости обработки.
В конце корневой спан обновляется с тремя результатами на уровне пайплайна: информацией о том, была ли обработка завершена, одобрен ли черновик и сколько было выполнено правок. Возвращаемый trace_id следует сохранять в журналах приложения или в интерфейсе, что позволяет напрямую перейти от обработки статьи к соответствующему треку Langfuse.
Спаны на уровне бизнес-логики внутри агентов
Автоматическое отслеживание подтверждает, что модель была вызвана. Оно не показывает, что именно каждый специалист передавал следующему. Для этого необходимо добавить дополнительный спан внутри узла агента и передавать аргумент config в вложенные вызовы LangChain или LangGraph, чтобы обработчик сохранял тот же контекст трека:
from langchain_core.runnables import RunnableConfig
def research_node(state: BlogState, config: RunnableConfig) -> dict:
with langfuse.start_as_current_observation(
as_type="span",
name="research-agent",
input={"topic": state["topic"], "audience": state["audience"]},
) as span:
result = research_agent.invoke(
{"topic": state["topic"], "audience": state["audience"]},
config=config,
)
span.update(
output={
"source_count": len(result["sources"]),
"open_question_count": len(result["open_questions"]),
"brief": result["research_brief"],
}
)
return {
"research_brief": result["research_brief"],
"sources": result["sources"],
"open_questions": result["open_questions"],
"status": "writing",
}
Этот элемент дополняет обработчик возврата вызова, но не заменяет его. Он добавляет смысл, который фреймворк не может определить самостоятельно, а именно что означает «исследование завершено успешно» в данной цепочке обработки. Здесь он фиксирует количество источников, число нерешенных вопросов и краткое описание, переданное автору.
Передача параметра config: RunnableConfig в метод research_agent.invoke() позволяет передавать обработчики возврата вызова и контекст трейсинга. Игнорирование этого параметра является классической проблемой с возможностями отслеживания: верхнеступенчатая структура обрабатывается, но вложенные вызовы отображаются как не связанные с ней трейсы или исчезают.
Сохраняйте постоянными имена элементов отслеживания: research-agent, writer-agent, citation-check. Название темы, имени модели или ID пользователя не следует включать в эти имена, поскольку такие данные относятся к входным данным, атрибутам модели или метаданным. Постоянные имена обеспечивают совместимость панелей управления и фильтров при изменениях в реализации.
Предупреждения и ошибки
Не каждый неожиданный результат означает сбой. Сомнительный источник, исчерпанный бюджет на редактирование или неподдерживаемая цитата также имеют значение и должны отображаться в качестве предупреждения.
def record_source_assessment(assessment: SourceAssessment) -> None:
if assessment.suspicious_content:
langfuse.update_current_span(
level="WARNING",
status_message="Untrusted source contained agent-directed instructions.",
)
def record_pipeline_failure(error: Exception) -> None:
langfuse.update_current_span(
level="ERROR",
status_message=f"Pipeline failed: {type(error).__name__}",
)
Langfuse поддерживает уровни логов для записей наблюдений. WARNING применяется к запускам, которые завершились, но требуют дополнительной проверки, в то время как ERROR используется для фактических сбоев. Сообщения о статусе должны описывать проблему без включения конфиденциального исходного текста или секретов в данные телеметрии.
Устройство для записи оценки исходного кода должно находиться в пути проверки исходного кода, а устройство для записи сбоев — в обработчике ошибок, способном их устранить или сообщить об них. Запись каждой исключения только для последующего логирования — плохая идея, поскольку неожиданные ошибки всё равно должны попадать в систему мониторинга ошибок и сигнализировать об этом явно.
Оценка результата
Трейс указывает, что произошло, а оценка показывает, был ли результат хорошим. После завершения запуска следует присвоить первоначальную детерминистичную оценку, полученную на основе состояния конвейера:
result, trace_id = run_blog_pipeline(
topic="How AI agents use tools",
blog_id="blog-ai-tools-001",
user_id="philip",
)
if trace_id:
langfuse.create_score(
trace_id=trace_id,
name="review_approved",
value=1 if result["approved"] else 0,
data_type="BOOLEAN",
comment=result["status"],
)
langfuse.create_score(
trace_id=trace_id,
name="revision_count",
value=float(result["revision_count"]),
data_type="NUMERIC",
)
Первая оценка отражает структурированное решение рецензента о согласии или отклонении, а вторая показывает, сколько правок потребовалась для черновика. Накопленные со временем данные раскрывают гораздо больше, чем просто отдельные случаи. Какая модель, версия запроса, категория темы или подход к поиску источников приводит к меньшему количеству правок и большему числу одобрений?
Ни то, ни другое не является абсолютной истиной, поскольку рецензент может ошибаться. Оценка, данная человеком после прочтения окончательного варианта, может поступить позже. Важно не доверять лишь нескольким запоминающимся примерам, а сделать качество измеримым и сравнивать его в разных случаях.
Процедура отладки
Когда статья оказывается слабой, откройте её запись и последовательно задайте себе следующие вопросы:
- Прошёл ли график ожидаемым путём? Обратите внимание на этапы исследования, написания текста, проверки цитат и рецензирования, а также на любые циклы правок.
Вот почему важна структура трассировки. Хорошо названные элементы структуры превращают общее замечание вроде «в блоге была ошибка» в точное утверждение, например «автор после второй корректировки добавил утверждение, отсутствовавшее в исходном описании задания».
Защита конфиденциальных данных
Функция отслеживания может записывать вводные данные, результаты обработки, выводы инструментов и метаданные, что также означает возможность сбора информации, которую не следует хранить. Ключи API, заголовки аутентификации и полные конфигурации секретов никогда не должны попадать в Langfuse. Если данные пользователей позже поступают в систему, необходимо проверить, что именно записывается, и настроить маскировку перед экспортом данных отслеживания. Langfuse предоставляет функции маскировки, которые скрывают конфиденциальные значения до их выхода из приложения.
Основное правило: записывайте столько информации, сколько необходимо для объяснения принятого решения, но не больше, чем можно безопасно сохранить.
Что получает система
Каждая операция теперь оставляет после себя операционный отчет:
- Каждая попытка обработки статьи представляет собой отдельный запись отслеживания.
- Агенты, модели и инструменты отображаются в виде вложенных записей.
- Фиксированные имена вместе с метаданными, версией, сессией и средой облегчают фильтрацию.
Отладка больше не сводится к просмотру логов терминала и догадкам, а становится проверкой точного момента, когда произошла ошибка.