Автономная AI-система обработки заявок под ключ

Разрабатываем автономные AI-системы обработки запросов. Это AI-оркестратор, который принимает входящие запросы из различных каналов: email, форм, API, мессенджеров. Система классифицирует их, извлекает данные, исполняет логику обработки и возвращает ответ. Или создаёт задачи в бизнес-системах — всё без участия оператора для типовых случаев.

В отличие от простого чат-бота или агента с одним инструментом, наша система включает полный цикл: приём → понимание → обогащение данных → исполнение → уведомление → мониторинг. Мы реализовали десятки таких проектов, и в этой статье разберём архитектуру на реальном примере.

Как работает классификация запросов?

Входящий запрос проходит через граф состояний LangGraph. Первый узел — классификатор на GPT-4o с Pydantic-моделью. Он определяет тип запроса (техподдержка, биллинг, новый заказ, статус, жалоба, возврат), срочность, уверенность и необходимость эскалации человеку. Если уверенность ниже 0.6 или запрос содержит триггеры (юридические угрозы, возвраты >$450–650., упоминание ущерба) — запрос передаётся оператору. Это наш опыт, подтверждённый сотнями проектов.

Архитектура системы

Входные каналы

• webhook (email-парсер)
• REST API
• Telegram/WhatsApp Bot
• web-форма

Ядро обработки

LangGraph-граф с состоянием, классификатор, исполнители, агрегатор.

Выходные каналы

• REST API внешних систем (CRM, ERP, Service Desk)
• email/push уведомления
• очередь задач (Celery/Redis)

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
from typing import TypedDict, Annotated, Optional
from datetime import datetime
import operator

class RequestState(TypedDict):
    # Входящий запрос
    raw_content: str
    channel: str  # "email", "api", "telegram", "form"
    sender_id: str
    received_at: datetime

    # Классификация
    request_type: Optional[str]  # "support", "order", "complaint", "inquiry", "refund"
    urgency: Optional[str]  # "critical", "high", "normal", "low"
    confidence: Optional[float]

    # Обогащение
    user_profile: Optional[dict]
    related_entities: Optional[list]  # Связанные заказы, договоры, тикеты

    # Обработка
    action_plan: Optional[list[dict]]
    executed_actions: Annotated[list, operator.add]
    requires_human: bool
    human_reason: Optional[str]

    # Результат
    response_draft: Optional[str]
    outcome: Optional[str]
    processing_time_ms: Optional[int]

from langchain_openai import ChatOpenAI
from pydantic import BaseModel
from typing import Literal

class RequestClassification(BaseModel):
    request_type: Literal["support_technical", "support_billing", "order_new", "order_status", "complaint", "refund_request", "general_inquiry"]
    urgency: Literal["critical", "high", "normal", "low"]
    confidence: float
    extracted_entities: dict
    requires_human: bool
    human_reason: Optional[str] = None
    summary: str

llm = ChatOpenAI(model="gpt-4o", temperature=0)

def classify_request(state: RequestState) -> RequestState:
    result = llm.with_structured_output(RequestClassification).invoke(
        f"""Классифицируй входящий запрос.
Канал: {state['channel']}
Запрос: {state['raw_content']}
Передай человеку если:
- Юридические угрозы или упоминание судебных разбирательств
- Запрос на возврат суммы > $450–650
- Упоминание о физическом ущербе
- Эмоционально заряженный отзыв с публичными угрозами"""
    )
    return {
        **state,
        "request_type": result.request_type,
        "urgency": result.urgency,
        "confidence": result.confidence,
        "requires_human": result.requires_human,
        "human_reason": result.human_reason,
    }

def plan_actions(state: RequestState) -> RequestState:
    """Агент составляет план действий на основе типа запроса"""
    action_templates = {
        "order_status": [
            {"action": "query_order_db", "params": {"order_id": "{extracted_order_id}"}},
            {"action": "generate_status_response", "params": {}},
            {"action": "send_response", "params": {}},
        ],
        "refund_request": [
            {"action": "verify_refund_eligibility", "params": {}},
            {"action": "create_refund_ticket", "params": {}},
            {"action": "notify_finance_team", "params": {}},
            {"action": "send_confirmation", "params": {}},
        ],
        "support_technical": [
            {"action": "search_knowledge_base", "params": {}},
            {"action": "generate_solution", "params": {}},
            {"action": "create_ticket_if_unsolved", "params": {}},
            {"action": "send_response", "params": {}},
        ],
    }
    base_plan = action_templates.get(state["request_type"], [
        {"action": "generate_generic_response", "params": {}},
        {"action": "create_manual_review_task", "params": {}},
    ])
    return {**state, "action_plan": base_plan}

