منطق توجيه التذاكر وتصعيدها

مشكلة التوجيه

تصنيف النوايا باستخدام LLM

import openai, json

client = openai.OpenAI(api_key='YOUR_OPENAI_KEY')

INTENTS = ['billing', 'technical_support', 'returns_refunds',
           'account_access', 'shipping', 'general_inquiry']

def classify_intent(message: str) -> dict:
    prompt = (
        f'Classify this customer message into exactly one intent.\n'
        f'Intents: {INTENTS}\n'
        f'Message: "{message}"\n'
        f'Respond with JSON: {{"intent": "...", "confidence": 0.0-1.0}}'
    )
    resp = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        response_format={'type': 'json_object'}
    )
    return json.loads(resp.choices[0].message.content)

التوجيه إلى قوائم انتظار المتخصصين

ROUTING_MAP = {
    'billing':           'queue_billing',
    'technical_support': 'queue_tech',
    'returns_refunds':   'queue_returns',
    'account_access':    'queue_tech',
    'shipping':          'queue_fulfillment',
    'general_inquiry':   'queue_general'
}

def route_ticket(ticket: dict) -> str:
    result = classify_intent(ticket['message'])
    intent = result['intent']
    confidence = result['confidence']
    queue = ROUTING_MAP.get(intent, 'queue_general')
    ticket['intent'] = intent
    ticket['confidence'] = confidence
    ticket['queue'] = queue
    return queue

حد الثقة للتصعيد

CONFIDENCE_THRESHOLD = 0.70

def route_with_escalation(ticket: dict) -> dict:
    result = classify_intent(ticket['message'])
    intent = result['intent']
    confidence = result['confidence']

    if confidence < CONFIDENCE_THRESHOLD:
        return {
            'ticket_id': ticket['id'],
            'action': 'escalate_to_human',
            'reason': f'Low confidence: {confidence:.2f}',
            'suggested_intent': intent
        }

    queue = ROUTING_MAP.get(intent, 'queue_general')
    return {
        'ticket_id': ticket['id'],
        'action': 'route_to_queue',
        'queue': queue,
        'intent': intent,
        'confidence': confidence
    }

التصعيد حسب أولوية SLA

from datetime import datetime, timezone

SLA_HOURS = {
    'queue_billing':      4,
    'queue_tech':         8,
    'queue_returns':      24,
    'queue_fulfillment':  12,
    'queue_general':      48
}

def check_sla_breach(ticket: dict) -> bool:
    created = datetime.fromisoformat(ticket['created_at'])
    age_hours = (datetime.now(timezone.utc) - created).total_seconds() / 3600
    sla = SLA_HOURS.get(ticket['queue'], 24)
    if age_hours > sla and ticket['status'] == 'open':
        ticket['queue'] = 'queue_supervisor_escalation'
        ticket['escalation_reason'] = f'SLA breach: {age_hours:.1f}h > {sla}h'
        return True
    return False

if __name__ == '__main__':
    from datetime import datetime, timedelta, timezone
    old_ticket = {
        'created_at': (datetime.now(timezone.utc) - timedelta(hours=10)).isoformat(),
        'queue': 'queue_billing',
        'status': 'open',
    }
    breached = check_sla_breach(old_ticket)
    print(f'SLA breached: {breached}')
    if breached:
        print('Escalation reason:', old_ticket['escalation_reason'])

التوجيه متعدد التصنيفات للتذاكر المعقدة

def classify_multi_intent(message: str) -> list[dict]:
    prompt = (
        f'A customer message may have multiple intents.\n'
        f'Intents: {INTENTS}\n'
        f'Message: "{message}"\n'
        f'Return JSON array: [{{"intent": "...", "confidence": 0.0}}]\n'
        f'Include only intents with confidence > 0.5'
    )
    resp = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        response_format={'type': 'json_object'}
    )
    data = json.loads(resp.choices[0].message.content)
    return data.get('intents', [])

# Route to multiple queues
def route_multi(ticket: dict) -> list[str]:
    intents = classify_multi_intent(ticket['message'])
    return [ROUTING_MAP.get(i['intent'], 'queue_general') for i in intents]

