Customer Support: Tested & Compared A Deep Dive

import json
import logging
import os
import sys
import time
from datetime import datetime
from typing import Optional

import joblib
import numpy as np
from confluent_kafka import Consumer, KafkaError, KafkaException, Producer
from sklearn.calibration import CalibratedClassifierCV
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LinearSVC
from sklearn.pipeline import Pipeline

# ---------------------------------------------------------------------------
# Configuration — pulled from environment with sane defaults
# ---------------------------------------------------------------------------
KAFKA_BOOTSTRAP = os.environ.get("KAFKA_BOOTSTRAP", "kafka-1:9092,kafka-2:9092,kafka-3:9092")
RAW_TOPIC = os.environ.get("RAW_TOPIC", "raw.tickets")
CLASSIFIED_TOPIC = os.environ.get("CLASSIFIED_TOPIC", "classified.tickets")
MODEL_PATH = os.environ.get("MODEL_PATH", "/models/ticket_classifier_v3.joblib")
GROUP_ID = os.environ.get("GROUP_ID", "classifier-cg-01")
CONFIDENCE_THRESHOLD = float(os.environ.get("CONFIDENCE_THRESHOLD", "0.72"))
POLL_TIMEOUT_S = float(os.environ.get("POLL_TIMEOUT_S", "1.0"))

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s — %(message)s",
)
logger = logging.getLogger("ticket_classifier")


def load_model(path: str) -> Pipeline:
    """Load the pre-trained TF-IDF + LinearSVC pipeline from disk.

    The pipeline bundles vectoriser and classifier so that raw text
    can be scored without manual feature construction.
    """
    try:
        pipe: Pipeline = joblib.load(path)
        logger.info("Loaded model from %s (vocab size: %d)", path, len(pipe.named_steps["tfidf"].vocabulary_))
        return pipe
    except FileNotFoundError:
        logger.error("Model file not found at %s — aborting.", path)
        sys.exit(1)
    except Exception as exc:
        logger.error("Unexpected error loading model: %s", exc)
        sys.exit(1)


def build_kafka_consumer() -> Consumer:
    """Create a Kafka consumer with at-least-once semantics."""
    try:
        c = Consumer({
            "bootstrap.servers": KAFKA_BOOTSTRAP,
            "group.id": GROUP_ID,
            "auto.offset.reset": "earliest",
            "enable.auto.commit": False,          # manual commit after successful classification
            "max.poll.interval.ms": 300_000,
            "session.timeout.ms": 30_000,
        })
        c.subscribe([RAW_TOPIC])
        logger.info("Consumer subscribed to '%s'", RAW_TOPIC)
        return c
    except KafkaException as exc:
        logger.error("Kafka consumer creation failed: %s", exc)
        sys.exit(1)


def build_kafka_producer() -> Producer:
    """Create a Kafka producer with idempotent delivery."""
    try:
        p = Producer({
            "bootstrap.servers": KAFKA_BOOTSTRAP,
            "enable.idempotence": True,
            "acks": "all",
            "retries": 10,
        })
        return p
    except KafkaException as exc:
        logger.error("Kafka producer creation failed: %s", exc)
        sys.exit(1)


def classify_ticket(pipeline: Pipeline, raw_text: str) -> dict:
    """Run inference on a single ticket body.

    Returns a dict with category, priority, confidence, and a
    fallback flag when confidence is below threshold.
    """
    # predict_proba returns (n_samples, n_classes)
    probabilities: np.ndarray = pipeline.predict_proba([raw_text])[0]
    best_idx: int = int(np.argmax(probabilities))
    confidence: float = float(probabilities[best_idx])
    category: str = pipeline.classes_[best_idx]  # type: ignore[attr-defined]

    # Priority heuristic: P1 for billing/outage, P2 for feature issues,
    # P3 for everything else.
    priority_map = {
        "billing": "P1",
        "outage": "P1",
        "bug": "P2",
        "feature_request": "P3",
        "general_inquiry": "P3",
    }
    priority: str = priority_map.get(category, "P3")

    return {
        "category": category,
        "priority": priority,
        "confidence": round(confidence, 4),
        "needs_human_review": confidence < CONFIDENCE_THRESHOLD,
    }


