背景教培机构的招生线索管理是一个典型的状态流转场景。一条线索从进入系统到最终成交或流失会经历分配、跟进、试听、转化等多个状态。我们最初用status字段加if-else实现随着业务复杂度增加代码变成了面条式的条件判断维护噩梦。后来重构为状态机模式今天分享这个过程。一、原始实现的问题最初的线索模型很简单class Lead:status models.CharField(max_length20, defaultnew)assigned_to models.ForeignKey(User, nullTrue)last_followup models.DateTimeField(nullTrue)# 分配逻辑散落在各处def assign_lead(lead, user):if lead.status new:lead.assigned_to userlead.status assignedlead.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 followinglead.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 Enumfrom transitions import Machineclass LeadState(Enum):NEW newASSIGNED assignedFOLLOWING followingTRIAL_SCHEDULED trial_scheduledTRIAL_COMPLETED trial_completedWON wonLOST lostRECYCLED recycledINVALID invalidclass 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 leadself.machine Machine(modelself,states[s.value for s in LeadState],transitionsself.transitions,initiallead.status,send_eventTrue)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 userself.lead.assigned_at timezone.now()def after_assign(self, event):# 发送通知给被分配人send_notification.delay(user_idself.lead.assigned_to_id,title新线索分配,contentf您收到一条新线索{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 configasync def route(self, lead, candidates):scores []for user in candidates:score await self._calculate_score(lead, user)scores.append((user, score))scores.sort(keylambda x: x[1], reverseTrue)# 如果最高分和第二高分差距小于阈值随机选一个避免总是分给同一个人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 chosenasync 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(secondsself.ASSIGN_TIMEOUT)leads Lead.objects.filter(statusassigned,assigned_at__ltcutoff)for lead in leads:fsm LeadStateMachine(lead)try:fsm.recycle()lead.status fsm.statelead.assigned_to Nonelead.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方便后续分析转化漏斗。
背景教培机构的招生线索管理是一个典型的状态流转场景。一条线索从进入系统到最终成交或流失会经历分配、跟进、试听、转化等多个状态。我们最初用status字段加if-else实现随着业务复杂度增加代码变成了面条式的条件判断维护噩梦。后来重构为状态机模式今天分享这个过程。一、原始实现的问题最初的线索模型很简单class Lead:status models.CharField(max_length20, defaultnew)assigned_to models.ForeignKey(User, nullTrue)last_followup models.DateTimeField(nullTrue)# 分配逻辑散落在各处def assign_lead(lead, user):if lead.status new:lead.assigned_to userlead.status assignedlead.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 followinglead.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 Enumfrom transitions import Machineclass LeadState(Enum):NEW newASSIGNED assignedFOLLOWING followingTRIAL_SCHEDULED trial_scheduledTRIAL_COMPLETED trial_completedWON wonLOST lostRECYCLED recycledINVALID invalidclass 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 leadself.machine Machine(modelself,states[s.value for s in LeadState],transitionsself.transitions,initiallead.status,send_eventTrue)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 userself.lead.assigned_at timezone.now()def after_assign(self, event):# 发送通知给被分配人send_notification.delay(user_idself.lead.assigned_to_id,title新线索分配,contentf您收到一条新线索{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 configasync def route(self, lead, candidates):scores []for user in candidates:score await self._calculate_score(lead, user)scores.append((user, score))scores.sort(keylambda x: x[1], reverseTrue)# 如果最高分和第二高分差距小于阈值随机选一个避免总是分给同一个人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 chosenasync 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(secondsself.ASSIGN_TIMEOUT)leads Lead.objects.filter(statusassigned,assigned_at__ltcutoff)for lead in leads:fsm LeadStateMachine(lead)try:fsm.recycle()lead.status fsm.statelead.assigned_to Nonelead.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方便后续分析转化漏斗。
