全网首个Python端的ClawBot原理大揭秘|在微信里对话分析组学数据

微信今天的更新，添加了ClawBot，于是我们根据官方提供的插件协议，使用Python重写了协议连接策略，并在OmicClaw和OmicVerse中实现了这一过程，感兴趣的可以往下翻翻：

在之前的内容中，很多人还不知道OmicClaw的具体使用方法，在本期的教程中，我们将带大家从0开始，一步一步地让大家在微信里就可以分析组学数据。

Steorra

写在前面，如果你对我们是如何基于Python实现的微信ClawBot连接，请你直接翻到本教程尾部（6. 基于Python实现的ClawBot连接的整体思路）

如果你对OmicClaw的使用感兴趣，你可以从头开始阅读（当然，你也可以把我们两个仓库的源码让Agent去阅读然后提取相关功能也是可以的）

在教程开始前，值得一提的是，在OmicClaw前天发行的版本里，我们将可视化窗口，代码编辑器Colab，Agent，网关全部统一成了一个后端，意味着无论在哪个环境，你所分析的都是同一个数据，这将大幅提升分析的便利度。

1. 安装OmicClaw

其实安装OmicClaw特别简单，只需要使用pip直接安装即可，但是OmicClaw的依赖OmicVerse包不太好装，需要先提前安装好torch和torch_geometric。

实际上对于Linux和MacOS用户，只需要运行下列命令，即可自动引导式地完成OmicVerse的安装了

curl -sSL omicverse.com/install | bash -s

安装OmicClaw特别简单，在前面的引导式脚本里其实就有OmicClaw作为可选包是否进行安装的判断，当然你也可以使用pip。

不过需要注意的是，如果你想使用微信进行对话，你需要确保omicverse的版本大于等于2.0.3,omicclaw的版本大于等于2.0.4

教程链接：https://omicclaw.readthedocs.io/zh-cn/latest/zh/index.html

pip install -U omicclaw

或者使用uv

pip install uv
uv pip install omicclaw

2. 配置及运行OmicClaw

你只需要在终端里输入omicclaw, 那么就会全自动运行omicclaw的网关环境，一键同时加载可视化窗口，Agent系统，代码编辑器以及多对话方式网关管理。

可以使用引导式配置网关模式

omicclaw --setup

当然也可以直接在网页里配置，只不过在网页里配置可能无法完成codex的oauth验证配置

omicclaw

3. 注册OmicClaw账户

目前OmicClaw还处于内测阶段，因此要求新用户在使用的时候都需要注册并登录，不过有趣的是，由于我的项目开源了，所以觉得麻烦的也可以自己fork下来把相关判断删掉即可注册登录。

账户登录不收集任何数据信息，大家可以在源码里看见，仅作为一个用户凭证，以便后续福利的发送。

3. 配置大模型的api-key

我们需要一个兼容openai的端口的api-key来进行配置

4. 微信扫码登录

我们点击通道，然后选择微信channel，然后点击获取登录二维码，即可获得二维码页面

我们扫码直接登录即可

需要注意的是，我们首先要更新微信到8.0.70版本，然后在插件里选择clawbot

5. 分析演示

由于我们的数据共用一个后端，那么我们可以在可视化窗口随便上传一个AnnData数据，在这里我们上传的是seqfish的数据。

我们可以看到数据有19416个细胞，315个基因，我们切到clawbot窗口，跟他📱招呼，你会发现它已经能自动识别在内存里的数据了。

然后我们让他可视化一下数据内容。

可以看到图片等内容都可以正常发送。此外，与别的Claw不同的是，作为组学设计的Claw，我们还可以回到网页里，继续对数据进行分析。

6. 基于Python的ClawBot连接

• OmicVerse仓库：https://github.com/Starlitnightly/omicverse
• OmicClaw仓库：https://github.com/Starlitnightly/omicclaw

微信企业号提供了一套叫iLink Bot的 HTTP API（域名：ilinkai.weixin.qq.com），允许第三方应用以"机器人"身份：

• 用长轮询（Long Polling）实时接收用户发来的消息
• 调用sendmessage回复文字或图片

我们的实现分成三层：

用户（微信）
    ↕  iLink HTTP API
WeChatApiClient          ← 纯 HTTP 封装（认证 / 收发消息 / 图片上传）
    ↕
WeChatJarvisBot          ← 业务逻辑（解析命令 / 调度 AI 分析任务）
    ↕