def delivery_report(err: Optional[KafkaError], msg):
    """Callback invoked by the producer on delivery success or failure."""
    if err is not None:
        logger.error("Message delivery failed: %s — topic %s partition %s", err, msg.topic(), msg.partition())
    else:
        logger.debug("Message delivered to %s [%d] @ offset %d", msg.topic(), msg.partition(), msg.offset())


def main():
    logger.info("Starting ticket classifier (pid=%d)", os.getpid())
    pipeline = load_model(MODEL_PATH)
    consumer = build_kafka_consumer()
    producer = build_kafka_producer()
    processed = 0
    errors = 0

    try:
        while True:
            msg = consumer.poll(timeout=POLL_TIMEOUT_S)
            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                logger.error("Consumer error: %s", msg.error())
                errors += 1
                continue

            try:
                payload = json.loads(msg.value().decode("utf-8"))
                raw_text: str = payload.get("body", "")
                if not raw_text.strip():
                    logger.warning("Empty ticket body, skipping offset %d", msg.offset())
                    consumer.commit(asynchronous=False)
                    continue

                result = classify_ticket(pipeline, raw_text)
                enriched = {
                    **payload,
                    **result,
                    "classified_at": datetime.utcnow().isoformat() + "Z",
                }
                producer.produce(
                    CLASSIFIED_TOPIC,
                    key=msg.key(),
                    value=json.dumps(enriched).encode("utf-8"),
                    on_delivery=delivery_report,
                )
                producer.flush()
                consumer.commit(asynchronous=False)
                processed += 1
                if processed % 500 == 0:
                    logger.info("Processed %d tickets (errors: %d)", processed, errors)
            except (json.JSONDecodeError, KeyError) as exc:
                logger.exception("Malformed message at offset %d: %s", msg.offset(), exc)
                errors += 1
                consumer.commit(asynchronous=False)
    except KeyboardInterrupt:
        logger.info("Shutting down — processed=%d errors=%d", processed, errors)
    finally:
        consumer.close()
        producer.flush()


if __name__ == "__main__":
    main()

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "os"
    "sync"
    "time"

    "github.com/segmentio/kafka-go"
    "github.com/gorilla/websocket"
)

// Ticket represents a classified ticket flowing through Kafka.
type Ticket struct {
    ID          string    `json:"id"`
    Category    string    `json:"category"`
    Priority    string    `json:"priority"`
    Body        string    `json:"body"`
    CustomerID  string    `json:"customer_id"`
    NeedsReview bool      `json:"needs_human_review"`
    ClassifiedAt string   `json:"classified_at"`
}

// AgentSession tracks a connected support agent.
type AgentSession struct {
    Conn     *websocket.Conn
    Send     chan []byte
    Skills   []string
    Region   string
    LastSeen time.Time
}

var upgrader = websocket.Upgrader{
    ReadBufferSize:  4096,
    WriteBufferSize: 4096,
    CheckOrigin: func(r *http.Request) bool {
        // In production, validate against a whitelist of origins.
        return true
    },
}

// Hub maintains the set of active agent sessions and dispatches tickets.
type Hub struct {
    agents    map[string]*AgentSession // keyed by agent ID
    mu        sync.RWMutex
    kafkaRdr  *kafka.Reader
    broadcast chan Ticket
}

func NewHub(kafkaBrokers []string) *Hub {
    return &Hub{
        agents:    make(map[string]*AgentSession),
        broadcast: make(chan Ticket, 1024), // buffered to absorb bursts
        kafkaRdr: kafka.NewReader(kafka.ReaderConfig{
            Brokers:  kafkaBrokers,
            Topic:    "classified.tickets",
            GroupID:  "router-group-01",
            MinBytes: 10e3, // 10 KB
            MaxBytes: 10e6, // 10 MB
        }),
    }
}

// RegisterAgent adds a new WebSocket-connected agent to the hub.
func (h *Hub) RegisterAgent(agentID string, session *AgentSession) {
    h.mu.Lock()
    defer h.mu.Unlock()
    h.agents[agentID] = session
    log.Printf("Agent %s registered (skills: %v, region: %s)", agentID, session.Skills, session.Region)
}

