Как вы обеспечиваете «человека в петле» (HITL) для критических действий агента?

Краткий тезис

Human-in-the-Loop (HITL) — это механизм, при котором перед выполнением опасного или необратимого действия AI-агент запрашивает подтверждение у человека. Это критически важно для безопасности, соблюдения политик и доверия пользователей. Реализация включает генерацию confirmation_prompt с контекстом действия, интерфейс для approve/deny, таймауты и логирование. Без HITL агент может случайно удалить данные, отправить письмо не тому адресату или совершить финансовую операцию.

1. Термин: Human-in-the-Loop (HITL)

Human-in-the-Loop — подход, при котором человек участвует в цикле принятия решений AI-системы. В контексте агентов HITL означает, что перед выполнением критического действия (action|dangerous action) агент приостанавливается и ждёт одобрения оператора.

Зачем нужен HITL

• Безопасность: предотвращение необратимых ошибок (удаление данных, отправка писем).
• Соответствие политикам: соблюдение корпоративных или юридических правил.
• Доверие: пользователь контролирует агента, а не наоборот.
• Отладка: возможность перехватить неправильное решение агента на раннем этапе.

Термин «action|Dangerous action» — действие, которое может привести к негативным последствиям: потеря данных, нарушение приватности, финансовый ущерб. Список таких действий определяется заранее и настраивается под сценарий.

2. Какие действия считаются критическими

Не все действия агента требуют HITL. Обычно выделяют категории:

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

3. Базовая архитектура HITL

Процесс HITL состоит из шагов:

1. Агент планирует действие — LLM решает, что нужно сделать (например, удалить файл).
2. Детектор опасных действий — проверяет, входит ли действие в список критических.
3. Генерация confirmation_prompt — формируется сообщение для пользователя: что будет сделано, с какими параметрами, какие риски.
4. Отправка запроса пользователю — через интерфейс (чат, веб-форма, API).
5. Ожидание ответа — агент блокируется до получения approve/deny (или таймаута).
6. Выполнение или отмена — если approve, действие выполняется; если deny или таймаут — отменяется с логированием.

Схема (текстовая):

Пользователь -> Запрос -> Агент (LLM) -> План действий
    |
    v
Детектор опасных действий
    |
    ├─ Безопасное действие -> Выполнить сразу
    └─ Опасное действие -> Создать confirmation_prompt
                              |
                              v
                         Интерфейс HITL -> Пользователь (approve/deny)
                              |
                              v
                         Выполнить или отменить

4. Confirmation prompt: структура и пример

Confirmation prompt — это сообщение, которое агент показывает пользователю. Оно должно быть информативным и однозначным.

Обязательные поля

• Действие (action): что именно будет выполнено (например, "Удалить файл report.pdf").
• Контекст (context): почему агент решил это сделать (например, "Пользователь попросил очистить папку Downloads").
• Риски (risks): последствия (например, "Файл будет безвозвратно удалён").
• Альтернативы (alternatives): что можно сделать вместо этого (например, "Переместить в корзину").
• Идентификатор запроса (request_id): для логирования и отслеживания.

Пример на Python

def build_confirmation_prompt(action: dict) -> str:
    return f"""
    ⚠️ Требуется подтверждение

    Действие: {action['name']}
    Параметры: {action['params']}
    **Контекст**: {action['context']}
    Риски: {action['risks']}

    Введите "approve" для выполнения или "deny" для отмены.
    """

5. Интерфейс для approve/deny

Способы взаимодействия с пользователем:

Рекомендация для агентов, работающих в реальном времени, лучше использовать чат или веб-форму с кнопками. Для асинхронных сценариев — email или push-уведомления.

Пример веб-интерфейса (FastAPI + Jinja2):

@app.post("/confirm")
async def confirm_action(request_id: str, decision: str):
    if decision == "approve":
        execute_action(request_id)
        return {"status": "executed"}
    elif decision == "deny":
        log_denial(request_id)
        return {"status": "cancelled"}

6. Таймауты и обработка отсутствия ответа

Пользователь может не ответить. Нужна стратегия:

• Таймаут (например, 5 минут) — если ответа нет, действие автоматически отклоняется (deny by default).
• Повторный запрос — через некоторое время отправить напоминание.
• Эскалация — если действие критично, уведомить другого ответственного.

Пример реализации таймаута

import asyncio

async def wait_for_approval(request_id: str, timeout: int = 300):
    try:
        decision = await asyncio.wait_for(
            get_user_decision(request_id), timeout=timeout
        )
        return decision
    except asyncio.TimeoutError:
        log_timeout(request_id)
        return "deny"  # deny by default

Почему deny by default Безопаснее отменить действие, чем выполнить его без подтверждения.

