ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Спроектировать escalation system (AI → Human при low confidence)

1. Цель задачи

Разработать и внедрить систему эскалации, которая передаёт сессию от AI-ассистента человеку-оператору, когда уверенность модели в ответе падает ниже порога 0.5. Ключевая метрика — доля эскалированных сессий не должна превышать 10% от общего числа обращений, что означает, что AI должен корректно обрабатывать 90% запросов без вмешательства человека. Вы научитесь проектировать надёжную escalation pipeline, интегрировать её в существующий AI-сервис и измерять её эффективность.

Ключевой результат Работающая escalation system с конфигурируемым порогом confidence, которая в production условиях эскалирует не более 10% сессий и гарантированно передаёт данные оператору при low confidence (<0.5).

2. Исходные данные

Перед началом необходимо иметь:

Если нет реального AI-ассистента — симулируем:

1. Создать простой RAG-чат на LangChain с эмбеддингами (sentence-transformers) и LLM (GPT-4o-mini / llama.cpp).
2. Добавить функцию estimate_confidence(response, context) на основе softmax логитов или правила: если длина ответа < 50 символов → confidence = 0.1; если есть маркеры неуверенности (“наверное”, “возможно”) → confidence -= 0.2.
3. Настроить генерацию synthetic-запросов с перемешанным confidence (например, 80% простые, 20% сложные).
4. Экспортировать логи в CSV (колонки: session_id, request, response, confidence, timestamp).

3. Технологический стек

4. Этапы выполнения

Этап 1: Разработка механизма оценки уверенности (2 часа)

Действия