AgentBridge              ← 驱动 OmicVerse AI Agent 执行代码、收割图表

所有核心代码都在：

omicverse/jarvis/channels/wechat.py

6.1：扫码登录（获取 Bot Token）

Bot Token 是每次调用 iLink API 的身份凭证。获取流程是扫二维码——这部分逻辑在OmicClaw 网关里。

相关文件：

• 后端：omicclaw/gateway/channel_config_routes.py
• 前端：omicclaw/static/js/sc-ui.js

后端：生成二维码 + 轮询状态

# channel_config_routes.py（简化）

# 1. 向 iLink 请求二维码内容（是一个 URL 字符串，不是图片）
GET https://ilinkai.weixin.qq.com/ilink/bot/get_bot_qrcode?bot_type=3
→ 返回 qrcode_img_content（一个 URL）

# 2. 用 Python qrcode 库把这个 URL 编码成真正的 PNG 图片
import qrcode, io, base64
img = qrcode.make(qrcode_img_content)
buf = io.BytesIO()
img.save(buf, format="PNG")
data_uri = "data:image/png;base64," + base64.b64encode(buf.getvalue()).decode()
# 把 data_uri 发给前端直接展示

# 3. 前端每隔 2 秒轮询一次扫码状态
GET https://ilinkai.weixin.qq.com/ilink/bot/get_qrcode_status?qrcode=<token>
→ status: "wait" / "confirmed" / "expired"

注意：GET 请求不能带Content-Length头，否则服务器会一直等待请求体，导致超时。只有 POST 请求才加这个头。

一旦状态变为confirmed，前端自动把返回的 Bot Token 填入配置框并关闭弹窗。

6.2：HTTP 请求认证格式

iLink 的每个 POST 请求都需要以下四个固定请求头：

# wechat.py → WeChatApiClient._headers()

headers = {
"Content-Type":      "application/json",
"AuthorizationType": "ilink_bot_token",
"Authorization":     f"Bearer {self._token}",
"X-WECHAT-UIN":      base64.b64encode(str(random_uint32).encode()).decode(),
"Content-Length":    str(len(body.encode("utf-8"))),
}

X-WECHAT-UIN是一个随机 32 位无符号整数转字符串再 Base64 编码的值，每次请求都不同，起防重放作用。

6.3：接收消息（长轮询）

iLink 用长轮询代替 WebSocket。原理是：我方发一个 POST，服务器最多挂 35 秒等有新消息再返回，没消息就返回空列表，我们再立刻发下一个请求。

# wechat.py → WeChatApiClient.get_updates()

POST /ilink/bot/getupdates
{
"get_updates_buf": "<上次返回的游标>",   # 空字符串表示从头开始
"base_info": {"channel_version": "jarvis"}
}

# 返回示例
{
"get_updates_buf": "<新游标>",   # 下次请求带上这个，保证消息不重复
"msgs": [
        {
"from_user_id": "xxx",
"context_token": "yyy",   # 回复时必须带上，关联对话
"item_list": [
                {"type": 1, "text_item": {"text": "帮我画 UMAP"}}
            ]
        }
    ]
}

游标（get_updates_buf）会持久化到本地文件，这样重启服务后不会重复处理旧消息：

# wechat.py → WeChatJarvisBot._cursor_store_path()
# 文件路径：~/.ovjarvis/wechat/<token_hash_前12位>_cursor.json

轮询主循环在WeChatJarvisBot._poll_loop()，连续失败 3 次才报错退出，普通超时直接忽略继续轮询。

6.4：发送文字回复

# wechat.py → WeChatApiClient.send_text()

POST /ilink/bot/sendmessage
{
"msg": {
"to_user_id":    "<用户 ID>",
"client_id":     "<随机 UUID，幂等键>",
"message_type":  2,
"message_state": 2,
"context_token": "<收消息时拿到的 token>",   # 必须原样带回
"item_list": [
            {"type": 1, "text_item": {"text": "这是回复内容"}}
        ]
    },
"base_info": {"channel_version": "jarvis"}
}

单条消息最多 3800 字，超长文本会自动按段落拆分后分批发送（_text_chunks()）。

6.5：发送图片（最复杂的部分）

直接发图片字节是不行的。iLink 要求先把图片加密上传到腾讯 CDN，再把 CDN 返回的下载凭证放进消息体。完整流程如下：

