Практические заметки: Создание многопроцессорной системы с нуля — Часть 6

В следующих заметках описывается практический подход к изучению темы «Создание многоподписческой системы с нуля — Часть 6: Наблюдаемость и отладка». Основное внимание уделяется контрактам, проверкам и шаблонам кода, которые можно легко вставить, а не мотивирующим аспектам. На этапе обзора сначала запишите контракт: необходимые входные данные, сигнал успешного выполнения и действия при частичной неудаче. Такой список поможет сохранять честность при последующих изменениях кода. Лучше использовать небольшие, тестируемые единицы кода вместо обширных скриптов. Если какой-то шаг не сработает, причина неудачи должна указывать на конкретную ответственность, а не на запутанную цепочку операций.

Что должно представлять собой трейсинг?

Этот этап «Что следует отслеживать» работает наилучшим образом, когда его рассматривают как измеримую поверхность. Запишите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объема работ. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия создаваемым элементам, определите критерии успешного выполнения и не соглашайтесь на молчаливое частичное завершение работы. Сохраняйте структуру графа простой и типизированной. Вложенные структуры скрывают информацию о том, какой узел заполнил тот или иной поле, что приводит к нарушению возобновления работы после перерывов.

blog-pipeline
├── research-agent
│   ├── search-web
│   └── research-model
├── writer-agent
│   └── writer-model
├── citation-check
│   └── citation-review-model
└── reviewer-agent
    └── reviewer-model

Подключение Langfuse к LangGraph

Этап подключения Langfuse к LangGraph работает наилучшим образом, если рассматривать его как измеримую среду. Соберите один идеальный пример обработки данных, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объема работ. Записывайте временные показатели, а также стоимость токенов или запросов рядом с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам. Сохраняйте структуру графа простой и типизированной; вложенные структуры данных маскируют информацию о том, какой узел заполнил тот или иной поле, и приводят к нарушению продолжения обработки после перерывов.

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"

from langfuse import get_client, propagate_attributes
from langfuse.langchain import CallbackHandler


langfuse = get_client()
langfuse_handler = CallbackHandler()

Отслеживание полного цикла обработки статьи

Этап Trace one complete article работает наилучшим образом, когда его рассматривают как измеримую поверхность. Сохраняйте один идеальный протокол, один случай сбоя и записку о возврате к предыдущему состоянию до расширения объёма работы. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь граф. Сохраняйте состояние графа простым и типизированным. Вложенные структуры скрывают информацию о том, какой узел записал какое поле, и мешают возобновлению работы после прерываний. Этап Trace one complete article работает наилучшим образом, когда его рассматривают как измеримую поверхность. Сохраняйте один идеальный протокол, один случай сбоя и записку о возврате к предыдущему состоянию до расширения объёма работы. Предпочитайте небольшие, тестируемые единицы кода вместо обширных скриптов. Когда какой-то шаг сбивается, причина сбоя должна указывать на конкретную ответственность, а не на запутанную цепочку операций.

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

Добавьте замечания, объясняющие процесс передачи данных

Для раздела «Добавить замечания, объясняющие этап», перед изменением кода необходимо определить входные данные, ответственного за шаг и критерии завершения. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия элементам, определите критерии успеха и не допускайте молчаливого частичного завершения работы. Внедряйте утверждение человека для операций, связанных с тратой денег или изменением производственных данных. Подключение на этапе компиляции не эквивалентно полноте выполнения бизнес-задач.

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",
    }

Отметьте сигналы, требующие внимания

Для маркировки этапов необходимо определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Записывайте время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости заранее предотвращает неожиданные счета при переходе с демо-среды в общедоступные среды. Вносите утверждение человека для операций, связанных с тратой денег или изменением производственных данных. Подключение на этапе компиляции не гарантирует полноты решения с точки зрения бизнеса.

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__}",
    )

Преобразование решений рецензентов в оценки

Чтобы преобразовать решения Turn Reviewer в конкретные этапы, необходимо определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, которое операторы могут проверять, не читая весь код. Вводите ручное утверждение для операций, связанных с тратой денег или изменением производственных данных. Подключение на этапе компиляции не гарантирует полноты бизнес-логики. Чтобы преобразовать решения Turn Reviewer в конкретные этапы, необходимо определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Предпочитайте небольшие, тестируемые единицы кода вместо обширных скриптов. При сбое шага он должен указывать на конкретную причину, а не на целый комплекс факторов.

угловатый трубопровод.

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",
    )

Отладка неудачной работы за пять вопросов

При работе над этапом отладки неудачной работы сначала запишите контракт: требуемые входные данные, сигнал о успехе и что происходит при частичной неудаче. Такой список помогает сохранять честность при последующих изменениях кода. Рассматривайте этот этап как контракт между входными данными и проверенными выходными данными. Дайте названия результатам обработки, определите критерии успеха и не допускайте беззвучного частичного завершения работы. Создавайте контрольные точки после дорогостоящих шагов. Система возобновления работы не должна снова взимать плату за один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий узел заново.

Возможность наблюдения также имеет границы конфиденциальности

При работе над возможностями наблюдаемости также существует определенный этап: сначала необходимо зафиксировать условия работы — требуемые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист помогает сохранять честность при последующих изменениях кода. Рядом с функциональными результатами следует записывать время выполнения и стоимость токенов или запросов. Отслеживание затрат с самого начала предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам. После дорогостоящих шагов следует делать контрольные точки. Функция возобновления работы не должна снова взимать плату за один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий элемент работы.

Что мы создали

При работе над этапом «Что мы создали» сначала запишите условия работы системы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист помогает сохранять честность последующих изменений в коде. Храните конфигурацию отдельно от кода приложения. Файлы с настройками, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь код. Создавайте контрольные точки после дорогостоящих операций. Функция возобновления работы не должна повторно взимать плату за один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий шаг. При работе над этапом «Что мы создали» сначала запишите условия работы системы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист помогает сохранять честность последующих изменений в коде. Предпочитайте небольшие, тестируемые модули большим скриптам. При сбое какого-либо шага причина неудачи должна указывать на конкретную ответственность, а не на запутанную структуру обработки данных.

Чек-лист операций

На этапе операционного чек-листа необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии.

Документируйте одновременно «идеальный» сценарий работы и сценарий восстановления. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки.

Внедряйте утверждение человеком для операций, связанных с тратой денег или изменением производственных данных. Настройки во время компиляции не заменяют полноты бизнес-логики.

Напишите краткое руководство: как обновлять ключи, как опустошать очередь, как откатить последнюю загрузку данных.

Предпочитайте небольшие, тестируемые модули большим скриптам. При сбое шага причина должна быть связана с конкретной ответственностью, а не с запутанной цепочкой операций.

Внедряйте человеческое утверждение для операций, связанных с расходованием средств или изменением производственных данных. Компиляционная настройка не гарантирует полноты функционала бизнес-приложения.

Перед внедрением всей стек-технологии заморозьте версии, сохраните эталонные записи для критически важных этапов и уточните шаги отката. В совместных средах необходимы ограничения на частоту запросов, проверки принадлежности пользователя и четко определенный ответственный за обновление секретов. Лучше выбирать надежность, чем креативные одноразовые демонстрации.

Примечание к заданию cf19385cb4a9: не храните ключи поставщика в репозитории, установите лимит токенов на сессию и сохраняйте записи рядом с фикстурами оценки, чтобы последующие замены моделей оставались сопоставимыми.