Розробка 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:
                # Менше 30 хвилин до SLA
                await self.send_urgent_reminder(task)
            elif time_to_sla < 1 and task["status"] == "assigned":
                # Менше 1 години до SLA, завдання не починалося
                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 тижнів