استخراج بيانات التذاكر الوصفية

def extract_metadata(message: str) -> dict:
    prompt = (
        f'Extract structured metadata from this customer message.\n'
        f'Return JSON: {{"order_id": null, "product": null, "account_email": null}}\n'
        f'Use null for fields not mentioned.\n'
        f'Message: "{message}"'
    )
    resp = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        response_format={'type': 'json_object'}
    )
    return json.loads(resp.choices[0].message.content)

meta = extract_metadata('My order #A12345 for the blue headphones never arrived.')
print(meta)  # {'order_id': 'A12345', 'product': 'blue headphones', 'account_email': None}

رفع الأولوية استنادًا إلى المشاعر

def detect_sentiment(message: str) -> str:
    prompt = f'Classify sentiment as positive/neutral/negative.\nMessage: "{message}"\nReturn JSON: {{"sentiment": "..."}}'
    resp = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        response_format={'type': 'json_object'}
    )
    return json.loads(resp.choices[0].message.content)['sentiment']

def set_priority(ticket: dict) -> str:
    sentiment = detect_sentiment(ticket['message'])
    if sentiment == 'negative':
        ticket['priority'] = 'high'
    elif sentiment == 'positive':
        ticket['priority'] = 'low'
    else:
        ticket['priority'] = 'normal'
    return ticket['priority']

تحديد سلسلة التصعيد

ESCALATION_CHAIN = {
    'queue_tech':              'queue_tech_tier2',
    'queue_tech_tier2':        'queue_tech_senior',
    'queue_tech_senior':       'queue_supervisor_escalation',
    'queue_billing':           'queue_billing_senior',
    'queue_billing_senior':    'queue_supervisor_escalation',
    'queue_returns':           'queue_supervisor_escalation',
    'queue_fulfillment':       'queue_supervisor_escalation',
    'queue_general':           'queue_supervisor_escalation',
    'queue_supervisor_escalation': None  # terminal — human manager
}

def escalate(ticket: dict) -> str | None:
    next_queue = ESCALATION_CHAIN.get(ticket['queue'])
    if next_queue:
        ticket['queue'] = next_queue
    return next_queue

if __name__ == '__main__':
    ticket = {'queue': 'queue_tech'}
    for _ in range(3):
        nxt = escalate(ticket)
        print(f"Escalated to: {ticket['queue']}")
        if nxt is None:
            break

خط أنابيب التوجيه الكامل

def process_ticket(raw_ticket: dict) -> dict:
    ticket = dict(raw_ticket)

    # Step 1: Classify and route
    routing = route_with_escalation(ticket)
    ticket.update(routing)

    # Step 2: Extract metadata
    ticket['metadata'] = extract_metadata(ticket['message'])

    # Step 3: Set priority from sentiment
    ticket['priority'] = set_priority(ticket)

    # Step 4: Check if already breaching SLA
    if ticket.get('created_at'):
        check_sla_breach(ticket)

    return ticket

result = process_ticket({
    'id': 'T001',
    'message': 'I was charged twice for my subscription last month!',
    'created_at': '2026-05-28T10:00:00+00:00',
    'status': 'open'
})
print(result)

مراقبة دقة التوجيه

import random

def log_routing_decision(ticket: dict, final_queue: str):
    was_correct = ticket.get('queue') == final_queue
    if not was_correct:
        print(f'[ROUTING_ERROR] ticket={ticket["id"]} '
              f'predicted={ticket["queue"]} actual={final_queue} '
              f'confidence={ticket.get("confidence", 0):.2f}')

# Specialist reassigns ticket: log the discrepancy
def specialist_reassign(ticket: dict, new_queue: str):
    log_routing_decision(ticket, new_queue)
    ticket['queue'] = new_queue
    return ticket

if __name__ == '__main__':
    ticket = {'id': 'T-1001', 'queue': 'queue_billing', 'confidence': 0.62}
    specialist_reassign(ticket, 'queue_tech')

ما حد الثقة الذي يجب أن يؤدي إلى التصعيد إلى موظف بشري بدلًا من التوجيه التلقائي؟

مراجعة توجيه التذاكر