单号机器人是玩具，几十个号一起跑才是生意。客服矩阵、私域矩阵、城市分号，一旦上量，核心矛盾就从"能不能收发"变成"怎么统一调度、怎么不串号、怎么知道哪个号掉了"。WTAPI的appId+instanceId二层模型天然支持多实例，工程上要做的是在网关层把多账号管理做好。这篇讲清四个关键设计。

一、标识体系：appId管租户，instanceId管号

WTAPI用二层标识：appId标识你的业务方（租户），instanceId标识一个具体在线微信号。一个appId下挂几十个instanceId，权限、统计、配额都按这两层切分。后台所有数据表都必须带instance_id，这是多账号不串数据的底线。

二、消息路由：哪号进，哪号回

Webhook回调带instanceId，投递队列时原样带上，回复时强制用同一个实例：

import redis, json, requests
from flask import Flask, request

rds = redis.Redis()
BASE = "https://wx.chuapi.com"
HEADERS = {
    "X-finder-TOKEN": "<平台Token>",
    "Authorization": "Bearer <Bearer>",
    "Content-Type": "application/json",
}
APP_ID = "<appId>"

app = Flask(__name__)

@app.route("/webhook", methods=["POST"])
def webhook():
    d = request.json
    rds.lpush("msg_queue", json.dumps({
        "instanceId": d["instanceId"],
        "fromWxid": d.get("fromWxid"),
        "content": d.get("content", ""),
    }))
    return {"code": "1000"}

def worker():
    while True:
        _, raw = rds.brpop("msg_queue")
        m = json.loads(raw)
        # 回复时用原instanceId，多账号才不会串号
        requests.post(
            f"{BASE}/finder/v2/api/postText",
            headers=HEADERS,
            json={"appId": APP_ID, "instanceId": m["instanceId"],
                  "toWxid": m["fromWxid"],
                  "content": generate_reply(m["content"])},
            timeout=10,
        )

会话状态、去重键、聊天记录全部按 instance_id:from_wxid 复合键隔离，同一个用户在不同号的对话绝不互通。

三、统一收件箱：一张表聚合所有号

CREATE TABLE messages (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  instance_id VARCHAR(64) NOT NULL,
  from_wxid   VARCHAR(64) NOT NULL,
  content TEXT, created_at DATETIME,
  INDEX idx_inst_time (instance_id, created_at)
);

后台按instance_id过滤就是单号视图，不加过滤就是全局收件箱，运营在一个页面处理所有号的会话。

四、健康调度：自动摘除掉线号

多账号最怕某个号悄悄掉线还在被分配任务。建一张实例状态表，定时轮询在线状态：离线的从发送池摘除并告警，恢复后自动加回。发送时若业务没指定号，按轮询或"队列最短"策略选在线实例，把负载和风控压力摊开。

class InstanceManager:
    def __init__(self, instance_ids):
        self.instances = {i: "unknown" for i in instance_ids}
        self._idx = 0

    def pick(self):
        online = [i for i, s in self.instances.items() if s == "online"]
        if not online:
            raise RuntimeError("无可用在线实例")
        inst = online[self._idx % len(online)]  # 轮询
        self._idx += 1
        return inst

    def mark(self, instance_id, status):
        self.instances[instance_id] = status
        if status != "online":
            alert_admin(f"实例掉线: {instance_id}")

把标识、路由、收件箱、健康调度这四件事做进统一后台，几十个号就像一个号一样好管，业务方甚至感知不到多账号的存在。