AstrBot 插件开发:从最佳实践到实战插件

作者:Administrator 发布时间: 2026-05-17 阅读量:5 评论数:0

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 会阻塞整个机器人。必须使用 aiohttphttpx

配置读取

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./-])')

设计要点:

  1. 模式互斥_REPO_PATTERN 末尾的 [^\w./-] 排除了 / 字符,因此 github.com/owner/repo/releases/tag/v1.0.0 中的 repo 之后是 /_REPO_PATTERN 不会误匹配。Release 标签和 Release 页面分别由前两个模式覆盖。

  2. 三层并行不起作用——这里故意用了串行:先匹配 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 方法论

  1. 设计文档先行:每个插件启动前先写 docs/superpowers/specs/ 设计文档,明确需求边界
  2. 实施计划拆解:拆分为 6-8 个可执行步骤,每个步骤有精确的文件路径和验收条件
  3. AI 子代理执行:通过子代理驱动开发(SDD)逐任务实现
  4. 代码审查:开发完成后进行系统性审查——正则优先级、连接管理、异常处理、跨平台兼容性
  5. 性能优化:审查发现问题后提交针对性优化(如 Gitparser 的 15c0c0a
  6. 规范化:对 PUBG 插件的仓库结构做了 PEP 8 规范化提交
  7. 反馈循环:插件实战中暴露的问题(如 repo 字段缺失、@star 兼容性)回归到技能中,丰富最佳实践

astrbot-plugin-dev 技能包到两个实战插件,这条路径展示了一个完整的"技能 → 实践 → 反哺"闭环。


相关链接

评论