手把手带你写个 ErisPulse 机器人

手把手带你写个 ErisPulse 机器人

这篇文章,我会从零开始,一步步带你写出一个能跑、能聊、能收集用户信息的 ErisPulse 机器人。不会贴一堆文档链接让你自己看,每一行代码我都会解释它干了什么。跟着走,出发吧。

框架的安装

框架安装的前提是安装 uv 或者 Python 3.10+,我这里默认你们都有这种环境(二选一即可)。

 1# 使用 uv 创建虚拟环境(python 3.12)
 2uv venv --python 3.12
 3
 4# 如果你没有 uv 这个工具链,并且在安装 Python 大于 3.10 的情况下,也可以直接开始
 5# python -m venv .venv
 6
 7# 激活虚拟环境
 8# windows: .venv\Scripts\activate
 9source .venv/bin/activate
10
11# 安装框架
12pip install ErisPulse

创建虚拟环境后如图:

firstbot-venv

安装这里我使用了 uv,你也可以按照上方的 pip install ErisPulse 直接安装:

firstbot-install

框架安装完成,接下来初始化一个基础机器人目录:

1epsdk init -q -n my_bot

firstbot-init

这里创建了一个名为 my_bot 的机器人目录,目录结构如下:

firstbot-init-dir

就两个文件,config/config.toml 是配置文件,main.py 是你写逻辑的地方。

安装沙箱适配器

正式接平台之前,我们先用 Sandbox 适配器来调试。它带了一个简单的 Web 界面,不需要配真实平台的 token 就能跑。

1epsdk install

运行后进入一个交互式界面:

firstbot-install-cmd

选择 1. 适配器,然后找到 sandbox

firstbot-install-sandbox

我这里是第 7 个,选 7 回车安装。你的序号可能不一样,找到 sandbox 就行。

装完之后的基础设施就齐全了,可以开始写代码了。

第一个命令:ping!

打开 main.py,你会看到自动生成的代码大概长这样:

 1"""
 2my_bot 主程序
 3
 4这是 ErisPulse 自动生成的主程序文件
 5您可以根据需要修改此文件
 6"""
 7
 8import asyncio
 9from ErisPulse import sdk
10
11async def main():
12    await sdk.run(keep_running=True)
13
14if __name__ == "__main__":
15    asyncio.run(main())

不用删任何东西,在 from ErisPulse import sdk 下面加一行导入,然后在 async def main() 上面加你的命令就行:

 1"""
 2my_bot 主程序
 3
 4这是 ErisPulse 自动生成的主程序文件
 5您可以根据需要修改此文件
 6"""
 7
 8import asyncio
 9from ErisPulse import sdk
10from ErisPulse.Core.Event import command
11
12@command("ping")
13async def ping(event):
14    await event.reply("pong!")
15
16async def main():
17    await sdk.run(keep_running=True)
18
19if __name__ == "__main__":
20    asyncio.run(main())

sdk.run(keep_running=True) 那行是框架帮你处理好的启动逻辑,初始化、启动适配器、保持运行全包了。你只管在上面用装饰器写命令,不用操心底层怎么跑的。

跑起来:

1epsdk run main.py

firstbot-run

