教培SaaS线索分配系统的状态机设计从人工抢单到智能路由的演进

背景

教培机构的招生线索管理是一个典型的状态流转场景。一条线索从进入系统到最终成交或流失，会经历分配、跟进、试听、转化等多个状态。我们最初用status字段加if-else实现，随着业务复杂度增加，代码变成了面条式的条件判断，维护噩梦。后来重构为状态机模式，今天分享这个过程。

一、原始实现的问题

最初的线索模型很简单：

class Lead:

status = models.CharField(max_length=20, default='new')

assigned_to = models.ForeignKey('User', null=True)

last_followup = models.DateTimeField(null=True)

# 分配逻辑散落在各处

def assign_lead(lead, user):

if lead.status == 'new':

lead.assigned_to = user

lead.status = 'assigned'

lead.save()

elif lead.status == 'assigned':

raise Exception('already assigned')

elif lead.status == 'lost':

raise Exception('cannot assign lost lead')

def followup_lead(lead, user, note):

if lead.status == 'assigned' and lead.assigned_to == user:

lead.status = 'following'

lead.last_followup = timezone.now()

lead.save()

elif lead.status == 'following':

lead.last_followup = timezone.now()

lead.save()

else:

raise Exception('invalid state transition')

这种写法的问题很明显：状态流转逻辑散落在十几个函数里，每加一个状态就要改十几处代码。而且没有防止非法状态转换，比如有人把'lost'状态的线索改成'following'也不会报错。

二、状态机建模

首先梳理线索的完整状态流转图：

new -> assigned -> following -> trial_scheduled -> trial_completed -> won/lost

还有几条特殊路径：

assigned -> recycled（超时未跟进回收）

following -> recycled（长时间未推进回收）

trial_completed -> following（试听后继续跟进）

任何状态 -> invalid（线索无效）

用状态机模式重构：

from enum import Enum

from transitions import Machine

class LeadState(Enum):

NEW = 'new'

ASSIGNED = 'assigned'

FOLLOWING = 'following'

TRIAL_SCHEDULED = 'trial_scheduled'

TRIAL_COMPLETED = 'trial_completed'

WON = 'won'

LOST = 'lost'

RECYCLED = 'recycled'

INVALID = 'invalid'

class LeadStateMachine:

transitions = [

# trigger, source, dest, conditions, before, after

['assign', 'new', 'assigned', 'can_assign', 'before_assign', 'after_assign'],

['assign', 'recycled', 'assigned', 'can_assign', 'before_assign', 'after_assign'],

['start_followup', 'assigned', 'following', None, None, 'notify_assignee'],

['schedule_trial', 'following', 'trial_scheduled', None, 'validate_trial_slot', 'notify_parent'],

['complete_trial', 'trial_scheduled', 'trial_completed', None, 'record_trial_feedback', None],

['continue_followup', 'trial_completed', 'following', None, None, 'notify_assignee'],

['win', ['following', 'trial_completed'], 'won', None, 'create_contract', 'notify_manager'],

['lose', ['following', 'trial_completed'], 'lost', None, 'record_loss_reason', None],

['recycle', ['assigned', 'following'], 'recycled', None, 'clear_assignee', 'add_to_pool'],

['invalidate', '*', 'invalid', None, 'record_invalid_reason', None],

]

def __init__(self, lead):

self.lead = lead

self.machine = Machine(

model=self,

states=[s.value for s in LeadState],

transitions=self.transitions,

initial=lead.status,

send_event=True

)

def can_assign(self, event):

# 检查当前用户是否有分配权限

user = event.kwargs.get('user')

return user.has_perm('assign_lead')

def before_assign(self, event):

user = event.kwargs.get('user')

self.lead.assigned_to = user

self.lead.assigned_at = timezone.now()

def after_assign(self, event):

# 发送通知给被分配人

send_notification.delay(

user_id=self.lead.assigned_to_id,

title='新线索分配',

content=f'您收到一条新线索：{self.lead.student_name}'

)