// UnregisterAgent removes an agent and drains their send channel.
func (h *Hub) UnregisterAgent(agentID string) {
    h.mu.Lock()
    defer h.mu.Unlock()
    if session, ok := h.agents[agentID]; ok {
        close(session.Send)
        session.Conn.Close()
        delete(h.agents, agentID)
        log.Printf("Agent %s unregistered", agentID)
    }
}

// routeScore computes a float64 suitability score for an agent.
// Higher is better. Factors: skill match (0.6), region match (0.2), idle time (0.2).
func routeScore(ticket Ticket, session *AgentSession) float64 {
    skillMatch := 0.0
    for _, s := range session.Skills {
        if s == ticket.Category {
            skillMatch = 1.0
            break
        }
    }
    regionMatch := 0.0
    if session.Region == ticket.CustomerID[:2] { // simplified region prefix
        regionMatch = 1.0
    }
    idleSeconds := time.Since(session.LastSeen).Seconds()
    idleScore := 1.0 / (1.0 + idleSeconds/60.0) // decays from 1.0 toward 0
    return 0.6*skillMatch + 0.2*regionMatch + 0.2*idleScore
}

// dispatch picks the best agent and sends the ticket JSON to their channel.
func (h *Hub) dispatch(ticket Ticket) {
    h.mu.RLock()
    defer h.mu.RUnlock()

    var bestAgent string
    var bestScore float64 = -1
    for id, session := range h.agents {
        score := routeScore(ticket, session)
        if score > bestScore {
            bestScore = score
            bestAgent = id
        }
    }

    if bestAgent == "" {
        log.Printf("No agents online for ticket %s — queuing for general pool", ticket.ID)
        // Fall back: write to PostgreSQL general queue (omitted for brevity).
        return
    }

    payload, _ := json.Marshal(ticket)
    select {
    case h.agents[bestAgent].Send <- payload:
        log.Printf("Dispatched ticket %s to agent %s (score %.2f)", ticket.ID, bestAgent, bestScore)
    case <-time.After(2 * time.Second):
        log.Printf("Agent %s channel full, re-queuing ticket %s", bestAgent, ticket.ID)
    }
}

// consumeKafka reads classified tickets from Kafka and fans them out.
func (h *Hub) consumeKafka(ctx context.Context) {
    defer h.kafkaRdr.Close()
    for {
        msg, err := h.kafkaRdr.FetchMessage(ctx)
        if err != nil {
            if ctx.Err() != nil {
                return
            }
            log.Printf("Kafka fetch error: %v", err)
            time.Sleep(2 * time.Second)
            continue
        }
        var ticket Ticket
        if err := json.Unmarshal(msg.Value, &ticket); err != nil {
            log.Printf("JSON decode error: %v", err)
            h.kafkaRdr.CommitMessages(ctx, msg)
            continue
        }
        h.broadcast <- ticket
        h.kafkaRdr.CommitMessages(ctx, msg)
    }
}

// run starts the dispatch loop and WebSocket upgrade handler.
func (h *Hub) run(ctx context.Context, addr string) {
    go h.consumeKafka(ctx)

    go func() {
        for ticket := range h.broadcast {
            h.dispatch(ticket)
        }
    }()

    http.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
        conn, err := upgrader.Upgrade(w, r, nil)
        if err != nil {
            log.Printf("WebSocket upgrade failed: %v", err)
            return
        }
        agentID := r.URL.Query().Get("agent_id")
        if agentID == "" {
            conn.Close()
            return
        }
        session := &AgentSession{
            Conn:   conn,
            Send:   make(chan []byte, 256),
            Skills: []string{"billing", "technical"}, // In prod, fetched from DB.
            Region: "us",
        }
        h.RegisterAgent(agentID, session)

        go func() {
            defer h.UnregisterAgent(agentID)
            for {
                select {
                case msg, ok := <-session.Send:
                    if !ok {
                        return
                    }
                    if err := conn.WriteMessage(websocket.TextMessage, msg); err != nil {
                        log.Printf("Write error to agent %s: %v", agentID, err)
                        return
                    }
                }
            }
        }()

        // Read pump — agents send heartbeats / acknowledgements.
        for {
            _, _, err := conn.NextReader()
            if err != nil {
                break
            }
            session.LastSeen = time.Now()
        }
    })

    log.Printf("Router listening on %s", addr)
    if err := http.ListenAndServe(addr, nil); err != nil {
        log.Fatalf("Server exited: %v", err)
    }
}

