背景
教培机构的招生线索管理是一个典型的状态流转场景。一条线索从进入系统到最终成交或流失，会经历分配、跟进、试听、转化等多个状态。我们最初用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，方便后续分析转化漏斗。
