Автономная 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+ лет реализуют систему под ключ. Для предварительной оценки вашего потока запросов закажите бесплатный аудит — мы определим потенциал автоматизации и сроки внедрения.