LangChain Human-in-the-Loop

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

Human-in-the-Loop (HITL) позволяет агенту приостанавливаться в критических моментах, ждать одобрения человека, а затем продолжить.

interrupt() — приостановка выполнения в инструментах

interrupt()Эта функция позволяет инструменту приостановиться на середине выполнения, дождаться внешнего ввода, а затем продолжить:

Пример

from langgraph.types import interrupt

# Пауза в инструменте с помощью interrupt()
def send_email(to: str, subject: str, body: str) -> str:
    """Отправить электронное письмо (требуется одобрение человека)"""
    # Приостановить выполнение и отправить запрос на одобрение наружу
    approval = interrupt({
        "action": "send_email",
        "to": to,
        "subject": subject,
        "body": body,
        "message": "Пожалуйста, подтвердите, отправлять ли это письмо?"
    })

    # Дождитесь внешнего одобрения, чтобы продолжить
    if approval.get("approved"):
        return f"Письмо отправлено на {to}"
    else:
        return f"Отправка письма отклонена: {approval.get('reason', 'Пользователь отменил')}"

Как работает interrupt():

  1. Инструмент вызывает interrupt(), Агент приостанавливает выполнение
  2. Внешняя система получает информацию о прерывании и отображает её пользователю
  3. После того как пользователь принимает решение, выполнение возобновляется через Command(resume=...)
  4. interrupt() возвращает значение, переданное пользователем, и инструмент продолжает выполнение

Полный пример — процесс одобрения

Пример

from dotenv import load_dotenv
load_dotenv()

from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage


@tool
def delete_course(course_name: str) -> str:
    """Удалить курс (требуется одобрение).

    Аргументы:
course_name: имя курса для удаления
    """
    # Приостановить и ждать одобрения
    approval = interrupt({
        "action": "delete_course",
        "course": course_name,
        "message": f"Подтвердите удаление курса '{course_name}'? Это действие нельзя отменить."
    })

    if approval.get("confirmed"):
        return f"Курс '{course_name}' удалён"
    else:
        return f"Удаление отменено"


checkpointer = InMemorySaver()
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    tools=[delete_course],
    checkpointer=checkpointer,
    system_prompt="Вы помощник администратора учебного руководства EXAMPLE.",
)

config = {"configurable": {"thread_id": "admin-001"}}

# Шаг 1: Инициировать запрос на удаление (вызовет прерывание)
print("=== Начало выполнения ===")
result = agent.invoke(
    {"messages": [HumanMessage(content="Пожалуйста, удалите курс 'Устаревшее руководство по Java'.")]},
    config=config,
)

# Проверить, приостановлен ли агент
state = agent.get_state(config)
print(f"Статус: {state.next}")  # ('tools',) означает паузу на узле tools
print(f"Информация о прерывании: {state.tasks)

# Шаг 2: Подтверждение человеком (имитировать нажатие пользователем "Подтвердить")
print("\n=== Подтверждение человеком ===)
resume_value = {"confirmed": True, "operator": "Администратор Чжан Сань"}
result = agent.invoke(
    Команда(resume=resume_value),
    config=config,
)
print(f"Окончательный ответ: {result['messages'][-1].content}")

Вывод:

=== 开始执行 ===
Status: ('tools',)
中断信息: (Interrupt(value={'action': 'delete_course', ...}),)

=== 人工审批 ===
Final reply: The course "Outdated Java Tutorial" has been deleted.

Параметры interrupt_before / interrupt_after

Помимо использования interrupt() в инструментах, вы также можете задать глобальные точки прерывания в create_agent():

Пример

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[some_tool],
    checkpointer=checkpointer,

    # Пауза перед узлами инструментов (требуется подтверждение перед каждым вызовом инструмента)
    interrupt_before=["tools"],

    # Пауза после узлов модели (можно проверить после каждого ответа модели)
    # interrupt_after=["model"],
)
ПараметрМомент паузыПрименимый сценарий
interrupt_before=["tools"]Перед каждым выполнением инструментаВсе вызовы инструментов требуют одобрения
interrupt_before=["model"]Перед каждым вызовом моделиВручную проверять сообщения перед обработкой моделью
interrupt_after=["model"]После каждого ответа моделиПроверить вывод модели перед решением, продолжать ли
interrupt_after=["tools"]После каждого выполнения инструментаПроверить результаты инструмента перед выбором следующего шага

Типичная архитектура HITL

В реальных веб-приложениях HITL обычно реализуется так:

Пример

# Бэкенд: получить сообщение пользователя, обработать до точки прерывания, вернуть информацию о прерывании
def handle_user_message(thread_id: str, message: str):
    config = {"configurable": {"thread_id": thread_id}}
    result = agent.invoke(
        {"messages": [HumanMessage(content=message)]},
        config=config,
    )
    state = agent.get_state(config)

    # Проверить, ожидается ли одобрение
    if state.tasks and state.tasks[0].interrupts:
        return {
            "status": "pending_approval",
            "interrupt": state.tasks[0].interrupts[0].value,
            "thread_id": thread_id,
        }

    return {
        "status": "completed",
        "reply": result["messages"][-1].content,
    }


# Бэкенд: обработать одобрение пользователя
def handle_approval(thread_id: str, approved: bool, reason: str = ""):
    config = {"configurable": {"thread_id": thread_id}}
    result = agent.invoke(
        Command(resume={"confirmed": approved, "reason": reason}),
        config=config,
    )
    return {"status": "completed", "reply": result["messages"][-1].content}

HITL требует Checkpointer. Потому что, когда Agent приостанавливается на interrupt(), его состояние должно быть сохранено, чтобы корректно возобновить выполнение.

Другие расширения