2026 年 5 月 · 开源项目 · 聊天机器人
一、什么是 AstrBot
AstrBot 是一个开源的多平台聊天机器人框架,支持接入 QQ(NapCat / AIOCQHTTP)、Telegram、Discord 等平台。它基于 Python 3.10+,提供了一套完整的插件开发 API,插件以类的形式继承 Star 基类,通过装饰器注册消息过滤器和命令。
项目地址:https://github.com/Soulter/AstrBot
插件的基本形态:
astrbot_plugin_<name>/
├── main.py # 入口,定义继承 Star 的插件类
├── metadata.yaml # 包名、版本、仓库地址(用于自动更新)
├── _conf_schema.json # WebUI 配置项定义
├── requirements.txt # pip 依赖
└── docs/superpowers/ # AI 辅助开发的设计文档和计划
二、astrbot-plugin-dev:插件开发技能包
为了避免每次写插件都踩同样的坑,项目沉淀了一套 OpenCode 技能(astrbot-plugin-dev),在 AI 辅助编码时自动激活。技能的核心覆盖了 4 个必须避免的坑 + 6 个开发场景速查。
四个必须避免的坑
坑 1:star 装饰器的版本兼容性
AstrBot v3 与 v4 的 API 存在差异:
# ❌ v3 会报 ImportError: cannot import name 'star' from 'astrbot.api.star'
from astrbot.api.star import Context, Star, star
@star
class MyPlugin(Star): ...
# ✅ 兼容 v3 和 v4:子类注册即可,无需装饰器
from astrbot.api.star import Context, Star
class MyPlugin(Star): ...
安全做法是不使用 @star,仅通过 class MyPlugin(Star) 继承。如果使用 @register("MyPlugin") 装饰器,可同时支持显式注册(新版本)和隐式发现(旧版本)。
坑 2:async generator 不能直接 await
# ❌ TypeError: object async_generator can't be used in 'await'
yield await self._handler(event, ...)
# ✅ async for 迭代委托
async for result in self._handler(event, ...):
yield result
AstrBot 的消息处理器使用 yield 返回消息,这使得它们成为 async generator。如果一个处理器调用另一个处理器,不能简单地 await——必须用 async for 展开。
坑 3:metadata.yaml 缺少 repo 字段
当插件管理界面提示"没有指定仓库地址"时,是因为 metadata.yaml 缺少 repo 字段:
name: myplugin
version: 1.0.0
desc: 插件描述
repo: https://github.com/user/astrbot_plugin_myplugin # 自动更新依赖此字段
没有 repo 字段,插件无法通过 WebUI 的"检查更新"功能升级。
坑 4:使用同步 requests 阻塞事件循环
AstrBot 基于 asyncio 事件循环,requests 的同步 I/O 会阻塞整个机器人。必须使用 aiohttp 或 httpx。
配置读取
from astrbot.api import AstrBotConfig
class MyPlugin(Star):
def __init__(self, context: Context, config: AstrBotConfig):
super().__init__(context)
self.config = config
token = config.get("api_key", "default")
消息过滤器速查
@filter.event_message_type(filter.EventMessageType.ALL) # 所有消息
@filter.event_message_type(filter.EventMessageType.GROUP_MESSAGE) # 仅群聊
@filter.command("hello") # /hello
@filter.command("add") # /add 1 2 → a=1, b=2
@filter.command_group("math") # /math add 1 2
@filter.permission_type(filter.PermissionType.ADMIN) # 仅管理员
LLM 调用
# 直接生成
resp = await self.context.llm_generate(
chat_provider_id=prov_id, prompt="Hello")
# Agent 模式(工具调用循环)
resp = await self.context.tool_loop_agent(
event=event, chat_provider_id=prov_id,
prompt="Search for X", tools=ToolSet([MyTool()]), max_steps=30)
# 全局注册工具
self.context.add_llm_tools(MyTool())
持久化存储
# KV 存储
await self.put_kv_data("key", value)
data = await self.get_kv_data("key", default)
# 文件存储(data/plugin_data/<name>/)
from astrbot.core.utils.astrbot_path import get_astrbot_data_path
path = Path(get_astrbot_data_path()) / "plugin_data" / self.name
三、Gitparser:GitHub 链接解析插件
功能与架构
Gitparser 自动检测聊天消息中的 GitHub 链接,调用 GitHub REST API 获取仓库或 Release 信息,以纯文本回复摘要。
整个插件是单文件 main.py,核心的数据流是:
消息 → @filter.event_message_type(ALL)
→ 正则匹配 URL(tag > releases > repo 三级,逐级串行)
→ 调用 GitHub REST API(可选 Token 提升速率限制)
→ 格式化纯文本 → yield event.plain_result(...)
三层正则匹配
URL 解析使用三层递进正则,优先级从精确到宽泛:
import re
# 第一级:Release 标签
_RELEASE_TAG_PATTERN = re.compile(
r'(?<![a-zA-Z0-9.-])github\.com/([a-zA-Z0-9._-]+)/([a-zA-Z0-9._-]+)/releases/tag/([^\s/]+)'
r'(?:\s|$|[^\w./-])')
# 第二级:Release 列表页
_RELEASES_PAGE_PATTERN = re.compile(
r'(?<![a-zA-Z0-9.-])github\.com/([a-zA-Z0-9._-]+)/([a-zA-Z0-9._-]+)/releases'
r'(?:\s|$|[^\w./-])')
# 第三级:普通仓库
_REPO_PATTERN = re.compile(
r'(?<![a-zA-Z0-9.-])github\.com/([a-zA-Z0-9._-]+)/([a-zA-Z0-9._-]+)'
r'(?:\.git)?'
r'(?:\s|$|[^\w./-])')
设计要点:
-
模式互斥:
_REPO_PATTERN末尾的[^\w./-]排除了/字符,因此github.com/owner/repo/releases/tag/v1.0.0中的repo之后是/→_REPO_PATTERN不会误匹配。Release 标签和 Release 页面分别由前两个模式覆盖。 -
三层并行不起作用——这里故意用了串行:先匹配 tag,再匹配 releases 页,最后才是仓库。因为一个 URL 可以是仓库也可以是 Release,精确优先确保
/releases/tag/v1.0.0不会被降级为普通仓库信息。
辅助函数:
def _find_first_url(text: str, pattern: re.Pattern) -> re.Match | None:
return pattern.search(text)
API 调用与错误处理
核心请求函数 _fetch_api:
async def _fetch_api(url: str, token: str = "") -> dict | None:
headers = {"Accept": "application/vnd.github.v3+json"}
if token:
headers["Authorization"] = f"token {token}"
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers, timeout=10) as resp:
if resp.status == 404:
return None # 项目不存在,不回复
if resp.status == 429:
return {"error": "rate_limited"} # 限流
if resp.status != 200:
logger.warning(f"GitHub API error: {resp.status} {url}")
return None
return await resp.json()
except asyncio.TimeoutError:
logger.warning(f"GitHub API timeout: {url}")
return None
except Exception as e:
logger.error(f"GitHub API error: {e}")
return None
关键设计决策:
- 404 静默不回复——避免在群聊中刷无用的错误消息
- 429 返回特殊标记——格式化时给出限流提示,而不是静默失败
- 所有异常兜底——网络错误、超时等统一日志记录,不向用户暴露异常堆栈
- Token 可选——不配置 Token 也可以查询,但有每小时 60 次的 IP 级限制
性能优化(提交 15c0c0a)
代码审查发现三个优化点,在一次提交中全部修复:
1. 复用 aiohttp.ClientSession
优化前:每次 _fetch_api 都新建 aiohttp.ClientSession(),意味着每次 API 调用都要新建 TCP 连接、完成 TLS 握手。
优化后:在 __init__ 中创建一次 session,所有请求复用,利用 HTTP Keep-Alive:
class GitparserPlugin(Star):
def __init__(self, context: Context, config: AstrBotConfig):
super().__init__(context)
self.config = config
token = config.get("github_token", "")
self._headers = {"Accept": "application/vnd.github.v3+json"}
if token:
self._headers["Authorization"] = f"token {token}"
self._session = aiohttp.ClientSession(headers=self._headers)
async def terminate(self):
await self._session.close()
2. 预构建请求头
Token 在插件初始化时读取一次并存入 self._headers,后续每次请求不需要再 config.get() + dict 分配。
3. 提取 rate-limit 检查
优化前 3 处重复:
if isinstance(data, dict) and data.get("error") == "rate_limited":
优化后:
def _check_rate_limited(data: dict | None) -> bool:
return isinstance(data, dict) and data.get("error") == "rate_limited"
4. 超时对象常量化:_REQUEST_TIMEOUT = aiohttp.ClientTimeout(total=10) 模块级常量。
5. 添加 @star 装饰器:显式注册兼容更多 AstrBot 版本。
6. 资源清理:实现 terminate() 方法,在插件卸载时关闭 session,防止资源泄漏。
28 insertions(+), 20 deletions(-) 改动量不大,但消除了每次请求的 TCP 握手、tls 协商、DNS 解析开销。
格式化输出
仓库信息格式化:
lines.append(f"📦 {owner}/{repo}")
if data.get("description"):
lines.append(data["description"])
parts = []
if data.get("stargazers_count"):
parts.append(f"⭐ Stars: {data['stargazers_count']}")
if data.get("forks_count"):
parts.append(f"🍴 Forks: {data['forks_count']}")
if data.get("language"):
parts.append(f"🔤 语言: {data['language']}")
lines.append(" | ".join(parts))
topics = data.get("topics", [])
if topics:
lines.append(f"🏷 {' '.join(f'#{t}' for t in topics[:8])}")
lines.append(f"📅 最后更新: {data.get('updated_at', '')[:10]} | 🔓 {data.get('license', {}).get('spdx_id', 'N/A')}")
Release 格式化:
lines.append(f"🚀 {owner}/{repo} - {tag_name}")
if release.get("name"):
lines.append(f"📝 {release['name']}")
lines.append(f"📅 发布于: {release.get('published_at', '')[:10]}")
lines.append(f"📦 下载: {release.get('zipball_url', 'N/A')}")
完整提交历史
af5dda4 feat: enrich repo info with avatar, topics, stats; bump to 1.1.0
91a3743 fix: add repo field for plugin update support
fbea066 revert: remove avatar image feature
15c0c0a perf: reuse ClientSession, prebuild headers, dedup rate-limit check ← 最新
版本从 1.0.0 到 1.1.0 经历了功能增强(头像、Topics、统计信息)→ 添加 repo 字段 → 简化(移除头像)→ 性能优化的完整迭代。
四、PUBG 战绩查询插件
功能与架构
PUBG 插件通过 developer.pubg.com 官方 API,查询玩家的终身战绩(各模式统计)和最近对局。输出支持两种格式:纯文本(不需要额外依赖)和 图片卡片(需要 Pillow)。
数据流
/player {name} → PUBG API /players → 获取 Player ID
→ /players/{id}/seasons/lifetime → 终身战绩
→ /players/{id}/matches → 最近 MATCH_LIMIT 场对局
→ 格式化(文本或图片) → yield event.plain_result / image_result
地图与模式映射
PUBG API 返回的原始数据是内部 ID,插件通过字典映射转换为人类可读的中文:
_MAP_NAMES = {
"Baltic_Main": "Erangel", # 经典艾伦格
"Desert_Main": "Miramar", # 米拉玛
"Savage_Main": "Sanhok", # 萨诺
"DihorOtok_Main": "Vikendi", # 维寒迪
"Summerland_Main": "Karakin", # 卡拉金
"Tiger_Main": "Taego", # 泰戈
"Kiki_Main": "Deston", # 帝斯顿
"Neon_Main": "Rondo", # 荣都
"Range_Main": "Camp Jackal", # 猎狐营地
"Chimera_Main": "Paramo", # 帕拉莫
"Heaven_Main": "Haven", # 褐湾
}
_MODE_LABELS = {
"squad-fpp": "四排FPP",
"squad": "四排TPP",
"duo-fpp": "双排FPP",
"duo": "双排TPP",
"solo-fpp": "单排FPP",
"solo": "单排TPP",
}
API 重试机制
PUBG API 有时会返回 429(限流)或临时 5xx,插件实现了指数退避重试:
API_TIMEOUT = 15 # 单次请求超时
API_MAX_RETRY = 2 # 最多重试 2 次
async def _api_request(method: str, url: str, headers: dict,
retry: int = API_MAX_RETRY) -> dict:
for attempt in range(retry + 1):
try:
async with aiohttp.ClientSession(timeout=API_TIMEOUT) as session:
async with session.request(method, url, headers=headers) as resp:
if resp.status == 429:
if attempt < retry:
await asyncio.sleep(2 ** attempt)
continue
raise PubgApiError("API rate limited")
if resp.status == 403:
raise PubgApiError("API key invalid or forbidden")
resp.raise_for_status()
return await resp.json()
except (asyncio.TimeoutError, aiohttp.ClientError) as e:
if attempt < retry:
await asyncio.sleep(2 ** attempt)
continue
raise PubgApiError(f"请求失败(已达最大重试次数): {e}")
raise PubgApiError("请求失败(已达最大重试次数)")
图片渲染引擎
插件内置了一个自包含的图像渲染引擎,用 Pillow 在内存中绘制战绩卡片。所有颜色、边距、行高都是常数量化的:
# ── 颜色主题 ──
BG = (15, 20, 30) # 深蓝黑背景
CARD = (25, 32, 48) # 卡片底色
ACCENT = (255, 180, 30) # 金色强调
ACCENT2 = (80, 160, 255) # 蓝色强调
WHITE = (240, 240, 240)
GRAY = (140, 150, 170)
WIN_CLR = (80, 220, 120) # 绿色(胜利/正数据)
BAN_CLR = (255, 70, 70) # 红色(封禁)
WARN_CLR = (255, 200, 50) # 黄色(警告)
PAD = 32
COL_W = 560
图片高度是动态计算的——根据实际数据的行数动态计算总高度:
total_h = (
PAD + H_HEADER + PAD + H_BAN_BAR
+ H_SEC_TITLE + len(mode_rows) * (H_MODE_ROW + 10)
+ (PAD + H_SEC_TITLE + len(match_rows) * (H_MATCH_ROW + 8) if match_rows else 0)
+ H_FOOTER + PAD
)
避免固定画布尺寸导致内容溢出或大量空白。
实际绘制采用了 PIL 的低级 API:
img = Image.new("RGB", (W, total_h), BG)
draw = ImageDraw.Draw(img)
# 圆角矩形效果:圆角用圆弧近似,此处用普通矩形 + 边框
draw.rectangle([PAD, y, W - PAD, y + H_HEADER - 10], fill=CARD, outline=ACCENT, width=2)
# 文本布局
draw.text((PAD + 18, y + 14), name, font=f_big, fill=ACCENT)
draw.text((PAD + 18, y + 50), f"[{platform.upper()}]", font=f_norm, fill=GRAY)
字体加载的多层回退
字体是跨平台图片渲染的最大难点。插件实现了多层自动发现:
def _load_font(size: int, bold: bool = False) -> FreeTypeFont | FontClass:
candidates = [
os.path.join(FONT_DIR, "NotoSansSC-Bold.ttf" if bold else "NotoSansSC-Regular.ttf"),
# Windows
"C:/Windows/Fonts/msyhbd.ttc" if bold else "C:/Windows/Fonts/msyh.ttc",
"C:/Windows/Fonts/simhei.ttf" if bold else "C:/Windows/Fonts/simsun.ttc",
# Linux
"/usr/share/fonts/truetype/noto/NotoSansCJK-Bold.ttc" if bold else ".../NotoSansCJK-Regular.ttc",
"/usr/share/fonts/opentype/noto/NotoSansCJK-Bold.ttc" if bold else ".../NotoSansCJK-Regular.ttc",
]
for path in candidates:
if os.path.exists(path):
try:
return ImageFont.truetype(path, size)
except Exception:
continue
return ImageFont.load_default() # 终极回退
搜索路径覆盖了嵌入式字体目录、Windows 系统字体、Linux 常见 CJK 字体路径。如果都找不到,PIL 的默认位图字体保证至少能显示英文和数字。
条件导入与类型标注
Pillow 不是必需的运行依赖——插件通过条件导入实现优雅降级:
try:
from PIL import Image, ImageDraw, ImageFont
PIL_OK = True
except ImportError:
PIL_OK = False
如果 PIL 不可用,player 命令切换到纯文本模式,不产生图片。
类型标注使用了 TYPE_CHECKING 技巧——PIL 的类型只在类型检查时导入,运行时不会因为 PIL 未安装而报 ImportError:
if TYPE_CHECKING:
from PIL import ImageDraw as IDraw
from PIL.ImageFont import FreeTypeFont, ImageFont as FontClass
def _load_font(size: int, bold: bool = False) -> FreeTypeFont | FontClass: ...
def _text_w(draw: IDraw.ImageDraw, text: str, font: FreeTypeFont | FontClass) -> int: ...
配合 from __future__ import annotations 启用 PEP 604 联合类型语法(X | Y 代替 Union[X, Y]),让类型标注更简洁。
仓库规范化(提交 f6d0c7c)
规范化变更的内容:
.gitignore — Python 标准忽略规则
__init__.py — 显式包导出 PubgPlugin、PubgApiError
requirements.txt — 依赖声明(astrbot>=3.0, aiohttp>=3.9, Pillow>=10.0)
main.py — 类型注解修复、import 排序、PEP 8 规范
metadata.yaml — 版本号统一(v1.2.0 → 1.3.0)
_conf_schema.json — 保持不变
目录结构从:
astrbot_plugin_pubg/
├── astrbot_plugin_pubg/ # 嵌套包名
│ ├── main.py
│ ├── metadata.yaml
│ └── _conf_schema.json
调整为:
astrbot_plugin_pubg/
├── main.py
├── metadata.yaml
├── _conf_schema.json
├── requirements.txt
├── __init__.py # from .main import PubgPlugin, PubgApiError
└── .gitignore
移除了嵌套的包名目录,改为平铺结构,与 AstrBot 插件的标准布局一致。
五、开发工作流复盘
这两个插件的开发完整实践了 Superpowers 方法论:
- 设计文档先行:每个插件启动前先写
docs/superpowers/specs/设计文档,明确需求边界 - 实施计划拆解:拆分为 6-8 个可执行步骤,每个步骤有精确的文件路径和验收条件
- AI 子代理执行:通过子代理驱动开发(SDD)逐任务实现
- 代码审查:开发完成后进行系统性审查——正则优先级、连接管理、异常处理、跨平台兼容性
- 性能优化:审查发现问题后提交针对性优化(如 Gitparser 的
15c0c0a) - 规范化:对 PUBG 插件的仓库结构做了 PEP 8 规范化提交
- 反馈循环:插件实战中暴露的问题(如
repo字段缺失、@star兼容性)回归到技能中,丰富最佳实践
从 astrbot-plugin-dev 技能包到两个实战插件,这条路径展示了一个完整的"技能 → 实践 → 反哺"闭环。
相关链接
- AstrBot:https://github.com/Soulter/AstrBot
- 开发文档:https://docs.astrbot.app/dev/star/plugin-new.html
- astrbot-plugin-dev:
https://github.com/sakuradairong/astrbot-plugin-dev - Gitparser:
https://github.com/sakuradairong/astrbot_plugin_Gitparser - PUBG 插件:
https://github.com/sakuradairong/astrbot_plugin_pubg