预计阅读时间：12 分钟

缘起

写机器人这件事，做过的人都知道一个痛点：事件源太多。

QQ 机器人、B站动态推送、直播弹幕监控……每个数据源都有自己的协议、自己的 SDK、自己的连接方式。每次接入一个新的数据源，都意味着写一堆样板代码：初始化连接、管理生命周期、消息循环、异常处理……而这些代码，无非是把数据从 A 搬到 B，中间夹着一点点业务逻辑。

能不能有一个框架，把"数据生产"和"数据处理"彻底解耦？让开发者只关心"收到了什么"和"要做什么"，而不必操心"怎么收到的"？

这就是 ButterBot 的诞生原因。

项目概况

ButterBot 是一个基于 Python 3.12+ 的事件驱动机器人框架，目前已迭代到 3.x 版本。项目开源在 GitHub 上，使用 GPL-3.0 协议。

核心思路：将数据输入抽象为事件源（Source），数据处理抽象为订阅者（Subscriber），通过事件总线（EventBus）连接双方。

这条主线贯穿了整个框架的设计。框架本身已经内置了 NapCat QQ 和 Bilibili 的事件源，开箱即用。

注意分支策略：项目的开发主要在 dev_main 分支上进行，main 分支通常会落后于 dev_main。如果你想尝鲜最新功能，请切换到 dev_main 分支；main 分支只在较为稳定的版本节点上合并更新。

安装

ButterBot 已发布到 PyPI。安装包名是 butterbot-python，Python 代码里的导入名是 butterbot。

如果你使用 pip：

pip install butterbot-python

如果你使用 uv：

uv add butterbot-python

安装完成后可以用下面的命令验证导入：

python -c "import butterbot; print(butterbot.__version__)"

架构一览

项目的模块结构非常清晰：

butterbot/
├── app/               # 用户入口
├── core/              # 框架核心
│   ├── event/         #   事件系统
│   ├── source/        #   事件源基类
│   ├── api/           #   API 基类
│   ├── data/          #   数据模型基类
│   ├── types/         #   枚举标签
│   ├── context/       #   依赖注入容器
│   └── filter/        #   过滤器
├── sources/           # 内置事件源实现
│   ├── bilibili/      #   B站（动态/直播/弹幕）
│   └── napcat/        #   QQ（OneBot 协议）
└── utils/             # 工具模块

用户只需要从 app/ 层导入即可，实现细节全部隐藏在 core/ 中。

核心设计模式

1. 事件生命周期：Source → EventBus → Subscriber

这是框架的核心路径，只有三步：

Source 产生数据 → EventBus.publish(uuid, Event)
    → SubscriberGroup 查找 uuid 对应的 Subscriber 列表
        → 匹配 event.status 与 subscriber.status_filter
            → asyncio.create_task(subscriber.callback(event)) 异步执行

每一步都设计得尽量精简。

2. 事件源（Source）：只管生产

事件源只有一个职责——产生数据。基类 BaseSource 只定义了三个抽象接口：start()、stop()，以及由框架自动调用的 bind()。

class MySource(BaseSource):
    async def start(self):
        # 从 API 拉数据、监听 WebSocket、读文件……随你
        # 通过 self.ctx.bus.publish(self.uuid, event) 发布事件
        ...

    async def stop(self):
        # 清理：取消 Task，关闭连接
        ...

事件源不关心数据被谁消费、消费了多少次。它只负责把数据包装成 Event 并发布。

3. 事件总线（EventBus）：只管分发

EventBus 管理着订阅者的注册和事件的发布，内部维护了一个 SubscriberGroup（按 Source UUID 索引的订阅者字典）。

发布时，EventBus 会根据发布者的 uuid 找到对应的 Subscriber 列表，逐个用 BaseType.matches() 判断状态是否匹配，匹配则通过 asyncio.create_task 异步执行回调。

class Event(Generic[BaseDataT]):
    data: BaseDataT       # 强类型数据
    status: BaseType      # 状态标签，如 "dynamic.new"
    id: str               # 唯一标识

事件本身只是一个简单的 @dataclass，携带数据、状态和标识符。

4. 订阅者（Subscriber）：只管消费

订阅者是一个 async def 函数，接收 Event 参数，返回 None。框架不限制你在回调里做什么——写日志、发消息、调 API、写数据库，全凭你。

@app.subscribe(source.uuid, DynamicType.NEW)
async def handle_new_dynamic(event: Event[DynamicData]):
    print(f"{event.data.author.name} 发布了新动态！")

这是整个框架中最直观的一层：看到 Event，做你想做的事。

5. 类型匹配（BaseType）：通配符支持的标签系统

框架提供了一套 BaseType(str, Enum)，使用 scope.state 格式（如 dynamic.new、live.open、danmaku.msg）。支持通配符订阅：

DynamicType.ALL    # "dynamic.all"  —— 匹配任何动态事件
DynamicType.NEW    # "dynamic.new"  —— 仅匹配新动态

匹配逻辑简单清晰：同类型的枚举，scope 相同，state 是 "all" 或精确匹配。

6. 数据模型分发（Discriminator）：自动路由到正确子类

这是框架中的一个亮点。基于 pydantic 和自定义元类 MetaDataModel，实现了多层 discriminator 自动分发：