1. Определить формулу confidence

   • Использовать смешанную метрику: confidence = w1 * logit_softmax + w2 * nli_score + w3 * heuristic_penalty.
   • Прочитать логи существующего AI-ассистента, выделить индикаторы неуверенности (например, шаблоны “I'm not sure”, “possibly”).
   • Задать веса: w1=0.5, w2=0.3, w3=0.2.

2. Реализовать функцию в коде Python

   import numpy as np
   
   def estimate_confidence(response: str, logits: np.ndarray, nli_prob: float) -> float:
       # 1. Softmax от максимального логита
       soft = np.exp(logits) / np.sum(np.exp(logits))
       confidence_llm = np.max(soft)
       
       # 2. Confidence от NLI (например, entailment вероятность)
       confidence_nli = nli_prob
       
       # 3. Heuristic penalty за маркеры неуверенности
       uncertainty_keywords = ['maybe', 'perhaps', 'i think', 'i guess', 'наверное']
       penalty = sum(kw in response.lower() for kw in uncertainty_keywords) * 0.1
       confidence_heuristic = max(0, 1.0 - penalty)
       
       # 4. Взвешенная сумма
       w1, w2, w3 = 0.5, 0.3, 0.2
       final = w1 * confidence_llm + w2 * confidence_nli + w3 * confidence_heuristic
       return round(final, 2)
   

3. Протестировать на 20 ручных примерах — проверить, что при заведомо сложных вопросах confidence падает ниже 0.5.

Ожидаемый результат этапа Функция estimate_confidence, корректно возвращающая значения в диапазоне [0,1], и тесты, показывающие, что threshold 0.5 действительно соответствует границе «требуется человек».

Этап 2: Интеграция escalation logic в AI-сервис (2 часа)

Действия

1. Добавить middleware в FastAPI-приложение (или расширить существующий endpoint /chat):

   @router.post("/chat")
   async def chat(request: ChatRequest):
       # Получить ответ и confidence
       response, logits, nli = await llm.generate_with_confidence(request.query)
       conf = estimate_confidence(response, logits, nli)
       
       if conf < 0.5:
           # Эскалация
           await escalate_to_human(session_id=request.session_id,
                                   query=request.query,
                                   response=response,
                                   confidence=conf)
           return {"status": "escalated", "message": "Запрос передан оператору."}
       else:
           return {"status": "ok", "response": response, "confidence": conf}
   

2. Реализовать функцию escalate_to_human — она должна:

   • Сериализовать данные (session_id, query, response, confidence, timestamp) в JSON.
   • Отправить сообщение в Redis Stream с ключом escalation_queue.
   • Записать событие в лог Prometheus (инкремент счётчика escalations_total).

3. Настроить Redis (локально, docker run -p 6379:6379 redis) или использовать Kafka (если опыт есть).

Ожидаемый результат этапа При каждом запросе с confidence < 0.5 сессия попадает в очередь эскалации, а пользователь получает уведомление о передаче оператору.

Этап 3: Интерфейс оператора (1.5 часа)

Действия

1. Создать Streamlit-приложение (файл operator_ui.py):

   import streamlit as st
   import redis
   import json
   
   r = redis.Redis(host='localhost', port=6379, decode_responses=True)
   
   st.title("Escalation Queue")
   
   if st.button("Get next escalation"):
       msg = r.xreadgroup("escalation_group", "operator1", {"escalation_queue": ">"}, count=1)
       if msg:
           session = json.loads(msg[0][1][0][1])
           st.json(session)
           st.text_area("Your response", key="operator_answer")
           if st.button("Submit reply"):
               # Сохранить ответ оператора в БД
               pass
       else:
           st.info("Queue empty")
   

2. Реализовать подтверждение приёма (acknowledge) по XACK, чтобы сессия не застревала в очереди.

3. Добавить дашборд с метриками (если используется Grafana) или прямо в Streamlit: процент эскалированных сессий за последние 15 минут.

Ожидаемый результат этапа Оператор может видеть очередь эскалированных сессий, просматривать запрос и ответ AI, а затем отвечать человеку.

Этап 4: Настройка мониторинга и измерение доли эскалаций (1 час)

Действия

1. Установить Prometheus client и экспортировать метрики из FastAPI:

   from prometheus_client import Counter, Histogram, generate_latest
   escalations = Counter('escalations_total', 'Total escalations', ['pipeline'])
   requests_total = Counter('requests_total', 'Total chat requests')
   

2. Запустить нагрузочное тестирование с помощью locust (100 виртуальных пользователей, 1000 запросов) с синтетическими данными, где ровно 15% запросов сложные (confidence < 0.5).

3. Собрать метрики и рассчитать долю эскалированных сессий:

   escalations_total / requests_total * 100%
   

4. Подобрать порог и/или веса так, чтобы доля опустилась до 10% (можно увеличить уверенность за счёт лучшего NLI или изменить heuristic penalty).

Ожидаемый результат этапа Доля эскалаций ≤10% при тестовом наборе; настроен дашборд (Grafana или Streamlit), показывающий ключевую метрику в реальном времени.

Этап 5: Документирование и финальная проверка (1 час)

Действия

1. Записать README c архитектурой (диаграмма очередности: User → AI → Confidence check → Human queue).
2. Сделать скриншоты работы операторского интерфейса.
3. Проверить все пункты Definition of Done.

Ожидаемый результат этапа Пакет артефактов: код (FastAPI, Streamlit, метрики), конфиги, результаты тестов, README.

5. Критерии приемки (Definition of Done)

• Реализована функция оценки confidence с порогом по умолчанию 0.5.
• AI-сервис эскалирует сессию при confidence < 0.5, отдавая пользователю сообщение «Запрос передан оператору».
• Данные эскалированных сессий сохраняются в очереди (Redis/Kafka) и доступны оператору через UI.
• Оператор может просмотреть запрос и ответ AI, а затем отправить свой ответ пользователю (хотя бы в симуляции).
• Метрики (число запросов, эскалаций) экспортируются в Prometheus.
• При нагрузочном тестировании с 15% сложных запросов доля эскалаций ≤10% (порог скорректирован экспериментально).
• Код покрыт unit-тестами на логику подсчёта confidence (минимум 5 тестовых кейсов).
• Написан README с описанием архитектуры и инструкцией по запуску.

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

Основной артефакт — репозиторий со следующим содержимым:

• src/ai_assistant.py — код AI-ассистента с функцией estimate_confidence.
• src/escalation.py — логика очереди (Redis Streams).
• src/main.py — FastAPI приложение с endpoint /chat.
• src/operator_ui.py — Streamlit-интерфейс оператора.
• tests/test_confidence.py — юнит-тесты.
• load_test/locustfile.py — скрипт для нагрузочного тестирования.
• prometheus.yml — конфигурация для сбора метрик.
• README.md — документация.

Дополнительно: скриншоты Grafana/Streamlit дашборда с метрикой доли эскалации.

7. Возможные сложности и их решение

8. Бюджет времени (оценка)

Примечание: При первом выполнении задачи заложите +20% времени на отладку (до 9 часов).

9. Связанные вопросы из базы знаний

10. Чек-лист самопроверки

• Я явно задал порог confidence (0.5) и проверил, что он конфигурируется через переменные окружения.
• Я протестировал логику эскалации на 10+ сессиях и убедился, что только действительно сложные запросы эскалируются.
• Я запустил нагрузочный тест и получил долю эскалаций ≤10% (если нет, скорректировал порог).
• Я написал юнит-тесты для функции оценки confidence, включая пограничные случаи.
• Я задокументировал архитектуру и шаги запуска в README, чтобы другой разработчик мог воспроизвести систему за 15 минут.