7. Логирование и аудит

Все HITL-события должны логироваться для:

• Аудита — кто и когда одобрил/отклонил действие.
• Отладки — почему агент предложил опасное действие.
• Улучшения — анализ частоты отказов, чтобы уменьшить количество ложных срабатываний.

Структура лога

{
  "timestamp": "2025-03-15T10:30:00Z",
  "request_id": "abc123",
  "user_id": "user_42",
  "action": "delete_file",
  "params": {"path": "/data/report.pdf"},
  "decision": "approve",
  "response_time_ms": 4500,
  "context": "Пользователь попросил очистить папку"
}

Инструменты ELK, Prometheus + Grafana, простое логирование в файл.

8. Многоуровневая HITL

Для разных уровней риска можно требовать разное количество подтверждений:

Dual control — механизм, при котором для выполнения действия нужно одобрение двух разных операторов. Это стандарт в финансах и безопасности.

9. Интеграция с политиками безопасности

HITL должен учитывать роли пользователей:

• Администратор может approve любые действия.
• Оператор — только действия в своей зоне ответственности.
• Наблюдатель — только просмотр, без права approve.

Пример ролевой модели

def can_approve(user_role: str, action: str) -> bool:
    if action in ["delete_db", "stop_server"]:
        return user_role == "admin"
    elif action in ["send_email", "create_user"]:
        return user_role in ["admin", "operator"]
    return True

10. Проблемы и пути решения

Лучшая практика дать пользователю возможность настроить уровень HITL (например, "всегда подтверждать", "только для действий с риском > 0.8", "отключить для доверенных действий").

11. Пример полной реализации на Python

import asyncio
import logging
from dataclasses import dataclass
from enum import Enum

class ActionRisk(Enum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3

@dataclass
class Action:
    name: str
    params: dict
    context: str
    risk: ActionRisk

class HITLManager:
    def __init__(self, timeout: int = 300):
        self.timeout = timeout
        self.pending = {}
        self.logger = logging.getLogger("hitl")

    async def request_confirmation(self, action: Action, user_id: str) -> bool:
        request_id = f"req_{id(action)}"
        prompt = self._build_prompt(action)
        self.pending[request_id] = {"action": action, "user_id": user_id}
        self.logger.info(f"Request {request_id}: {prompt}")

        # Отправка пользователю (заглушка)
        decision = await self._wait_for_decision(request_id)
        self.logger.info(f"Request {request_id}: decision={decision}")
        return decision == "approve"

    def _build_prompt(self, action: Action) -> str:
        return (
            f"⚠️ Подтвердите действие: {action.name}\n"
            f"Параметры: {action.params}\n"
            f"Контекст: {action.context}\n"
            f"Риск: {action.risk.name}\n"
            f"Введите 'approve' или 'deny'."
        )

    async def _wait_for_decision(self, request_id: str) -> str:
        # В реальной системе здесь ожидание ответа от пользователя
        try:
            decision = await asyncio.wait_for(
                self._get_user_input(request_id), timeout=self.timeout
            )
            return decision
        except asyncio.TimeoutError:
            self.logger.warning(f"Timeout for {request_id}")
            return "deny"

    async def _get_user_input(self, request_id: str) -> str:
        # Заглушка: имитация ввода пользователя
        await asyncio.sleep(1)
        return "approve"

Пет-проект для закрепления

Задача Создать агента для управления файлами (удаление, переименование, перемещение) с HITL. Агент принимает текстовые команды, но перед удалением файла запрашивает подтверждение.

Инструменты

• Python (asyncio, pathlib)
• FastAPI (веб-интерфейс)
• Streamlit (простой UI для демонстрации)
• SQLite (логирование)

Шаги:

1. Определить список опасных действий: delete_file, move_file (если перезаписывает), rename_file (если меняет расширение).
2. Реализовать агента, который парсит команду и вызывает HITLManager.
3. Создать FastAPI-эндпоинт для получения решения пользователя (approve/deny).
4. Добавить таймаут 30 секунд, после которого действие отклоняется.
5. Логировать все запросы в SQLite.
6. Сделать Streamlit-интерфейс: поле ввода команды, список ожидающих подтверждений, кнопки Approve/Deny.

Ожидаемый результат

• Пользователь пишет "удали файл test.txt".
• Агент не удаляет сразу, а показывает в интерфейсе: "Подтвердите удаление test.txt".
• Пользователь нажимает Approve → файл удаляется, лог записывается.
• Если пользователь не отвечает 30 секунд → действие отклоняется.

Связь с другими вопросами

Навигация

• Предыдущий: 145
• Следующий: 147
• Индекс: 00. Индекс разборов