Понимание того, как настроить рабочие процессы с участием человека (HITL) в LangGraph.
Делиться
Недавно разработанные LLM-системы, такие как GPT-5.4 от OpenAI и Opus 4.6 от Anthropic, продемонстрировали выдающиеся возможности в выполнении длительных задач, решаемых агентами.
В результате мы наблюдаем рост использования агентов LLM как в индивидуальном, так и в корпоративном контексте для выполнения сложных задач, таких как проведение финансового анализа, разработка приложений и проведение масштабных исследований.
Эти агенты, будь то часть высокоавтономной системы или заранее определенного рабочего процесса, могут выполнять многоэтапные задачи, используя инструменты для достижения целей с минимальным участием человека.
Однако «минимальный» не означает полное отсутствие человеческого контроля.
Напротив, проверка человеком остается важной из-за присущей LLM-системам вероятностной природы и потенциальной возможности ошибок.
Эти ошибки могут распространяться и накапливаться на протяжении всего рабочего процесса, особенно когда мы объединяем множество компонентов, выполняющих различные функции.
Вы, наверное, заметили впечатляющий прогресс, которого агенты достигли в области программирования. Причина в том, что код относительно легко проверить (то есть он либо работает, либо выдает ошибку, и обратная связь видна немедленно).
Однако в таких областях, как создание контента, исследования или принятие решений, правильность часто носит субъективный характер и её сложнее оценить автоматически.
Именно поэтому проектирование с участием человека (HITL) остается критически важным.
В этой статье мы рассмотрим, как использовать LangGraph для настройки агентного рабочего процесса с участием человека для генерации и публикации контента на платформе Bluesky.
Содержание
(1) Введение в LangGraph
(2) Пример рабочего процесса
(3) Ключевые понятия
(4) Пошаговое описание кода
(5) Лучшие практики прерываний
Соответствующий репозиторий на GitHub можно найти здесь.
(1) Введение в LangGraph
LangGraph (часть экосистемы LangChain) — это низкоуровневая среда управления агентами и среда выполнения для построения рабочих процессов с участием агентов.
Это мой основной фреймворк, учитывая высокую степень контроля и настраиваемости, что крайне важно для решений производственного уровня.
Хотя LangChain предлагает объект промежуточного программного обеспечения (HumanInTheLoopMiddleware) для простого начала работы с участием человека в звонках агентов, это реализовано на высоком уровне абстракции, который скрывает лежащие в основе механизмы.
LangGraph, напротив, не абстрагирует подсказки или архитектуру, что обеспечивает нам необходимую более тонкую степень контроля. Он явно позволяет нам определять:
- Как происходит обмен данными между этапами
- Здесь принимаются решения и выполняется код.
- Там, где требуется вмешательство человека
Поэтому мы будем использовать LangGraph для демонстрации концепции HITL в рамках агентного рабочего процесса .
Также полезно различать рабочие процессы, управляемые агентами , и автономные агенты искусственного интеллекта.
Рабочие процессы агентов имеют заранее определенные пути и предназначены для выполнения в заданном порядке, при этом LLM-ы и/или агенты интегрированы в один или несколько компонентов. С другой стороны, агенты ИИ автономно планируют, выполняют итеративно продвигаются к цели.
В этой статье мы сосредоточимся на агентных рабочих процессах , в которых мы намеренно включаем контрольные точки, выполняемые человеком, в заранее определенный поток.
(2) Пример рабочего процесса
В качестве примера мы построим следующий рабочий процесс генерации контента для социальных сетей:
- Пользователь вводит интересующую его тему (например, «последние новости об антропологии»).
- Узел веб-поиска использует инструмент Tavily для поиска в интернете статей, соответствующих заданным критериям.
- Выбран лучший результат поиска, который передается в модуль LLM в узле создания контента для генерации публикации в социальных сетях.
- В узле проверки есть два контрольных пункта, в которых проводится проверка человеком:
(i) Представлять сгенерированный контент для утверждения, отклонения или редактирования людьми;
(ii) После утверждения рабочий процесс запускает инструмент API Bluesky и запрашивает окончательное подтверждение перед публикацией в сети.
Вот как это выглядит при запуске из терминала:
А вот и опубликованная запись в моем профиле на Bluesky:
Bluesky — это социальная платформа, похожая на Twitter (X), и она выбрана для этой демонстрации, потому что её API гораздо проще в использовании и доступе.
(3) Ключевые понятия
Основной механизм, лежащий в основе настройки HITL в LangGraph, — это концепция прерываний .
Прерывания (с использованием interrupt() и Command в LangGraph) позволяют приостанавливать выполнение графа в определенных точках, отображать определенную информацию пользователю и ожидать его ввода перед возобновлением рабочего процесса.
Объект `command` — это универсальный объект, позволяющий обновлять состояние графа (
update), указывать следующий узел для выполнения (goto) или получать значение для возобновления выполнения графа (`resume).
Вот как выглядит этот процесс:
(1) При достижении функции interrupt() выполнение приостанавливается, и пользователю отображается полезная нагрузка, переданная в нее. Полезная нагрузка, передаваемая в interrupt обычно должна быть в формате JSON или строки, например,
decision = interrupt("Should we get KFC for lunch?") # String shown to user (2) После ответа пользователя мы передаем значения ответа в граф для возобновления выполнения. Это включает в себя использование Command и ее параметра resume в рамках повторного вызова графа:
if human_response == "yes": return graph.invoke(Command(resume="KFC")) else: return graph.invoke(Command(resume="McDonalds")) (3) Значение ответа в resume возвращается в переменной decision , которую узел будет использовать для дальнейшего выполнения узла и последующего потока графа:
if decision == "KFC": return Command(goto="kfc_order_node", update={"lunch_choice": "KFC") else: return Command(goto="mcd_order_node", update={"lunch_choice": "McDonalds")Прерывания являются динамическими и могут быть размещены в любом месте кода, в отличие от статических точек останова, которые фиксируются до или после определенных узлов.
При этом мы обычно размещаем прерывания либо внутри узлов, либо внутри инструментов, вызываемых во время выполнения графа.
Наконец, поговорим о контрольных точках . Когда рабочий процесс приостанавливается при прерывании, нам нужен способ сохранить его текущее состояние , чтобы он мог возобновиться позже.
Поэтому нам необходима контрольная точка для сохранения состояния, чтобы оно не было потеряно во время паузы прерывания. Контрольную точку можно рассматривать как снимок состояния графа в определенный момент времени.
В процессе разработки допустимо сохранять состояние в памяти с помощью контрольной точки InMemorySaver .
Для продакшена лучше использовать хранилища данных, такие как Postgres или Redis. Поэтому в этом примере мы будем использовать SQLite Checkpoint вместо хранилища в оперативной памяти.
Чтобы гарантировать возобновление работы графика точно в точке , где произошло прерывание, необходимо передать и использовать тот же идентификатор потока .
Представьте себе поток как единую сессию выполнения (подобную отдельному индивидуальному разговору), где каждый поток имеет уникальный идентификатор и поддерживает собственное состояние и историю.
Идентификатор потока передается в config при каждом вызове графа, чтобы LangGraph знал, из какого состояния следует возобновить работу после прерывания.
Теперь, когда мы рассмотрели понятия прерываний, Command , контрольных точек и потоков, давайте перейдем к разбору кода.
Поскольку основное внимание будет уделено механизмам взаимодействия человека с системой, мы не будем рассматривать подробную настройку кода. Полную реализацию можно найти в репозитории GitHub.
(4) Пошаговое описание кода
(4.1) Начальная настройка
Начнём с установки необходимых зависимостей и генерации API-ключей для Bluesky, OpenAI, LangChain, LangGraph и Tavily.
# requirements.txt langchain-openai>=1.1.9 langgraph>=1.0.8 langgraph-checkpoint-sqlite>=3.0.3 openai>=2.20.0 tavily-python>=0.7.21 # env.example export OPENAI_API_KEY=your_openai_api_key export TAVILY_API_KEY=your_tavily_api_key export BLUESKY_HANDLE=yourname.bsky.social export BLUESKY_APP_PASSWORD=your_bluesky_app_password(4.2) Определить государство
Мы создали State , который представляет собой общий структурированный объект данных, служащий центральной памятью графа. Он включает поля, содержащие ключевую информацию, такую как содержимое публикации и статус одобрения.
Ключ post_data указывает, где будет храниться сгенерированное содержимое записи.
(4.3) Прерывание на уровне узла
Ранее мы упоминали, что прерывания могут происходить на уровне узла или внутри вызовов инструментов. Давайте посмотрим, как работает первый вариант, настроив узел для проверки человеком.
Цель узла проверки — приостановить выполнение и представить пользователю черновой вариант содержимого для ознакомления.
Здесь мы видим работу функции interrupt() (строки с 8 по 13), где выполнение графа приостанавливается на первом участке функции узла.
Ключ details , передаваемый в interrupt() содержит сгенерированное содержимое, а ключ action запускает функцию обработчика ( handle_content_interrupt() ) для поддержки проверки:
Сгенерированный контент выводится на терминал для просмотра пользователем, и он может утвердить его как есть, отклонить сразу или отредактировать непосредственно в терминале перед утверждением.
В зависимости от принятого решения функция-обработчик возвращает одно из трех значений:
-
True(одобрено), -
False(отклонено), или - Строковое значение, соответствующее контенту, отредактированному пользователем (
edited).
Возвращаемое значение передается обратно узлу проверки с помощью graph.invoke(Command=resume…) , который возобновляет выполнение с того места, где был вызван interrupt() (строка 15), и определяет, к какому узлу перейти дальше: утвердить, отклонить или отредактировать контент и перейти к утверждению.
(4.4) Прерывание на уровне инструмента
Прерывания также можно определять на уровне вызова инструмента. Это демонстрируется на следующем этапе проверки человеком в узле утверждения перед публикацией контента в сети Bluesky.
Вместо того чтобы размещать interrupt() внутри узла, мы размещаем её внутри инструмента publish_post , который создаёт публикации через API Bluesky:
Как и на уровне узла, мы вызываем функцию обработчика ( handle_publish_interrupt ), чтобы зафиксировать решение, принятое человеком:
Результатом данного этапа проверки может быть следующее:
-
{"action": "confirm"}или -
{"action": "cancel},
Вторая часть кода (т.е., начиная со строки 19) в инструменте publish_post использует это возвращаемое значение для определения того, следует ли продолжать публикацию записи в Bluesky или нет.
(4.5) Настройка графа с помощью Checkpointer
Далее мы соединяем узлы в граф для компиляции и вводим контрольную точку SQLite для захвата снимков состояния при каждом прерывании.
По умолчанию SQLite разрешает использовать соединение только тому потоку, который его создал. Поскольку LangGraph использует пул потоков для записи контрольных точек, нам необходимо установить
check_same_thread=Falseчтобы разрешить доступ к соединению и этим потокам.
(4.6) Настройка полного рабочего процесса с конфигурацией
После того как график готов, мы помещаем его в рабочий процесс, который запускает конвейер генерации контента.
Этот рабочий процесс включает в себя настройку идентификатора потока, который передается каждому graph.invoke() . Этот идентификатор является связующим звеном между вызовами, так что граф приостанавливается при прерывании и возобновляет работу с того места, где остановился.
Возможно, вы заметили ключ __interrupt__ в приведенном выше коде. Это просто специальный ключ, который LangGraph добавляет к результату всякий раз, когда срабатывает interrupt() .
Иными словами, это основной сигнал, указывающий на то, что выполнение графа приостановлено и ожидает ввода данных от человека, прежде чем продолжить.
Включение __interrupt__ в цикл while означает, что цикл постоянно проверяет, продолжается ли прерывание. Как только прерывание разрешается, клавиша исчезает, и цикл while завершается.
После завершения процесса мы можем запустить его следующим образом:
run_hitl_workflow(query="latest news about Anthropic")(5) Лучшие практики прерываний
Хотя прерывания играют важную роль в обеспечении рабочих процессов HITL, их неправильное использование может привести к сбоям.
Поэтому я рекомендую ознакомиться с документацией LangGraph. Вот несколько практических правил, которые следует помнить:
- Не следует заключать вызовы прерываний в блоки try/except, иначе они не смогут корректно приостановить выполнение.
- Всегда сохраняйте порядок вызовов прерываний неизменным, не пропускайте и не меняйте их порядок.
- Передайте в прерывания только значения, безопасные для работы с JSON, и избегайте сложных объектов.
- Убедитесь, что любой код, предшествующий прерыванию, может безопасно выполняться несколько раз (т.е., обеспечивается идемпотентность), или переместите его после прерывания.
Например, я столкнулся с проблемой в узле веб-поиска, где я разместил прерывание сразу после поиска по Тавили. Цель заключалась в том, чтобы приостановить поиск и дать пользователям возможность просмотреть результаты поиска для генерации контента.
Но поскольку прерывания работают путем повторного запуска узлов, с которых они были вызваны, узел просто повторно запустил веб-поиск и передал другой набор результатов поиска, отличный от тех, которые я одобрил ранее.
Поэтому прерывания лучше всего работают в качестве промежуточного этапа перед действием, но если мы используем их после недетерминированного шага (например, поиска), нам необходимо сохранить результат, иначе мы рискуем получить что-то другое при возобновлении работы.
Подводя итоги
В задачах, выполняемых с помощью агентов, проверка человеком может показаться узким местом, но она по-прежнему имеет решающее значение, особенно в областях, где результаты субъективны или их трудно проверить.
LangGraph упрощает создание рабочих процессов HITL с использованием прерываний и контрольных точек.
Таким образом, задача состоит в том, чтобы решить, где разместить эти точки принятия решений человеком, чтобы найти оптимальный баланс между контролем и эффективностью.
Кеннет Люн. Все работы Кеннета Люна.
Источник: towardsdatascience.com
