这份 notebook 由 Donne Martin 准备。Source 和 license info 在 GitHub。

设计 Call Center

Constraints & assumptions

• Call Center 里有哪些 level 的员工？
  • Operator / Supervisor / Director
• Call 先给 Operator 处理，可以假设吗？
  • Yes
• Operator 无法处理时，是否升级给 Supervisor？
  • Yes
• Supervisor 无法处理时，是否升级给 Director？
  • Yes
• 可以假设 Director 能处理所有 calls？
  • Yes
• 如果没人可接听怎么办？
  • 进入 queue
• 需要处理 VIP calls（插队）吗？
  • No
• 输入可以假设 valid 吗？
  • Assume they're valid

Solution

设计核心是两件事：

1. 分级处理：Operator → Supervisor → Director
2. 排队兜底：所有人忙时进入 queue

核心对象

• Rank：员工 level（Operator/Supervisor/Director）
• CallState：READY / IN_PROGRESS / COMPLETE
• Call：一次 call，包含当前 rank 与分配的 employee
• Employee（abstract）：通用员工模型，支持 take_call、complete_call、escalate_call
• Operator / Supervisor / Director：不同层级的实现
• CallCenter：负责 dispatch、queue、escalation、完成回收

主要流程

• 新 call 进来时，按 rank 从低到高找可用 employee
• 找不到则进入 queued_calls
• operator/supervisor 处理不了时，调用 escalate_call 升级
• call 完成后，Call Center 可以把队列里的 call 分配给刚释放的 employee

Class 设计（示例）

class Rank(Enum):
    OPERATOR = 0
    SUPERVISOR = 1
    DIRECTOR = 2

class CallState(Enum):
    READY = 0
    IN_PROGRESS = 1
    COMPLETE = 2

class Call(object):
    def __init__(self, rank):
        self.state = CallState.READY
        self.rank = rank
        self.employee = None

class Employee(metaclass=ABCMeta):
    def __init__(self, employee_id, name, rank, call_center):
        self.employee_id = employee_id
        self.name = name
        self.rank = rank
        self.call = None
        self.call_center = call_center

    def take_call(self, call):
        self.call = call
        self.call.employee = self
        self.call.state = CallState.IN_PROGRESS

    def complete_call(self):
        self.call.state = CallState.COMPLETE
        self.call_center.notify_call_completed(self.call)

    @abstractmethod
    def escalate_call(self):
        pass

    def _escalate_call(self):
        self.call.state = CallState.READY
        call = self.call
        self.call = None
        self.call_center.notify_call_escalated(call)

class Operator(Employee):
    def escalate_call(self):
        self.call.rank = Rank.SUPERVISOR
        self._escalate_call()

class Supervisor(Employee):
    def escalate_call(self):
        self.call.rank = Rank.DIRECTOR
        self._escalate_call()

class Director(Employee):
    def escalate_call(self):
        raise NotImplementedError('Directors handle any call')

class CallCenter(object):
    def __init__(self, operators, supervisors, directors):
        self.operators = operators
        self.supervisors = supervisors
        self.directors = directors
        self.queued_calls = deque()

    def dispatch_call(self, call):
        employee = self._dispatch_call(call, self.operators)
        if employee is None:
            employee = self._dispatch_call(call, self.supervisors)
        if employee is None:
            employee = self._dispatch_call(call, self.directors)
        if employee is None:
            self.queued_calls.append(call)

    def _dispatch_call(self, call, employees):
        for employee in employees:
            if employee.call is None:
                employee.take_call(call)
                return employee
        return None

可以扩展的点

• VIP queue：优先级队列（priority queue）
• Round-robin：分配策略更公平
• Metrics：等待时间、平均处理时间、escalation rate
• Working hours：只在 shift 内接听

参考与下载

下载 Python 源码