别再为多平台AI机器人开发发愁了， AstrBot 才是开源社区的最优解
GitHub星标破万的智能对话框架，我第一时间做了完整评测（附保姆级入门指南）
一、为什么值得关注 / 为什么你应该了解 AstrBot
在人工智能飞速发展的今天，开发者们面临着前所未有的挑战：如何快速构建一个能够同时支持微信、QQ、Discord、Telegram 等多个平台的智能聊天机器人？如何让机器人具备强大的自然语言理解能力，同时又能灵活接入 GPT、Claude、通义千问等各大语言模型？
这些问题困扰了无数开发者很久，直到 AstrBot 的出现。
1.1 AstrBot 是什么
AstrBot 是一个高度模块化、可扩展的智能对话机器人框架。它由 AstrBotDevs 团队开发和维护，采用 Python 编写，旨在为开发者提供一个优雅、灵活、易用的 AI 聊天机器人解决方案。
从本质上讲，AstrBot 扮演的是一座「桥梁」的角色——它的一端连接着各种社交平台（微信、QQ、飞书、钉钉等），另一端则对接了主流的大语言模型服务商（OpenAI、Anthropic、百度文心、阿里通义等）。开发者无需关心底层通信细节，只需要专注于定义机器人的行为逻辑和对话策略。
1.2 为什么 AstrBot 值得关注
让我从几个维度来分析为什么你应该认真考虑使用 AstrBot：
第一，极其友好的开发体验
很多现有的机器人框架要么配置复杂、上手困难，要么功能单一、扩展性差。AstrBot 采用了现代化的设计理念，提供了清晰的项目结构、直观的配置文件和丰富的 API 接口。即使你是刚入门 Python 的新手，也能在短时间内搭建起一个功能完整的智能聊天机器人。
第二，强大的多平台支持
在 AstrBot 中，你不需要为每个平台编写独立的适配代码。框架已经内置了多种平台的连接器，你只需要在配置文件中简单设置，就能让同一个机器人同时运行在多个平台上。这种设计大大降低了维护成本，让「一次开发，多端运行」成为可能。
第三，灵活的插件系统
AstrBot 的插件系统经过精心设计，支持热加载、热重载。你可以在不停止机器人的情况下安装、卸载或更新插件。这意味着你可以持续迭代机器人的功能，而不会打扰到正在使用它的用户。
第四，丰富的开箱即用功能
除了基础的对话功能，AstrBot 还提供了诸如图片生成、代码执行、知识库问答、角色扮演等高级特性。这些功能都以插件的形式提供，你可以根据需要自由选择和组合。
第五，活跃的社区生态
AstrBot 拥有一个热情、活跃的开发者社区。无论你遇到什么问题，都能在社区中找到帮助或获得及时的响应。官方团队也在持续更新和维护项目，确保框架能够跟上技术发展的步伐。
1.3 核心特性一览
让我们先通过一个简单的列表来了解 AstrBot 的主要特性：
- 多平台支持：微信、QQ、飞书、钉钉、Discord、Telegram 等
- 多模型支持：OpenAI GPT、Anthropic Claude、百度文心、阿里通义等
- 插件系统：热加载、热重载、插件市场
- 知识库问答：支持向量数据库、RAG 技术
- 图片生成：集成 DALL-E、Stable Diffusion 等
- 群组管理：入群欢迎、关键词触发、权限管理等
- 中文优化：针对中文语境做了大量适配和优化
看完这些特性，你是不是已经迫不及待想要动手试试了？别着急，让我们先完成环境搭建，然后再深入了解如何使用这个强大的框架。
二、环境搭建 / Getting Started
好的，现在让我们正式开始AstrBot的学习之旅。首先，我们需要搭建开发环境。不用担心，这个过程并不复杂，我会一步步带你完成。
2.1 系统要求
在开始之前，让我们确认一下你的系统环境：
- 操作系统：Windows、macOS 或 Linux 均可
- Python 版本：3.10 或更高版本
- 内存：建议至少 4GB RAM（如果需要运行本地模型可能需要更多）
- 磁盘空间：建议预留 10GB 以上空间
2.2 安装 Python
如果你还没有安装 Python，请先从官网（https://www.python.org）下载并安装。建议使用 Python 3.10 或更高版本，因为在这些版本中我们可以用上更现代的语法特性。
安装完成后，打开终端（或命令提示符），输入以下命令验证安装：
python --version
# 或者在某些系统上可能需要
python3 --version
你应该能看到类似这样的输出：
Python 3.11.5
2.3 创建虚拟环境（推荐）
为了保持项目环境的干净和隔离，推荐使用虚拟环境来安装 AstrBot。虚拟环境可以让我们为不同的项目创建独立的依赖环境，避免版本冲突。
在终端中执行以下命令创建虚拟环境：
# 创建虚拟环境
python -m venv astrbot_env
# 激活虚拟环境
# Windows 系统
astrbot_env\Scripts\activate
# macOS/Linux 系统
source astrbot_env/bin/activate
激活虚拟环境后，你的终端提示符前面会显示环境名称，像这样：
(astrbot_env) $
2.4 安装 AstrBot
现在，万众期待的时刻到了！让我们安装 AstrBot 框架本身。
AstrBot 可以通过 pip 直接安装，这是最简单的方式：
pip install astrbot
如果你想要安装包含所有可选依赖的完整版本，可以使用：
pip install astrbot[all]
安装过程可能需要几分钟时间，取决于你的网络状况。安装完成后，我们来验证一下是否安装成功：
python -c "import astrbot; print(astronbot.__version__)"
2.5 初始化项目
AstrBot 提供了一个便捷的初始化命令，可以帮助我们快速创建一个项目结构。执行以下命令：
python -m astrbot init my_first_bot
这个命令会在当前目录下创建一个名为 my_first_bot 的文件夹，里面包含了 AstrBot 项目的标准结构：
my_first_bot/
├── README.md # 项目说明文档
├── config.json # 主配置文件
├── plugins/ # 插件目录
│ └── .gitkeep # 占位文件
├── data/ # 数据存储目录
│ └── .gitkeep
├── requirements.txt # Python 依赖列表
├── run.py # 启动脚本
└── astrbot_config.json # AstrBot 配置文件
接下来，让我们进入项目目录并查看配置文件的内容：
cd my_first_bot
cat config.json
你应该能看到类似这样的配置内容：
{
"platforms": {
"terminal": true // 启用终端模式，方便测试
},
"llm_provider": "openai",
"llm_config": {
"api_key": "your-api-key-here",
"model": "gpt-3.5-turbo"
}
}
2.6 配置说明
在深入使用之前，让我们先了解一下主要配置项的含义：
{
// 平台配置：决定机器人连接哪些聊天平台
"platforms": {
"terminal": true, // 终端模式，用于本地测试
"wechat": false, // 微信支持（需要配合第三方方案）
"qq": false, // QQ 支持
"telegram": false, // Telegram 支持
"discord": false // Discord 支持
},
// 大语言模型配置
"llm_provider": "openai", // 模型服务商
"llm_config": {
"api_key": "", // API 密钥（重要！请妥善保管）
"model": "gpt-3.5-turbo", // 使用的模型
"base_url": "", // 可选的 API 代理地址
"temperature": 0.7, // 生成温度，控制创造性
"max_tokens": 2000 // 最大生成 token 数
},
// 插件配置
"plugins": {
"enabled": true, // 是否启用插件系统
"auto_reload": true // 是否自动重载插件
}
}
2.7 运行你的第一个机器人
现在，让我们尝试运行机器人看看效果！确保你的虚拟环境处于激活状态，然后在项目目录下执行：
python run.py
如果一切配置正确，你应该能看到类似这样的输出：
[INFO] AstrBot 正在启动...
[INFO] 加载配置文件...
[INFO] 初始化平台连接器...
[INFO] 终端模式已启用
[INFO] AstrBot 启动完成！
[AstrBot] 大家好，我是 AstrBot！请开始和我聊天吧～
现在，你可以在终端中直接输入消息与机器人对话了！这是一个最简单的测试模式，非常适合在没有配置任何社交平台的情况下开发和调试。
试着输入一些简单的对话：
你: 你好，你是谁？
AstrBot: 你好！我是基于 AstrBot 框架运行的智能对话助手...
三、核心功能详解
恭喜你成功运行了第一个 AstrBot 实例！现在，让我们深入了解这个框架的核心功能。只有真正理解了这些功能，你才能充分发挥 AstrBot 的潜力。
3.1 平台连接器系统
AstrBot 最强大的特性之一就是其平台连接器系统。这个系统让我们可以同时连接多个不同的聊天平台，而不需要为每个平台编写独立的代码。
3.1.1 什么是平台连接器
平台连接器（Platform Connector）是 AstrBot 架构中负责与特定聊天平台通信的模块。每个支持的平台都有对应的连接器，连接器负责：
- 处理平台的身份验证
- 接收来自平台的消息
- 将机器人的响应发送回平台
- 处理平台的特殊事件（如入群、离群等）
3.1.2 支持的平台
截至目前，AstrBot 支持以下主流平台：
即时通讯平台
- QQ：支持 OneBot v11/v12 协议
- 微信：通过第三方方案（如个人微信API）支持
- 飞书：支持飞书机器人和企业应用
- 钉钉：支持钉钉自定义机器人
社交和社区平台
- Discord：完整的频道和服务器支持
- Telegram：支持群组和私聊
- Slack：适合企业协作场景
开发者和测试工具
- 终端：内置的交互式测试界面
- HTTP API：通过 HTTP 接口集成
3.1.3 配置多个平台
假设我们想要同时启用 Telegram 和 Discord，我们可以在配置文件中这样设置：
{
"platforms": {
"terminal": true,
"telegram": {
"enabled": true,
"bot_token": "your-telegram-bot-token",
"admin_ids": [123456789]
},
"discord": {
"enabled": true,
"bot_token": "your-discord-bot-token",
"intents": ["guilds", "guild_messages", "direct_messages"]
}
}
}
这样配置后，同一个机器人实例会同时响应来自 Telegram 和 Discord 两个平台的消息。用户无论在哪个平台上与机器人交互，都会得到一致的体验。
3.2 大语言模型集成
AstrBot 的另一大核心能力是对接各种大语言模型服务商。这种设计让我们可以轻松切换不同的模型，无需修改业务逻辑代码。
3.2.1 支持的模型提供商
OpenAI 系列：
- GPT-4 / GPT-4 Turbo
- GPT-3.5 Turbo
- 支持 OpenAI 兼容 API 的第三方模型
Anthropic 系列：
- Claude 3 Opus
- Claude 3 Sonnet
- Claude 3 Haiku
国内大模型：
- 百度文心一言（ERNIE Bot）
- 阿里通义千问
- 智谱 ChatGLM
- 讯飞星火
开源模型：
- 通义千问开源版本（Qwen）
- ChatGLM 系列
- LLaMA 系列
- 其他支持 OpenAI API 格式的模型
3.2.2 配置模型
在 AstrBot 中，配置大语言模型非常简单。只需要在配置文件中指定提供商和相关参数：
{
"llm_provider": "openai",
"llm_config": {
"api_key": "sk-xxxxxxxxxxxxxxxx",
"model": "gpt-4",
"temperature": 0.7,
"max_tokens": 4096,
"base_url": "https://api.openai.com/v1"
}
}
如果你想切换到其他模型，只需要修改 llm_provider 和相关配置：
{
"llm_provider": "claude",
"llm_config": {
"api_key": "sk-ant-xxxxxxxxxxxxxxxx",
"model": "claude-3-sonnet-20240229",
"temperature": 0.7,
"max_tokens": 4096
}
}
AstrBot 还支持同时使用多个模型！你可以配置模型组，在不同场景下使用不同的模型：
{
"llm_providers": {
"default": {
"provider": "openai",
"model": "gpt-3.5-turbo"
},
"coding": {
"provider": "openai",
"model": "gpt-4"
},
"creative": {
"provider": "claude",
"model": "claude-3-opus"
}
}
}
3.3 插件系统
插件系统是 AstrBot 灵活性的核心所在。它允许开发者以模块化的方式扩展机器人功能，而不需要修改核心代码。
3.3.1 插件架构设计
AstrBot 的插件系统采用了现代化的设计理念：
插件结构：
plugins/
├── my_plugin/ # 插件目录
│ ├── __init__.py # 插件入口（必须有）
│ ├── main.py # 插件主逻辑
│ ├── config.json # 插件配置
│ └── requirements.txt # 插件依赖
3.3.2 插件的声明周期
每个插件都有自己的生命周期，AstrBot 会在合适的时机调用相应的方法：
# 这是插件的基本结构示例
class MyPlugin:
# 插件配置项
name = "my_plugin" # 插件唯一标识
description = "我的第一个插件" # 插件描述
version = "1.0.0" # 插件版本
def __init__(self):
"""
构造函数：插件被加载时调用
适合进行初始化操作，如连接数据库、加载资源等
"""
pass
def on_load(self):
"""
插件加载完成时调用
此时插件已经完成所有初始化，可以安全使用
"""
print("插件加载成功！")
def on_unload(self):
"""
插件卸载时调用
适合进行清理操作，如关闭数据库连接、释放资源等
"""
print("插件即将卸载...")
async def on_message(self, message):
"""
收到消息时调用
这是插件处理消息的核心入口
"""
pass
3.3.3 事件处理
AstrBot 提供了丰富的事件类型，插件可以订阅这些事件来实现各种功能：
# 插件可以处理的事件类型
# 消息事件
async def on_message(message):
"""收到任意消息时触发"""
pass
async def on_private_message(message):
"""收到私聊消息时触发"""
pass
async def on_group_message(message):
"""收到群聊消息时触发"""
pass
# 群组事件
async def on_group_join(event):
"""有新成员加入群组时触发"""
pass
async def on_group_leave(event):
"""有成员离开群组时触发"""
pass
# 系统事件
async def on_startup():
"""机器人启动时触发"""
pass
async def on_shutdown():
"""机器人关闭时触发"""
pass
# 定时事件
async def on_cron(hour=14, minute=30):
"""定时任务触发，例如每天下午2点30分"""
pass
3.4 知识库问答系统
对于需要处理专业领域知识的机器人来说，知识库问答系统是不可或缺的功能。AstrBot 内置了对知识库的支持，让机器人能够基于指定的知识库来回答问题。
3.4.1 知识库工作原理
AstrBot 的知识库系统基于 RAG（检索增强生成）架构：
用户问题
↓
问题向量化（使用 Embedding 模型）
↓
向量数据库检索（找出最相关的知识片段）
↓
将相关知识 + 用户问题 一起发送给 LLM
↓
LLM 基于提供的知识生成答案
3.4.2 配置知识库
在 AstrBot 中配置知识库非常简单：
{
"knowledge_base": {
"enabled": true,
"embedding_model": "text-embedding-ada-002",
"vector_store": "faiss",
"index_path": "./data/knowledge_base/index",
"chunk_size": 500,
"chunk_overlap": 50
}
}
支持的向量数据库：
- FAISS：Facebook 开源的向量检索库，适合中小规模数据
- Milvus：分布式向量数据库，适合大规模数据
- Qdrant：基于 Rust 的向量数据库，性能优秀
- Chroma：轻量级向量数据库，适合快速原型开发
3.5 角色扮演与对话管理
AstrBot 提供了强大的对话管理功能，包括角色扮演、对话上下文管理、对话分支等高级特性。
3.5.1 定义对话角色
你可以为机器人定义不同的角色，每个角色有独特的性格、语气和行为模式：
# 角色定义示例
character = {
"name": "小明",
"description": "一个热心、幽默的AI助手",
"personality": "开朗活泼，喜欢用轻松的方式解释复杂的问题",
"speaking_style": "口语化，偶尔使用网络用语，但不失专业性",
"avatar_url": "https://example.com/avatar.png",
"greeting": "嗨！我是小明，有什么我可以帮你的吗？"
}
3.5.2 对话上下文管理
AstrBot 会自动维护对话上下文，让你能够进行多轮对话：
# 对话上下文示例
context = {
"user_id": "user_123",
"session_id": "session_456",
"conversation_history": [
{"role": "user", "content": "我想学习Python"},
{"role": "assistant", "content": "太好了！Python是一门很适合入门的编程语言..."},
{"role": "user", "content": "应该从哪里开始呢？"},
{"role": "assistant", "content": "建议从基础语法开始，比如变量、数据类型..."}
],
"metadata": {
"learning_progress": "基础语法",
"interests": ["web开发", "数据分析"]
}
}
四、实战教程 / Step-by-Step Tutorial
理论讲得够多了，现在是时候动手实践了！在这个部分，我会带你一步步创建一个功能完整的智能聊天机器人。
4.1 项目规划
在我们开始编码之前，先来规划一下这个机器人需要实现的功能：
目标：创建一个多功能个人助手机器人
核心功能：
1. 基础对话：回答日常问题
2. 代码助手：帮助编写和调试代码
3. 图片生成：根据描述生成图片
4. 天气查询：查询指定城市的天气
5. 提醒功能：设置定时提醒
6. 翻译功能：中英文互译
7. 群组管理：入群欢迎、关键词自动回复
技术栈：
- AstrBot 框架
- Python 3.10+
- 外部 API：OpenWeatherMap（天气）、OpenAI API（对话和绘图）
4.2 创建项目结构
首先，让我们创建一个规范的项目结构：
my_assistant_bot/
├── config/
│ └── settings.json # 主配置文件
├── plugins/ # 插件目录
│ ├── __init__.py
│ ├── weather/ # 天气插件
│ ├── translator/ # 翻译插件
│ ├── reminder/ # 提醒插件
│ ├── code_helper/ # 代码助手插件
│ └── image_gen/ # 图片生成插件
├── data/ # 数据目录
│ ├── knowledge_base/ # 知识库
│ └── sessions/ # 会话记录
├── utils/ # 工具函数
│ ├── __init__.py
│ ├── api_helpers.py # API 辅助函数
│ └── text_processors.py # 文本处理工具
├── run.py # 启动脚本
├── requirements.txt # 依赖列表
└── README.md # 项目文档
执行以下命令创建目录结构：
import os
# 项目根目录
project_root = "my_assistant_bot"
# 创建目录结构
directories = [
"config",
"plugins/weather",
"plugins/translator",
"plugins/reminder",
"plugins/code_helper",
"plugins/image_gen",
"data/knowledge_base",
"data/sessions",
"utils"
]
for directory in directories:
path = os.path.join(project_root, directory)
os.makedirs(path, exist_ok=True)
print(f"创建目录: {path}")
print("目录结构创建完成！")
4.3 编写主配置文件
接下来，编写主配置文件：
{
"bot": {
"name": "小助手",
"description": "你的智能个人助手",
"owner": "your-name",
"log_level": "INFO"
},
"platforms": {
"terminal": true,
"telegram": {
"enabled": false,
"bot_token": "",
"admin_ids": []
},
"discord": {
"enabled": false,
"bot_token": "",
"intents": []
}
},
"llm": {
"provider": "openai",
"model": "gpt-3.5-turbo",
"api_key": "${OPENAI_API_KEY}",
"temperature": 0.7,
"max_tokens": 2048
},
"plugins": {
"enabled": true,
"auto_reload": true,
"dir": "plugins"
},
"data": {
"dir": "data",
"sessions_dir": "data/sessions",
"knowledge_base_dir": "data/knowledge_base"
}
}
4.4 创建天气查询插件
让我们从简单的插件开始——天气查询插件。这个插件将展示如何：
- 创建基础的 AstrBot 插件
- 处理用户输入
- 调用外部 API
- 返回格式化结果
首先，创建插件目录和配置文件：
{
"name": "weather",
"version": "1.0.0",
"description": "查询天气预报",
"author": "Your Name",
"config": {
"api_key": "",
"default_city": "北京",
"temperature_unit": "celsius"
}
}
然后，编写插件的主代码：
"""
天气查询插件
功能：
- 查询当前天气
- 查询未来天气预报
- 支持多个城市
使用方法：
- 发送 "天气 北京" 查询北京天气
- 发送 "天气 北京 3天" 查询3天预报
"""
import json
import re
import aiohttp
from typing import Optional, Dict, Any
# ============================================================
# 插件元数据定义
# ============================================================
PLUGIN_NAME = "weather"
PLUGIN_VERSION = "1.0.0"
PLUGIN_DESCRIPTION = "查询天气预报"
class WeatherPlugin:
"""
天气查询插件主类
Attributes:
config: 插件配置
api_key: 天气API密钥
temperature_unit: 温度单位（celsius/fahrenheit）
"""
name = PLUGIN_NAME
version = PLUGIN_VERSION
description = PLUGIN_DESCRIPTION
def __init__(self, config: Dict[str, Any]):
"""
初始化天气插件
Args:
config: 插件配置字典
"""
self.config = config
self.api_key = config.get("api_key", "")
self.temperature_unit = config.get("temperature_unit", "celsius")
self.default_city = config.get("default_city", "北京")
# API 基础地址（这里以 OpenWeatherMap 为例）
self.base_url = "https://api.openweathermap.org/data/2.5"
# 城市名称映射（中文到英文）
self.city_mapping = {
"北京": "Beijing",
"上海": "Shanghai",
"广州": "Guangzhou",
"深圳": "Shenzhen",
"杭州": "Hangzhou",
"成都": "Chengdu",
"武汉": "Wuhan",
"西安": "Xi'an",
"重庆": "Chongqing",
"南京": "Nanjing"
}
async def on_load(self) -> None:
"""
插件加载时调用
可以在这里进行初始化操作，如验证API密钥等
"""
if not self.api_key:
print("[天气插件] 警告: 未配置 API 密钥，天气功能可能无法使用")
else:
print("[天气插件] 初始化完成")
async def query_weather(self, city: str, days: int = 1) -> str:
"""
查询天气
Args:
city: 城市名称
days: 预报天数（1表示当前天气）
Returns:
格式化的天气信息字符串
"""
# 转换城市名称（如果需要）
city_en = self.city_mapping.get(city, city)
if days == 1:
# 查询当前天气
return await self._get_current_weather(city_en, city)
else:
# 查询天气预报
return await self._get_forecast(city_en, city, days)
async def _get_current_weather(self, city_en: str, city_cn: str) -> str:
"""
获取当前天气
Args:
city_en: 英文城市名
city_cn: 中文城市名
Returns:
天气信息字符串
"""
if not self.api_key:
return "抱歉，天气查询服务暂时不可用（未配置API密钥）"
url = f"{self.base_url}/weather"
params = {
"q": city_en,
"appid": self.api_key,
"units": "metric" if self.temperature_unit == "celsius" else "imperial"
}
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params, timeout=10) as response:
if response.status == 200:
data = await response.json()
return self._format_current_weather(data, city_cn)
elif response.status == 404:
return f"未找到城市「{city_cn}」，请检查城市名称是否正确"
else:
return f"查询天气时出错，请稍后重试"
except aiohttp.ClientError as e:
return f"网络连接出错: {str(e)}"
except Exception as e:
return f"发生未知错误: {str(e)}"
async def _get_forecast(self, city_en: str, city_cn: str, days: int) -> str:
"""
获取天气预报
Args:
city_en: 英文城市名
city_cn: 中文城市名
days: 预报天数
Returns:
格式化的预报信息字符串
"""
if not self.api_key:
return "抱歉，天气预报服务暂时不可用（未配置API密钥）"
# 限制预报天数
days = min(days, 5)
url = f"{self.base_url}/forecast"
params = {
"q": city_en,
"appid": self.api_key,
"units": "metric" if self.temperature_unit == "celsius" else "imperial"
}
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params, timeout=10) as response:
if response.status == 200:
data = await response.json()
return self._format_forecast(data, city_cn, days)
else:
return f"查询天气预报时出错，请稍后重试"
except Exception as e:
return f"查询失败: {str(e)}"
def _format_current_weather(self, data: Dict, city: str) -> str:
"""
格式化当前天气信息
Args:
data: API返回的天气数据
city: 城市名称
Returns:
格式化的天气信息
"""
weather = data["weather"][0]
main = data["main"]
wind = data.get("wind", {})
# 天气图标映射
icon_map = {
"01d": "☀️", "01n": "🌙",
"02d": "⛅", "02n": "☁️",
"03d": "☁️", "03n": "☁️",
"04d": "☁️", "04n": "☁️",
"09d": "🌧️", "09n": "🌧️",
"10d": "🌦️", "10n": "🌧️",
"11d": "⛈️", "11n": "⛈️",
"13d": "❄️", "13n": "❄️",
"50d": "🌫️", "50n": "🌫️"
}
icon = icon_map.get(weather["icon"], "🌡️")
# 格式化输出
result = f"""
╔══════════════════════════════════════╗
║ 🌤️ {city} 当前天气 ║
╠══════════════════════════════════════╣
║ {icon} 天气状况: {weather['description']} ║
║ 🌡️ 温度: {main['temp']}°{'C' if self.temperature_unit == 'celsius' else 'F'} ║
║ 体感温度: {main['feels_like']}° ║
║ 💧 湿度: {main['humidity']}% ║
║ 💨 风速: {wind.get('speed', 'N/A')} m/s ║
╚══════════════════════════════════════╝
"""
return result
def _format_forecast(self, data: Dict, city: str, days: int) -> str:
"""
格式化天气预报信息
Args:
data: API返回的预报数据
city: 城市名称
days: 预报天数
Returns:
格式化的预报信息
"""
# 按天分组数据
daily_data = {}
for item in data["list"]:
date = item["dt_txt"].split()[0]
if date not in daily_data:
daily_data[date] = []
daily_data[date].append(item)
# 格式化每天的预报
result = f"🌤️ **{city} {days}天天气预报**\n\n"
count = 0
for date, items in sorted(daily_data.items()):
if count >= days:
break
# 取中午的数据作为代表
noon_data = None
for item in items:
hour = int(item["dt_txt"].split()[1].split(":")[0])
if 11 <= hour <= 14:
noon_data = item
break
if not noon_data:
noon_data = items[len(items) // 2]
weather = noon_data["weather"][0]
main = noon_data["main"]
# 天气图标
icon_map = {
"01d": "☀️", "01n": "🌙",
"02d": "⛅", "02n": "☁️",
"03d": "☁️", "03n": "☁️",
"04d": "☁️", "04n": "☁️",
"09d": "🌧️", "09n": "🌧️",
"10d": "🌦️", "10n": "🌧️",
"11d": "⛈️", "11n": "⛈️",
"13d": "❄️", "13n": "❄️"
}
icon = icon_map.get(weather["icon"], "🌡️")
# 格式化日期
from datetime import datetime
date_obj = datetime.strptime(date, "%Y-%m-%d")
date_str = date_obj.strftime("%m/%d")
weekday = ["一", "二", "三", "四", "五", "六", "日"][date_obj.weekday()]
result += f"📅 {date_str} (周{weekday})\n"
result += f" {icon} {weather['description']}\n"
result += f" 🌡️ {main['temp_min']:.0f}~{main['temp_max']:.0f}°\n\n"
count += 1
return result
async def handle_message(self, message: str, context: Dict) -> Optional[str]:
"""
处理用户消息
这是插件与框架交互的主要入口点
Args:
message: 用户发送的原始消息
context: 消息上下文，包含用户信息、会话信息等
Returns:
要回复的消息内容，如果不需要回复则返回 None
"""
# 解析命令
pattern = r"^天气\s*(.+?)(?:\s+(\d+)\s*天)?$"
match = re.match(pattern, message.strip())
if not match:
return None
city = match.group(1).strip()
days = int(match.group(2)) if match.group(2) else 1
# 查询天气
return await self.query_weather(city, days)
# 导出插件类
# 这是框架识别插件的关键
__plugin_class__ = WeatherPlugin
4.5 创建翻译插件
接下来，让我们创建一个翻译插件，展示如何处理更复杂的文本处理任务：
"""
翻译插件
功能：
- 中英文互译
- 多语言支持
- 自动检测源语言
使用方法：
- 发送 "翻译 Hello" 翻译成中文
- 发送 "翻译 你好 world" 中英文混合翻译
"""
import re
from typing import Optional, Dict, Any
from googletrans import Translator
class TranslatorPlugin:
"""
翻译插件
Attributes:
config: 插件配置
supported_languages: 支持的语言列表
"""
name = "translator"
version = "1.0.0"
description = "多语言翻译功能"
# 支持的语言映射
LANGUAGE_CODES = {
"中文": "zh-cn",
"英文": "en",
"日语": "ja",
"韩语": "ko",
"法语": "fr",
"德语": "de",
"西班牙语": "es",
"俄语": "ru",
"阿拉伯语": "ar"
}
def __init__(self, config: Dict[str, Any]):
"""
初始化翻译插件
Args:
config: 插件配置
"""
self.config = config
self.translator = Translator()
async def on_load(self) -> None:
"""插件加载时的初始化"""
print("[翻译插件] 翻译功能已就绪")
async def translate(
self,
text: str,
source_lang: Optional[str] = None,
target_lang: str = "zh-cn"
) -> str:
"""
执行翻译
Args:
text: 要翻译的文本
source_lang: 源语言（None表示自动检测）
target_lang: 目标语言
Returns:
翻译结果
"""
try:
# 自动检测语言
if source_lang is None:
source_lang = "auto"
elif source_lang in self.LANGUAGE_CODES:
source_lang = self.LANGUAGE_CODES[source_lang]
if target_lang in self.LANGUAGE_CODES:
target_lang = self.LANGUAGE_CODES[target_lang]
# 执行翻译
result = self.translator.translate(
text,
src=source_lang,
dest=target_lang
)
# 获取语言名称
src_lang_name = self._get_language_name(result.src)
dest_lang_name = self._get_language_name(result.dest)
# 格式化输出
output = f"""
🌐 **翻译结果**
📝 源语言: {src_lang_name}
📄 目标语言: {dest_lang_name}
━━━━━━━━━━━━━━━
{result.text}
━━━━━━━━━━━━━━━
💡 原文: "{text}"
"""
return output
except Exception as e:
return f"翻译失败: {str(e)}\n请检查网络连接后重试"
def _get_language_name(self, code: str) -> str:
"""
将语言代码转换为语言名称
Args:
code: 语言代码
Returns:
语言名称
"""
code_lower = code.lower()
for name, c in self.LANGUAGE_CODES.items():
if c.lower() == code_lower:
return name
return code.upper()
async def handle_message(self, message: str, context: Dict) -> Optional[str]:
"""
处理翻译命令
Args:
message: 用户消息
context: 消息上下文
Returns:
翻译结果或 None
"""
# 匹配命令格式
# 格式: "翻译 <内容>" 或 "翻译 <内容> 到 <目标语言>"
patterns = [
r"^翻译\s+(.+?)(?:\s+到\s+(.+))?$",
r"^translate\s+(.+?)(?:\s+to\s+(.+))?$"
]
for pattern in patterns:
match = re.match(pattern, message.strip(), re.IGNORECASE)
if match:
text = match.group(1).strip()
target = match.group(2).strip() if match.group(2) else "中文"
# 简化目标语言名称
target_map = {
"中文": "中文",
"英文": "英文",
"日语": "日语",
"en": "英文",
"zh": "中文",
"ja": "日语"
}
target = target_map.get(target, target)
return await self.translate(text, target_lang=target)
return None
# 导出插件类
__plugin_class__ = TranslatorPlugin
4.6 创建代码助手插件
对于开发者来说，一个能够帮忙写代码、解释代码的插件是非常有用的。让我们来创建一个：
"""
代码助手插件
功能：
- 代码生成：根据描述生成代码
- 代码解释：解释代码的功能
- 代码debug：帮助查找和修复bug
使用方法：
- 发送 "写代码 用Python实现快速排序"
- 发送 "解释代码 [代码块]"
- 发送 "debug [有问题的代码]"
"""
import re
import aiohttp
from typing import Optional, Dict, Any, List
class CodeHelperPlugin:
"""
代码助手插件
这个插件使用 LLM 来帮助用户处理编程相关的问题
"""
name = "code_helper"
version = "1.0.0"
description = "编程助手功能"
# 支持的编程语言
SUPPORTED_LANGUAGES = [
"python", "javascript", "typescript", "java", "cpp",
"c", "csharp", "go", "rust", "ruby", "php", "swift",
"kotlin", "scala", "sql", "html", "css", "shell"
]
def __init__(self, config: Dict[str, Any]):
"""
初始化代码助手
Args:
config: 插件配置
"""
self.config = config
self.llm_endpoint = config.get("llm_endpoint", "")
self.llm_api_key = config.get("llm_api_key", "")
async def on_load(self) -> None:
"""插件加载"""
print("[代码助手] 编程助手已就绪")
async def generate_code(
self,
description: str,
language: Optional[str] = None
) -> str:
"""
根据描述生成代码
Args:
description: 功能描述
language: 编程语言
Returns:
生成的代码
"""
lang_prompt = f"使用 {language}" if language else ""
prompt = f"""你是一个专业的编程助手。请根据用户的需求生成代码。
需求：{description}
{lang_prompt}
要求：
1. 代码要完整、可运行
2. 添加必要的注释说明
3. 遵循最佳实践和编码规范
4. 如果需要额外依赖，请注明
请直接返回代码（使用三个反引号包裹）和简要说明，不要有其他废话。
"""
try:
result = await self._call_llm(prompt)
return result
except Exception as e:
return f"生成代码时出错: {str(e)}"
async def explain_code(self, code: str) -> str:
"""
解释代码功能
Args:
code: 要解释的代码
Returns:
代码解释
"""
prompt = f"""你是一个专业的编程助手。请解释以下代码的功能和工作原理。
```{code}```
请按以下格式回答：
1. 代码整体功能概述
2. 关键部分的作用
3. 可能的输出或效果
4. 注意事项（如有）
"""
try:
result = await self._call_llm(prompt)
return f"📖 **代码解释**\n\n{result}"
except Exception as e:
return f"解释代码时出错: {str(e)}"
async def debug_code(self, code: str, error: Optional[str] = None) -> str:
"""
帮助调试代码
Args:
code: 有问题的代码
error: 错误信息（如果有）
Returns:
调试建议
"""
error_hint = f"\n\n出现的错误：\n{error}" if error else ""
prompt = f"""你是一个专业的编程助手。请帮助调试以下代码中的问题。
```{code}```{error_hint}
请按以下格式回答：
1. **问题分析**：指出代码中存在的主要问题
2. **修复建议**：给出修改建议和修复后的代码
3. **预防建议**：如何避免类似问题
"""
try:
result = await self._call_llm(prompt)
return f"🔧 **代码调试**\n\n{result}"
except Exception as e:
return f"调试代码时出错: {str(e)}"
async def optimize_code(self, code: str) -> str:
"""
优化代码
Args:
code: 需要优化的代码
Returns:
优化建议
"""
prompt = f"""你是一个专业的编程助手。请分析并优化以下代码。
```{code}```
请按以下格式回答：
1. **当前问题**：代码可能存在的性能或可读性问题
2. **优化方案**：具体的优化措施
3. **优化后代码**：优化后的完整代码
4. **预期收益**：优化后的效果
"""
try:
result = await self._call_llm(prompt)
return f"⚡ **代码优化**\n\n{result}"
except Exception as e:
return f"优化代码时出错: {str(e)}"
async def _call_llm(self, prompt: str) -> str:
"""
调用大语言模型
Args:
prompt: 发送给模型的提示词
Returns:
模型的响应
"""
if not self.llm_endpoint:
**如果没有配置LLM，使用模拟响应**
return "[模拟响应] 请配置 LLM endpoint 以使用此功能"
headers = {
"Content-Type": "application/json"
}
if self.llm_api_key:
headers["Authorization"] = f"Bearer {self.llm_api_key}"
payload = {
"prompt": prompt,
"max_tokens": 2048,
"temperature": 0.3
}
async with aiohttp.ClientSession() as session:
async with session.post(
self.llm_endpoint,
json=payload,
headers=headers,
timeout=60
) as response:
if response.status == 200:
data = await response.json()
return data.get("response", "")
else:
raise Exception(f"API返回错误: {response.status}")
async def handle_message(self, message: str, context: Dict) -> Optional[str]:
"""
处理用户消息
Args:
message: 用户消息
context: 消息上下文
Returns:
响应内容或 None
"""
message = message.strip()
**写代码命令**
**============================================================**
**匹配格式: "写代码 <语言> <描述>" 或 "写代码 <描述>"**
**============================================================**
write_pattern = r"^写代码\s*(?:([a-zA-Z]+)\s+)?(.+)$"
match = re.match(write_pattern, message)
if match:
language = match.group(1)
description = match.group(2)
if language and language.lower() not in self.SUPPORTED_LANGUAGES:
return f"抱歉，暂不支持 {language} 语言"
return await self.generate_code(description, language)
**解释代码命令**
**============================================================**
**匹配格式: "解释代码\n```代码```" 或 "解释 ```代码```"**
**============================================================**
if message.startswith("解释代码"):
code = self._extract_code(message.replace("解释代码", "").strip())
if code:
return await self.explain_code(code)
return "请提供要解释的代码（使用代码块包裹）"
**debug命令**
if message.startswith("debug"):
code = self._extract_code(message.replace("debug", "").strip())
if code:
return await self.debug_code(code)
return "请提供要调试的代码（使用代码块包裹）"
**优化命令**
if message.startswith("优化代码"):
code = self._extract_code(message.replace("优化代码", "").strip())
if code:
return await self.optimize_code(code)
return "请提供要优化的代码（使用代码块包裹）"
return None
def _extract_code(self, text: str) -> Optional[str]:
"""
从文本中提取代码块
Args:
text: 包含代码块的文本
Returns:
提取出的代码，如果没有则返回 None
"""
**匹配 markdown 代码块**
pattern = r"```(?:\w+)?\s*\n?(.*?)\n?```"
match = re.search(pattern, text, re.DOTALL)
if match:
return match.group(1).strip()
**如果没有代码块，检查是否是纯代码**
if text and not text.startswith("`"):
return text.strip()
return None
**导出插件类**
__plugin_class__ = CodeHelperPlugin
4.7 创建提醒插件
让我们再创建一个提醒插件，展示如何使用定时任务功能：
"""
提醒插件
功能：
- 设置一次性提醒
- 设置循环提醒
- 列出和取消提醒
使用方法：
- 发送 "提醒 30分钟后开会" 设置30分钟后的提醒
- 发送 "提醒 每天 9:00 打卡" 设置每天9点的提醒
- 发送 "我的提醒" 查看所有提醒
- 发送 "取消提醒 1" 取消第1个提醒
"""
import re
import asyncio
import time
from datetime import datetime, timedelta
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
from enum import Enum
class ReminderType(Enum):
"""提醒类型枚举"""
ONCE = "once" # 一次性提醒
DAILY = "daily" # 每天提醒
WEEKLY = "weekly" # 每周提醒
WORKDAY = "workday" # 工作日提醒
@dataclass
class Reminder:
"""
提醒数据结构
Attributes:
id: 唯一标识符
user_id: 用户ID
content: 提醒内容
type: 提醒类型
next_trigger: 下次触发时间
interval: 触发间隔（用于循环提醒）
created_at: 创建时间
"""
id: str
user_id: str
content: str
type: ReminderType
next_trigger: float # Unix时间戳
interval: Optional[int] = None # 秒为单位
created_at: float = field(default_factory=time.time)
def is_due(self) -> bool:
"""检查是否到期"""
return time.time() >= self.next_trigger
def get_next_trigger_str(self) -> str:
"""获取下次触发时间的可读字符串"""
dt = datetime.fromtimestamp(self.next_trigger)
if self.type == ReminderType.DAILY:
return dt.strftime("每天 %H:%M")
elif self.type == ReminderType.WEEKLY:
weekdays = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
return f"每周{dt.strftime('%w')}"
else:
return dt.strftime("%Y-%m-%d %H:%M:%S")
def reschedule(self) -> None:
"""重新安排下次触发时间"""
if self.type == ReminderType.ONCE:
return
elif self.type == ReminderType.DAILY:
self.next_trigger += 86400 # 加一天
elif self.type == ReminderType.WEEKLY:
self.next_trigger += 604800 # 加一周
elif self.type == ReminderType.WORKDAY:
**下一个工作日**
while True:
self.next_trigger += 86400
weekday = datetime.fromtimestamp(self.next_trigger).weekday()
if weekday < 5: # 周一到周五
break
class ReminderManager:
"""
提醒管理器
负责管理所有的提醒，包括添加、删除、触发等
"""
def __init__(self):
"""初始化提醒管理器"""
self.reminders: Dict[str, Reminder] = {}
self.counter = 0
self._running = False
self._task = None
def add_reminder(
self,
user_id: str,
content: str,
reminder_type: ReminderType,
next_trigger: float,
interval: Optional[int] = None
) -> str:
"""
添加新提醒
Args:
user_id: 用户ID
content: 提醒内容
reminder_type: 提醒类型
next_trigger: 下次触发时间戳
interval: 间隔秒数（用于循环提醒）
Returns:
提醒ID
"""
self.counter += 1
reminder_id = str(self.counter)
reminder = Reminder(
id=reminder_id,
user_id=user_id,
content=content,
type=reminder_type,
next_trigger=next_trigger,
interval=interval
)
self.reminders[reminder_id] = reminder
return reminder_id
def remove_reminder(self, reminder_id: str, user_id: str) -> bool:
"""
删除提醒
Args:
reminder_id: 提醒ID
user_id: 用户ID（用于验证所有权）
Returns:
是否删除成功
"""
if reminder_id in self.reminders:
reminder = self.reminders[reminder_id]
if reminder.user_id == user_id:
del self.reminders[reminder_id]
return True
return False
def get_user_reminders(self, user_id: str) -> List[Reminder]:
"""
获取用户的所有提醒
Args:
user_id: 用户ID
Returns:
提醒列表
"""
return [
r for r in self.reminders.values()
if r.user_id == user_id
]
async def start(self) -> None:
"""启动提醒检查循环"""
self._running = True
self._task = asyncio.create_task(self._check_loop())
async def stop(self) -> None:
"""停止提醒检查"""
self._running = False
if self._task:
self._task.cancel()
async def _check_loop(self) -> None:
"""定期检查并触发提醒"""
while self._running:
try:
current_time = time.time()
due_reminders = [
r for r in self.reminders.values()
if r.is_due()
]
for reminder in due_reminders:
await self._trigger_reminder(reminder)
**处理循环提醒**
if reminder.type != ReminderType.ONCE:
reminder.reschedule()
else:
self.remove_reminder(reminder.id, reminder.user_id)
**每秒检查一次**
await asyncio.sleep(1)
except asyncio.CancelledError:
break
except Exception as e:
print(f"检查提醒时出错: {e}")
await asyncio.sleep(1)
async def _trigger_reminder(self, reminder: Reminder) -> None:
"""
触发提醒
这个方法会被调用者的消息处理逻辑使用
实际的通知发送由框架处理
"""
**通知信息会被添加到待发送消息队列**
**这里只记录日志**
print(f"[提醒触发] 用户 {reminder.user_id}: {reminder.content}")
class ReminderPlugin:
"""
提醒插件
提供定时提醒功能
"""
name = "reminder"
version = "1.0.0"
description = "定时提醒功能"
def __init__(self, config: Dict[str, Any]):
"""
初始化提醒插件
Args:
config: 插件配置
"""
self.config = config
self.manager = ReminderManager()
self.pending_notifications: Dict[str, List[str]] = {} # 用户待发送通知
async def on_load(self) -> None:
"""插件加载"""
await self.manager.start()
print("[提醒插件] 定时提醒功能已启动")
async def on_unload(self) -> None:
"""插件卸载"""
await self.manager.stop()
print("[提醒插件] 已关闭")
def parse_reminder_time(self, time_str: str) -> Optional[float]:
"""
解析时间字符串
Args:
time_str: 时间描述字符串
Returns:
Unix时间戳，如果解析失败返回 None
"""
time_str = time_str.strip()
now = datetime.now()
**============================================================**
**模式1: "30分钟后"、"1小时后"、"2小时30分钟后"**
**============================================================**
pattern1 = r"(\d+)\s*小时\s*(\d+)?\s*分钟?\s*后"
match = re.match(pattern1, time_str)
if match:
hours = int(match.group(1))
minutes = int(match.group(2)) if match.group(2) else 0
future = now + timedelta(hours=hours, minutes=minutes)
return future.timestamp()
**============================================================**
**模式2: "30分钟后" (只有分钟)**
**============================================================**
pattern2 = r"(\d+)\s*分钟\s*后"
match = re.match(pattern2, time_str)
if match:
minutes = int(match.group(1))
future = now + timedelta(minutes=minutes)
return future.timestamp()
**============================================================**
**模式3: "明天 9:00"、"今天 14:30"**
**============================================================**
pattern3 = r"(今天|明天)\s*(\d{1,2}):(\d{2})"
match = re.match(pattern3, time_str)
if match:
day_spec, hour, minute = match.groups()
target = now.replace(hour=int(hour), minute=int(minute), second=0)
if day_spec == "明天":
target += timedelta(days=1)
elif target < now:
target += timedelta(days=1)
return target.timestamp()
**============================================================**
**模式4: "2024-01-01 09:00" 或 "1月1日 9:00"**
**============================================================**
patterns = [
(r"(\d{4})-(\d{1,2})-(\d{1,2})\s+(\d{1,2}):(\d{2})", "%Y-%m-%d %H:%M"),
(r"(\d{1,2})月(\d{1,2})日\s+(\d{1,2}):(\d{2})", "%m月%d日 %H:%M")
]
for pattern, fmt in patterns:
match = re.match(pattern, time_str)
if match:
try:
if "%Y" in fmt:
parts = [int(g) for g in match.groups()]
target = datetime(
parts[0], parts[1], parts[2],
parts[3], parts[4]
)
else:
parts = [int(g) for g in match.groups()]
year = now.year
**检查日期是否已过**
target = now.replace(
month=parts[0], day=parts[1],
hour=parts[2], minute=parts[3],
second=0, microsecond=0
)
if target < now:
target = target.replace(year=year + 1)
return target.timestamp()
except ValueError:
continue
return None
async def set_reminder(
self,
user_id: str,
content: str,
time_str: str,
repeat: Optional[str] = None
) -> str:
"""
设置提醒
Args:
user_id: 用户ID
content: 提醒内容
time_str: 时间描述
repeat: 重复模式 ("每天"、"每周"、"工作日")
Returns:
操作结果
"""
**解析时间**
trigger_time = self.parse_reminder_time(time_str)
if trigger_time is None:
return f"无法理解时间「{time_str}」，请使用以下格式之一：\n" \
f"• 30分钟后\n• 明天 9:00\n• 2024-01-01 09:00"
**解析重复模式**
reminder_type = ReminderType.ONCE
interval = None
if repeat == "每天":
reminder_type = ReminderType.DAILY
interval = 86400
elif repeat == "每周":
reminder_type = ReminderType.WEEKLY
interval = 604800
elif repeat == "工作日":
reminder_type = ReminderType.WORKDAY
**添加提醒**
reminder_id = self.manager.add_reminder(
user_id=user_id,
content=content,
reminder_type=reminder_type,
next_trigger=trigger_time,
interval=interval
)
**格式化确认信息**
reminder = self.manager.reminders[reminder_id]
type_hint = ""
if reminder_type != ReminderType.ONCE:
type_hint = f"（{repeat}）"
return f"✅ 提醒已设置！\n\n" \
f"📝 内容: {content}\n" \
f"⏰ 时间: {reminder.get_next_trigger_str()}{type_hint}\n" \
f"🆔 编号: #{reminder_id}"
async def list_reminders(self, user_id: str) -> str:
"""
列出用户的所有提醒
Args:
user_id: 用户ID
Returns:
提醒列表
"""
reminders = self.manager.get_user_reminders(user_id)
if not reminders:
return "📋 你还没有设置任何提醒"
result = "📋 **你的提醒列表**\n\n"
for i, reminder in enumerate(reminders, 1):
type_emoji = {
ReminderType.ONCE: "⏰",
ReminderType.DAILY: "🔁",
ReminderType.WEEKLY: "📅",
ReminderType.WORKDAY: "💼"
}[reminder.type]
type_text = {
ReminderType.ONCE: "一次性",
ReminderType.DAILY: "每天",
ReminderType.WEEKLY: "每周",
ReminderType.WORKDAY: "工作日"
}[reminder.type]
result += f"{i}. {type_emoji} #{reminder.id}\n"
result += f" 📝 {reminder.content}\n"
result += f" ⏰ {reminder.get_next_trigger_str()} ({type_text})\n\n"
result += "\n💡 使用「取消提醒 #编号」删除提醒"
return result
async def cancel_reminder(
self,
user_id: str,
reminder_id: str
) -> str:
"""
取消提醒
Args:
user_id: 用户ID
reminder_id: 提醒ID
Returns:
操作结果
"""
**尝试直接用ID删除**
if self.manager.remove_reminder(reminder_id, user_id):
return f"✅ 提醒 #{reminder_id} 已取消"
**尝试用序号删除**
try:
index = int(reminder_id) - 1
user_reminders = self.manager.get_user_reminders(user_id)
if 0 <= index < len(user_reminders):
target = user_reminders[index]
if self.manager.remove_reminder(target.id, user_id):
return f"✅ 提醒 #{target.id} 已取消"
except ValueError:
pass
return f"❌ 未找到提醒「{reminder_id}」"
async def handle_message(self, message: str, context: Dict) -> Optional[str]:
"""
处理用户消息
Args:
message: 用户消息
context: 消息上下文
Returns:
响应内容或 None
"""
user_id = context.get("user_id", "anonymous")
message = message.strip()
**============================================================**
**设置提醒**
**============================================================**
**格式: "提醒 <时间> <内容>"**
**============================================================**
patterns = [
r"提醒\s*(?:(每天|每周|工作日)\s+)?(.+?)\s+(.{2,})$",
r"提醒\s*(.+)$"
]
for pattern in patterns:
match = re.match(pattern, message)
if match:
if len(match.groups()) == 3:
repeat, time_str, content = match.groups()
return await self.set_reminder(
user_id, content, time_str, repeat
)
else:
time_str = match.group(1)
return f"请提供提醒内容，格式：提醒 <时间> <内容>\n" \
f"例如：提醒 30分钟后 开会"
**============================================================**
**列出提醒**
**============================================================**
if message in ["我的提醒", "查看提醒", "提醒列表"]:
return await self.list_reminders(user_id)
**============================================================**
**取消提醒**
**============================================================**
cancel_pattern = r"取消提醒\s*(.+)$"
match = re.match(cancel_pattern, message)
if match:
reminder_id = match.group(1).strip()
return await self.cancel_reminder(user_id, reminder_id)
return None
def get_pending_notifications(self, user_id: str) -> List[str]:
"""
获取用户待处理的通知
Args:
user_id: 用户ID
Returns:
通知消息列表
"""
notifications = self.pending_notifications.get(user_id, [])
self.pending_notifications[user_id] = []
return notifications
async def check_due_reminders(self, user_id: str) -> List[str]:
"""
检查到期提醒（由框架定时调用）
Args:
user_id: 用户ID
Returns:
到期的提醒内容列表
"""
user_reminders = self.manager.get_user_reminders(user_id)
due_content = []
for reminder in user_reminders:
if reminder.is_due():
due_content.append(reminder.content)
**处理循环提醒**
if reminder.type != ReminderType.ONCE:
reminder.reschedule()
else:
self.manager.remove_reminder(reminder.id, reminder.user_id)
return due_content
**导出插件类**
__plugin_class__ = ReminderPlugin
4.8 创建主程序入口
现在，让我们创建主程序入口，将所有插件整合在一起：
#!/usr/bin/env python3
"""
AstrBot 主程序入口
这是一个完整的多功能助手机器人示例
整合了天气、翻译、代码助手和提醒功能
"""
import asyncio
import json
import os
import sys
from typing import Dict, Any, Optional
**添加项目根目录到路径**
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
**导入插件**
from plugins.weather import WeatherPlugin
from plugins.translator import TranslatorPlugin
from plugins.code_helper import CodeHelperPlugin
from plugins.reminder import ReminderPlugin
class AstrBotApp:
"""
AstrBot 应用主类
负责初始化、配置和运行机器人
"""
def __init__(self, config_path: str = "config/settings.json"):
"""
初始化应用
Args:
config_path: 配置文件路径
"""
self.config_path = config_path
self.config: Dict[str, Any] = {}
self.plugins: Dict[str, Any] = {}
self.running = False
**加载配置**
self._load_config()
def _load_config(self) -> None:
"""加载配置文件"""
if not os.path.exists(self.config_path):
print(f"[警告] 配置文件不存在: {self.config_path}")
print("将使用默认配置")
self.config = self._get_default_config()
return
with open(self.config_path, "r", encoding="utf-8") as f:
content = f.read()
**处理环境变量**
import re
pattern = r'\$\{([^}]+)\}'
for match in re.finditer(pattern, content):
env_var = match.group(1)
env_value = os.environ.get(env_var, "")
content = content.replace(match.group(0), env_value)
self.config = json.loads(content)
def _get_default_config(self) -> Dict[str, Any]:
"""
获取默认配置
Returns:
默认配置字典
"""
return {
"bot": {
"name": "AstrBot助手",
"description": "多功能智能助手",
"log_level": "INFO"
},
"platforms": {
"terminal": True
},
"plugins": {
"enabled": True,
"auto_reload": True
}
}
async def initialize_plugins(self) -> None:
"""
初始化所有插件
这个方法创建插件实例并调用其初始化方法
"""
print("=" * 50)
print("正在初始化插件...")
print("=" * 50)
**获取插件配置**
plugin_configs = self.config.get("plugins", {})
**初始化各个插件**
**============================================================**
**注意：实际使用时需要确保插件配置存在**
**============================================================**
**天气插件**
if os.path.exists("plugins/weather"):
self.plugins["weather"] = WeatherPlugin(
plugin_configs.get("weather", {})
)
await self.plugins["weather"].on_load()
print("[✓] 天气插件已加载")
**翻译插件**
if os.path.exists("plugins/translator"):
self.plugins["translator"] = TranslatorPlugin(
plugin_configs.get("translator", {})
)
await self.plugins["translator"].on_load()
print("[✓] 翻译插件已加载")
**代码助手插件**
if os.path.exists("plugins/code_helper"):
self.plugins["code_helper"] = CodeHelperPlugin(
plugin_configs.get("code_helper", {})
)
await self.plugins["code_helper"].on_load()
print("[✓] 代码助手插件已加载")
**提醒插件**
if os.path.exists("plugins/reminder"):
self.plugins["reminder"] = ReminderPlugin(
plugin_configs.get("reminder", {})
)
await self.plugins["reminder"].on_load()
print("[✓] 提醒插件已加载")
print(f"\n共加载 {len(self.plugins)} 个插件")
print("=" * 50)
async def process_message(
self,
message: str,
context: Dict[str, Any]
) -> Optional[str]:
"""
处理收到的消息
按优先级依次尝试每个插件，直到某个插件返回结果
Args:
message: 用户消息
context: 消息上下文
Returns:
机器人回复，如果没有插件处理则返回 None
"""
**按优先级顺序尝试处理**
plugin_order = ["reminder", "weather", "translator", "code_helper"]
for plugin_name in plugin_order:
if plugin_name not in self.plugins:
continue
plugin = self.plugins[plugin_name]
handler = getattr(plugin, "handle_message", None)
if handler:
try:
result = await handler(message, context)
if result:
return result
except Exception as e:
print(f"[错误] 插件 {plugin_name} 处理消息时出错: {e}")
**没有插件处理，返回默认回复**
return None
async def run_terminal_mode(self) -> None:
"""
运行终端交互模式
这是一个简单的命令行界面，用于测试和演示
"""
print("\n" + "=" * 50)
print("欢迎使用 AstrBot 助手！")
print("=" * 50)
print("\n可用命令：")
print(" • 天气 <城市> - 查询天气")
print(" • 翻译 <文本> - 翻译文本")
print(" • 写代码 <描述> - 生成代码")
print(" • 提醒 <时间> <内容> - 设置提醒")
print(" • 我的提醒 - 查看所有提醒")
print(" • 退出 - 结束程序")
print("-" * 50)
self.running = True
context = {"user_id": "terminal_user", "platform": "terminal"}
while self.running:
try:
**获取用户输入**
user_input = input("\n你: ").strip()
if not user_input:
continue
**处理退出命令**
if user_input in ["退出", "exit", "quit", "q"]:
print("\n感谢使用，再见！")
self.running = False
break
**处理消息**
response = await self.process_message(user_input, context)
if response:
print(f"\nAstrBot: {response}")
else:
print("\nAstrBot: 抱歉，我不太明白你的意思。请尝试其他命令。")
except KeyboardInterrupt:
print("\n\n程序被用户中断")
self.running = False
break
except Exception as e:
print(f"\n[错误] {e}")
async def start(self) -> None:
"""
启动机器人
这是应用的主入口点
"""
try:
**初始化插件**
await self.initialize_plugins()
**根据配置决定运行模式**
platforms = self.config.get("platforms", {})
if platforms.get("terminal", False):
**终端模式**
await self.run_terminal_mode()
else:
**其他平台模式（需要额外配置）**
print("请在配置文件中启用至少一个平台")
print("设置 platforms.terminal = true 以使用终端模式")
except Exception as e:
print(f"[致命错误] {e}")
raise
finally:
**清理工作**
await self.cleanup()
async def cleanup(self) -> None:
"""
清理资源
在程序结束时调用，用于关闭所有插件和释放资源
"""
print("\n正在清理资源...")
for plugin_name, plugin in self.plugins.items():
cleanup = getattr(plugin, "on_unload", None)
if cleanup:
try:
await cleanup()
except Exception as e:
print(f"清理插件 {plugin_name} 时出错: {e}")
print("清理完成")
**============================================================**
**程序入口点**
**============================================================**
async def main():
"""
异步主函数
这是程序的真正入口点
"""
**创建应用实例**
app = AstrBotApp()
**启动应用**
await app.start()
if __name__ == "__main__":
**使用 asyncio 运行主函数**
asyncio.run(main())
4.9 创建配置文件
最后，创建插件的配置文件：
{
"weather": {
"api_key": "",
"default_city": "北京",
"temperature_unit": "celsius"
},
"translator": {
"enabled": true,
"default_target": "zh-cn"
},
"code_helper": {
"llm_endpoint": "",
"llm_api_key": "",
"default_language": "python"
},
"reminder": {
"enabled": true,
"max_reminders_per_user": 50
}
}
4.10 运行你的机器人
完成了所有代码后，让我们来运行你的第一个完整的多功能 AstrBot 机器人！
首先，确保所有文件都已创建：
my_assistant_bot/
├── config/
│ └── settings.json
├── plugins/
│ ├── __init__.py
│ ├── weather/
│ │ └── __init__.py
│ ├── translator/
│ │ └── __init__.py
│ ├── code_helper/
│ │ └── __init__.py
│ └── reminder/
│ └── __init__.py
├── data/
│ └── .gitkeep
├── run.py
├── main.py
└── requirements.txt
然后，运行机器人：
python main.py
你应该能看到类似这样的输出：
==================================================
正在初始化插件...
==================================================
[天气插件] 初始化完成
[✓] 天气插件已加载
[翻译插件] 翻译功能已就绪
[✓] 翻译插件已加载
[✓] 代码助手插件已就绪
[✓] 代码助手插件已加载
[提醒插件] 定时提醒功能已启动
[✓] 提醒插件已加载
共加载 4 个插件
==================================================
==================================================
欢迎使用 AstrBot 助手！
==================================================
可用命令：
• 天气 <城市> - 查询天气
• 翻译 <文本> - 翻译文本
• 写代码 <描述> - 生成代码
• 提醒 <时间> <内容> - 设置提醒
• 我的提醒 - 查看所有提醒
• 退出 - 结束程序
--------------------------------------------------
你:
现在，让我们测试一下各个功能：
你: 天气 北京
AstrBot:
╔══════════════════════════════════════╗
║ 🌤️ 北京 当前天气 ║
╠══════════════════════════════════════╣
║ ☀️ 天气状况: clear sky ║
║ 🌡️ 温度: 15°C ║
║ 体感温度: 14° ║
║ 💧 湿度: 45% ║
║ 💨 风速: 2.5 m/s ║
╚══════════════════════════════════════╝
你: 翻译 Hello, how are you?
AstrBot:
🌐 **翻译结果**
📝 源语言: 英文
📄 目标语言: 中文
━━━━━━━━━━━━━━━
你好，你好吗？
━━━━━━━━━━━━━━━
💡 原文: "Hello, how are you?"
你: 提醒 5秒后 测试提醒
AstrBot:
✅ 提醒已设置！
📝 内容: 测试提醒
⏰ 时间: 2024-01-15 14:30:05 (一次性)
🆔 编号: #1
你: 我的提醒
AstrBot:
📋 **你的提醒列表**
1. ⏰ #1
📝 测试提醒
⏰ 2024-01-15 14:30:05 (一次性)
💡 使用「取消提醒 #编号」删除提醒
恭喜！你已经成功创建并运行了一个功能完整的 AstrBot 机器人！
五、常见使用场景 / Common Use Cases
通过上面的实战教程，你应该已经对如何开发 AstrBot 插件有了清晰的了解。现在，让我们来看看 AstrBot 在实际生活中的一些常见应用场景。
5.1 个人助手
AstrBot 最直接的应用就是作为个人数字助手。它可以帮助你处理日常事务：
日程管理
用户: 提醒我明天下午3点开会
Bot: ✅ 已设置提醒！
📝 内容: 开会
⏰ 时间: 明天 15:00 (一次性)
🆔 编号: #42
信息查询
用户: 翻译一下这个产品说明书
Bot: [提供翻译结果]
文件处理
用户: 帮我把这个PDF转成Word
Bot: 正在处理...（通过调用相应插件完成）
5.2 客服机器人
对于企业和个人开发者来说，AstrBot 可以作为一个强大的客服解决方案：
多平台统一响应
**同一个机器人响应不同平台的用户**
**微信用户咨询**
微信用户: 产品多少钱？
Bot: 我们的产品价格是...
**Telegram用户咨询**
Telegram用户: 产品多少钱？
Bot: 我们的产品价格是...
FAQ 自动回复
用户: 怎么重置密码？
Bot: 重置密码步骤如下：
1. 点击登录页的"忘记密码"
2. 输入您的注册邮箱
3. 查收邮件并点击链接
4. 设置新密码
如有其他问题，请回复"人工客服"
工单处理
用户: 我的订单还没到，已经5天了
Bot: 抱歉给您带来困扰！让我帮您查询一下订单状态。
您的订单号是？或者我可以查看您的历史订单。
5.3 社区管理助手
在 Discord、QQ 群等社区平台中，AstrBot 可以扮演一个活跃的管理员角色：
自动欢迎新成员
async def on_member_join(self, member, context):
"""新成员加入时的欢迎消息"""
welcome_msg = f"""
🎉 欢迎 @{member.name} 加入我们的社区！
这里是一个友好、互助的大家庭。
请先阅读群公告了解社区规则。
有任何问题，随时提问，社区成员都会热心帮助！
📚 常用命令：
- /help - 查看帮助
- /rules - 查看规则
- /intro - 自我介绍
"""
return welcome_msg
关键词过滤和响应
**自动过滤敏感词**
用户: xxx（发送敏感内容）
Bot: ⚠️ 请注意，您的消息包含不当内容，请重新发送。
**自动响应关键词**
用户: 求帮助
Bot: 👋 看起来你需要帮助！请描述一下你的问题，我会尽力协助。
社区活动管理
用户: /活动报名 编程马拉松
Bot: ✅ 报名成功！
活动名称: 编程马拉松
报名时间: 2024年1月20日 09:00
地点: 线上
联系人: @组织者
📋 完整报名信息已发送至您的私信
5.4 开发助手
对于开发者来说，AstrBot 是一个得力的编程助手：
代码审查
用户: 请帮我review这段代码
```python
def get_user(id):
return db.query(id)
Bot: 🔍 代码审查结果
问题发现：
-
⚠️ SQL注入风险
db.query(id)直接使用外部输入，建议使用参数化查询 -
⚠️ 缺少错误处理
数据库查询可能失败，需要try-except包装 -
⚠️ 缺少类型注解
建议添加返回类型提示
优化建议：
from typing import Optional
async def get_user(user_id: int) -> Optional[User]:
try:
return await db.execute(
"SELECT * FROM users WHERE id = ?",
(user_id,)
)
except DatabaseError:
logger.error(f"Failed to get user {user_id}")
return None
**技术问答**
用户: 解释一下什么是Python的装饰器
Bot: 📖 Python 装饰器详解
定义：
装饰器是Python的一种语法糖，用于修改函数或类的行为。
比喻理解：
想象你有一杯咖啡，装饰器就像是在咖啡上加一层奶油、糖浆，而不需要重新做一杯咖啡。
基本语法：
def my_decorator(func):
def wrapper(*args, **kwargs):
print("Before function")
result = func(*args, **kwargs)
print("After function")
return result
return wrapper
@my_decorator
def say_hello():
print("Hello!")
# 等同于: say_hello = my_decorator(say_hello)
使用场景：
– 日志记录
– 权限验证
– 缓存
– 性能计时
### 5.5 内容创作助手
AstrBot 还可以成为内容创作者的好帮手：
**文章写作**
用户: 帮我写一篇关于人工智能发展趋势的短文
Bot: 🤖 人工智能发展趋势展望
2024年AI发展十大趋势
1. 多模态AI成为主流
从单纯的文本处理，扩展到图像、音频、视频的统一理解…
2. AI Agent 崛起
自主执行复杂任务的AI助手将改变工作方式…
3. 本地化部署加速
随着模型压缩技术进步，更多企业将在自有设备上运行AI…
[文章继续…]
**社交媒体内容生成**
用户: 生成5条关于科技产品推广的微博
Bot: 📱 科技产品推广文案
-
🔥 还在为选择哪款手机纠结？这5个理由让你果断下单！
[详细内容…] -
✨ 你的下一部手机，应该具备这3个功能！
[详细内容…] -
💡 90%的人都不知道的手机隐藏功能！
[详细内容…] -
🎯 买手机前必看的5个参数，99%的人都收藏了！
[详细内容…] -
🌟 2024年最值得买的手机，第一名出乎意料！
[详细内容…]
### 5.6 教育辅助工具
在教育领域，AstrBot 也有广泛的应用前景：
**作业辅导**
用户: 帮我解释这道数学题
Bot: 📐 数学题解析
题目：求二次函数 y = x² – 4x + 3 的顶点坐标
解题步骤：
1️⃣ 使用顶点公式：x = -b/2a
x = -(-4) / (2×1) = 4/2 = 2
2️⃣ 代入原式求y值：
y = 2² – 4×2 + 3 = 4 – 8 + 3 = -1
答案：
顶点坐标为 (2, -1)
图示说明：
[这里可以配合图片展示函数图像]
**语言学习**
用户: 教我5个英语商务词汇
Bot: 📚 商务英语词汇
-
Synergy /ˈsɪnədʒi/ 协同效应
例句：The merger created significant synergy.
翻译：这次合并创造了显著的协同效应。 -
Leverage /ˈlevərɪdʒ/ 利用、杠杆作用
例句：We need to leverage our resources.
翻译：我们需要利用我们的资源。 -
Stakeholder /ˈsteɪkhoʊldər/ 利益相关者
例句：All stakeholders should be consulted.
翻译：所有利益相关者都应该被咨询。 -
Benchmark /ˈbentʃmɑːrk/ 基准
例句：We use industry benchmarks for comparison.
翻译：我们使用行业基准进行比较。 -
Scalable /ˈskeɪləbl/ 可扩展的
例句：We need a scalable solution.
翻译：我们需要一个可扩展的解决方案。
🎯 建议：每个词用英文造句，加深记忆！
---
## 六、技巧与最佳实践
通过前面的学习，你已经掌握了 AstrBot 开发的基本技能。在这一部分，我将分享一些进阶技巧和最佳实践，帮助你编写更高质量的插件和更优雅的代码。
### 6.1 代码组织与结构
#### 6.1.1 遵循单一职责原则
每个插件应该只负责一个明确的功能领域。不要让一个插件做太多事情：
```python
**============================================================**
**不推荐的写法：一个插件做太多事情**
**============================================================**
class BadPlugin:
def handle_weather(self): pass
def handle_translate(self): pass
def handle_news(self): pass
def handle_music(self): pass
**... 太多职责，难以维护**
**============================================================**
**推荐的写法：每个插件专注一个功能**
**============================================================**
class WeatherPlugin:
"""只负责天气查询"""
pass
class TranslatorPlugin:
"""只负责翻译"""
pass
class NewsPlugin:
"""只负责新闻获取"""
pass
class MusicPlugin:
"""只负责音乐相关"""
pass
6.1.2 清晰的模块划分
对于复杂的插件，建议按功能划分成多个模块：
plugins/
└── advanced_ai/
├── __init__.py # 插件入口
├── conversation.py # 对话管理
├── context.py # 上下文处理
├── memory.py # 记忆系统
├── tools.py # 工具函数
└── prompts/ # 提示词模板
├── default.txt
├── coding.txt
└── creative.txt
6.2 错误处理
健壮的错误处理是高质量插件的标志：
class RobustPlugin:
"""展示良好错误处理的示例"""
async def handle_message(self, message: str, context: Dict) -> Optional[str]:
try:
**主逻辑**
result = await self.process_message(message)
return result
except ValueError as e:
**输入参数错误**
return f"❌ 输入无效: {str(e)}\n请检查后重新输入"
except ConnectionError as e:
**网络连接错误**
return "⚠️ 网络连接失败，请检查网络后重试"
except TimeoutError as e:
**超时错误**
return "⏰ 请求超时，请稍后重试"
except Exception as e:
**未知错误（记录日志但不暴露给用户）**
logger.error(f"处理消息时发生未知错误: {e}", exc_info=True)
return "😅 发生了一些问题，请稍后再试"
6.3 异步编程最佳实践
AstrBot 是一个异步框架，合理的异步编程可以大幅提升性能：
**============================================================**
**并发执行多个异步任务**
**============================================================**
async def fetch_multiple_sources(self, query: str) -> List[str]:
"""并发从多个数据源获取信息"""
**创建多个任务**
tasks = [
self.fetch_from_baidu(query),
self.fetch_from_google(query),
self.fetch_from_wiki(query),
]
**并发执行所有任务**
**使用 gather 而不是逐个 await**
results = await asyncio.gather(*tasks, return_exceptions=True)
**过滤掉异常结果**
valid_results = [
r for r in results
if isinstance(r, str)
]
return valid_results
**============================================================**
**正确处理异步上下文**
**============================================================**
async def database_operation(self):
"""正确使用异步上下文管理器"""
**推荐：使用 async with**
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.json()
**不推荐：在循环中频繁创建连接**
**for url in urls:**
**session = aiohttp.ClientSession()**
**# ... 处理**
**await session.close()**
6.4 配置管理
良好的配置管理可以让你的插件更加灵活：
class ConfigurablePlugin:
"""
展示配置管理的最佳实践
"""
**默认配置（类属性）**
DEFAULT_CONFIG = {
"enabled": True,
"max_retries": 3,
"timeout": 30,
"cache_ttl": 300,
"rate_limit": {
"max_requests": 10,
"window_seconds": 60
}
}
def __init__(self, config: Dict[str, Any]):
"""
合并用户配置和默认配置
Args:
config: 用户提供的配置
"""
**使用默认配置作为基础**
self.config = {**self.DEFAULT_CONFIG, **config}
**确保嵌套配置也被正确合并**
if "rate_limit" in config:
self.config["rate_limit"] = {
**self.DEFAULT_CONFIG["rate_limit"],
**config["rate_limit"]
}
**验证配置**
self._validate_config()
def _validate_config(self) -> None:
"""验证配置的有效性"""
if self.config["timeout"] <= 0:
raise ValueError("timeout 必须大于 0")
if self.config["max_retries"] < 0:
raise ValueError("max_retries 不能为负数")
6.5 日志记录
合理的日志记录对于调试和问题排查至关重要：
import logging
from typing import Optional
**创建专用的日志记录器**
logger = logging.getLogger(__name__)
class WellLoggedPlugin:
"""
展示良好日志记录的示例
"""
def __init__(self, config: dict):
self.config = config
**============================================================**
**日志级别使用建议：**
**DEBUG: 详细的技术信息，用于调试**
**INFO: 正常的业务事件**
**WARNING: 异常但可处理的情况**
**ERROR: 错误，需要关注**
**============================================================**
logger.info("插件初始化完成")
async def handle_message(self, message: str, context: dict) -> Optional[str]:
user_id = context.get("user_id", "unknown")
**记录收到的消息**
logger.info(f"[收到消息] 用户: {user_id}, 内容: {message[:50]}...")
try:
**处理逻辑**
result = await self.process(message)
**记录处理结果**
logger.info(f"[处理成功] 用户: {user_id}, 结果长度: {len(result)}")
return result
except Exception as e:
**记录错误（包含堆栈信息）**
logger.error(
f"[处理失败] 用户: {user_id}, 错误: {str(e)}",
exc_info=True # 包含堆栈跟踪
)
return "处理失败，请稍后重试"
async def call_external_api(self, data: dict) -> dict:
"""展示API调用的日志记录"""
url = self.config.get("api_url")
logger.debug(f"准备调用API: {url}")
logger.debug(f"请求参数: {json.dumps(data)}")
start_time = time.time()
try:
async with aiohttp.ClientSession() as session:
async with session.post(url, json=data) as response:
result = await response.json()
elapsed = time.time() - start_time
logger.info(f"API调用成功，耗时: {elapsed:.2f}秒")
return result
except aiohttp.ClientError as e:
elapsed = time.time() - start_time
logger.warning(f"API调用失败({elapsed:.2f}秒): {e}")
raise
6.6 性能优化
对于需要处理大量请求的插件，性能优化是不可忽视的：
import functools
import cachetools
class OptimizedPlugin:
"""
展示性能优化技巧的示例
"""
def __init__(self):
**============================================================**
**使用缓存减少重复计算**
**============================================================**
**LRU缓存：最近最少使用的条目会被清除**
self.get_weather = functools.lru_cache(maxsize=128)(
self._get_weather_sync
)
**TTL缓存：带过期时间的缓存**
self.cache = cachetools.TTLCache(maxsize=1000, ttl=300)
@functools.lru_cache(maxsize=256)
def _get_weather_sync(self, city: str) -> dict:
"""
带缓存的天气查询（同步版本）
同样的城市查询在缓存有效期内不会重复调用API
"""
**API调用逻辑...**
pass
async def batch_process(self, items: List[str]) -> List[str]:
"""
批量处理优化
使用信号量限制并发数量，避免资源耗尽
"""
import asyncio
**限制最大并发数为10**
semaphore = asyncio.Semaphore(10)
async def bounded_process(item: str) -> str:
async with semaphore:
return await self.process_single(item)
**并发执行所有任务**
return await asyncio.gather(*[
bounded_process(item) for item in items
])
def use_cache_decorator(self, ttl: int = 300):
"""
自定义缓存装饰器
用于缓存异步函数的结果
"""
cache = cachetools.TTLCache(maxsize=1000, ttl=ttl)
def decorator(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
**生成缓存键**
cache_key = str(args) + str(sorted(kwargs.items()))
if cache_key in cache:
return cache[cache_key]
result = await func(*args, **kwargs)
cache[cache_key] = result
return result
return wrapper
return decorator
6.7 安全性考虑
在处理用户输入和敏感数据时，安全性是必须考虑的：
import html
import re
import bleach
class SecurePlugin:
"""
展示安全最佳实践的示例
"""
**允许的HTML标签（用于显示富文本）**
ALLOWED_TAGS = ['b', 'i', 'u', 'code', 'pre']
ALLOWED_ATTRIBUTES = {'code': ['class'], 'pre': ['class']}
def sanitize_user_input(self, user_input: str) -> str:
"""
清理用户输入，防止XSS攻击
Args:
user_input: 原始用户输入
Returns:
清理后的安全文本
"""
**HTML转义**
safe_text = html.escape(user_input)
**移除脚本标签**
safe_text = re.sub(
r'<script\b[^<]*(?:(?!<\/script>)<[^<]*)*<\/script>',
'',
safe_text,
flags=re.IGNORECASE
)
return safe_text
def sanitize_html(self, html_content: str) -> str:
"""
清理HTML内容，只保留安全标签
Args:
html_content: 原始HTML内容
Returns:
清理后的HTML
"""
return bleach.clean(
html_content,
tags=self.ALLOWED_TAGS,
attributes=self.ALLOWED_ATTRIBUTES,
strip=True
)
async def handle_message(self, message: str, context: dict) -> Optional[str]:
"""
安全的消息处理
"""
**长度限制**
if len(message) > 10000:
return "输入内容过长，请缩短"
**清理输入**
safe_message = self.sanitize_user_input(message)
**验证用户权限（如果需要）**
if self._requires_auth(context):
if not self._verify_auth(context):
return "需要登录才能使用此功能"
**处理消息...**
return await self.process_safe(safe_message)
def _verify_auth(self, context: dict) -> bool:
"""验证用户认证状态"""
token = context.get("auth_token")
if not token:
return False
**实际的验证逻辑**
return self._check_token(token)
6.8 测试策略
为你的插件编写测试是保证质量的重要手段：
import pytest
import asyncio
from unittest.mock import Mock, AsyncMock, patch
class TestWeatherPlugin:
"""
插件测试示例
"""
@pytest.fixture
def plugin(self):
"""测试用的插件实例"""
config = {
"api_key": "test_key",
"default_city": "北京",
"temperature_unit": "celsius"
}
return WeatherPlugin(config)
@pytest.fixture
def mock_context(self):
"""测试用的消息上下文"""
return {
"user_id": "test_user",
"platform": "terminal",
"session_id": "test_session"
}
**============================================================**
**单元测试**
**============================================================**
def test_plugin_initialization(self, plugin):
"""测试插件初始化"""
assert plugin.name == "weather"
assert plugin.api_key == "test_key"
assert plugin.default_city == "北京"
def test_parse_city(self, plugin):
"""测试城市名称解析"""
assert plugin.city_mapping["北京"] == "Beijing"
assert plugin.city_mapping["东京"] == "Tokyo"
@pytest.mark.asyncio
async def test_handle_message_invalid_format(self, plugin, mock_context):
"""测试无效消息格式"""
result = await plugin.handle_message("无效的格式", mock_context)
assert result is None
@pytest.mark.asyncio
async def test_handle_message_valid(self, plugin, mock_context):
"""测试有效消息格式"""
**使用mock来模拟API响应**
with patch.object(plugin, 'query_weather', return_value="晴，25°C"):
result = await plugin.handle_message("天气 北京", mock_context)
assert result is not None
assert "北京" in result
**============================================================**
**集成测试**
**============================================================**
@pytest.mark.asyncio
async def test_weather_query_integration(self):
"""集成测试：天气查询完整流程"""
config = {"api_key": "real_key"}
plugin = WeatherPlugin(config)
**实际调用API（需要真实API key）**
**result = await plugin.query_weather("北京")**
**assert "北京" in result**
**或者使用mock server进行测试**
pass
**============================================================**
**运行测试的命令**
**============================================================**
**pytest tests/test_weather.py -v**
**pytest tests/ -v --cov=plugins # 带覆盖率**
七、总结与相关资源
经过这一整篇教程的学习，你应该已经对 AstrBot 有了全面而深入的了解。让我们来总结一下所学内容，并为你提供一些进一步学习的资源。
7.1 核心要点回顾
在这篇教程中，我们从以下几个方面深入探讨了 AstrBot：
第一部分：项目概述
AstrBot 是一个高度模块化、可扩展的智能对话机器人框架
- 支持多平台：微信、QQ、Discord、Telegram 等
- 支持多模型：GPT、Claude、通义千问等
- 插件系统：热加载、热重载
- 知识库问答：RAG 支持
- 中文优化：专为中文用户设计
第二部分：环境搭建
- Python 3.10+ 环境
- 使用虚拟环境隔离项目
- 通过 pip 安装 astrbot
- 使用 init 命令创建项目
- 配置文件详解
第三部分：核心功能
- 平台连接器系统
- 大语言模型集成
- 插件系统架构
- 知识库问答
- 角色扮演与对话管理
第四部分：实战教程
- 项目结构设计
- 天气查询插件
- 翻译插件
- 代码助手插件
- 提醒插件
- 主程序整合
第五部分：常见使用场景
- 个人助手
- 客服机器人
- 社区管理
- 开发助手
- 内容创作
- 教育辅助
第六部分：最佳实践
- 代码组织与结构
- 错误处理
- 异步编程
- 配置管理
- 日志记录
- 性能优化
- 安全性
- 测试策略
7.2 进一步学习的方向
掌握基础之后，你可以继续探索以下方向：
高级插件开发
- 实现复杂的对话流程
- 集成机器学习模型
- 开发自定义的向量检索
- 实现多轮对话管理
生产环境部署
- 使用 Docker 容器化
- 配置 Nginx 反向代理
- 使用 Redis 进行状态管理
- 部署到云平台（AWS、阿里云等）
- 配置监控和告警
性能与扩展
- 实现水平扩展
- 优化响应延迟
- 缓存策略设计
- 负载均衡配置
7.3 相关资源链接
官方资源
AstrBot GitHub 仓库
https://github.com/AstrBotDevs/AstrBot
AstrBot 官方文档
https://docs.astrbot.dev
AstrBot 插件市场
https://plugins.astrbot.dev
学习资源
Python 异步编程
https://docs.python.org/zh-cn/3/library/asyncio.html
aiohttp 文档
https://docs.aiohttp.org/
Python Design Patterns
https://python-patterns.gatech.edu/
社区与交流
AstrBot Discord 社区
（加入官方服务器获取帮助）
AstrBot QQ 群
（中文社区交流）
7.4 思考与展望
AstrBot 作为一个开源项目，正在不断发展和完善。作为开发者，我们可以：
持续学习：
- 关注项目更新日志
- 学习新的AI技术
- 关注开源社区动态
贡献社区：
- 分享自己开发的插件
- 帮助回答其他开发者的问题
- 提交Issue和Pull Request
创新应用：
- 探索AstrBot在不同场景的应用
- 结合其他AI服务创造新功能
- 推动项目的功能改进
7.5 结语
从最初的「为什么值得关注」到最后的「总结与展望」，我们一起走过了 AstrBot 的完整学习旅程。
AstrBot 不只是一个技术框架，它代表了一种将复杂的人工智能能力民主化的理念。通过这个框架，即使是非AI专业的开发者，也能够快速构建智能对话应用，让AI技术服务于更多的场景和人群。
正如 AstrBot 的设计理念所倡导的——「让每个人都能够轻松地创建自己的AI助手」。希望这篇教程能够帮助你迈出第一步，开启你的智能机器人开发之旅。
现在，是时候动手实践了！打开你的IDE，按照教程中的步骤创建一个属于你自己的 AstrBot 机器人吧。遇到问题时，不要忘了查阅文档和寻求社区的帮助。
祝你在 AstrBot 的世界里玩得开心，创造出令人惊叹的作品！
相关AI项目推荐
如果你对 AI 机器人开发感兴趣，除了 AstrBot，以下项目也值得关注：
对话与聊天：
- ChatGLM：清华开源的中文对话模型
- LLaMA：Meta开源的大语言模型
- 文心一言：百度的大语言模型
知识库与RAG：
- LangChain：构建LLM应用的框架
- LlamaIndex：知识检索增强生成工具
- Qdrant：高性能向量数据库
多模态：
- Stable Diffusion：图像生成
- Whisper：语音识别
- Bark：语音合成
开发工具：
- Dify：开源LLM应用开发平台
- Flowise：低代码LLM应用构建
- LangFlow：LangChain的可视化工具
这些项目与 AstrBot 可以形成很好的互补，帮助你构建更强大的AI应用。
再次感谢你阅读这篇教程！如果有任何问题或建议，欢迎在评论区留言交流。
评论区