func main() {
    brokers := []string{
        os.Getenv("KAFKA_BROKER_1"),
        os.Getenv("KAFKA_BROKER_2"),
        os.Getenv("KAFKA_BROKER_3"),
    }
    hub := NewHub(brokers)
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    hub.run(ctx, ":8080")
}

import (
    "context"
    "encoding/json"
    "fmt"
    "time"

    "github.com/go-redis/redis/v8"
)

// TicketSummary is the minimal payload stored in Redis sorted sets.
type TicketSummary struct {
    ID        string  `json:"id"`
    Customer  string  `json:"customer_id"`
    Priority  int64   `json:"priority_score"` // 1=high, 5=low
    Category  string  `json:"category"`
    Timestamp int64   `json:"ts"`
}

// QueueManager handles Redis-backed agent queues.
type QueueManager struct {
    client *redis.Client
    ctx    context.Context
}

func NewQueueManager(addr, password string, db int) *QueueManager {
    client := redis.NewClient(&redis.Options{
        Addr:     addr,
        Password: password,
        DB:       db,
        PoolSize: 200, // sized for ~2000 concurrent dashboard polls
    })
    // Verify connectivity.
    if err := client.Ping(context.Background()).Err(); err != nil {
        panic(fmt.Sprintf("cannot connect to Redis at %s: %v", addr, err))
    }
    return &QueueManager{
        client: client,
        ctx:    context.Background(),
    }
}

// queueKey returns the Redis key for a given skill/category queue.
func queueKey(category string) string {
    return fmt.Sprintf("queue:%s", category)
}

// dedupKey returns the key used to track recent submissions for dedup.
func dedupKey(customerID, category string) string {
    return fmt.Sprintf("dedup:%s:%s", customerID, category)
}

// Enqueue adds a ticket to the appropriate skill queue.
// Priority score is inverted so lower numbers (P1) sort first in the sorted set.
// Deduplication: if the same customer submitted the same category within
// the last 5 minutes, the existing ticket is re-scored instead of duplicated.
func (qm *QueueManager) Enqueue(t TicketSummary) error {
    // Atomic dedup check + enqueue via Lua script to avoid race conditions.
    script := `
        local dedup = KEYS[1]
        local queue = KEYS[2]
        local ticketID = ARGV[1]
        local customerID = ARGV[2]
        local priority = tonumber(ARGV[3])
        local category = ARGV[4]
        local ts = tonumber(ARGV[5])
        local ttl = tonumber(ARGV[6])

        -- Check dedup: if a recent ticket from same customer+category exists, bump score.
        local existing = redis.call('GET', dedup)
        if existing then
            local existingID = existing
            -- Re-score the existing ticket to boost priority.
            redis.call('ZADD', queue, priority, existingID)
            return {0, existingID}
        end

        -- No duplicate: add to queue and set dedup key.
        redis.call('ZADD', queue, priority, ticketID)
        redis.call('SETEX', dedup, ttl, ticketID)
        return {1, ticketID}
    `
    dedupTTL := int64(300) // 5 minutes
    result, err := qm.client.Eval(
        qm.ctx,
        script,
        []string{dedupKey(t.Customer, t.Category), queueKey(t.Category)},
        t.ID, t.Customer, t.Priority, t.Category, t.Timestamp, dedupTTL,
    ).Result()
    if err != nil {
        return fmt.Errorf("enqueue lua failed: %w", err)
    }
    added := result.([]interface{})[0].(int64)
    if added == 0 {
        log.Printf("Deduplicated ticket %s for customer %s", t.ID, t.Customer)
    } else {
        log.Printf("Enqueued ticket %s into '%s' with priority %d", t.ID, t.Category, t.Priority)
    }
    return nil
}

