Відстеження пайплайну блогу LangGraph за допомогою Langfuse: фрагменти, оцінки, попередження
Оснастити багатоагентний пайплайн LangGraph інструментами Langfuse: один слід за статтею, вкладені фрагменти, попередження, оцінки та процедура виправлення помилок для слабких чернеток.
Джерела: Building Multi-Agent System From Scratch — Part 6: Observability and Debugging
Механізми захисту (перевірка джерел, перевірка посилань, обмеження кількості спроб, передача справи до людини) допомагають лише тоді, коли можна пізніше відтворити невдалу спробу. У такому разі багатоагентний конвеєр для блогів створює безліч запитань: чи були джерела дослідження ненадійними? Чи проігнорував автор попередження щодо посилань? Чи було правильним відхилення рецензентом? Чи закінчився тайм-аут пошуку? Скільки коштувала спроба у вигляді викликів моделей та токенів?
Без можливості спостереження система є чорною скринькою, оточеною ще більшою кількістю чорних скриньок. Відстеження багатьох агентів є надзвичайно складною частиною цієї роботи. Ця конфігурація додає Langfuse до конвеєра для блогів LangGraph, створюючи один запис відстеження за кожну обробку статті, який містить інформацію про передачу завдань, виклики моделей та інструментів, час виконання, витрати, помилки та результати оцінки.
Визначення того, що таке запис відстеження
Перш ніж переходити до коду, виберіть одиницю, яку потрібно перевірити. У цьому конвеєрному процесі один запис відповідає одній спробі створити одну статтю. Він починається з моменту подання теми та закінчується, коли процес завершується, зупиняється через втручання людини або зазнає невдачі. Усередині нього містяться спостереження за кожною виконаною роботою:
blog-pipeline
├── research-agent
│ ├── search-web
│ └── research-model
├── writer-agent
│ └── writer-model
├── citation-check
│ └── citation-review-model
└── reviewer-agent
└── reviewer-model
Langfuse працює з трьома поняттями. Трек — це один повний запуск від початку до кінця. Спостереження — це окремий крок у межах цього треку, який може бути фрагментом (логікою додатку), операцією генерації (викликом LLM) або подією (маленьким маркером). Сесія об’єднує кілька треків, наприклад, усі спроби щодо однієї теми блогу.
Ієрархія відображає процес дебаггінгу: спочатку розглядається повний запуск, а потім детально аналізується конкретний агент, інструмент чи виклик моделі, які спричинили проблему.
Налаштування 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. Вона наносить однакові мітки на кожне спостереження під час одного запуску, що дозволяє фільтрувати виклики пошуку, генерацію тексту та перевірку як частини одного процесу обробки статті.
Виклик графа обгорнутий у кореневий span, причому обробник передається через конфігурацію 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 визначає процес роботи та версію результату, тож у разі змін поведінки сліди можна фільтрувати за версією, а не за припущеннями.
Обробник працює зі звичайними налаштуваннями LangGraph у форматі config, і звідти автоматично збирає деталі на рівні фреймворку: кожен виклик вузла, генерація моделлю 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. Тему, назву моделі чи ідентифікатор користувача не слід вбудовувати в назви, оскільки ці дані належать до вхідних даних, атрибутів моделі чи метаданих. Постійні назви забезпечують те, що панелі керування та фільтри залишаються функціональними після змін у реалізації.
Попередження та помилки
Не кожен незвичайний результат є збоєм. Сумнівне джерело, вичерпаний бюджет на редагування чи непідтримуваний посилання також мають значення та повинні відображатися як попередження.
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 пропонує функції маскування, які приховують конфіденційні значення перед їх виходом з програми.
Основне правило: записуйте достатньо інформації для пояснення рішення, але не більше, ніж можна безпечно зберігати.
Що отримує пайплайн
Кожен запуск тепер залишає оперативний запис:
- Кожна спроба обробки статті є окремим записом трекування.
- Агенти, моделі та інструменти відображаються як вкладені записи спостережень.
- Константні назви разом із метаданами, версією, сеансом та середовищем полегшують фільтрацію.
Виправлення помилок більше не полягає у прогляданні журналів термінала та здогадках, а стає перевіркою саме того моменту, де щось пішло не так.