Human-in-the-loop в LangChain | Курс LangChain Agents урок 8

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

Теория

Когда агенту нельзя доверять автономное выполнение

Агент решает сам, какие инструменты вызывать и с какими аргументами. В большинстве задач это удобно. Но есть класс действий, где ошибка необратима:

удаление записей из базы данных
отправка email или сообщений
публикация контента
перевод денег или изменение прав доступа

В таких случаях нужен человек, агент должен остановиться, показать что он собирается сделать, дождаться решения и только потом продолжить.

Как работает HumanInTheLoopMiddleware

Middleware сверяет каждый вызов инструмента с настроенной политикой. Если инструмент входит в список interrupt_on, выполнение прерывается через LangGraph interrupt. Граф сохраняет состояние через checkpointer, агент "засыпает".

Результат первого invoke содержит ключ __interrupt__ со списком ожидающих решений. Человек проверяет аргументы и принимает одно из трёх решений:

approve - выполнить как есть
edit - изменить аргументы и выполнить
reject - отклонить и передать агенту объяснение

Для возобновления нужно вызвать invoke ещё раз с тем же thread_id, передав решение через Command(resume=...).

Примеры кода

Базовая настройка

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

load_dotenv()


@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Отправить email."""
    return f"Email отправлен на {to}."


@tool
def read_data(query: str) -> str:
    """Прочитать данные из базы."""
    return f"Данные по запросу: {query}"


model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))

agent = create_agent(
    model=model,
    tools=[send_email, read_data],
    checkpointer=InMemorySaver(),  # обязателен для HITL
    system_prompt="Выполняй запросы через инструменты напрямую, без уточнений.",
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "send_email": True,   # прерывать, разрешить approve/edit/reject
                "read_data": False,   # не прерывать, выполнять автоматически
            },
        ),
    ],
)

config = {"configurable": {"thread_id": "hitl-demo"}}

# Запускаем, агент остановится перед send_email
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Отправь email на test@example.com, тема: Привет, текст: Добро пожаловать!"}]},
    config=config,
)

interrupts = result.get("__interrupt__", [])
print("Прерываний:", len(interrupts))
if interrupts:
    action = interrupts[0].value["action_requests"][0]
    print(f"Инструмент: {action['name']}")
    print(f"Аргументы: {action['args']}")

Прерывание сохраняется в result["__interrupt__"] как список объектов Interrupt. Каждый содержит .value с полями action_requests и review_configs.

Approve: разрешить выполнение

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

load_dotenv()


@tool
def delete_records(table: str, condition: str) -> str:
    """Удалить записи из таблицы."""
    return f"Удалены записи из {table} где {condition}."


model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))

agent = create_agent(
    model=model,
    tools=[delete_records],
    checkpointer=InMemorySaver(),
    system_prompt="Для удаления данных всегда используй инструмент delete_records.",
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "delete_records": {"allowed_decisions": ["approve", "reject"]},
            },
        ),
    ],
)

config = {"configurable": {"thread_id": "approve-demo"}}

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Удали записи из таблицы logs где created_at < '2025-01-01'."}]},
    config=config,
)

interrupts = result.get("__interrupt__", [])
if interrupts:
    action = interrupts[0].value["action_requests"][0]
    print(f"Ожидает решения: {action['name']}({action['args']})")

    # Подтверждаем выполнение
    final = agent.invoke(
        Command(resume={"decisions": [{"type": "approve"}]}),
        config=config,
    )
    print(final["messages"][-1].content)

Edit: изменить аргументы перед выполнением

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

load_dotenv()


@tool
def send_report(recipient: str, content: str) -> str:
    """Отправить отчёт получателю."""
    return f"Отчёт отправлен на {recipient}."


model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))

agent = create_agent(
    model=model,
    tools=[send_report],
    checkpointer=InMemorySaver(),
    system_prompt="Выполняй запросы через инструменты, без уточнений.",
    middleware=[
        HumanInTheLoopMiddleware(interrupt_on={"send_report": True}),
    ],
)

config = {"configurable": {"thread_id": "edit-demo"}}

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Отправь отчёт на boss@company.com, содержание: задача выполнена."}]},
    config=config,
)

interrupts = result.get("__interrupt__", [])
if interrupts:
    action = interrupts[0].value["action_requests"][0]
    print(f"Агент хочет: {action['name']}({action['args']})")

    # Меняем получателя перед отправкой
    final = agent.invoke(
        Command(resume={
            "decisions": [{
                "type": "edit",
                "edited_action": {
                    "name": "send_report",
                    "args": {
                        "recipient": "team@company.com",   # изменили
                        "content": action["args"]["content"],
                    },
                },
            }]
        }),
        config=config,
    )
    # Смотрим результат инструмента, а не ответ модели
    tool_result = next(m for m in reversed(final["messages"]) if hasattr(m, "tool_call_id"))
    print(tool_result.content)

Reject: отклонить и объяснить

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

load_dotenv()


@tool
def publish_post(title: str, content: str) -> str:
    """Опубликовать пост."""
    return f"Пост '{title}' опубликован."


model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))

agent = create_agent(
    model=model,
    tools=[publish_post],
    checkpointer=InMemorySaver(),
    system_prompt="Выполняй запросы через инструменты, без уточнений.",
    middleware=[
        HumanInTheLoopMiddleware(interrupt_on={"publish_post": True}),
    ],
)

config = {"configurable": {"thread_id": "reject-demo"}}

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Опубликуй пост, заголовок: LangChain Agents, содержание: руководство по агентам."}]},
    config=config,
)

from langchain_core.messages import ToolMessage

interrupts = result.get("__interrupt__", [])
if interrupts:
    action = interrupts[0].value["action_requests"][0]
    print(f"Агент хочет опубликовать: {action['args'].get('title')}")

    # Отклоняем, агент переработает пост и попробует снова
    after_reject = agent.invoke(
        Command(resume={
            "decisions": [{
                "type": "reject",
                "message": "Пост слишком короткий. Добавь примеры кода и перепубликуй.",
            }]
        }),
        config=config,
    )

    # Агент снова попадает в HITL, подтверждаем повторную публикацию
    interrupts2 = after_reject.get("__interrupt__", [])
    if interrupts2:
        action2 = interrupts2[0].value["action_requests"][0]
        print(f"Повторная попытка: {action2['args'].get('title')}")
        print(f"Новый контент: {action2['args'].get('content', '')[:80]}")

        final = agent.invoke(
            Command(resume={"decisions": [{"type": "approve"}]}),
            config=config,
        )
        tool_result = next(m for m in reversed(final["messages"]) if isinstance(m, ToolMessage))
        print(tool_result.content)

Сообщение из reject добавляется в историю как обратная связь. Агент видит объяснение и может переделать работу.

Сквозной проект: подтверждение перед отправкой отчёта

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

load_dotenv()


@tool
def search_web(query: str) -> str:
    """Поиск информации по теме."""
    return (
        f"Результаты поиска по '{query}': компания основана в 2021, "
        f"разрабатывает AI-системы, известна моделью Claude."
    )


@tool
def send_report(recipient: str, summary: str) -> str:
    """Отправить финальный отчёт получателю."""
    return f"Отчёт отправлен на {recipient}."


model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))

agent = create_agent(
    model=model,
    tools=[search_web, send_report],
    checkpointer=InMemorySaver(),
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "search_web": False,    # выполнять автоматически
                "send_report": True,    # прерывать для подтверждения
            },
            description_prefix="Требуется подтверждение",
        ),
    ],
    system_prompt=(
        "Ты агент-аналитик. Исследуй тему через search_web, "
        "затем отправь краткий отчёт через send_report на manager@company.com."
    ),
)

config = {"configurable": {"thread_id": "report-1"}}

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Исследуй Anthropic и отправь отчёт."}]},
    config=config,
)

interrupts = result.get("__interrupt__", [])
if interrupts:
    action = interrupts[0].value["action_requests"][0]
    args = action["args"]
    print("--- Ожидает подтверждения ---")
    print(f"Получатель: {args.get('recipient')}")
    print(f"Резюме: {str(args.get('summary', ''))[:100]}")

    # В реальном приложении здесь input() от пользователя
    final = agent.invoke(
        Command(resume={"decisions": [{"type": "approve"}]}),
        config=config,
    )
    print(final["messages"][-1].content)
else:
    print(result["messages"][-1].content)

Частые ошибки

Нет checkpointer? состояние не сохраняется между вызовами

# Неправильно: без checkpointer нельзя возобновить выполнение
agent = create_agent(model=model, tools=[...], middleware=[HumanInTheLoopMiddleware(...)])

# Правильно
agent = create_agent(
    model=model,
    tools=[...],
    checkpointer=InMemorySaver(),
    middleware=[HumanInTheLoopMiddleware(...)],
)

Разный thread_id при возобновлении

# Неправильно: другой thread_id = другой тред, агент начнёт заново
agent.invoke({"messages": [...]}, config={"configurable": {"thread_id": "t1"}})
agent.invoke(Command(resume={...}), config={"configurable": {"thread_id": "t2"}})

# Правильно: тот же config при resume
config = {"configurable": {"thread_id": "t1"}}
agent.invoke({"messages": [...]}, config=config)
agent.invoke(Command(resume={...}), config=config)

Обращение к аргументам через arguments вместо args

# Неправильно
action["arguments"]["recipient"]

# Правильно
action["args"]["recipient"]

Задание

Создайте агента с двумя инструментами: search_web не требует подтверждения и write_file(filename, content) требует подтверждения, разрешены только approve и reject.

Запросите у агента: "Найди информацию об OpenAI и сохрани в файл report.txt". Проверьте что прерывание происходит только перед write_file. Попробуйте сначала reject с объяснением, затем запустите снова и сделайте approve.

<< урок 7

урок 9 >>

Подписывайтесь на мой Telegram канал

Если вам нужен ментор и вы хотите научиться разрабатывать AI агентов, пишите, обсудим условия