// Dequeue pops the highest-priority ticket for a given category.
// Returns nil when the queue is empty.
func (qm *QueueManager) Dequeue(category string) (*TicketSummary, error) {
    key := queueKey(category)
    // ZPOPMIN atomically removes and returns the lowest-score member.
    result, err := qm.client.ZPopMin(qm.ctx, key, 1).Result()
    if err != nil {
        return nil, fmt.Errorf("zpopmin failed for %s: %w", key, err)
    }
    if len(result) == 0 {
        return nil, nil // empty queue
    }
    var ticket TicketSummary
    if err := json.Unmarshal([]byte(result[0].Member.(string)), &ticket); err != nil {
        return nil, fmt.Errorf("JSON unmarshal failed: %w", err)
    }
    return &ticket, nil
}

// QueueDepth returns the number of pending tickets in a category queue.
func (qm *QueueManager) QueueDepth(category string) (int64, error) {
    return qm.client.ZCard(qm.ctx, queueKey(category)).Result()
}

func main() {
    qm := NewQueueManager("redis-cluster.internal:6379", "", 0)

    // Simulate enqueueing 10 tickets.
    for i := 0; i < 10; i++ {
        t := TicketSummary{
            ID:        fmt.Sprintf("tkn-%05d", i),
            Customer:  fmt.Sprintf("cust-%d", i%3), // deliberate dupes
            Priority:  int64((i % 5) + 1),
            Category:  "billing",
            Timestamp: time.Now().Unix(),
        }
        if err := qm.Enqueue(t); err != nil {
            fmt.Printf("ERROR: %v\n", err)
        }
    }

    depth, _ := qm.QueueDepth("billing")
    fmt.Printf("Billing queue depth after enqueue: %d\n", depth)

    // Drain queue.
    for {
        ticket, err := qm.Dequeue("billing")
        if err != nil {
            fmt.Printf("ERROR: %v\n", err)
            break
        }
        if ticket == nil {
            fmt.Println("Queue empty.")
            break
        }
        fmt.Printf("Processing: %s (priority %d)\n", ticket.ID, ticket.Priority)
    }
}

from pydantic import BaseModel, Field, model_validator, ValidationError
from typing import Literal, Optional
from datetime import datetime

class TicketIn(BaseModel):
    id: str = Field(..., min_length=1, max_length=64)
    customer_id: str = Field(..., pattern=r'^cust_[a-zA-Z0-9]{6,20}$')
    channel: Literal["email", "chat", "api", "phone"]
    body: str = Field(..., min_length=1, max_length=50_000)
    priority_override: Optional[Literal["P1", "P2", "P3", "P4"]] = None
    created_at: datetime

    @model_validator(mode='after')
    def validate_priority_for_channel(self):
        if self.channel == "phone" and self.priority_override != "P1":
            self.priority_override = "P1"  # phone tickets are always urgent
        return self

def validate_ticket(raw_json: dict) -> TicketIn:
    try:
        return TicketIn(**raw_json)
    except ValidationError as e:
        raise ValueError(f"Invalid ticket payload: {e}")

[General]
hostname = "0.0.0.0"
port = 8000

[Zookeeper]
servers = "zk-1:2181,zk-2:2181,zk-3:2181"

[Kafka "support"]
broker = "kafka-1:9092,kafka-2:9092,kafka-3:9092"
class-name = "kafka"

[Consumer "classifier"]
class-name = "kafka_zk"
cluster = "support"
topic = ["raw.tickets", "classified.tickets"]

func (r *DegradedRouter) Route(t Ticket) error {
    if r.classifierHealthy {
        result, err := r.classifyViaML(t)
        if err != nil {
            r.logger.Warn("ML classification failed, falling back", zap.Error(err))
            r.classifierHealthy = false
            return r.routeWithRules(t)
        }
        t.Category = result.Category
        t.Priority = result.Priority
        return r.dispatchToAgent(t)
    }
    return r.routeWithRules(t)
}

func (r *DegradedRouter) healthCheck() {
    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()
    for range ticker.C {
        ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
        _, err := r.cli.Health(ctx)
        cancel()
        r.classifierHealthy = (err == nil)
        if !r.classifierHealthy {
            r.logger.Warn("Classifier unhealthy, using rule-based fallback")
        } else {
            r.logger.Info("Classifier recovered, resuming ML routing")
        }
    }
}