本地 PNG
    │
    ├─ 1. 用 AES-128-ECB 加密（PKCS7 填充）
    │
    ├─ 2. POST /ilink/bot/getuploadurl
    │      → 拿到 upload_param（上传凭证）
    │
    ├─ 3. POST https://novac2c.cdn.weixin.qq.com/c2c/upload
    │       ?encrypted_query_param=<upload_param>
    │       &filekey=<随机16字节hex>
    │      Body: 加密后的字节流
    │      → 响应头 x-encrypted-param = download_param（下载凭证）
    │
    └─ 4. POST /ilink/bot/sendmessage
           item_list[type=2].image_item.media = {
               encrypt_query_param: download_param,
               aes_key: base64(hex字符串的UTF-8字节),  ← 注意：不是 base64(原始密钥字节)
               encrypt_type: 1
           }

代码实现：

# wechat.py → WeChatApiClient.upload_image() + send_image()

# 加密
aeskey = secrets.token_bytes(16)          # 随机 16 字节密钥
ciphertext = _encrypt_aes_ecb(plaintext, aeskey)

# 关键：aes_key 字段的编码方式
aeskey_hex = aeskey.hex()                 # e.g. "a3f1..."
aes_key_b64 = base64.b64encode(
    aeskey_hex.encode("ascii")            # 先把 hex 字符串转 bytes
).decode("ascii")                         # 再 base64 → 这才是发给 API 的值

容易踩坑：很多人会直接base64(raw_key_bytes)，但 iLink 要求的是base64(hex字符串的bytes)，对应 TypeScript 里的Buffer.from(aeskey.toString('hex')).toString('base64')。

图片上传封装在WeChatApiClient.upload_image()，发送封装在send_image()，上层调用在WeChatJarvisBot._send_figure()/_send_figures()。

6.6：消息处理与 AI 分析

收到消息后，WeChatJarvisBot._on_message()会：

1. 过滤群消息（当前只处理私聊）、非白名单用户、bot 自身消息
2. 解析命令：/help/status/reset/cancel/model
3. 普通消息：交给 AI Agent 分析

AI 分析是异步运行的，如果上一个任务还没结束，新消息会进入队列等待：

# wechat.py → WeChatJarvisBot._on_message()

if running andnot running.task.done():
# 加入队列，等当前分析完成后继续
    self._pending.setdefault(route, []).append(text)

AI 分析完成后，图表会自动上传 CDN 并发送为图片消息，文字摘要单独发出。

6.7 图表自动发送原理

Agent 执行代码后，图表通过以下路径流转：

execute_code 运行 matplotlib 代码
    │
    ├─ InProcessKernelExecutor 捕获新增 figure number
    │   → 存为 base64 PNG，写入 current_session["notebook"]
    │
    └─ AgentBridge._harvest_figures() 从 notebook 中读取
        → result.figures: List[bytes]
        → WeChatJarvisBot._send_figures()
        → upload_image() + send_image()

为防止重复发送（同一张图既在内存里被捕获、也被plt.savefig()写到磁盘），使用 SHA-256 内容哈希去重：

# agent_bridge.py → _content_key()
def_content_key(data: bytes) -> str:
return"sha:" + hashlib.sha256(data[:1024]).hexdigest()[:20]

6.8 模块关系一览

6.9 常见问题

Q：为什么 GET 请求会超时 30 秒？

GET 接口（获取二维码、查询扫码状态）不能带Content-Length头。如果带了，服务器会等待一个从未到来的请求体，直到超时。

Q：为什么图片发送了两次？

Agent 代码里plt.savefig("x.png")不关闭 figure 时，内存里的 figure 会被 kernel executor 捕获一次，磁盘上的文件还会被文件扫描器捕获一次。通过对 PNG 内容计算 SHA-256 前缀做跨路径去重来解决（见agent_bridge.py → _content_key()）。

Q：消息重启后会不会重复处理？

不会。游标get_updates_buf会持久化到~/.ovjarvis/wechat/<hash>_cursor.json，重启后继续从上次的位置读取。

Q：Agent 需要特别设置才能正确发图片吗？

需要在请求里注入 channel hint，告诉 Agent：用plt.show()或plt.savefig()生成图表即可，系统会自动发送，不要自己生成下载链接。这段 prompt 定义在wechat.py的_WECHAT_CHANNEL_HINT常量中。

7. 交流群

此前的交流比较活跃，但是小助手也是我本人，分身乏术，所以又新建了一个群，希望大家不要重复加群，每个群的消息是一样同步的，如果群满了也请添加小助手FernandoZeng。