class NapcatEvent(BaseDataModel):
    discriminator_field = "post_type"  # 分发依据的字段

    class MessageEvent(NapcatEvent):
        discriminator_value = "message"     # 当 post_type=="message" 时，自动路由到这里

    class NoticeEvent(NapcatEvent):
        discriminator_value = "notice"

以 NapcatEvent 为例，通过 NapcatEvent.from_dict(raw_dict) 即可根据 post_type 字段自动构建对应子类。多层嵌套也支持——MessageEvent 的子类还可以根据 message_type 继续分发到 GroupMessageEvent 和 PrivateMessageEvent。这套机制在对接动态类型的外部 API 时极其好用。

7. 过滤器组合（Filter）：& 和 | 操作

BaseFilter 支持通过 & 和 | 组合多个过滤条件：

combined = filter_a & filter_b  # 两者都通过
combined = filter_a | filter_b  # 任一通过

可以链式调用，形成一个灵活的过滤链。

8. API 单例管理（APIContext）：按需懒加载

每个 API 实例（如 BilibiliApi）按 (类型, 配置键) 缓存为单例，source 中只需 self.ctx.api_ctx.get(BilibiliApi, config_key) 即可获取。线程安全，懒加载，避免重复创建连接的开销。

完整的使用示例

以下是一个完整的 B站动态 + 直播监听示例：

import asyncio
from butterbot.app import BotApp, Event
from butterbot.sources.bilibili import (
    BiliDynamicSource, BiliLiveSource,
    DynamicType, LiveType,
)
from butterbot.sources.bilibili.data import DynamicData, LiveRoomData

# 创建应用（自动从 config.yaml 加载配置）
app = BotApp()

# 注册事件源
dynamic_source = app.add_source(
    BiliDynamicSource, watch_targets=[1802011210], poll_interval=100
)
live_source = app.add_source(
    BiliLiveSource, watch_targets=[22758221], poll_interval=100
)

# 订阅新动态
@app.subscribe(dynamic_source.uuid, DynamicType.NEW)
async def on_new_dynamic(event: Event[DynamicData]):
    print(f"[新动态] {event.data.author.name}: {event.data.text}")

# 订阅开播
@app.subscribe(live_source.uuid, LiveType.OPEN)
async def on_live_open(event: Event[LiveRoomData]):
    print(f"[开播] {event.data.anchor_info.name} 开播了！")

# 启动
async with app:
    await asyncio.sleep(300)

就这么简单——一行代码注册事件源，一个装饰器订阅事件，剩下的全交给框架。不需要手动管理轮询线程、不需要处理重连逻辑、不需要操心线程安全问题。

三个内置事件源

B站动态（轮询）

监控 UP 主的动态发布，通过 DataPair 模式比对新旧数据来判断是 NEW、DELETED 还是 NULL。支持自定义轮询间隔。

B站直播（轮询）

监控直播间状态变化，可检测 ONLINE、OFFLINE、OPEN（开播）、CLOSE（下播）四种状态。

B站弹幕（WebSocket）

基于 bilibili-api 的 WebSocket 实时接收弹幕消息，延迟低。

NapCat QQ（WebSocket）

通过 NapCatQQ 的 OneBot 协议接入 QQ Bot，支持群消息、私聊消息、通知事件、请求事件和元事件（心跳/生命周期）。

事件数据通过多层 discriminator 自动分发——从 post_type 到 message_type，无需手动解析 JSON 判断类型。

何为"万物皆可 API/SDK"？

框架将每个平台的数据访问封装为 API 层，通过 BaseApi 基类和 APIContext 的单例管理，外部开发者无需关心 HTTP 请求、认证、Cookie 管理等细节。需要接入一个新平台时，只需编写一套 Api → Data → Type → Source 即可。

你甚至可以编写一个同时监听 B站 和 QQ 的 Bot，在 QQ 群里转发 B站 开播通知——不同事件源之间完全解耦，通过同一个 EventBus 各自运作。

如何参与

开发过程中最期待的就是有人适配新的事件源后提交 PR。目前框架的结构已经很好扩展了，按照 sources/ 下的分层模式：

new_platform/
├── api/ ...          # API 封装
├── data/ ...         # 数据模型（使用 discriminator 自动分发）
├── types/ ...        # 状态枚举
└── source/ ...       # 事件源实现

写一套下来大概几百行代码，却能让整个生态受益。

小结

从设计哲学上看，ButterBot 做的其实不是传统的"机器人框架"，而是一个通用的、事件驱动的、面向事件源的生产-消费框架。只不过恰好第一个、第二个事件源都跟机器人相关——但也正因为这种抽象，它其实可以用于任何"收集-处理"场景。

得益于底层基于 asyncio 的事件循环和异步架构，你可以轻松地将它嵌入 Web 服务（如 FastAPI、Sanic 等异步框架）中运行。例如，在 Web 后端启动时一并启动 BotApp，让机器人事件处理和 HTTP 接口服务于同一个进程——B站开播通知直接推送到前端 WebSocket、QQ 群消息指令触发数据库操作、直播弹幕实时写入监控面板。框架不挑运行环境，能跑 asyncio 的地方就能跑 ButterBot。

如果你也在写机器人、做数据采集、或者需要一个简单的发布-订阅框架，欢迎来看看：

• GitHub: github.com/GEYUANwuqi/ButterBot
• PyPI 安装包名：butterbot-python
• Python 导入名：butterbot
• 仓库开发：uv sync --locked --dev
• 开源协议：GPL-3.0

本文由 Wuqi 原创，转载请注明出处。