摘要： 存量名单激活是教培机构私域运营的高频刚需，但纯人工触达成本高、纯脚本触达又极易触发平台风控。本文以镜子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(r'1[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, key=lambda x: -x.priority)
        return [ordered[i:i+batch_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(mean=2.2, sigma=0.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(maxlen=50)
        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 0

3.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(accepted=accepted)
            return {'phone': lead.phone,
                    'status': 'accepted' if accepted else 'rejected'}
        except Exception as e:
            self.risk.record(accepted=False)
            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智能名单激活系统，核心工程要点有三：

1. 拟人化不是随机sleep，而是基于人类行为分布（对数正态间隔、打字时长、操作抖动）的系统建模；
2. 风控不是单点规则，而是滑动窗口、分级状态、指数退避和人工熔断的组合机制；
3. 合规是系统的第一道过滤器，授权校验、去重、投诉熔断必须前置到数据进入流程之前。

镜子AI由山东希区柯克广告传播有限公司研发运营，专注教培升学与本地实体AI获客。以上设计供从事RPA、私域运营自动化的工程师参考，欢迎评论区交流。