使用transitions库后，状态流转逻辑集中在一处定义，非法转换会被自动拦截，新增状态只需在transitions列表里加一条。

三、智能分配路由

状态机解决了"怎么转"的问题，但没解决"转给谁"的问题。最初是人工抢单，后来客户要求智能分配。

分配策略需要考虑多个因素：销售老师的转化率、当前待跟进线索数、线索来源渠道与销售老师的匹配度、时间段（有些老师晚上效率高）。

class LeadRouter:

def __init__(self, config):

self.weights = {

'conversion_rate': 0.35,

'workload': 0.25,

'channel_match': 0.20,

'response_speed': 0.20

}

self.config = config

async def route(self, lead, candidates):

scores = []

for user in candidates:

score = await self._calculate_score(lead, user)

scores.append((user, score))

scores.sort(key=lambda x: x[1], reverse=True)

# 如果最高分和第二高分差距小于阈值，随机选一个避免总是分给同一个人

if len(scores) >= 2 and (scores[0][1] - scores[1][1]) < 0.05:

top_two = scores[:2]

chosen = random.choice(top_two)[0]

else:

chosen = scores[0][0]

return chosen

async def _calculate_score(self, lead, user):

# 转化率得分

conv_rate = await self._get_conversion_rate(user)

conv_score = min(conv_rate / 0.3, 1.0) # 30%转化率得满分

# 工作量得分（待跟进越少分越高）

pending = await self._get_pending_count(user)

workload_score = max(1 - pending / 30, 0) # 30条待跟进得0分

# 渠道匹配度

channel_match = await self._get_channel_match(user, lead.channel)

# 响应速度

avg_response = await self._get_avg_response_time(user)

response_score = max(1 - avg_response / 3600, 0) # 1小时响应得0分

total = (

self.weights['conversion_rate'] * conv_score +

self.weights['workload'] * workload_score +

self.weights['channel_match'] * channel_match +

self.weights['response_speed'] * response_score

)

return total

这里踩了一个坑：最初把转化率权重设为0.6，结果导致所有高转化率的销售老师线索堆积，而新老师一直分不到线索，能力无法提升。后来把转化率权重降到0.35，加了随机选择机制，线索分配更均衡了。

四、超时回收机制

线索分配后如果销售老师不及时跟进，需要在超时后自动回收。这需要一个定时任务扫描超时线索：

class LeadRecycler:

ASSIGN_TIMEOUT = 2 * 3600 # 分配后2小时未跟进

FOLLOWUP_TIMEOUT = 48 * 3600 # 跟进后48小时未推进

async def run(self):

while True:

await self._check_assignment_timeout()

await self._check_followup_timeout()

await asyncio.sleep(300) # 每5分钟检查一次

async def _check_assignment_timeout(self):

cutoff = timezone.now() - timedelta(seconds=self.ASSIGN_TIMEOUT)

leads = Lead.objects.filter(

status='assigned',

assigned_at__lt=cutoff

)

for lead in leads:

fsm = LeadStateMachine(lead)

try:

fsm.recycle()

lead.status = fsm.state

lead.assigned_to = None

lead.save()

await self._notify_recycle(lead, 'assignment_timeout')

except MachineError:

pass # 状态转换失败，跳过

五、总结

从面条式if-else到状态机加智能路由的演进，核心收获：

1. 状态机模式把散落的状态流转逻辑集中管理，新增状态和转换规则只需改一处配置。

2. transitions库的send_event模式可以在转换前后插入钩子函数，做通知、校验、日志等操作。

3. 智能分配需要多维度加权打分，不能只看单一指标，权重要跟业务方一起调。

4. 超时回收是线索管理容易被忽略的一环，不回收就等于线索浪费。

5. 状态机要有完整的日志记录，每次状态转换都记录who、when、from、to、reason，方便后续分析转化漏斗。