AI Dispatcher – Digital Employee for Task Coordination
A dispatch service drowning in tickets: operators manually combing through emails and chats, assigning engineers on a hunch, SLA breaches piling up. We've seen this firsthand in telecom, handling 800 daily field service requests. Manual dispatching couldn't keep up: average task assignment time was 12 minutes, individual engineer overload reached 30%, and SLA compliance lingered at 71%. After deploying an AI dispatcher, assignment time dropped to 45 seconds, SLA compliance rose to 89%, and engineer load evened out. The classification model fine-tunes on your historical data, improving accuracy each month.
Our digital dispatcher automatically receives incoming requests, classifies them, assigns assignees based on workload and competencies, tracks execution, and escalates overdue tasks. All without human intervention. We implement such systems turnkey—from classifier configuration to CRM integration.
How does the AI dispatcher route requests?
At the core is intelligent intake and routing. The process consists of three steps: request classification, assignee scoring, and execution monitoring. Let's walk through the code:
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
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:
cls = await client.beta.chat.completions.parse(
model="gpt-4o",
messages=[{
"role": "system",
"content": "Classify the incoming request for dispatching. Determine priority, category, required competencies, and expected duration."
}, {
"role": "user",
"content": f"Request: {request['description']}\nFrom: {request['client']}\nContact: {request['contact']}"
}],
response_format=TaskClassification,
temperature=0,
)
task_class = cls.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 from {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, location):
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)
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
In the first step, the GPT-4o model analyzes the request description and extracts category, priority, and required skills. The scoring algorithm uses a base of 100 points: penalties for workload (10 points per active task) and distance (up to 30 points), bonus for specialization (20 points). This ensures optimal distribution given the current situation.
Scoring algorithm details
The algorithm also considers the assignee's task history, rating, and schedule availability. Business rules can be added, such as VIP client priority or mandatory certification.
Why is SLA monitoring critical?
Every task has a Service Level Agreement (SLA)—a completion deadline. Breaching it damages business reputation. Our AI dispatcher doesn't just create tasks; it monitors them in real time. Every 15 minutes, the system checks remaining time until deadline: if less than an hour, a reminder is sent; under 30 minutes, an urgent notification; a breach triggers escalation to management. Monitoring code:
class SLAMonitor:
async def check_and_escalate(self):
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:
await self.handle_sla_breach(task)
elif time_to_sla < 0.5:
await self.send_urgent_reminder(task)
elif time_to_sla < 1 and task["status"] == "assigned":
await self.escalate_to_supervisor(task, reason="task_not_started")
async def handle_sla_breach(self, task):
breach_message = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "system",
"content": "Create a concise, concrete SLA breach escalation message for the manager."
}, {
"role": "user",
"content": f"Task: {task['title']}, SLA breached by {abs(int((datetime.now() - task['sla_deadline']).total_seconds() / 60))} min"
}],
)
await slack_client.post_message(
channel="#dispatcher-escalations",
text=f"🚨 SLA breached:\n{breach_message.choices[0].message.content}",
)
await self.task_db.update(task["id"], {"sla_status": "breached"})
How to train the request classifier?
To train the classifier, we use your historical data. Steps:
- Dataset collection: at least 500 labeled requests.
- Preprocessing: cleaning, lemmatization.
- Model tuning: fine-tuning GPT-4o or other LLM on classification tasks.
- Validation: check accuracy on a held-out set (minimum 85%).
- Deployment: upload the model to SageMaker or Vertex AI.
What results did the AI dispatcher show in telecom?
From our practice: a telecom operator with field engineers, 800 field service requests daily. Before: manual distribution, 12 minutes per assignment, 71% SLA compliance. After: AI dispatcher.
| Metric | Before | After |
|---|---|---|
| Task assignment time | 12 min | 45 sec |
| SLA compliance | 71% | 89% |
| Engineer overload | ~30% imbalance | -34% (balanced distribution) |
The AI dispatcher assigns tasks 16 times faster than a human, and SLA compliance improves by 18 percentage points. The system accepts requests via chat, email, CRM API; classifies them into 8 types (connection, repair, equipment replacement); assigns an engineer based on skills, load, geolocation; monitors SLA (4 hours for critical, 24 hours for standard); automatically reminds and escalates overdue tasks.
What's included in AI dispatcher development
- Request classifier: configuring categories, priorities, required skills
- Assignment algorithm: scoring assignees by workload, skills, geolocation
- SLA monitoring: automatic checks, reminders, escalations
- Integration: with CRM, messengers (Slack/Telegram), notification APIs
- Documentation: architecture description, admin guide
- Training: classification model fine-tuning on your data
- Support: 3 months warranty after launch
Implementation timelines
| Stage | Duration |
|---|---|
| Request classifier | 1–2 weeks |
| Assignment algorithm | 1–2 weeks |
| SLA monitoring and escalations | 1 week |
| CRM and notification integration | 1–2 weeks |
| Total | 4–7 weeks |
Contact us to discuss implementing an AI dispatcher in your service. Get a consultation on dispatching optimization. Our experience: certified engineers, work with major telecom operators.