def route_after_classification(state: RequestState) -> str:
    if state["requires_human"]:
        return "escalate_to_human"
    if state["confidence"] < 0.6:
        return "escalate_to_human"
    return "enrich"

def route_after_enrichment(state: RequestState) -> str:
    if state.get("user_profile", {}).get("tier") == "vip" and state["urgency"] in ("high", "critical"):
        return "plan_premium"
    return "plan"

graph = StateGraph(RequestState)
graph.add_node("classify", classify_request)
graph.add_node("enrich", enrich_request)
graph.add_node("plan", plan_actions)
graph.add_node("plan_premium", plan_premium_actions)
graph.add_node("execute", execute_actions)
graph.add_node("generate_response", generate_final_response)
graph.add_node("escalate_to_human", create_human_task)
graph.add_node("send_response", send_response_to_channel)
graph.set_entry_point("classify")
graph.add_conditional_edges("classify", route_after_classification)
graph.add_conditional_edges("enrich", route_after_enrichment)
graph.add_edge("plan", "execute")
graph.add_edge("plan_premium", "execute")
graph.add_edge("execute", "generate_response")
graph.add_edge("generate_response", "send_response")
graph.add_edge("send_response", END)
graph.add_edge("escalate_to_human", END)
processor = graph.compile(checkpointer=PostgresSaver(conn))

Почему система работает без оператора?

Ключевое отличие — способность выполнять действия в бизнес-системах: создавать заказы, проверять статусы, возвраты, отправлять уведомления. Каждое действие — это готовый модуль, который система вызывает по плану. Планировщик действий формирует последовательность шагов на основе типа запроса и контекста пользователя. Если план успешно выполнен — ответ отправляется автоматически. В противном случае система эскалирует задачу оператору с подробным логом ошибок.

Практический кейс: онлайн-ретейлер, 2500 запросов/день

До внедрения: среднее время первого ответа 4.2 часа, 12 операторов работают в три смены, 60% времени тратится на типовые статусные запросы.

После внедрения:

• Автономная обработка без участия оператора: 74%
• Среднее время первого ответа: с 4.2 часов до 2.1 минуты (в 120 раз быстрее)
• Ночная смена: сокращена с 4 до 1 оператора (мониторинг эскалаций)
• Точность ответов (выборка 500 запросов): 94.1%
• Ложные эскалации: 8.3%
• Ошибочное автоматическое закрытие: 2.1%

Экономия на фонде оплаты труда операторов составила более $18k–26k. в год, а стоимость обработки одной заявки снизилась в 10 раз. Первые две недели после запуска ушли на дообучение классификатора на реальных данных — точность выросла с 81% до 94% после 500 корректировок.

Что входит в работу

• Архитектура и проектирование: моделирование графа состояний, определение типов запросов, сценариев обработки.
• Разработка классификатора: подбор промптов, fine-tuning GPT-4o при необходимости, тестирование на исторических данных.
• Интеграция с каналами: webhook, API, мессенджеры, веб-формы.
• Исполнители действий: подключение к CRM, ERP, Service Desk, написание кода для типовых операций.
• Система мониторинга: метрики в Prometheus, дашборды в Grafana, алерты по SLA.
• Документация: описание графа, API, инструкции для операторов.
• Обучение команды: воркшоп по дообучению модели и администрированию.
• Поддержка на запуске: 2 недели сопровождения после ввода в эксплуатацию.

Как мы это делаем: пошаговый план

1. Аналитика (1–2 недели): собираем логи, выявляем типовые запросы, определяем критерии эскалации.
2. Проектирование графа (1–2 недели): создаём StateGraph, определяем узлы и рёбра.
3. Разработка классификатора (2–3 недели): тренируем модель, тестируем на выборке.
4. Реализация исполнителей (2–4 недели): код для каждого типа запросов.
5. Интеграция каналов (1–2 недели): подключаем email, API, мессенджеры.
6. Тестирование и калибровка (2 недели): прогон на реальных данных, корректировка порогов.
7. Деплой и мониторинг: развёртывание на Kubernetes, настройка алертов.

Сроки

• Архитектура системы и граф: 1–2 недели
• Классификатор + обогащение данных: 2–3 недели
• Исполнители для каждого типа запросов: 2–4 недели
• Интеграция с каналами (email, мессенджеры): 1–2 недели
• Калибровка и запуск в production: 2 недели
• Итого: 8–13 недель

Оценим ваш проект — просто напишите нам. Мы гарантируем прозрачность на каждом этапе и передаём полную документацию. Сертифицированные инженеры с опытом 10+ лет реализуют систему под ключ. Для предварительной оценки вашего потока запросов закажите аудит — мы определим потенциал автоматизации и сроки внедрения.