预计阅读时间：13 分钟

缘起

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

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

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

这就是 ButterBot 的诞生原因。

项目概况

ButterBot 是一个基于 Python 3.12+ 的事件驱动机器人框架。当前 3.1.0 已作为正式稳定版发布到 PyPI，项目开源在 GitHub 上，使用 GPL-3.0 协议。

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

这条主线贯穿了整个框架的设计。框架内置 NapCat QQ 和 Bilibili 适配，并提供稳定的应用、事件源、API、数据模型、类型和 Filter 扩展契约。

分支策略：dev_main 用于日常开发；main 承载已验证的稳定版本与公开文档。GitHub Pages 只从 main 构建，因此开发中的文档不会进入公共站点。正式文档见 https://geyuanwuqi.github.io/ButterBot/。

安装

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

如果你使用 pip：

python -m pip install butterbot-python

如果你使用 uv：

uv add butterbot-python

基础包不安装平台适配的网络依赖。需要接入 NapCat、Bilibili，或同时使用两者时，按需安装 extra：

python -m pip install "butterbot-python[napcat]"
python -m pip install "butterbot-python[bilibili]"
python -m pip install "butterbot-python[all]"

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

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() 的状态转换和失败回滚；自定义 Source 只实现 on_start()、on_stop()，并由框架在启动时自动调用 bind() 注入上下文。

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

    async def on_stop(self) -> None:
        # 清理：取消 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 相同，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) 即可获取，避免在同一配置下重复创建连接。

完整的使用示例

以下示例使用 YAML 创建 B站动态和直播状态 Source，再在应用中注册处理器。先安装 butterbot-python[bilibili]，并在项目根目录创建 config.yaml：

sources:
  bili_account:
    source_name: bilibili
    kwarg:
      BiliDynamicSource:
        watch_targets: [621240130]
        poll_interval: 60
      BiliLiveSource:
        watch_targets: [26498147]
        poll_interval: 30
    sessdata: "${BILI_SESSDATA:-}"
    bili_jct: "${BILI_JCT:-}"
    buvid3: "${BILI_BUVID3:-}"

然后编写应用：

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 创建应用和 Source
app = BotApp()

dynamic_source = app.get_source(BiliDynamicSource, "bili_account")
live_source = app.get_source(BiliLiveSource, "bili_account")
assert dynamic_source is not None
assert live_source is not None

# 订阅新动态
@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} 开播了！")

# 启动并在 SIGINT/SIGTERM 时正常关闭
app.run()

就这么简单——配置声明事件源，装饰器注册订阅，app.run() 负责应用生命周期。你仍应为自定义 Source 管理自己创建的任务和连接；框架会在关闭时停止 Source、排空回调并释放 API 资源。

三个内置事件源

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 基类和 API 容器管理实例。接入新平台通常需要实现 Api → Data → Type → Source，并在应用侧显式装配或注册对应的配置工厂。

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

如何参与

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

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

优先通过公开的 butterbot.app 和 butterbot.core 门面导入契约；深层实现路径不属于稳定 API。稳定 API 在 3.x 内遵守语义化版本兼容承诺，详细边界见稳定性说明。

小结

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

得益于底层基于 asyncio 的事件循环和异步架构，可以将它嵌入 FastAPI、Sanic 等异步 Web 服务。例如，在 Web 后端的生命周期中启动 BotApp，让机器人事件处理和 HTTP 接口运行在同一进程：B站开播通知推送到前端 WebSocket、QQ 群消息指令触发数据库操作、直播弹幕实时写入监控面板。嵌入宿主时应使用 await app.start() 或 async with app，并避免由库重复安装进程级 signal handler。

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

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

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