Как вы делаете агента, который может «просить помощи» у другого агента или человека?

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

Агент, способный запрашивать помощь, строится на механизме confidence threshold (порога уверенности) и escalation policy (политики эскалации). Когда внутренняя оценка уверенности]] агента в своём ответе падает ниже заданного порога, он не генерирует финальный ответ, а инициирует запрос к более компетентному агенту или человеку-оператору. Ключевые компоненты: детектор неуверенности, маршрутизатор запросов и система очередей с приоритетами для эскалации.

1. Термин: Confidence Threshold (Порог уверенности)

Confidence threshold — это числовое значение (обычно от 0.0 до 1.0), которое агент использует для принятия решения: отвечать самостоятельно или запрашивать помощь.

Как это работает:

• Агент генерирует ответ и параллельно вычисляет score уверенности (confidence score)
• Если score >= threshold → агент выдаёт ответ пользователю
• Если score < threshold → запускается процесс эскалации

Откуда берётся confidence score

• Logit-based: из softmax-вероятностей последнего токена LLM (например, max probability среди токенов)
• Verbalized confidence: LLM просят явно оценить уверенность в ответе (например, «Оцени свою уверенность от 0 до 1»)
• Ensemble methods: несколько запусков одной модели с разными seed → дисперсия ответов как мера неуверенности
• Separate classifier: отдельная модель (часто маленький BERT), обученная предсказывать, правилен ли ответ агента

Пример кода: логит-основанная уверенность

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

model = AutoModelForCausalLM.from_pretrained("microsoft/Phi-3-mini-4k-instruct")
tokenizer = AutoTokenizer.from_pretrained("microsoft/Phi-3-mini-4k-instruct")

def get_confidence(prompt: str) -> float:
    inputs = tokenizer(prompt, return_tensors="pt")
    with torch.no_grad():
        outputs = model(**inputs)
    # Берём логиты последнего токена
    logits = outputs.logits[0, -1, :]
    probs = torch.softmax(logits, dim=-1)
    # Уверенность = максимальная вероятность среди всех токенов
    confidence = torch.max(probs).item()
    return confidence

2. Термин: Escalation Policy (Политика эскалации)

Escalation policy — это набор правил, определяющих, кому и при каких условиях передаётся запрос. Обычно реализуется как иерархия уровней:

Типы эскалации

• Vertical escalation: передача выше по иерархии (агент → супервайзер → человек)
• Horizontal escalation: передача агенту того же уровня, но с другой специализацией (например, от общего агента к агенту по финансам)

Важно: политика должна включать timeout — если агент не ответил за N секунд, запрос автоматически эскалируется на следующий уровень.

3. Архитектура: Supervisor Agent (Агент-супервайзер)

Ключевой паттерн — supervisor agent, который управляет роутингом запросов между специализированными агентами и человеком.

Компоненты

1. Router: определяет, к какому агенту направить запрос на основе тематики
2. Confidence evaluator: вычисляет уверенность ответа
3. Escalation manager: реализует политику эскалации
4. Queue system: управляет приоритетами запросов (например, Redis Queue)

Схема работы

Пользователь → Supervisor Agent
    ↓
Confidence Evaluator
    ↓
[score >= threshold] → Ответ пользователю
[score < threshold] → Router → Специализированный агент
    ↓
[специалист ответил] → Проверка уверенности → [score >= threshold] → Ответ
    ↓
[score < threshold] → Escalation Manager → Человек-оператор

4. Реализация: Handoff Protocol (Протокол передачи)

Handoff protocol — это механизм передачи контекста между агентами или агентом и человеком. Должен включать:

1. Полный контекст диалога: история сообщений, результаты предыдущих попыток
2. Причина эскалации: почему агент не смог ответить (низкая уверенность, нехватка данных, противоречие в документах)
3. Предложенный черновик ответа: даже если агент неуверен, он может дать набросок
4. Метаданные: версия модели, использованные инструменты, время выполнения

Пример структуры handoff

@dataclass
class HandoffPackage:
    user_query: str
    conversation_history: list[dict]
    draft_response: str | None
    confidence_score: float
    escalation_reason: str
    attempted_agents: list[str]
    metadata: dict

5. Эскалация человеку: Human-in-the-Loop (HITL)

Когда агент не справляется, запрос передаётся человеку-оператору. Ключевые аспекты:

Интерфейс оператора

• Дашборд с очередью запросов (приоритет по уровню эскалации)
• Полный контекст: история, попытки агентов, черновик ответа
• Инструменты для быстрого ответа: шаблоны, поиск по базе знаний
• Кнопка «вернуть агенту» после исправления ошибки

Feedback loop

• Человек исправляет ответ → этот исправленный ответ используется для fine-tuning агента
• Метрика: escalation rate (доля запросов, ушедших к человеку) — должна снижаться со временем

Когда эскалировать человеку обязательно

• Юридические/комплаенс-вопросы (например, «Могу ли я уволить сотрудника без предупреждения?»)
• Запросы с персональными данными (PII)
• Ситуации, где неверный ответ может нанести вред (медицина, финансы)

6. Метрики для оценки системы

Confidence calibration — важнейшая метрика. Если агент говорит «уверен на 0.9», но ошибается в 30% случаев — калибровка плохая. Исправляется с помощью temperature scaling или Platt scaling.

7. Проблемы и компромиссы

8. Пример реализации: Supervisor Agent с эскалацией

import asyncio
from dataclasses import dataclass
from enum import Enum

class AgentLevel(Enum):
    GENERAL = 1
    SPECIALIST = 2
    HUMAN = 3

@dataclass
class AgentResponse:
    text: str
    confidence: float
    level: AgentLevel

class SupervisorAgent:
    def __init__(self, threshold: float = 0.7):
        self.threshold = threshold
        self.specialists = {
            "finance": FinanceAgent(),
            "legal": LegalAgent(),
            "tech": TechSupportAgent()
        }
        self.human_queue = HumanQueue()
    
    async def handle_query(self, query: str) -> str:
        # Шаг 1: пробует генералист
        response = await self.general_agent.respond(query)
        if response.confidence >= self.threshold:
            return response.text
        
        # Шаг 2: определяет тему и направляет специалисту
        topic = await self.classify_topic(query)
        specialist = self.specialists.get(topic)
        if specialist:
            response = await specialist.respond(query)
            if response.confidence >= self.threshold:
                return response.text
        
        # Шаг 3: эскалация человеку
        handoff = HandoffPackage(
            user_query=query,
            conversation_history=[],
            draft_response=response.text,
            confidence_score=response.confidence,
            escalation_reason="Low confidence after specialist attempt",
            attempted_agents=["general", topic],
            metadata={}
        )
        return await self.human_queue.enqueue(handoff)

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

Задача: Создать агента поддержки, который эскалирует сложные запросы человеку.

Инструменты: Python, LangChain, FastAPI, Redis Queue, Streamlit (дашборд оператора)

Шаги:

1. Реализуй базового агента на LangChain с tool calling (поиск по документации)
2. Добавь confidence evaluator на основе logit-вероятностей
3. Настрой порог 0.7: при уверенности ниже — запрос в Redis Queue
4. Создай Streamlit-дашборд для оператора: очередь запросов, контекст, кнопка ответа
5. Добавь feedback loop: исправленные ответы сохраняются в базу для fine-tuning

Ожидаемый результат: Работающая система, где 70% запросов решаются агентом, 20% — специализированным агентом, 10% — человеком. Дашборд показывает escalation rate и среднее время ответа.

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

Навигация

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