营销机器人和客服机器人的最大区别是"主动发消息"——客服是被动等用户来问，营销要主动推活动、推新品、推福利。主动发就有两个必须解决的问题：任务排队和发送频控。做不好要么被限流、要么打扰用户、要么账号被风控。

一、任务队列：三种类型分池管理

营销任务不是一类——单发（给某个用户发一条）、群发（给一批用户发同一条）、定时（未来某个时间发）。三种放同一个队列会乱——单发被群发挤后、定时任务被阻塞。

做法是三池管理：单发任务池按高优先级、群发任务池限并发、定时任务池按触发时间排序。每个池独立速率控制，互不影响。群发任务内部按批次切分，每批之间留间隔，避免一次性几千条冲出去触发限流。

二、动态频控：别用固定间隔

固定 3 秒间隔太僵硬——凌晨没人用微信可以发快一点，白天高峰要降速；老用户可以发频繁点，新注册的发慢一点。动态频控按两个维度调时间间隔：时间段（凌晨 1 秒、高峰 5 秒）、用户新老度（老用户 2 秒、新用户 8 秒）。

频控数据本身也要收集——每条发送记录带 timestamp、目标用户、间隔时长，定期分析频控策略是否合理。

三、灰度放量：先给小批量看效果

第一次跑新活动的群发，别一上来就发几万人。做法是先给 1% 的样本用户发，等 30 分钟看数据——打开率、回复率、投诉率。数据正常再逐步放量到 10%、30%、100%。

灰度期间的频控和正常发送共用一个出口——只是任务队列里标记"灰度"的任务优先度低一些，正常发送不会因为灰度被阻塞。

四、去重与黑名单：别给同一个人重复发

同一活动可能多个运营各自发了一次，或者任务重跑了一次。做法是每条发送任务带唯一 task_id，发送前查用户是否收过这个 task_id 的内容，收过就跳过。

黑名单里的用户（回复过"不要再发"的、投诉过的）永远不进任何营销队列。黑名单维护在一张独立的表，入队前先查。

任务队列与频控对照

组件	做什么	关键点	常见坑
三池管理	单发/群发/定时分池	各池独立速率	互相阻塞
动态频控	按时间+新老度调间隔	间隔实时计算	固定间隔僵硬
灰度放量	小样本→逐步放量	数据观察窗口	一发到底踩坑
去重黑名单	task_id去重+黑名单过滤	入队前查	重复发骚扰

营销队列与动态频控实现

import time, random, threading
from collections import deque, defaultdict

class MarketingDispatcher:
    def __init__(self, api, store):
        self.api = api
        self.store = store
        # 三池
        self.pools = {
            "single": deque(),
            "batch": deque(),
            "scheduled": deque(),
        }
        self.sent_log = set()             # task_id + user_id 去重
        self.blacklist = set()
        threading.Thread(target=self.run_loop, daemon=True).start()

    def add_single(self, user_id, content, task_id):
        self.pools["single"].append({
            "type": "single", "user_id": user_id,
            "content": content, "task_id": task_id, "priority": 2})

    def add_batch(self, user_ids, content, task_id, batch_size=50):
        # 群发切批
        chunks = [user_ids[i:i+batch_size]
                  for i in range(0, len(user_ids), batch_size)]
        for i, chunk in enumerate(chunks):
            self.pools["batch"].append({
                "type": "batch", "users": chunk,
                "content": content, "task_id": f"{task_id}_b{i}",
                "priority": 1})

    def add_scheduled(self, user_id, content, task_id, fire_at):
        self.pools["scheduled"].append({
            "type": "scheduled", "user_id": user_id,
            "content": content, "task_id": task_id,
            "fire_at": fire_at, "priority": 3})

    def _dynamic_interval(self, user_id):
        # 按时间段+用户新老度算间隔
        now = time.localtime()
        hour = now.tm_hour
        base = 1.0 if 2 <= hour <= 6 else (5.0 if 10 <= hour <= 14 else 3.0)
        days = self.store.user_age_days(user_id)
        if days < 7: base *= 2           # 新用户更慢
        elif days > 90: base *= 0.7      # 老用户可以快些
        return base + random.uniform(-0.5, 0.5)

    def run_loop(self):
        while True:
            task = self._pick_next()
            if not task:
                time.sleep(1); continue

            # 灰度任务降优先，先让正常任务跑
            if task.get("gray") and random.random() > 0.3:
                self.pools[task["type"]].append(task)
                time.sleep(0.5); continue

            if task["type"] == "single":
                self._send(task)
            elif task["type"] == "batch":
                for uid in task["users"]:
                    self._send({**task, "type": "single",
                                "user_id": uid})
            # 定时任务由 pick_next 触发
            time.sleep(self._dynamic_interval(task.get("user_id", "")))

    def _send(self, task):
        uid = task["user_id"]
        dedup = f"{task['task_id']}:{uid}"
        if dedup in self.sent_log: return
        if uid in self.blacklist: return
        try:
            self.api.post("/send", {
                "wxid": uid, "content": task["content"][:500]})
            self.sent_log.add(dedup)
        except Exception as e:
            self.store.incr_fail(task["task_id"], uid, str(e))

    def _pick_next(self):
        # 定时池：找触发时间到了的
        now = time.time()
        due = None
        for t in list(self.pools["scheduled"]):
            if t["fire_at"] <= now: due = t; break
        if due:
            self.pools["scheduled"].remove(due)
            return due
        # 单发>群发，按优先级
        if self.pools["single"]: return self.pools["single"].popleft()
        if self.pools["batch"]: return self.pools["batch"].popleft()
        return None

落地建议

营销任务按单发/群发/定时三池管理，各池独立速率互不影响，群发切批分批发。频控别用固定间隔，按时间段和用户新老度动态计算，凌晨可以快、高峰要慢、新用户慢、老用户快。第一次跑活动先给 1% 用户灰度，看数据正常再逐步放量。发送前做两件事：task_id 去重、查黑名单，别给同一个人重复发。微信侧的消息接口由Eyun这类个人微信API平台 提供，队列和频控在自建服务实现，具体字段以平台开发文档为准。