Разработка AI-цифрового диспетчера (AI Dispatcher)

AI Диспетчер — цифровой сотрудник для координации задач

AI Диспетчер автоматически принимает входящие заявки, классифицирует их, назначает исполнителей с учётом загрузки и компетенций, отслеживает выполнение и эскалирует просроченные задачи. Применяется в службах технической поддержки, полевых операциях, транспортной диспетчеризации, управлении заявками на обслуживание.

Ядро: intake и маршрутизация заявок

from openai import AsyncOpenAI
from pydantic import BaseModel
from typing import Literal, Optional

client = AsyncOpenAI()

class TaskClassification(BaseModel):
    category: str
    priority: Literal["critical", "high", "normal", "low"]
    required_skill: str
    estimated_duration_minutes: int
    location: Optional[str]
    sla_hours: float             # SLA в часах
    special_requirements: list[str]

class DispatcherAgent:

    def __init__(self, team_db, task_db):
        self.team_db = team_db
        self.task_db = task_db

    async def process_incoming_request(self, request: dict) -> dict:
        """Принимает заявку и назначает исполнителя"""

        # Классификация
        classification = await client.beta.chat.completions.parse(
            model="gpt-4o",
            messages=[{
                "role": "system",
                "content": "Классифицируй входящую заявку для диспетчеризации. Определи приоритет, категорию, необходимые компетенции и ожидаемое время выполнения."
            }, {
                "role": "user",
                "content": f"Заявка: {request['description']}\nОт: {request['client']}\nКонтакт: {request['contact']}"
            }],
            response_format=TaskClassification,
            temperature=0,
        )

        task_class = classification.choices[0].message.parsed

        # Выбор исполнителя
        assignee = await self.select_best_assignee(task_class, request.get("location"))

        # Создание задачи
        task = await self.task_db.create({
            "title": request.get("title", f"Заявка от {request['client']}"),
            "description": request["description"],
            "category": task_class.category,
            "priority": task_class.priority,
            "required_skill": task_class.required_skill,
            "assignee_id": assignee["id"] if assignee else None,
            "sla_deadline": datetime.now() + timedelta(hours=task_class.sla_hours),
            "status": "assigned" if assignee else "pending",
        })

        # Уведомление исполнителя
        if assignee:
            await self.notify_assignee(assignee, task)

        return {"task_id": task["id"], "assignee": assignee, "sla": task["sla_deadline"]}

    async def select_best_assignee(
        self,
        task: TaskClassification,
        location: Optional[str],
    ) -> Optional[dict]:
        """Выбирает оптимального исполнителя"""

        available = await self.team_db.get_available(
            skill=task.required_skill,
            shift="current",
        )

        if not available:
            return None

        # Скоринг исполнителей
        scored = []
        for person in available:
            score = 100

            # Загрузка (меньше задач = выше приоритет)
            current_load = await self.task_db.count_active(person["id"])
            score -= current_load * 10

            # Близость к локации (для полевых задач)
            if location and person.get("current_location"):
                distance = calculate_distance(location, person["current_location"])
                score -= min(distance / 10, 30)  # Максимум -30 за дистанцию

            # Специализация
            if task.required_skill in person.get("specializations", []):
                score += 20

            scored.append({**person, "score": score})

        # Лучший кандидат
        return max(scored, key=lambda x: x["score"]) if scored else None

SLA-мониторинг и эскалации

class SLAMonitor:

    async def check_and_escalate(self):
        """Проверяет SLA каждые 15 минут"""

        tasks = await self.task_db.get_active_tasks()
        now = datetime.now()

        for task in tasks:
            sla_deadline = task["sla_deadline"]
            time_to_sla = (sla_deadline - now).total_seconds() / 3600

            if time_to_sla < 0:
                # SLA нарушен
                await self.handle_sla_breach(task)
            elif time_to_sla < 0.5:
                # До SLA < 30 минут
                await self.send_urgent_reminder(task)
            elif time_to_sla < 1 and task["status"] == "assigned":
                # До SLA < 1 час, задача не начата
                await self.escalate_to_supervisor(task, reason="task_not_started")

    async def handle_sla_breach(self, task: dict):
        """Обработка нарушения SLA"""

        breach_message = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "system",
                "content": "Создай сообщение об эскалации нарушения SLA для руководителя. Краткое, конкретное."
            }, {
                "role": "user",
                "content": f"Задача: {task['title']}, нарушение SLA на {abs(int((datetime.now() - task['sla_deadline']).total_seconds() / 60))} мин"
            }],
        )

        await slack_client.post_message(
            channel="#dispatcher-escalations",
            text=f"🚨 SLA нарушен:\n{breach_message.choices[0].message.content}",
        )
        await self.task_db.update(task["id"], {"sla_status": "breached"})

Практический кейс: телеком, 800 заявок/день

Компания: телеком с полевыми инженерами, 800 заявок на технические выезды в день.

AI Диспетчер:

• Приём заявок через чат, email, API CRM
• Классификация: 8 типов (подключение, ремонт, замена оборудования и др.)
• Назначение инженера с учётом навыков, загрузки, геолокации
• SLA-контроль: 4ч для критических, 24ч для стандартных
• Автоматические напоминания инженерам
• Эскалация просроченных руководителю

Результаты:

• SLA compliance: 71% → 89%
• Время назначения задачи: 12 мин → 45 сек
• Перегрузка отдельных инженеров: -34% (равномерное распределение)

Сроки

• Классификатор заявок: 1–2 недели
• Алгоритм назначения исполнителей: 1–2 недели
• SLA мониторинг и эскалации: 1 неделя
• Интеграция с CRM и уведомлениями: 1–2 недели
• Итого: 4–7 недель