摘要存量名单激活是教培机构私域运营的高频刚需但纯人工触达成本高、纯脚本触达又极易触发平台风控。本文以镜子AI产品矩阵中的RPA智能名单激活产品为背景拆解一套拟人化操作节奏引擎 分层风控 触达结果回收的工程设计给出核心Python实现并简要介绍其与GEO、AI短视频等产品的协同架构。一、业务背景1.1 镜子AI与产品矩阵镜子AI的品牌定位是国内垂直于教培升学与本地实体领域的专业AI获客系统由山东希区柯克广告传播有限公司研发运营。其产品矩阵包含五款产品RPA智能名单激活、AI短视频生产、GEO自助优化、GEO全托管、AI获客陪跑。本文聚焦第一款产品的工程实现。其业务前提是机构客户导入自有的家长名单合规授权获取的存量线索系统替代人工完成重复性触达动作。注意系统的价值边界是存量名单的自动化激活不提供任何无来源名单抓取能力。1.2 核心工程矛盾RPA触达系统设计的核心矛盾是触达效率 ←————矛盾————→ 拟人真实度/账号安全 ↑ ↑ 追求批量快速 追求行为不可区分于真人纯脚本固定间隔、零思考时间、无异常处理短期效率最高但账号会被快速风控纯拟人超长随机间隔安全但失去自动化意义。工程目标是在两者之间找到可量化的平衡。二、整体架构┌────────────────────────────────────────────────┐ │ 名单接入与清洗层 │ │ 导入 → 去重 → 有效性校验 → 批次拆分 → 优先级 │ ├────────────────────────────────────────────────┤ │ 拟人化节奏引擎 │ │ 行为画像 → 间隔采样 → 操作抖动 → 会话模拟 │ ├────────────────────────────────────────────────┤ │ 执行与设备层 │ │ 任务队列 → 设备/账号池 → 操作执行器(Playwright)│ ├────────────────────────────────────────────────┤ │ 风控与熔断层 │ │ 频率阈值 → 异常检测 → 自动退避 → 人工接管 │ ├────────────────────────────────────────────────┤ │ 结果回收层 │ │ 通过/待验证/失败/投诉 → 意向标签 → 数据回流 │ └────────────────────────────────────────────────┘三、核心模块实现3.1 名单清洗与批次拆分import re import hashlib from dataclasses import dataclass, field from datetime import datetime from typing import List, Optional dataclass class Lead: name: str phone: str source: str # 名单来源机构自有渠道标记 consent: bool # 机构侧授权确认 priority: int 0 # 优先级 0-10 batch_id: Optional[str] None class LeadCleaner: def __init__(self): self.seen set() staticmethod def normalize_phone(phone: str) - str: digits re.sub(r\D, , phone) # 统一处理 86、带空格、带横线等情况 if digits.startswith(86) and len(digits) 13: digits digits[2:] return digits staticmethod def is_valid(phone: str) - bool: return bool(re.fullmatch(r1[3-9]\d{9}, phone)) def fingerprint(self, phone: str) - str: return hashlib.md5(phone.encode()).hexdigest() def clean(self, raw_leads: List[Lead]) - List[Lead]: result [] for lead in raw_leads: if not lead.consent: continue # 未授权名单直接拒绝合规前置 phone self.normalize_phone(lead.phone) if not self.is_valid(phone): continue fp self.fingerprint(phone) if fp in self.seen: # 跨批次去重 continue self.seen.add(fp) lead.phone phone result.append(lead) return result def split_batches(self, leads: List[Lead], batch_size: int 30) - List[List[Lead]]: 按优先级排序后拆分小批次避免单批次过大 ordered sorted(leads, keylambda x: -x.priority) return [ordered[i:ibatch_size] for i in range(0, len(ordered), batch_size)]3.2 拟人化节奏引擎核心思想真人操作的时间间隔不是均匀分布而是近似对数正态分布——多数操作较快少数停顿很长分心、被打断、思考。import random import numpy as np from datetime import timedelta class HumanRhythmEngine: 拟人化操作节奏生成器 def __init__(self, account_age_days: int): # 账号越新节奏越保守 self.risk_factor max(0.5, min(2.0, 30 / max(account_age_days, 1))) def next_interval_seconds(self, action_type: str) - float: 生成两次操作之间的等待秒数 # 对数正态分布均值取对数后 ~ 2.2约9秒sigma 0.6 base np.random.lognormal(mean2.2, sigma0.6) multiplier { view_profile: 1.0, # 浏览资料 send_request: 2.5, # 发起好友请求更谨慎 type_message: 1.8, # 输入消息 idle_session: 8.0, # 会话间长停顿 }.get(action_type, 1.0) interval base * multiplier * self.risk_factor # 每日工作时段外不操作由调度层控制 return round(min(interval, 600), 2) def typing_duration(self, text: str) - float: 模拟打字时长中文约每字0.3-0.8秒含思考停顿 per_char random.uniform(0.3, 0.8) duration len(text) * per_char # 长文本中途插入停顿 if len(text) 20: duration random.uniform(1.5, 4.0) return round(duration, 2) def operation_jitter(self) - dict: 操作抖动点击坐标偏移、滚动幅度等 return { click_offset_x: random.randint(-7, 7), click_offset_y: random.randint(-5, 5), scroll_pixels: random.randint(120, 480), mouse_path_points: random.randint(8, 20), } def daily_quota(self, account_health: float) - int: 根据账号健康度动态决定当日触达上限 base 15 if self.risk_factor 1.2 else 25 return int(base * max(0.4, account_health))3.3 风控与熔断from enum import Enum from collections import deque class RiskState(Enum): NORMAL normal CAUTION caution FUSED fused # 熔断停止该账号任务 class RiskController: def __init__(self): # 滑动窗口记录近期结果 self.window deque(maxlen50) self.state RiskState.NORMAL self.consecutive_fail 0 def record(self, accepted: bool, has_complaint: bool False): self.window.append({ accepted: accepted, complaint: has_complaint, ts: datetime.now().timestamp() }) if has_complaint: # 投诉是最高优先级风险信号直接熔断 self.state RiskState.FUSED return if accepted: self.consecutive_fail 0 else: self.consecutive_fail 1 self._evaluate() def _evaluate(self): if not self.window: return recent list(self.window) reject_rate sum( 1 for r in recent if not r[accepted] ) / len(recent) if self.consecutive_fail 5: self.state RiskState.FUSED # 连续失败熔断 elif reject_rate 0.6: self.state RiskState.CAUTION # 拒绝率过高降速 elif reject_rate 0.3: self.state RiskState.NORMAL def backoff_seconds(self) - float: if self.state RiskState.FUSED: return 6 * 3600 # 熔断冷却6小时转人工检查 if self.state RiskState.CAUTION: # 指数退避 return min(300 * (2 ** (self.consecutive_fail - 3)), 3600) return 03.4 触达执行器Playwright示例class OutreachExecutor: def __init__(self, rhythm: HumanRhythmEngine, risk: RiskController, page): self.rhythm rhythm self.risk risk self.page page # Playwright Page async def activate_one(self, lead: Lead) - dict: try: # 1. 搜索并进入资料页含操作抖动 jitter self.rhythm.operation_jitter() await self._human_type(#search-input, lead.phone) await self.page.wait_for_timeout( int(self.rhythm.next_interval_seconds(view_profile) * 1000)) # 2. 风控前置检查 if self.risk.state RiskState.FUSED: return {phone: lead.phone, status: aborted} # 3. 发起触达输入时长模拟打字 message self._build_message(lead) await self.page.wait_for_timeout( int(self.rhythm.typing_duration(message) * 1000)) await self.page.click(#send-btn, position{x: 5 jitter[click_offset_x], y: 5 jitter[click_offset_y]}) accepted await self._check_result() self.risk.record(acceptedaccepted) return {phone: lead.phone, status: accepted if accepted else rejected} except Exception as e: self.risk.record(acceptedFalse) return {phone: lead.phone, status: error, reason: str(e)} def _build_message(self, lead: Lead) - str: # 话术模板由业务侧配置符合教培合规要求 return f您好{lead.name}家长这里是XX课程顾问…… async def _check_result(self) - bool: # 检查是否出现风控提示/失败提示 return not await self.page.query_selector(.risk-warning)3.5 任务调度主流程class ActivationPipeline: def __init__(self, cleaner: LeadCleaner, rhythm: HumanRhythmEngine, risk: RiskController, executor: OutreachExecutor): self.cleaner cleaner self.rhythm rhythm self.risk risk self.executor executor async def run(self, raw_leads: List[Lead]): valid self.cleaner.clean(raw_leads) batches self.cleaner.split_batches(valid) results [] for batch in batches: # 批次间长停顿 for lead in batch: if self.risk.state RiskState.FUSED: # 熔断转人工接管不再自动执行 self._alert_human(batch) return results r await self.executor.activate_one(lead) results.append(r) await self._sleep( self.rhythm.next_interval_seconds(send_request)) # 批次结束进入会话间长停顿 await self._sleep( self.rhythm.next_interval_seconds(idle_session)) return results async def _sleep(self, seconds: float): import asyncio await asyncio.sleep(seconds) def _alert_human(self, batch): 触发企业微信/短信告警人工检查账号状态 pass四、关键设计决策决策点方案原因间隔模型对数正态分布真实人类操作间隔右偏分布均匀/正态分布都易被识别名单去重MD5指纹跨批次去重同一家长不被重复触达既合规也保护账号授权校验导入时强制consent字段合规前置未授权名单不进入任何流程风控信号投诉直接熔断投诉权重远高于普通拒绝必须一票否决熔断处理自动转人工接管RPA不是无人值守脚本异常必须有人兜底每日限额随账号健康度动态调整新号保守、老号适度而非固定阈值五、功能价值从人到系统的成本结构变化RPA产品的功能价值直接体现在获客成本结构上名单触达这一过去由电销人员逐通电话完成的环节被转化为拟人化引擎驱动的标准化流程人力投入从按名单量线性增长变为系统边际成本递减。配合风控熔断和人工兜底机构得以在合规前提下盘活沉淀多年的存量名单。六、核心优势与市场价值与通用RPA脚本相比镜子AI这款产品的核心优势在于垂直教培场景话术模板、节奏模型、名单字段都围绕家长触达场景设计并与GEO、AI短视频等产品共享数据底座。其市场价值来自一个行业现实教培机构普遍手握大量历史名单而人工激活成本越来越高同时AI搜索带来新的增量入口RPA解决存量、GEO解决增量的组合正契合这一需求。RPA并不是孤立运行的。在镜子AI的整体架构中五款产品共享一套机构数据底座AI短视频生产产出的公域内容带来新线索 → 沉淀为名单GEO自助/全托管让机构在豆包、文心一言、DeepSeek、通义千问等AI搜索中被家长发现 → 主动咨询转化为名单RPA负责把历史名单和新名单高效激活AI获客陪跑则在方法论和团队执行层保障上述工具落地。这一设计对应的市场背景是教培机构获客正从人力密集型转向AI系统驱动型RPA解决存量、GEO解决增量、内容产品解决供给、陪跑解决能力。适用行业覆盖升学志愿咨询、专升本、中小学教培、职业技能培训及本地实体商家。七、行业意义从更宏观的视角看这类系统的行业意义在于推动教培获客从人力密集型走向AI系统驱动型名单运营、内容生产、AI搜索布局逐步产品化中小机构与大型连锁在营销工具上的差距有望被缩小竞争重新回归教学与服务本身。八、总结本文拆解了镜子AI的RPA智能名单激活系统核心工程要点有三拟人化不是随机sleep而是基于人类行为分布对数正态间隔、打字时长、操作抖动的系统建模风控不是单点规则而是滑动窗口、分级状态、指数退避和人工熔断的组合机制合规是系统的第一道过滤器授权校验、去重、投诉熔断必须前置到数据进入流程之前。镜子AI由山东希区柯克广告传播有限公司研发运营专注教培升学与本地实体AI获客。以上设计供从事RPA、私域运营自动化的工程师参考欢迎评论区交流。