打开 Sandbox 的 Web 调试界面(默认 http://localhost:8000/sandbox),输入 /ping,机器人回你一句 pong!

firstbot-ping

就五行代码,框架自己搞定命令解析和路由。你只关心收到命令后回什么就行了。

让机器人记住事

光回 pong 没意思。改一下 ping,让它记住你 ping 了多少次:

1@command("ping")
2async def ping(event):
3    count = sdk.storage.get("ping_count", 0)
4    count += 1
5    sdk.storage.set("ping_count", count)
6    await event.reply(f"pong!这是你第 {count} 次 ping 我了")

sdk.storage 是框架内置的键值存储,基于 SQLite。不用建数据库,不用写 SQL,getset 就完事。机器人重启了数据还在。

firstbot-counter

真正有用的:收集用户信息

前面都是小打小闹。现在写一个用户注册功能——机器人依次问名字、年龄、邮箱,格式不对会要求重填:

 1@command("register")
 2async def register(event):
 3    data = await event.collect([
 4        {"key": "name", "prompt": "你的名字是?"},
 5        {"key": "age", "prompt": "你的年龄?",
 6         "validator": lambda e: e.get_text().isdigit()},
 7        {"key": "email", "prompt": "你的邮箱?",
 8         "validator": lambda e: "@" in e.get_text()}
 9    ])
10
11    if data:
12        await event.reply(
13            f"注册完成!\n"
14            f"姓名:{data['name']}\n"
15            f"年龄:{data['age']}\n"
16            f"邮箱:{data['email']}"
17        )
18    else:
19        await event.reply("超时了,下次再来吧!")

/register 试试。机器人挨个问问题,年龄输"abc"会让你重填,邮箱没 @ 也会重来。整个过程像这样:

firstbot-register

注意看这段代码——你没有写任何状态机,没有维护用户当前在第几步的字典,没有写中间件拦截消息,没有搞超时定时器。collect 一个方法全干完了。

我以前在别的框架手写这种表单收集,光状态管理就两百行,各种边界情况防不胜防。这东西一个方法调用就解决了,说实话第一次看到的时候我是有点破防的。

发消息也要写得爽

ErisPulse 发消息用的是链式语法,从左往右读就是消息的组装过程:

1from ErisPulse import sdk
2
3adapter = sdk.adapter.get("sandbox")
4
5# 普通文本
6await adapter.Send.To("group", "room_123").Text("大家好")
7
8# @用户 + 回复消息
9await adapter.Send.To("group", "room_123").At("user_456").Reply("msg_789").Text("回复你的消息")

修饰方法(AtReply)返回 self,让你一直往后链。发送方法(TextImage)返回 asyncio.Task,你可以 await 等结果,也可以不 await 让它后台发:

 1# 不等,后台发
 2adapter.Send.To("user", "123").Text("发了不管了")
 3
 4# 等结果
 5result = await adapter.Send.To("user", "123").Text("等等结果")
 6
 7# 存起来,过会儿再等
 8task = adapter.Send.To("user", "123").Text("存起来")
 9# 干点别的……
10result = await task

对比一下别的框架怎么发的:

1await bot.send(target_type="group", target_id="room_123", message=[
2    {"type": "at", "data": {"user_id": "user_456"}},
3    {"type": "reply", "data": {"message_id": "msg_789"}},
4    {"type": "text", "data": {"text": "回复你的消息"}}
5])

现在的完整 main.py

到这儿,你的 main.py 长这样:

 1"""
 2my_bot 主程序
 3
 4这是 ErisPulse 自动生成的主程序文件
 5您可以根据需要修改此文件
 6"""
 7
 8import asyncio
 9from ErisPulse import sdk
10from ErisPulse.Core.Event import command
11
12@command("ping")
13async def ping(event):
14    count = sdk.storage.get("ping_count", 0)
15    count += 1
16    sdk.storage.set("ping_count", count)
17    await event.reply(f"pong!这是你第 {count} 次 ping 我了")
18
19@command("register")
20async def register(event):
21    data = await event.collect([
22        {"key": "name", "prompt": "你的名字是?"},
23        {"key": "age", "prompt": "你的年龄?",
24         "validator": lambda e: e.get_text().isdigit()},
25        {"key": "email", "prompt": "你的邮箱?",
26         "validator": lambda e: "@" in e.get_text()}
27    ])
28    if data:
29        await event.reply(
30            f"注册完成!\n"
31            f"姓名:{data['name']}\n"
32            f"年龄:{data['age']}\n"
33            f"邮箱:{data['email']}"
34        )
35    else:
36        await event.reply("超时了,下次再来吧!")
37
38@command("hello")
39async def hello(event):
40    name = event.get_user_nickname() or "朋友"
41    await event.reply(f"你好,{name}!")
42
43async def main():
44    await sdk.run(keep_running=True)
45
46if __name__ == "__main__":
47    asyncio.run(main())

三十多行,一个带计数器的 ping、一个完整的用户注册系统、一个打招呼命令。都是能直接跑的。

1epsdk run main.py

换个平台?改个配置就行

上面一直在用 Sandbox 调试。等你准备好接真实平台:

1# Telegram
2epsdk install Telegram
3
4# 云湖
5epsdk install Yunhu
6
7# OneBot11 (QQ)
8epsdk install OneBot11

然后去 config/config.toml 里填上对应平台的 token:

1[Telegram_Adapter]
2token = "你的_bot_token"

firstbot-config

重启,main.py 一行不改,所有命令照常工作。

这背后是 ErisPulse 把每个平台的事件都转成了 OneBot12 统一标准。你的代码只跟标准打交道,适配器负责翻译各平台的方言。换平台就是换配置,不是换代码。

装个 Dashboard,告别黑框框

1epsdk install Dashboard

访问 http://localhost:8000/Dashboard,你能看到:

  • 机器人在线状态
  • 实时日志和事件流
  • 已安装的模块和适配器
  • 在线改配置
  • 模块商店,点一下就能装新模块

firstbot-dashboard

不用再 SSH 进服务器翻日志了。


跟着走完这套流程,你应该能感受到 ErisPulse 的几个特点了:

  • collectconfirmchoose 让你写对话像写脚本,不用管状态机
  • 链式 Send 语法从左往右就是消息结构,不用拼 JSON
  • 换平台只换配置不换代码
  • Dashboard 让管理变得可视

动手试试吧:

1pip install ErisPulse
2epsdk init -n my_bot

评论