LangGraph human-in-the-loop: подтверждение действий локального агента
На странице · 4 разделов
Для подтверждения действия в LangGraph поставьте `interrupt()` перед вызывающим последствия инструментом, сохраните состояние графа с checkpointer и возобновите тот же `thread_id` через `Command(resume=...)`. Отказ должен обходить инструмент, а разрешение — запускать только заранее проверенное действие. Пример ниже исполняет оба пути на синтетической записи; модель Ollama и внешняя CRM в нём не запускались.
Документация проверена 29 сентября 2026 года. Синтетическая проверка ниже выполнена с
langgraph==1.0.9на двух тестовых потоках. Она не использует Ollama, реальные данные или производственную CRM. Полная проверка локального агента требует отдельно испытать выбранную модель, её доступ к сети и настоящий инструмент с ограниченными правами.
Что именно подтверждает человек, а что остаётся за системой?
Локальная модель может предложить вызов инструмента, но не должна исполнять его сама без политики приложения. ChatOllama описывает bind_tools() для модели с поддержкой tool calling и возвращаемые структурные tool_calls; документация Ollama различает ответ с вызовом и последующее исполнение инструмента приложением s5s6. Передавайте в approval-граф только разобранную заявку с разрешённым именем инструмента и допустимыми аргументами. Текст «я обновил карточку» не является следом вызова s5s6.
В примере ниже действие mark_demo относится к вымышленной записи DEMO-1 и вызывает только локальную тестовую функцию. На практике цепочка такова: задача сотрудника → выбранная и проверенная на tools локальная модель → runtime Ollama → предложение структурного вызова → interrupt() → решение уполномоченного сотрудника → ограниченный инструмент → проверенный результат. Первая часть — интеграция читателя, а не подтверждённый запуск Modelium. Для подготовки модели см. общий гайд агента и проверку tool calling.
Точка контроля — до внешнего действия. LangGraph сохраняет состояние графа при interrupt(), показывает сериализуемый payload проверяющему и принимает решение через Command(resume=...) в том же thread_id s1s2. Отказ не должен попадать в исполнительный узел; разрешение не должно автоматически давать доступ ко всем инструментам. Подтверждение человеком не заменяет allowlist, права пользователя и журнал действий: отдельно проверьте безопасность MCP-инструментов.
Какие предусловия нужны до теста?
Нужны Python с venv, установленный langgraph==1.0.9 и возможность локально исполнить безвредную функцию. Для демонстрации Ollama, сеть, скачивание модели, GPU и VRAM не нужны. Это намеренно изолирует механику паузы от качества модели. Если затем подключаете ChatOllama, выбирайте конкретную модель, умеющую tools по её документации, проверяйте версии и условия весов; сам факт установки модели не гарантирует корректность её предложений s5s6. Для чисел памяти или качества этот материал данных не имеет.
python3 -m venv .venv
. .venv/bin/activate
python -m pip install 'langgraph==1.0.9'
InMemorySaver нужен только для теста в одном процессе: при рестарте контрольные точки исчезнут. Для локального восстановления между запусками существует отдельный langgraph-checkpoint-sqlite, а для production документация предлагает постоянный backend, например Postgres; выбор backend и доступ к нему нужно спроектировать до реального пилота s2s7.
Как проверить паузу, отказ и разрешение в исполняемом графе?
Сохраните код как hitl_demo.py. Узел review не выполняет действия до interrupt(). После возобновления он принимает только булево True как разрешение; любой другой JSON-ответ ведёт к отказу. Две ветки графа ведут к тестовому исполнению либо к отказу. Этот контроль решает только одну заранее составленную заявку, не предлагает автоматически редактировать произвольные аргументы s1s3.
from typing import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
class State(TypedDict, total=False):
proposal: dict[str, str]
approved: bool
result: str
def review(state: State):
decision = interrupt({"question": "Разрешить тестовое действие?", "proposal": state["proposal"]})
return {"approved": decision is True}
def route(state: State) -> str:
return "execute" if state["approved"] else "deny"
def execute(state: State):
# Безвредный тестовый инструмент вместо настоящей записи в CRM.
assert state["proposal"]["action"] == "mark_demo"
return {"result": "Инструмент mark_demo вызван для " + state["proposal"]["record_id"]}
def deny(state: State):
return {"result": "Действие отклонено; инструмент не вызывался"}
builder = StateGraph(State)
builder.add_node("review", review)
builder.add_node("execute", execute)
builder.add_node("deny", deny)
builder.add_edge(START, "review")
builder.add_conditional_edges("review", route, {"execute": "execute", "deny": "deny"})
builder.add_edge("execute", END)
builder.add_edge("deny", END)
graph = builder.compile(checkpointer=InMemorySaver())
for thread_id, choice in (("demo-deny", False), ("demo-approve", True)):
config = {"configurable": {"thread_id": thread_id}}
paused = graph.invoke({"proposal": {"action": "mark_demo", "record_id": "DEMO-1"}}, config=config)
assert "__interrupt__" in paused and "result" not in paused
print("PAUSED", thread_id, paused["__interrupt__"][0].value)
final = graph.invoke(Command(resume=choice), config=config)
assert final["approved"] is choice
assert ("вызван" in final["result"]) is choice
print("RESULT", thread_id, final["result"])
Запустите python hitl_demo.py. В тесте Modelium для demo-deny сначала показана заявка mark_demo/DEMO-1, затем сообщение «Действие отклонено; инструмент не вызывался». Для demo-approve показана такая же пауза, затем «Инструмент mark_demo вызван для DEMO-1». Оба assert прошли; это проверка кода и ветвления, а не успешной работы языковой модели или записи в CRM.
Как довести образец до пилота без обхода контроля?
Оставьте один инструмент с минимальными полномочиями и валидируйте на сервере тип и содержимое предложения модели до показа оператору; модельный текст и документы считайте недоверенным вводом. Определите, кто вправе возобновлять конкретный thread_id: одного знания ID недостаточно для авторизации. Отдельно фиксируйте заявленное действие, версию записи, идентификатор одобряющего, решение, момент исполнения и его результат. Перед критичным действием проверьте состояние целевой записи повторно; устаревшее одобрение не должно исполняться над изменившимися данными. Это проектные требования безопасности, а не свойства автоматически включённой демонстрации.
При resume узел с interrupt() перезапускается с начала, поэтому запись в БД, отправка письма или вызов API до паузы может повториться s1. Вынесите эффект в отдельный узел после решения, добавьте идемпотентный ключ и серверную проверку прав; даже действие после паузы надо защищать от повторного запуска после сбоя s1s3. Не оборачивайте interrupt() широким try/except: по документации это мешает механизму остановки s1. Для готового агента LangChain предлагает HumanInTheLoopMiddleware с вариантами approve, edit, reject, respond; последний — ответ человеку от имени инструмента, а не отказ от побочного действия s4. В учебном графе выше используется свой строгий boolean-контракт, не этот middleware.
Если заявка возникает при поиске по документам, сперва проверьте RAG-трассу и цитаты, затем разрешайте действия. Для реального пилота недостаточно двух демонстрационных строк: запустите разрешение и отказ на тестовом стенде, перезапустите процесс с постоянным checkpointer, затем сверяйте журнал, права и реальное состояние целевой записи. Основной следующий шаг: проверьте структурный вызов инструмента выбранной локальной моделью; без этого одобренное моделью действие остаётся только учебной фикстурой.
Источники
- LangGraph — Interrupts. Получено 29.09.2026.
- LangGraph — Persistence. Получено 29.09.2026.
- LangGraph — Graph API overview. Получено 29.09.2026.
- LangChain — Human-in-the-loop middleware. Получено 29.09.2026.
- LangChain — ChatOllama integration. Получено 29.09.2026.
- Ollama — Tool calling. Получено 29.09.2026.
- LangGraph — Checkpointers. Получено 29.09.2026.