技术支持 wechatapi.net

初始版本的最小中转器

最早的微信机器人脚本仅验证了消息中转的基本流程，通过简单的HTTP回调接收微信消息，调用OpenClaw CLI处理并返回结果。伪代码如下：

@app.post("/wechat/callback")
async def handle_wechat(request: Request):
    body = await request.body()
    data = json.loads(body)
    text = extract_text(data)
    result = subprocess.run(["openclaw", "agent", "--session-id", "demo", "--message", text], capture_output=True, text=True)
    reply_text("某个wxid", result.stdout)
    return {"status": "ok"}

此版本虽能快速验证可行性，但缺乏会话管理、并发控制、错误处理等关键设计，仅停留在脚本层面。

---

产品化初始化流程

改进方向从提升用户体验入手，通过交互式命令行引导用户完成关键配置：

def interactive_init():
    print("WeChat OpenClaw Gateway 首次初始化")
    api_token = input("请输入 WX_API_TOKEN: ").strip()
    public_url = input("请输入公网回调地址 PUBLIC_URL（不能为空）: ").strip()
    group_trigger = input("请输入群触发词（默认 狗子）: ").strip() or "狗子"
    cfg = DEFAULTS.copy()
    cfg.update({"api_token": api_token, "public_url": public_url, "group_trigger": group_trigger})
    write_config(cfg)

此步骤生成标准化配置文件，使项目从个人脚本转向可分发工具。

---

统一消息模型设计

通过深度解析微信回调数据，构建独立于协议层的消息抽象：

def parse_wechat_payload(data):
    if not isinstance(data, dict):
        return None
    type_name = data.get("TypeName")
    wxid = str(data.get("Wxid", "") or "").strip()
    msg_data = data.get("Data", {})
    # 解析字段并处理群聊/私聊差异
    return {
        "event_type": type_name,
        "wxid": wxid,
        "msg_type": msg_data.get("MsgType"),
        "is_group": from_user.endswith("@chatroom"),
        "actual_text": processed_text
    }

该模型屏蔽了微信协议细节，为多通道接入奠定基础。

---

会话隔离策略

针对群聊和私聊设计不同的会话ID生成规则：

def build_session_id(chat_id: str, sender_wxid: str, is_group: bool, config: dict) -> str:
    if not is_group:
        return f"wechat_dm_{norm(chat_id)}"
    if config["GROUP_SESSION_MODE"] == "per_user":
        return f"wechat_group_{norm(chat_id)}_user_{norm(sender_wxid)}"
    return f"wechat_group_{norm(chat_id)}"

通过norm函数处理特殊字符，确保不同场景下的会话隔离性。

---

并发分片方案

采用会话级哈希分片实现并行处理：

def shard_index_for_session(session_id: str, worker_count: int) -> int:
    h = int(hashlib.md5(session_id.encode("utf-8")).hexdigest(), 16)
    return h % worker_count

将任务投递到对应Worker队列：

worker_queues[shard_idx].put_nowait(task)

此设计平衡了并发性能与消息顺序性需求。

---

消息类型过滤

在入口层明确边界，初期仅处理文本消息：

if msg_type != 1:
    logger.info("📭 暂时只处理文本消息，已忽略 MsgType=%s", msg_type)
    return {"status": "ignored_msg_type"}

避免过早支持多媒体消息带来的复杂度爆炸。

---

性能瓶颈分析

最终发现延迟主要来自OpenClaw CLI的冷启动开销：

subprocess.run(["openclaw", "agent", "--session-id", sid, "--message", text])

即使优化了网关层设计，底层调用方式仍制约整体响应速度。