可扩展的多平台消息处理框架
Core Architecture
事件驱动架构
所有交互通过事件传递,从平台接收到模块处理形成完整的异步事件流
@message.on_message()
async def handler(event):
text = event.get_text()
await event.reply(f"收到: {text}")
OneBot12 · Multi-Platform
多平台适配
一套代码同时对接云湖、Telegram、OneBot11/12、邮件等平台,适配器自动处理协议差异
@command("hello")
async def hello(event):
# 云湖 · Telegram · QQ · 邮件...
# 同一份代码,所有平台运行
platform = event.get_platform()
await event.reply(f"Hello from {platform}!")
Messaging Pipeline
链式发送与事件管线
SendDSL 链式发送、同优先级并行处理与 done(claim, stop) 链路控制,构成可组合、可覆写的异步事件管线
# 规则组合:重试 + 超时 + 回调
await (adapter.Send.To("user", "123")
.Retry(3).Timeout(10)
.Hook(on_success)
.OnError(on_error)
.Text("可靠发送"))
# Build 批量构建:统一执行
await (adapter.Send.To("group", "G001")
.Build()
.Text("第一句").Image("pic.jpg")
.send_all())
Interactive
交互式对话
内置确认、选择菜单、表单收集、多轮对话等交互原语,轻松构建复杂交互流程
欢迎使用问卷调查!请选择你喜欢的颜色:
绿色
你选择了绿色,确认提交吗?
确认
取消
确认
感谢参与!你的选择已记录。
Modular
模块化与声明式开发
模块以独立 PyPI 包分发,支持懒加载、热更新与依赖拓扑排序;ConfigClass 声明式配置实时热读,get_meta 元信息接入 help 与仪表盘
# pip install ErisPulse-MyModule
# or: epsdk install MyModule
class Main(BaseModule):
@staticmethod
def get_load_strategy():
return ModuleLoadStrategy(
lazy_load=True, # 首次访问时才初始化
priority=10 # 数值越大越先加载
)
async def on_load(self, event):
self.logger.info("模块已就绪")
Dashboard
仪表盘
实时监控适配器状态、模块加载进度与 Bot 在线情况,全局运行状态一目了然
运行中
99.8% uptime
适配器
3 online
云湖
Telegram
QQ
模块
7 loaded
WeatherQuery
AIChat
实时事件
Live
12:04:32yunhu/Y1001 → message.group
12:04:33module.load → WeatherQuery
12:04:34所有模块加载完毕
Scope & Lifecycle
统一作用域与生命周期
从模块、身份、出站三个维度统一控制可用范围;在框架运行的每个关键阶段插入自定义逻辑,模块卸载时注册的资源自动归属回收
[ErisPulse.scope.platforms.onebot11]
modules = ["Chat", "Tool*"] # 白名单:精确/glob/正则
blocked = ["re:^Danger"]
[ErisPulse.scope.identity.users.onebot11]
allow = ["u_admin"]
deny = ["spam_*"]
[ErisPulse.scope.actions.MyModule]
send = { deny = true } # 全禁发送
api = { allow = ["get_*"] } # 仅允许查询类 API
AI Powered
AI 辅助开发
提供完整的开发文档与规范,让 AI 直接生成可用模块,支持 Vibe Coding 工作流
# 将 ErisPulse 物料投喂给 AI
# 即可直接生成可用模块
class AIModule(BaseModule):
async def on_load(self):
self.logger.info("AI 生成就绪")