gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix：修复日志占用和文件层级

Browse Source
This commit is contained in:
SengokuCola
2025-06-15 13:14:39 +08:00
parent 38b71f74f5
commit 83ae078d6e
12 changed files with 958 additions and 1733 deletions
Download Patch File Download Diff File Expand all files Collapse all files

241
HEARFLOW_API_说明文档.md
View File

@@ -1,241 +0,0 @@
# HearflowAPI 使用说明
## 概述
HearflowAPI 是一个新增的插件API模块，提供了与心流和子心流相关的操作接口。通过这个API，插件开发者可以方便地获取和操作sub_hearflow实例。
## 主要功能
### 1. 获取子心流实例
#### `get_sub_hearflow_by_chat_id(chat_id: str) -> Optional[SubHeartflow]`
根据chat_id获取指定的sub_hearflow实例（仅获取已存在的）。
**参数：**
- `chat_id`: 聊天ID，与sub_hearflow的subheartflow_id相同
**返回值：**
- `SubHeartflow`: sub_hearflow实例，如果不存在则返回None
**示例：**
```python
# 获取当前聊天的子心流实例
current_subflow = await self.get_sub_hearflow_by_chat_id(self.observation.chat_id)
if current_subflow:
print(f"找到子心流: {current_subflow.chat_id}")
else:
print("子心流不存在")
```
#### `get_or_create_sub_hearflow_by_chat_id(chat_id: str) -> Optional[SubHeartflow]`
根据chat_id获取或创建sub_hearflow实例。
**参数：**
- `chat_id`: 聊天ID
**返回值：**
- `SubHeartflow`: sub_hearflow实例，创建失败时返回None
**示例：**
```python
# 获取或创建子心流实例
subflow = await self.get_or_create_sub_hearflow_by_chat_id("some_chat_id")
if subflow:
print("成功获取或创建子心流")
```
### 2. 获取子心流列表
#### `get_all_sub_hearflow_ids() -> List[str]`
获取所有活跃子心流的ID列表。
**返回值：**
- `List[str]`: 所有活跃子心流的ID列表
#### `get_all_sub_hearflows() -> List[SubHeartflow]`
获取所有活跃的子心流实例。
**返回值：**
- `List[SubHeartflow]`: 所有活跃的子心流实例列表
**示例：**
```python
# 获取所有活跃的子心流ID
all_chat_ids = self.get_all_sub_hearflow_ids()
print(f"共有 {len(all_chat_ids)} 个活跃的子心流")
# 获取所有活跃的子心流实例
all_subflows = self.get_all_sub_hearflows()
for subflow in all_subflows:
print(f"子心流 {subflow.chat_id} 状态: {subflow.chat_state.chat_status.value}")
```
### 3. 心流状态操作
#### `get_sub_hearflow_chat_state(chat_id: str) -> Optional[ChatState]`
获取指定子心流的聊天状态。
**参数：**
- `chat_id`: 聊天ID
**返回值：**
- `ChatState`: 聊天状态，如果子心流不存在则返回None
#### `set_sub_hearflow_chat_state(chat_id: str, target_state: ChatState) -> bool`
设置指定子心流的聊天状态。
**参数：**
- `chat_id`: 聊天ID
- `target_state`: 目标状态
**返回值：**
- `bool`: 是否设置成功
**示例：**
```python
from src.chat.heart_flow.sub_heartflow import ChatState
# 获取当前状态
current_state = await self.get_sub_hearflow_chat_state(self.observation.chat_id)
print(f"当前状态: {current_state.value}")
# 设置状态
success = await self.set_sub_hearflow_chat_state(self.observation.chat_id, ChatState.FOCUS)
if success:
print("状态设置成功")
```
### 4. Replyer和Expressor操作
#### `get_sub_hearflow_replyer_and_expressor(chat_id: str) -> Tuple[Optional[Any], Optional[Any]]`
根据chat_id获取指定子心流的replyer和expressor实例。
**参数：**
- `chat_id`: 聊天ID
**返回值：**
- `Tuple[Optional[Any], Optional[Any]]`: (replyer实例, expressor实例)，如果子心流不存在或未处于FOCUSED状态，返回(None, None)
#### `get_sub_hearflow_replyer(chat_id: str) -> Optional[Any]`
根据chat_id获取指定子心流的replyer实例。
**参数：**
- `chat_id`: 聊天ID
**返回值：**
- `Optional[Any]`: replyer实例，如果不存在则返回None
#### `get_sub_hearflow_expressor(chat_id: str) -> Optional[Any]`
根据chat_id获取指定子心流的expressor实例。
**参数：**
- `chat_id`: 聊天ID
**返回值：**
- `Optional[Any]`: expressor实例，如果不存在则返回None
**示例：**
```python
# 获取replyer和expressor
replyer, expressor = await self.get_sub_hearflow_replyer_and_expressor(self.observation.chat_id)
if replyer and expressor:
print(f"获取到replyer: {type(replyer).__name__}")
print(f"获取到expressor: {type(expressor).__name__}")
# 检查属性
print(f"Replyer聊天ID: {replyer.chat_id}")
print(f"Expressor聊天ID: {expressor.chat_id}")
print(f"是否群聊: {replyer.is_group_chat}")
# 单独获取replyer
replyer = await self.get_sub_hearflow_replyer(self.observation.chat_id)
if replyer:
print("获取到replyer实例")
# 单独获取expressor
expressor = await self.get_sub_hearflow_expressor(self.observation.chat_id)
if expressor:
print("获取到expressor实例")
```
## 可用的聊天状态
```python
from src.chat.heart_flow.sub_heartflow import ChatState
ChatState.FOCUS # 专注模式
ChatState.NORMAL # 普通模式
ChatState.ABSENT # 离开模式
```
## 完整插件示例
```python
from typing import Tuple
from src.plugin_system.base.base_action import BaseAction as PluginAction, register_action
from src.chat.heart_flow.sub_heartflow import ChatState
@register_action
class MyHearflowPlugin(PluginAction):
"""我的心流插件"""
activation_keywords = ["心流信息"]
async def process(self) -> Tuple[bool, str]:
try:
# 获取当前聊天的chat_id
current_chat_id = self.observation.chat_id
# 获取子心流实例
subflow = await self.get_sub_hearflow_by_chat_id(current_chat_id)
if not subflow:
return False, "未找到子心流实例"
# 获取状态信息
current_state = await self.get_sub_hearflow_chat_state(current_chat_id)
# 构建回复
response = f"心流信息：\n"
response += f"聊天ID: {current_chat_id}\n"
response += f"当前状态: {current_state.value}\n"
response += f"是否群聊: {subflow.is_group_chat}\n"
return True, response
except Exception as e:
return False, f"处理出错: {str(e)}"
```
## 注意事项
1. **线程安全**: API内部已处理锁机制，确保线程安全。
2. **错误处理**: 所有API方法都包含异常处理，失败时会记录日志并返回安全的默认值。
3. **性能考虑**: `get_sub_hearflow_by_chat_id` 只获取已存在的实例，性能更好；`get_or_create_sub_hearflow_by_chat_id` 会在需要时创建新实例。
4. **状态管理**: 修改心流状态时请谨慎，确保不会影响系统的正常运行。
5. **日志记录**: 所有操作都会记录适当的日志，便于调试和监控。
6. **Replyer和Expressor可用性**:
- 这些实例仅在子心流处于**FOCUSED状态**时可用
- 如果子心流处于NORMAL或ABSENT状态，将返回None
- 需要确保HeartFC实例存在且正常运行
7. **使用Replyer和Expressor时的注意事项**:
- 直接调用这些实例的方法需要谨慎，可能影响系统正常运行
- 建议主要用于监控、信息获取和状态检查
- 不建议在插件中直接调用回复生成方法，这可能与系统的正常流程冲突
## 相关类型和模块
- `SubHeartflow`: 子心流实例类
- `ChatState`: 聊天状态枚举
- `DefaultReplyer`: 默认回复器类
- `DefaultExpressor`: 默认表达器类
- `HeartFChatting`: 专注聊天主类
- `src.chat.heart_flow.heartflow`: 主心流模块
- `src.chat.heart_flow.subheartflow_manager`: 子心流管理器
- `src.chat.focus_chat.replyer.default_replyer`: 回复器模块
- `src.chat.focus_chat.expressors.default_expressor`: 表达器模块

717
MaiBot插件开发文档.md Normal file
View File

@@ -0,0 +1,717 @@
# MaiBot 插件开发文档
## 📖 总体介绍
MaiBot 是一个基于大语言模型的智能聊天机器人，采用现代化的插件系统架构，支持灵活的功能扩展和定制。插件系统提供了统一的开发框架，让开发者可以轻松创建和管理各种功能组件。
### 🎯 插件系统特点
- **组件化架构**：支持Action（动作）和Command（命令）两种主要组件类型
- **统一API接口**：提供丰富的API功能，包括消息发送、数据库操作、LLM调用等
- **配置驱动**：支持TOML配置文件，实现灵活的参数配置
- **热加载机制**：支持动态加载和卸载插件
- **依赖管理**：内置依赖检查和解析机制
- **拦截控制**：Command组件支持消息拦截控制
## 🧩 主要组件
### 1. 插件（Plugin）
插件是功能的容器，每个插件可以包含多个组件。插件通过继承 `BasePlugin` 类实现：
```python
from src.plugin_system import BasePlugin, register_plugin
@register_plugin
class MyPlugin(BasePlugin):
plugin_name = "my_plugin"
plugin_description = "我的插件"
plugin_version = "1.0.0"
plugin_author = "开发者"
config_file_name = "config.toml" # 可选配置文件
```
### 2. Action组件
Action是给麦麦在回复之外提供额外功能的组件，由麦麦的决策系统自主选择是否使用，具有随机性和拟人化的调用特点。Action不是直接响应用户命令，而是让麦麦根据聊天情境智能地选择合适的动作，使其行为更加自然和真实。
```python
from src.plugin_system import BaseAction, ActionActivationType, ChatMode
class MyAction(BaseAction):
# 激活设置 - 麦麦会根据这些条件决定是否使用此Action
focus_activation_type = ActionActivationType.KEYWORD
normal_activation_type = ActionActivationType.RANDOM
activation_keywords = ["关键词1", "关键词2"]
mode_enable = ChatMode.ALL
async def execute(self) -> Tuple[bool, str]:
# 麦麦决定使用此Action时执行的逻辑
await self.send_text("这是麦麦主动执行的动作")
return True, "执行成功"
```
### 3. Command组件
Command是直接响应用户明确指令的组件，与Action不同，Command是被动触发的，当用户输入特定格式的命令时立即执行。Command支持正则表达式匹配和消息拦截：
```python
from src.plugin_system import BaseCommand
class MyCommand(BaseCommand):
command_pattern = r"^/hello\s+(?P<name>\w+)$"
command_help = "打招呼命令"
command_examples = ["/hello 世界"]
intercept_message = True # 拦截后续处理
async def execute(self) -> Tuple[bool, Optional[str]]:
name = self.matched_groups.get("name", "世界")
await self.send_text(f"你好，{name}！")
return True, f"已向{name}问候"
```
> **Action vs Command 区别**：
> - **Action**：麦麦主动决策使用，具有随机性和智能性，让麦麦行为更拟人化
> - **Command**：用户主动触发，确定性执行，用于提供具体功能和服务
## 🚀 快速开始
### 1. 创建插件目录
在项目的 `src/plugins/` 文件夹下创建你的插件目录：
```
src/plugins/
└── my_plugin/
├── plugin.py # 插件主文件
├── config.toml # 配置文件（可选）
└── README.md # 说明文档（可选）
```
### 2. 编写插件主文件
创建 `plugin.py` 文件：
```python
from typing import List, Tuple, Type
from src.plugin_system import (
BasePlugin, register_plugin, BaseAction, BaseCommand,
ComponentInfo, ActionActivationType, ChatMode
)
# 定义一个简单的Action
class GreetingAction(BaseAction):
focus_activation_type = ActionActivationType.KEYWORD
activation_keywords = ["你好", "hello"]
async def execute(self) -> Tuple[bool, str]:
await self.send_text("你好！很高兴见到你！")
return True, "执行问候动作"
# 定义一个简单的Command
class InfoCommand(BaseCommand):
command_pattern = r"^/info$"
command_help = "显示插件信息"
command_examples = ["/info"]
async def execute(self) -> Tuple[bool, str]:
await self.send_text("这是我的第一个插件！")
return True, "显示插件信息"
# 注册插件
@register_plugin
class MyFirstPlugin(BasePlugin):
plugin_name = "first_plugin"
plugin_description = "我的第一个插件"
plugin_version = "1.0.0"
plugin_author = "我的名字"
def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]:
return [
(GreetingAction.get_action_info(
name="greeting",
description="问候用户"
), GreetingAction),
(InfoCommand.get_command_info(
name="info",
description="显示插件信息"
), InfoCommand),
]
```
### 3. 创建配置文件（可选）
创建 `config.toml` 文件：
```toml
[plugin]
name = "first_plugin"
version = "1.0.0"
enabled = true
[greeting]
enable_emoji = true
custom_message = "欢迎使用我的插件！"
[logging]
level = "INFO"
```
### 4. 启动机器人
将插件放入 `src/plugins/` 目录后，启动MaiBot，插件会自动加载。
## 📚 完整说明
### 插件生命周期
1. **发现阶段**：系统扫描 `src/plugins/` 目录，查找Python文件
2. **加载阶段**：导入插件模块，注册插件类
3. **实例化阶段**：创建插件实例，加载配置文件
4. **注册阶段**：注册插件及其包含的组件
5. **运行阶段**：组件根据条件被激活和执行
### Action组件详解
Action组件是麦麦智能决策系统的重要组成部分，它们不是被动响应用户输入，而是由麦麦根据聊天情境主动选择执行。这种设计使麦麦的行为更加拟人化和自然，就像真人聊天时会根据情况做出不同的反应一样。
#### 激活类型
Action的激活类型决定了麦麦在什么情况下会考虑使用该Action：
- `NEVER`：从不激活，通常用于临时禁用
- `ALWAYS`：麦麦总是会考虑使用此Action
- `LLM_JUDGE`：通过LLM智能判断当前情境是否适合使用
- `RANDOM`：基于随机概率决定是否使用，增加行为的不可预测性
- `KEYWORD`：当检测到特定关键词时会考虑使用
#### 聊天模式
- `FOCUS`：专注聊天模式
- `NORMAL`：普通聊天模式
- `ALL`：所有模式
#### Action示例
```python
class AdvancedAction(BaseAction):
# 激活设置
focus_activation_type = ActionActivationType.LLM_JUDGE
normal_activation_type = ActionActivationType.KEYWORD
activation_keywords = ["帮助", "help"]
llm_judge_prompt = "当用户需要帮助时回答'是'，否则回答'否'"
random_activation_probability = 0.3
mode_enable = ChatMode.ALL
parallel_action = True
# 动作参数（用于LLM规划）
action_parameters = {
"query": "用户的问题或需求"
}
# 使用场景描述
action_require = [
"用户明确请求帮助",
"检测到用户遇到困难"
]
async def execute(self) -> Tuple[bool, str]:
query = self.action_data.get("query", "")
# 麦麦主动决定帮助用户时执行的逻辑
await self.send_text(f"我来帮助你解决：{query}")
await self.send_type("emoji", "😊")
# 存储执行记录
await self.api.store_action_info(
action_build_into_prompt=True,
action_prompt_display=f"麦麦主动帮助用户：{query}",
action_done=True,
thinking_id=self.thinking_id
)
return True, f"麦麦已主动帮助处理：{query}"
```
### Command组件详解
#### 正则表达式匹配
Command使用正则表达式匹配用户输入，支持命名组捕获：
```python
class UserCommand(BaseCommand):
# 匹配 /user add 用户名
command_pattern = r"^/user\s+(?P<action>add|del|info)\s+(?P<username>\w+)$"
command_help = "用户管理命令"
command_examples = [
"/user add 张三",
"/user del 李四",
"/user info 王五"
]
intercept_message = True
async def execute(self) -> Tuple[bool, str]:
action = self.matched_groups.get("action")
username = self.matched_groups.get("username")
if action == "add":
return await self._add_user(username)
elif action == "del":
return await self._delete_user(username)
elif action == "info":
return await self._show_user_info(username)
return False, "无效的操作"
```
#### 消息拦截控制
- `intercept_message = True`：拦截消息，不进行后续处理
- `intercept_message = False`：不拦截，继续处理其他组件
### 配置系统
插件支持TOML配置文件，配置会自动加载到插件实例：
```python
class ConfigurablePlugin(BasePlugin):
config_file_name = "config.toml"
def some_method(self):
# 获取配置值，支持嵌套键访问
max_items = self.get_config("limits.max_items", 10)
custom_message = self.get_config("messages.greeting", "默认消息")
```
配置文件格式：
```toml
[limits]
max_items = 20
timeout = 30
[messages]
greeting = "欢迎使用配置化插件！"
error = "操作失败"
[features]
enable_debug = true
```
### 错误处理
插件应该包含适当的错误处理：
```python
async def execute(self) -> Tuple[bool, str]:
try:
# 执行逻辑
result = await self._do_something()
return True, "操作成功"
except ValueError as e:
logger.error(f"{self.log_prefix} 参数错误: {e}")
await self.send_text("参数错误，请检查输入")
return False, f"参数错误: {e}"
except Exception as e:
logger.error(f"{self.log_prefix} 执行失败: {e}")
await self.send_text("操作失败，请稍后重试")
return False, f"执行失败: {e}"
```
## 🔌 API说明
### 消息API
插件可以通过 `self.api` 访问各种API功能：
#### 基础消息发送
```python
# 发送文本消息
await self.send_text("这是文本消息")
# 发送特定类型消息
await self.send_type("emoji", "😊")
await self.send_type("image", image_url)
# 发送命令消息
await self.send_command("命令名", {"参数": "值"})
```
#### 高级消息发送
```python
# 向指定群聊发送消息
await self.api.send_text_to_group("消息内容", "群ID", "qq")
# 向指定用户发送私聊消息
await self.api.send_text_to_user("消息内容", "用户ID", "qq")
# 向指定目标发送任意类型消息
await self.api.send_message_to_target(
message_type="text",
content="消息内容",
platform="qq",
target_id="目标ID",
is_group=True,
display_message="显示消息"
)
```
#### 消息查询
```python
# 获取聊天类型
chat_type = self.api.get_chat_type() # "group" 或 "private"
# 获取最近消息
recent_messages = self.api.get_recent_messages(count=5)
```
### 数据库API
插件可以使用数据库API进行数据持久化：
#### 通用查询
```python
# 查询数据
results = await self.api.db_query(
model_class=SomeModel,
query_type="get",
filters={"field": "value"},
limit=10,
order_by=["-time"]
)
# 创建记录
new_record = await self.api.db_query(
model_class=SomeModel,
query_type="create",
data={"field1": "value1", "field2": "value2"}
)
# 更新记录
updated_count = await self.api.db_query(
model_class=SomeModel,
query_type="update",
filters={"id": 123},
data={"field": "new_value"}
)
# 删除记录
deleted_count = await self.api.db_query(
model_class=SomeModel,
query_type="delete",
filters={"id": 123}
)
# 计数
count = await self.api.db_query(
model_class=SomeModel,
query_type="count",
filters={"active": True}
)
```
#### 原始SQL查询
```python
# 执行原始SQL
results = await self.api.db_raw_query(
sql="SELECT * FROM table WHERE condition = ?",
params=["value"],
fetch_results=True
)
```
#### Action记录存储
```python
# 存储Action执行记录
await self.api.store_action_info(
action_build_into_prompt=True,
action_prompt_display="显示的动作描述",
action_done=True,
thinking_id="思考ID",
action_data={"key": "value"}
)
```
### LLM API
插件可以调用大语言模型：
```python
# 获取可用模型
models = self.api.get_available_models()
# 使用模型生成内容
success, response, reasoning, model_name = await self.api.generate_with_model(
prompt="你的提示词",
model_config=models["某个模型"],
request_type="plugin.generate",
temperature=0.7,
max_tokens=1000
)
if success:
await self.send_text(f"AI回复：{response}")
else:
await self.send_text("AI生成失败")
```
### 配置API
```python
# 获取全局配置
global_config = self.api.get_global_config()
# 获取插件配置
plugin_config = self.api.get_config("section.key", "默认值")
```
### 工具API
```python
# 获取当前时间戳
timestamp = self.api.get_current_timestamp()
# 格式化时间
formatted_time = self.api.format_timestamp(timestamp, "%Y-%m-%d %H:%M:%S")
# JSON处理
json_str = self.api.dict_to_json({"key": "value"})
data = self.api.json_to_dict(json_str)
# 生成UUID
uuid = self.api.generate_uuid()
# 哈希计算
hash_value = self.api.calculate_hash("text", "md5")
```
### 流API
```python
# 获取当前聊天流信息
chat_stream = self.api.get_service("chat_stream")
if chat_stream:
stream_id = chat_stream.stream_id
platform = chat_stream.platform
# 群聊信息
if chat_stream.group_info:
group_id = chat_stream.group_info.group_id
group_name = chat_stream.group_info.group_name
# 用户信息
user_id = chat_stream.user_info.user_id
user_name = chat_stream.user_info.user_nickname
```
### 心流API
```python
# 等待新消息
has_new_message = await self.api.wait_for_new_message(timeout=30)
# 获取观察信息
observations = self.api.get_service("observations")
```
## 🔧 高级功能
### 插件依赖管理
```python
@register_plugin
class DependentPlugin(BasePlugin):
plugin_name = "dependent_plugin"
plugin_description = "依赖其他插件的插件"
dependencies = ["core_actions", "example_plugin"] # 依赖列表
def get_plugin_components(self):
# 只有依赖满足时才会加载
return [...]
```
### 并行Action
```python
class ParallelAction(BaseAction):
parallel_action = True # 允许与其他Action并行执行
async def execute(self) -> Tuple[bool, str]:
# 这个Action可以与其他并行Action同时执行
return True, "并行执行完成"
```
### 动态配置更新
```python
class DynamicPlugin(BasePlugin):
def get_plugin_components(self):
# 根据配置动态决定加载哪些组件
components = []
if self.get_config("features.enable_greeting", True):
components.append((GreetingAction.get_action_info(), GreetingAction))
if self.get_config("features.enable_commands", True):
components.append((SomeCommand.get_command_info(), SomeCommand))
return components
```
### 自定义元数据
```python
class MetadataAction(BaseAction):
@classmethod
def get_action_info(cls, name=None, description=None):
info = super().get_action_info(name, description)
# 添加自定义元数据
info.metadata = {
"category": "utility",
"priority": "high",
"custom_field": "custom_value"
}
return info
```
## 📋 开发规范
### 1. 命名规范
- 插件名使用小写字母和下划线：`my_plugin`
- 类名使用大驼峰：`MyPlugin`、`GreetingAction`
- 方法名使用小写字母和下划线：`execute`、`send_message`
### 2. 文档规范
- 所有插件类都应该有完整的文档字符串
- Action和Command的描述要清晰明确
- 提供使用示例和配置说明
### 3. 错误处理
- 所有异步操作都要包含异常处理
- 使用日志记录错误信息
- 向用户返回友好的错误消息
### 4. 配置管理
- 敏感配置不要硬编码在代码中
- 提供合理的默认值
- 支持配置热更新
### 5. 性能考虑
- 避免在初始化时执行耗时操作
- 合理使用缓存减少重复计算
- 及时释放不需要的资源
## 🎯 最佳实践
### 1. 插件结构
```
src/plugins/my_plugin/
├── __init__.py # 空文件或简单导入
├── plugin.py # 主插件文件
├── actions/ # Action组件目录
│ ├── __init__.py
│ ├── greeting.py
│ └── helper.py
├── commands/ # Command组件目录
│ ├── __init__.py
│ ├── admin.py
│ └── user.py
├── utils/ # 工具函数
│ ├── __init__.py
│ └── helpers.py
├── config.toml # 配置文件
└── README.md # 说明文档
```
### 2. 模块化设计
```python
# actions/greeting.py
from src.plugin_system import BaseAction
class GreetingAction(BaseAction):
# ... 实现细节
# commands/admin.py
from src.plugin_system import BaseCommand
class AdminCommand(BaseCommand):
# ... 实现细节
# plugin.py
from .actions.greeting import GreetingAction
from .commands.admin import AdminCommand
@register_plugin
class MyPlugin(BasePlugin):
def get_plugin_components(self):
return [
(GreetingAction.get_action_info(), GreetingAction),
(AdminCommand.get_command_info(), AdminCommand),
]
```
### 3. 配置分层
```toml
# config.toml
[plugin]
name = "my_plugin"
version = "1.0.0"
enabled = true
[components]
enable_greeting = true
enable_admin = false
[greeting]
message_template = "你好，{username}！"
enable_emoji = true
[admin]
allowed_users = ["admin", "moderator"]
```
### 4. 日志实践
```python
from src.common.logger import get_logger
logger = get_logger("my_plugin")
class MyAction(BaseAction):
async def execute(self):
logger.info(f"{self.log_prefix} 开始执行动作")
try:
# 执行逻辑
result = await self._do_something()
logger.debug(f"{self.log_prefix} 执行结果: {result}")
return True, "成功"
except Exception as e:
logger.error(f"{self.log_prefix} 执行失败: {e}", exc_info=True)
return False, str(e)
```
---
## 🎉 总结
MaiBot的插件系统提供了强大而灵活的扩展能力，通过Action和Command两种组件类型，开发者可以轻松实现各种功能。系统提供了丰富的API接口、完善的配置管理和错误处理机制，让插件开发变得简单高效。
遵循本文档的指导和最佳实践，你可以快速上手MaiBot插件开发，为机器人添加强大的自定义功能。
如有问题或建议，欢迎提交Issue或参与讨论！

12
bot.py
View File

@@ -10,7 +10,7 @@ from dotenv import load_dotenv
from rich.traceback import install
# 最早期初始化日志系统，确保所有后续模块都使用正确的日志格式
from src.common.logger import initialize_logging, get_logger
from src.common.logger import initialize_logging, get_logger, shutdown_logging
from src.main import MainSystem
from src.manager.async_task_manager import async_task_manager
@@ -130,6 +130,9 @@ async def graceful_shutdown():
logger.error(f"等待任务取消时发生异常: {e}")
logger.info("麦麦优雅关闭完成")
# 关闭日志系统，释放文件句柄
shutdown_logging()
except Exception as e:
logger.error(f"麦麦关闭失败: {e}", exc_info=True)
@@ -271,6 +274,13 @@ if __name__ == "__main__":
if "loop" in locals() and loop and not loop.is_closed():
loop.close()
logger.info("事件循环已关闭")
# 关闭日志系统，释放文件句柄
try:
shutdown_logging()
except Exception as e:
print(f"关闭日志系统时出错: {e}")
# 在程序退出前暂停，让你有机会看到输出
# input("按 Enter 键退出...") # <--- 添加这行
sys.exit(exit_code) # <--- 使用记录的退出码

752
docs/plugin_detailed_guide.md
View File

@@ -1,752 +0,0 @@
# MaiBot 插件详细解析指南
## 📋 目录
1. [插件基类详解](#插件基类详解)
2. [Action组件深入](#action组件深入)
3. [Command组件深入](#command组件深入)
4. [API系统详解](#api系统详解)
5. [配置系统](#配置系统)
6. [注册中心机制](#注册中心机制)
7. [高级功能](#高级功能)
8. [最佳实践](#最佳实践)
---
## 插件基类详解
### BasePlugin 核心功能
`BasePlugin` 是所有插件的基类，提供插件的生命周期管理和基础功能。
```python
@register_plugin
class MyPlugin(BasePlugin):
# 必需的基本信息
plugin_name = "my_plugin" # 插件唯一标识
plugin_description = "插件功能描述" # 简短描述
plugin_version = "1.0.0" # 版本号
plugin_author = "作者名称" # 作者信息
enable_plugin = True # 是否启用
# 可选配置
dependencies = ["other_plugin"] # 依赖的其他插件
config_file_name = "config.toml" # 配置文件名
def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]:
"""返回插件包含的组件列表（必须实现）"""
return [
(MyAction.get_action_info(), MyAction),
(MyCommand.get_command_info(), MyCommand)
]
```
### 插件生命周期
1. **加载阶段** - 插件管理器扫描插件目录
2. **实例化阶段** - 创建插件实例，传入 `plugin_dir`
3. **配置加载** - 自动加载配置文件（如果指定）
4. **依赖检查** - 验证依赖的插件是否存在
5. **组件注册** - 注册所有组件到注册中心
6. **运行阶段** - 组件响应用户交互
### 配置访问
```python
class MyPlugin(BasePlugin):
config_file_name = "config.toml"
def some_method(self):
# 获取配置值
max_retry = self.get_config("network.max_retry", 3)
api_key = self.get_config("api.key", "")
# 配置支持嵌套结构
db_config = self.get_config("database", {})
```
---
## Action组件深入
### Action激活机制
Action组件支持多种激活方式，可以组合使用：
#### 1. 关键词激活
```python
class KeywordAction(BaseAction):
focus_activation_type = ActionActivationType.KEYWORD
normal_activation_type = ActionActivationType.KEYWORD
activation_keywords = ["天气", "weather", "温度"]
keyword_case_sensitive = False # 是否区分大小写
async def execute(self) -> Tuple[bool, str]:
# 获取触发的关键词
triggered_keyword = self.action_data.get("triggered_keyword")
return True, f"检测到关键词: {triggered_keyword}"
```
#### 2. LLM智能判断
```python
class SmartAction(BaseAction):
focus_activation_type = ActionActivationType.LLM_JUDGE
llm_judge_prompt = """
判断用户消息是否表达了情感支持的需求。
如果用户显得沮丧、焦虑或需要安慰，返回True，否则返回False。
"""
async def execute(self) -> Tuple[bool, str]:
# LLM判断为需要情感支持
user_emotion = self.action_data.get("emotion", "neutral")
return True, "我理解你现在的感受，有什么可以帮助你的吗？ 🤗"
```
#### 3. 随机激活
```python
class RandomAction(BaseAction):
focus_activation_type = ActionActivationType.RANDOM
random_activation_probability = 0.1 # 10%概率触发
async def execute(self) -> Tuple[bool, str]:
import random
responses = ["今天天气不错呢！", "你知道吗，刚才想到一个有趣的事...", "随便聊聊吧！"]
return True, random.choice(responses)
```
#### 4. 始终激活
```python
class AlwaysAction(BaseAction):
focus_activation_type = ActionActivationType.ALWAYS
parallel_action = True # 允许与其他Action并行
async def execute(self) -> Tuple[bool, str]:
# 记录所有消息到数据库
await self.api.store_user_data("last_message", self.action_data.get("message"))
return True, "" # 静默执行，不发送回复
```
### Action数据访问
```python
class DataAction(BaseAction):
async def execute(self) -> Tuple[bool, str]:
# 访问消息数据
message = self.action_data.get("message", "")
username = self.action_data.get("username", "用户")
user_id = self.action_data.get("user_id", "")
platform = self.action_data.get("platform", "")
# 访问系统数据
thinking_id = self.thinking_id
reasoning = self.reasoning # 执行该动作的理由
# 访问计时器信息
timers = self.cycle_timers
return True, f"处理来自 {platform} 的用户 {username} 的消息"
```
### 聊天模式支持
```python
class ModeAwareAction(BaseAction):
mode_enable = ChatMode.PRIVATE # 只在私聊中启用
# mode_enable = ChatMode.GROUP # 只在群聊中启用
# mode_enable = ChatMode.ALL # 在所有模式中启用
async def execute(self) -> Tuple[bool, str]:
current_mode = self.action_data.get("chat_mode", ChatMode.PRIVATE)
return True, f"当前聊天模式: {current_mode.name}"
```
---
## Command组件深入
### 高级正则表达式模式
Command使用正则表达式进行精确匹配，支持复杂的参数提取：
#### 1. 基础命令
```python
class BasicCommand(BaseCommand):
command_pattern = r"^/hello$"
command_help = "简单的问候命令"
async def execute(self) -> Tuple[bool, Optional[str]]:
await self.send_reply("Hello!")
return True, "Hello!"
```
#### 2. 带参数命令
```python
class ParameterCommand(BaseCommand):
command_pattern = r"^/user\s+(?P<action>add|remove|list)\s+(?P<name>\w+)?$"
command_help = "用户管理命令，用法：/user <add|remove|list> [用户名]"
command_examples = ["/user add alice", "/user remove bob", "/user list"]
async def execute(self) -> Tuple[bool, Optional[str]]:
action = self.matched_groups.get("action")
name = self.matched_groups.get("name")
if action == "add" and name:
# 添加用户逻辑
await self.api.store_user_data(f"user_{name}", {"name": name, "created": self.api.get_current_time()})
response = f"用户 {name} 已添加"
elif action == "remove" and name:
# 删除用户逻辑
await self.api.delete_user_data(f"user_{name}")
response = f"用户 {name} 已删除"
elif action == "list":
# 列出用户逻辑
users = await self.api.get_user_data_pattern("user_*")
response = f"用户列表: {', '.join(users.keys())}"
else:
response = "参数错误，请查看帮助信息"
await self.send_reply(response)
return True, response
```
#### 3. 复杂参数解析
```python
class AdvancedCommand(BaseCommand):
command_pattern = r"^/remind\s+(?P<time>\d{1,2}:\d{2})\s+(?P<date>\d{4}-\d{2}-\d{2})?\s+(?P<message>.+)$"
command_help = "设置提醒，用法：/remind <时间> [日期] <消息>"
command_examples = [
"/remind 14:30 买牛奶",
"/remind 09:00 2024-12-25 圣诞节快乐"
]
async def execute(self) -> Tuple[bool, Optional[str]]:
time_str = self.matched_groups.get("time")
date_str = self.matched_groups.get("date")
message = self.matched_groups.get("message")
# 解析时间
from datetime import datetime, date
try:
hour, minute = map(int, time_str.split(":"))
if date_str:
reminder_date = datetime.strptime(date_str, "%Y-%m-%d").date()
else:
reminder_date = date.today()
# 创建提醒
reminder_time = datetime.combine(reminder_date, datetime.min.time().replace(hour=hour, minute=minute))
await self.api.store_user_data("reminder", {
"time": reminder_time.isoformat(),
"message": message,
"user_id": self.api.get_current_user_id()
})
response = f"已设置提醒：{reminder_time.strftime('%Y-%m-%d %H:%M')} - {message}"
except ValueError as e:
response = f"时间格式错误: {e}"
await self.send_reply(response)
return True, response
```
### 命令权限控制
```python
class AdminCommand(BaseCommand):
command_pattern = r"^/admin\s+(?P<operation>\w+)"
command_help = "管理员命令（需要权限）"
async def execute(self) -> Tuple[bool, Optional[str]]:
# 检查用户权限
user_id = self.api.get_current_user_id()
user_role = await self.api.get_user_info(user_id, "role", "user")
if user_role != "admin":
await self.send_reply("❌ 权限不足，需要管理员权限")
return False, "权限不足"
operation = self.matched_groups.get("operation")
# 执行管理员操作...
return True, f"管理员操作 {operation} 已执行"
```
---
## API系统详解
### MessageAPI - 消息发送
```python
class MessageExampleAction(BaseAction):
async def execute(self) -> Tuple[bool, str]:
# 发送文本消息
await self.api.send_message("text", "这是一条文本消息")
# 发送带格式的消息
await self.api.send_message("text", "**粗体文本** *斜体文本*")
# 发送图片（如果支持）
await self.api.send_message("image", "/path/to/image.jpg")
# 发送文件（如果支持）
await self.api.send_message("file", "/path/to/document.pdf")
# 获取消息发送状态
success = await self.api.send_message("text", "测试消息")
if success:
logger.info("消息发送成功")
return True, "消息发送演示完成"
```
### LLMAPI - 大模型调用
```python
class LLMExampleAction(BaseAction):
async def execute(self) -> Tuple[bool, str]:
# 获取可用模型
models = self.api.get_available_models()
if not models:
return False, "没有可用的模型"
# 选择第一个可用模型
model_name, model_config = next(iter(models.items()))
# 生成文本
prompt = "写一首关于春天的诗"
success, response, usage, model_used = await self.api.generate_with_model(
prompt=prompt,
model_config=model_config,
max_tokens=200,
temperature=0.7
)
if success:
logger.info(f"使用模型 {model_used} 生成了 {usage.get('total_tokens', 0)} 个token")
return True, f"生成的诗歌：\n{response}"
else:
return False, f"生成失败：{response}"
```
### DatabaseAPI - 数据库操作
```python
class DatabaseExampleAction(BaseAction):
async def execute(self) -> Tuple[bool, str]:
user_id = self.api.get_current_user_id()
# 存储用户数据
await self.api.store_user_data("user_score", 100)
await self.api.store_user_data("user_level", "beginner")
# 存储复杂数据结构
user_profile = {
"name": "Alice",
"age": 25,
"interests": ["music", "reading", "coding"],
"settings": {
"theme": "dark",
"language": "zh-CN"
}
}
await self.api.store_user_data("profile", user_profile)
# 读取数据
score = await self.api.get_user_data("user_score", 0)
profile = await self.api.get_user_data("profile", {})
# 删除数据
await self.api.delete_user_data("old_key")
# 批量查询
all_user_data = await self.api.get_user_data_pattern("user_*")
# 存储Action执行记录
await self.api.store_action_info(
action_build_into_prompt=True,
action_prompt_display="用户查询了个人信息",
action_done=True
)
return True, f"用户数据操作完成，当前积分：{score}"
```
### ConfigAPI - 配置访问
```python
class ConfigExampleAction(BaseAction):
async def execute(self) -> Tuple[bool, str]:
# 读取全局配置
bot_name = self.api.get_global_config("bot.name", "MaiBot")
debug_mode = self.api.get_global_config("system.debug", False)
# 获取用户信息
current_user = self.api.get_current_user_id()
platform, user_id = await self.api.get_user_id_by_person_name("Alice")
# 获取特定用户信息
user_nickname = await self.api.get_person_info(current_user, "nickname", "未知用户")
user_language = await self.api.get_person_info(current_user, "language", "zh-CN")
return True, f"配置信息获取完成，机器人名称：{bot_name}"
```
### UtilsAPI - 工具函数
```python
class UtilsExampleAction(BaseAction):
async def execute(self) -> Tuple[bool, str]:
# 时间相关
current_time = self.api.get_current_time()
formatted_time = self.api.format_time(current_time, "%Y-%m-%d %H:%M:%S")
# ID生成
unique_id = self.api.generate_unique_id()
random_string = self.api.generate_random_string(length=8)
# 文件操作
if self.api.file_exists("/path/to/file.txt"):
content = self.api.read_file("/path/to/file.txt")
self.api.write_file("/path/to/backup.txt", content)
# JSON处理
data = {"key": "value", "number": 42}
json_str = self.api.to_json(data)
parsed_data = self.api.from_json(json_str)
# 安全字符串处理
safe_filename = self.api.sanitize_filename("用户文件 (1).txt")
return True, f"工具函数演示完成，时间：{formatted_time}"
```
---
## 配置系统
### TOML配置文件
```toml
# config.toml
[plugin]
name = "my_plugin"
description = "插件描述"
enabled = true
debug = false
[features]
enable_ai = true
enable_voice = false
max_users = 100
[api]
timeout = 30
retry_count = 3
base_url = "https://api.example.com"
[database]
cache_size = 1000
auto_cleanup = true
[messages]
welcome = "欢迎使用插件！"
error = "操作失败，请重试"
success = "操作成功完成"
[advanced]
custom_settings = { theme = "dark", language = "zh-CN" }
feature_flags = ["beta_feature", "experimental_ui"]
```
### 配置使用示例
```python
class ConfigurablePlugin(BasePlugin):
config_file_name = "config.toml"
def get_plugin_components(self):
# 根据配置决定加载哪些组件
components = []
if self.get_config("features.enable_ai", False):
components.append((AIAction.get_action_info(), AIAction))
if self.get_config("features.enable_voice", False):
components.append((VoiceCommand.get_command_info(), VoiceCommand))
return components
class ConfigurableAction(BaseAction):
async def execute(self) -> Tuple[bool, str]:
# 注意：这里不能直接创建插件实例获取配置
# 应该通过其他方式访问配置，比如从API或全局配置中获取
# 使用默认值或硬编码配置
welcome_message = "欢迎使用插件！" # 应该从配置获取
timeout = 30 # 应该从配置获取
return True, welcome_message
```
---
## 注册中心机制
### 组件查询
```python
from src.plugin_system.core.component_registry import component_registry
# 获取所有注册的Action
actions = component_registry.get_components_by_type(ComponentType.ACTION)
# 获取所有注册的Command
commands = component_registry.get_components_by_type(ComponentType.COMMAND)
# 查找特定命令
command_info = component_registry.find_command_by_text("/help")
# 获取插件信息
plugin_info = component_registry.get_plugin_info("simple_plugin")
# 获取插件的所有组件
plugin_components = component_registry.get_plugin_components("simple_plugin")
```
### 动态组件操作
```python
# 注册新组件
component_info = ActionInfo(name="dynamic_action", ...)
component_registry.register_component(component_info, DynamicAction)
# 注销组件
component_registry.unregister_component("dynamic_action")
# 检查组件是否存在
exists = component_registry.component_exists("my_action")
```
---
## 高级功能
### 组件依赖管理
```python
class DependentPlugin(BasePlugin):
plugin_name = "dependent_plugin"
dependencies = ["simple_plugin", "core_plugin"] # 依赖其他插件
def get_plugin_components(self):
# 只有在依赖满足时才会被调用
return [(MyAction.get_action_info(), MyAction)]
```
### 动态组件创建
```python
def create_dynamic_action(keyword: str, response: str):
"""动态创建Action组件"""
class DynamicAction(BaseAction):
focus_activation_type = ActionActivationType.KEYWORD
activation_keywords = [keyword]
async def execute(self) -> Tuple[bool, str]:
return True, response
return DynamicAction
# 使用
WeatherAction = create_dynamic_action("天气", "今天天气很好！")
```
### 组件生命周期钩子
```python
class LifecycleAction(BaseAction):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.on_initialize()
def on_initialize(self):
"""组件初始化时调用"""
logger.info("Action组件初始化")
async def execute(self) -> Tuple[bool, str]:
result = await self.on_execute()
self.on_complete()
return result
async def on_execute(self) -> Tuple[bool, str]:
"""实际执行逻辑"""
return True, "执行完成"
def on_complete(self):
"""执行完成后调用"""
logger.info("Action执行完成")
```
---
## 最佳实践
### 1. 错误处理
```python
class RobustAction(BaseAction):
async def execute(self) -> Tuple[bool, str]:
try:
# 主要逻辑
result = await self.process_main_logic()
return True, result
except ValueError as e:
# 参数错误
logger.warning(f"参数错误: {e}")
return False, "参数格式不正确"
except ConnectionError as e:
# 网络错误
logger.error(f"网络连接失败: {e}")
return False, "网络连接异常，请稍后重试"
except Exception as e:
# 未知错误
logger.error(f"未知错误: {e}", exc_info=True)
return False, "处理失败，请联系管理员"
async def process_main_logic(self):
# 具体业务逻辑
pass
```
### 2. 性能优化
```python
class OptimizedAction(BaseAction):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._cache = {} # 本地缓存
async def execute(self) -> Tuple[bool, str]:
cache_key = self.generate_cache_key()
# 检查缓存
if cache_key in self._cache:
logger.debug("使用缓存结果")
return True, self._cache[cache_key]
# 计算结果
result = await self.compute_result()
# 存储到缓存
self._cache[cache_key] = result
return True, result
def generate_cache_key(self) -> str:
# 根据输入生成缓存键
message = self.action_data.get("message", "")
return f"result_{hash(message)}"
```
### 3. 资源管理
```python
class ResourceAction(BaseAction):
async def execute(self) -> Tuple[bool, str]:
# 使用上下文管理器确保资源正确释放
async with self.api.get_resource_manager() as resources:
# 获取资源
db_connection = await resources.get_database()
file_handle = await resources.get_file("data.txt")
# 使用资源进行处理
result = await self.process_with_resources(db_connection, file_handle)
return True, result
# 资源会自动释放
```
### 4. 测试友好设计
```python
class TestableAction(BaseAction):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.dependencies = self.create_dependencies()
def create_dependencies(self):
"""创建依赖对象，便于测试时注入mock"""
return {
'weather_service': WeatherService(),
'user_service': UserService()
}
async def execute(self) -> Tuple[bool, str]:
weather = await self.dependencies['weather_service'].get_weather()
user = await self.dependencies['user_service'].get_current_user()
return True, f"今天{weather}，{user}！"
```
### 5. 日志记录
```python
class LoggedAction(BaseAction):
async def execute(self) -> Tuple[bool, str]:
start_time = self.api.get_current_time()
logger.info(f"{self.log_prefix} 开始执行，用户: {self.action_data.get('username')}")
try:
result = await self.process()
duration = self.api.get_current_time() - start_time
logger.info(f"{self.log_prefix} 执行成功，耗时: {duration}ms")
return True, result
except Exception as e:
logger.error(f"{self.log_prefix} 执行失败: {e}", exc_info=True)
raise
```
---
## 总结
通过本详细指南，你已经深入了解了MaiBot插件系统的各个方面：
- **插件基类** - 生命周期管理和配置系统
- **Action组件** - 多种激活机制和智能交互
- **Command组件** - 强大的正则表达式匹配和参数处理
- **API系统** - 7大模块提供完整功能支持
- **高级功能** - 依赖管理、动态创建、生命周期钩子
- **最佳实践** - 错误处理、性能优化、资源管理
现在你已经具备了开发复杂插件的所有知识！
## 📚 相关文档
- [系统总览](plugin_guide_overview.md) - 了解整体架构
- [快速开始](plugin_quick_start.md) - 5分钟创建第一个插件
- [示例插件](../src/plugins/examples/simple_plugin/) - 完整功能参考
---
> 💡 **持续学习**: 插件开发是一个实践的过程，建议边学边做，逐步掌握各种高级特性！

155
docs/plugin_guide_overview.md
View File

@@ -1,155 +0,0 @@
# MaiBot 插件编写指南 - 总览
## 📋 目录结构
本指南分为三个部分：
- **[总览](plugin_guide_overview.md)** - 插件系统架构和设计理念（当前文档）
- **[快速开始](plugin_quick_start.md)** - 5分钟创建你的第一个插件
- **[详细解析](plugin_detailed_guide.md)** - 深入理解各个组件和API
## 🎯 插件系统概述
MaiBot 采用组件化的插件系统，让开发者可以轻松扩展机器人功能。系统支持两种主要组件类型：
- **Action组件** - 智能动作，基于关键词、LLM判断等条件自动触发
- **Command组件** - 命令处理，基于正则表达式匹配用户输入的命令
## 🏗️ 系统架构
```
src/
├── plugin_system/ # 🔧 系统核心（框架代码）
│ ├── core/ # 插件管理和注册中心
│ ├── apis/ # 统一API接口（7大模块）
│ ├── base/ # 插件和组件基类
│ └── registry/ # 组件注册和查询
└── plugins/ # 🔌 插件内容（用户代码）
├── built_in/ # 内置插件
└── examples/ # 示例插件
```
### 核心设计理念
1. **分离关注点** - 系统框架与插件内容完全分离
2. **组件化设计** - 一个插件可包含多个Action和Command组件
3. **统一API访问** - 通过PluginAPI统一访问所有系统功能
4. **声明式配置** - 通过类属性声明组件行为，简化开发
5. **类型安全** - 完整的类型定义，IDE友好
## 🧩 组件类型详解
### Action组件 - 智能动作
Action用于实现智能交互逻辑，支持多种激活方式：
- **关键词激活** - 消息包含特定关键词时触发
- **LLM判断激活** - 使用大模型智能判断是否需要触发
- **随机激活** - 按概率随机触发
- **始终激活** - 每条消息都触发（谨慎使用）
**适用场景：**
- 智能问候、闲聊互动
- 情感分析和回应
- 内容审核和提醒
- 数据统计和分析
### Command组件 - 命令处理
Command用于处理结构化的用户命令，基于正则表达式匹配：
- **精确匹配** - 支持参数提取和验证
- **灵活模式** - 正则表达式的完整威力
- **帮助系统** - 自动生成命令帮助信息
**适用场景：**
- 功能性操作（查询、设置、管理）
- 工具类命令（计算、转换、搜索）
- 系统管理命令
- 游戏和娱乐功能
## 🔌 API系统概览
系统提供7大API模块，涵盖插件开发的所有需求：
| API模块 | 功能描述 | 主要用途 |
|---------|----------|----------|
| **MessageAPI** | 消息发送和交互 | 发送文本、图片、语音等消息 |
| **LLMAPI** | 大模型调用 | 文本生成、智能判断、创意创作 |
| **DatabaseAPI** | 数据库操作 | 存储用户数据、配置、历史记录 |
| **ConfigAPI** | 配置访问 | 读取全局配置和用户信息 |
| **UtilsAPI** | 工具函数 | 文件操作、时间处理、ID生成 |
| **StreamAPI** | 流管理 | 聊天流控制和状态管理 |
| **HearflowAPI** | 心流系统 | 与消息处理流程集成 |
## 🎨 开发体验
### 简化的导入接口
```python
from src.plugin_system import (
BasePlugin, register_plugin, BaseAction, BaseCommand,
ComponentInfo, ActionInfo, CommandInfo, ActionActivationType, ChatMode
)
```
### 声明式组件定义
```python
class HelloAction(BaseAction):
# 🎯 直接通过类属性定义行为
focus_activation_type = ActionActivationType.KEYWORD
activation_keywords = ["你好", "hello", "hi"]
mode_enable = ChatMode.ALL
async def execute(self) -> Tuple[bool, str]:
return True, "你好！我是MaiBot 😊"
```
### 统一的API访问
```python
class MyCommand(BaseCommand):
async def execute(self) -> Tuple[bool, Optional[str]]:
# 💡 通过self.api访问所有系统功能
await self.api.send_message("text", "处理中...")
models = self.api.get_available_models()
await self.api.store_user_data("key", "value")
return True, "完成！"
```
## 🚀 开发流程
1. **创建插件目录** - 在 `src/plugins/` 下创建插件文件夹
2. **定义插件类** - 继承 `BasePlugin`，设置基本信息
3. **创建组件类** - 继承 `BaseAction` 或 `BaseCommand`
4. **注册组件** - 在插件的 `get_plugin_components()` 中返回组件列表
5. **测试验证** - 启动系统测试插件功能
## 📚 学习路径建议
1. **初学者** - 从[快速开始](plugin_quick_start.md)开始，5分钟体验插件开发
2. **进阶开发** - 阅读[详细解析](plugin_detailed_guide.md)，深入理解各个组件
3. **实战练习** - 参考 `simple_plugin` 示例，尝试开发自己的插件
4. **API探索** - 逐步尝试各个API模块的功能
## 💡 设计亮点
- **零配置启动** - 插件放入目录即可自动加载
- **热重载支持** - 开发过程中可动态重载插件（规划中）
- **依赖管理** - 支持插件间依赖关系声明
- **配置系统** - 支持TOML配置文件，灵活可定制
- **完整API** - 覆盖机器人开发的各个方面
- **类型安全** - 完整的类型注解，IDE智能提示
## 🎯 下一步
选择适合你的起点：
- 🚀 [立即开始 →](plugin_quick_start.md)
- 📖 [深入学习 →](plugin_detailed_guide.md)
- 🔍 [查看示例 →](../src/plugins/examples/simple_plugin/)
---
> 💡 **提示**: 插件系统仍在持续改进中，欢迎提出建议和反馈！

119
docs/plugin_loading_paths.md
View File

@@ -1,119 +0,0 @@
# 插件加载路径说明
## 概述
MaiBot-Core 现在支持从多个路径加载插件，为插件开发者提供更大的灵活性。
## 支持的插件路径
系统会按以下优先级顺序搜索和加载插件：
### 1. 项目根目录插件路径：`/plugins`
- **路径**: 项目根目录下的 `plugins/` 文件夹
- **优先级**: 最高
- **用途**: 用户自定义插件、第三方插件
- **特点**:
- 与项目源码分离
- 便于版本控制管理
- 适合用户添加个人插件
### 2. 源码目录插件路径：`/src/plugins`
- **路径**: src目录下的 `plugins/` 文件夹
- **优先级**: 次高
- **用途**: 系统内置插件、官方插件
- **特点**:
- 与项目源码集成
- 适合系统级功能插件
## 插件结构支持
两个路径都支持相同的插件结构：
### 传统结构（推荐用于复杂插件）
```
plugins/my_plugin/
├── __init__.py
├── actions/
│ ├── __init__.py
│ └── my_action.py
├── commands/
│ ├── __init__.py
│ └── my_command.py
└── config.toml
```
### 简化结构（推荐用于简单插件）
```
plugins/my_plugin/
├── __init__.py
├── my_action.py
├── my_command.py
└── config.toml
```
## 文件命名约定
### 动作文件
- `*_action.py`
- `*_actions.py`
- 包含 `action` 字样的文件名
### 命令文件
- `*_command.py`
- `*_commands.py`
- 包含 `command` 字样的文件名
## 加载行为
1. **顺序加载**: 先加载 `/plugins`，再加载 `/src/plugins`
2. **重名处理**: 如果两个路径中有同名插件，优先加载 `/plugins` 中的版本
3. **错误隔离**: 单个插件加载失败不会影响其他插件的加载
4. **详细日志**: 系统会记录每个插件的来源路径和加载状态
## 最佳实践
### 用户插件开发
- 将自定义插件放在 `/plugins` 目录
- 使用清晰的插件命名
- 包含必要的 `__init__.py` 文件
### 系统插件开发
- 将系统集成插件放在 `/src/plugins` 目录
- 遵循项目代码规范
- 完善的错误处理
### 版本控制
- 将 `/plugins` 目录添加到 `.gitignore`（如果是用户自定义插件）
- 或者为插件创建独立的git仓库
## 示例插件
参考 `/plugins/example_root_plugin/` 中的示例插件，了解如何在根目录创建插件。
## 故障排除
### 常见问题
1. **插件未被加载**
- 检查插件目录是否有 `__init__.py` 文件
- 确认文件命名符合约定
- 查看启动日志中的加载信息
2. **导入错误**
- 确保插件依赖的模块已安装
- 检查导入路径是否正确
3. **重复注册**
- 检查是否有同名的动作或命令
- 避免在不同路径放置相同功能的插件
### 调试日志
启动时查看日志输出：
```
[INFO] 正在从 plugins 加载插件...
[INFO] 正在从 src/plugins 加载插件...
[SUCCESS] 插件加载完成: 总计 X 个动作, Y 个命令
[INFO] 插件加载详情:
[INFO] example_plugin (来源: plugins): 1 动作, 1 命令
```

270
docs/plugin_quick_start.md
View File

@@ -1,270 +0,0 @@
# MaiBot 插件快速开始指南
## 🚀 5分钟创建你的第一个插件
本指南将带你快速创建一个功能完整的插件，体验MaiBot插件开发的简单和强大。
## 📋 前置要求
- 已克隆MaiBot项目到本地
- Python 3.8+ 环境
- 基本的Python编程知识
## 🎯 我们要做什么
我们将创建一个名为 `my_first_plugin` 的插件，包含：
- 一个Action组件：自动回应"Hello"
- 一个Command组件：计算器功能
## 📁 第一步：创建插件目录
在 `src/plugins/examples/` 下创建你的插件目录：
```bash
mkdir src/plugins/examples/my_first_plugin
```
## 📝 第二步：创建插件文件
在插件目录中创建 `plugin.py` 文件：
```python
# src/plugins/examples/my_first_plugin/plugin.py
from typing import List, Tuple, Type, Optional
import re
# 导入插件系统核心
from src.plugin_system import (
BasePlugin, register_plugin, BaseAction, BaseCommand,
ComponentInfo, ActionInfo, CommandInfo, ActionActivationType, ChatMode
)
from src.common.logger import get_logger
logger = get_logger("my_first_plugin")
class HelloAction(BaseAction):
"""自动问候Action - 当用户说Hello时自动回应"""
# 🎯 声明式配置：只需设置类属性
focus_activation_type = ActionActivationType.KEYWORD
normal_activation_type = ActionActivationType.KEYWORD
activation_keywords = ["hello", "Hello", "HELLO"]
keyword_case_sensitive = False
mode_enable = ChatMode.ALL
parallel_action = False
async def execute(self) -> Tuple[bool, str]:
"""执行问候动作"""
username = self.action_data.get("username", "朋友")
response = f"Hello, {username}! 很高兴见到你！ 🎉"
logger.info(f"向 {username} 发送问候")
return True, response
class CalculatorCommand(BaseCommand):
"""计算器命令 - 执行简单数学运算"""
# 🎯 声明式配置：定义命令模式
command_pattern = r"^/calc\s+(?P<expression>[\d\+\-\*/\(\)\s\.]+)$"
command_help = "计算器，用法：/calc <数学表达式>"
command_examples = ["/calc 1+1", "/calc 2*3+4", "/calc (10-5)*2"]
async def execute(self) -> Tuple[bool, Optional[str]]:
"""执行计算命令"""
# 获取匹配的表达式
expression = self.matched_groups.get("expression", "").strip()
if not expression:
await self.send_reply("❌ 请提供数学表达式！")
return False, None
try:
# 安全计算（只允许基本数学运算）
allowed_chars = set("0123456789+-*/.() ")
if not all(c in allowed_chars for c in expression):
await self.send_reply("❌ 表达式包含不允许的字符！")
return False, None
# 执行计算
result = eval(expression) # 在实际项目中应使用更安全的计算方法
response = f"🧮 计算结果：\n`{expression} = {result}`"
await self.send_reply(response)
logger.info(f"计算: {expression} = {result}")
return True, response
except Exception as e:
error_msg = f"❌ 计算错误: {str(e)}"
await self.send_reply(error_msg)
logger.error(f"计算失败: {expression}, 错误: {e}")
return False, error_msg
@register_plugin
class MyFirstPlugin(BasePlugin):
"""我的第一个插件 - 展示基本功能"""
# 🏷️ 插件基本信息
plugin_name = "my_first_plugin"
plugin_description = "我的第一个MaiBot插件，包含问候和计算功能"
plugin_version = "1.0.0"
plugin_author = "你的名字"
enable_plugin = True
def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]:
"""返回插件包含的组件"""
return [
# Action组件：自动问候
(HelloAction.get_action_info(
name="hello_action",
description="自动回应Hello问候"
), HelloAction),
# Command组件：计算器
(CalculatorCommand.get_command_info(
name="calculator_command",
description="简单计算器，支持基础数学运算"
), CalculatorCommand)
]
```
## 📊 第三步：创建配置文件（可选）
创建 `config.toml` 文件来配置插件：
```toml
# src/plugins/examples/my_first_plugin/config.toml
[plugin]
name = "my_first_plugin"
description = "我的第一个插件"
enabled = true
[hello_action]
enable_emoji = true
greeting_message = "Hello, {username}! 很高兴见到你！"
[calculator]
max_expression_length = 100
allow_complex_math = false
```
如果你创建了配置文件，需要在插件类中指定：
```python
@register_plugin
class MyFirstPlugin(BasePlugin):
# ... 其他属性 ...
config_file_name = "config.toml" # 添加这一行
```
## 🔄 第四步：测试插件
1. **启动MaiBot**：
```bash
cd /path/to/MaiBot-Core
python -m src.main
```
2. **测试Action组件**：
- 在聊天中发送 "Hello" 或 "hello"
- 应该收到自动回复："Hello, [用户名]! 很高兴见到你！ 🎉"
3. **测试Command组件**：
- 发送 `/calc 1+1`
- 应该收到回复："🧮 计算结果：\n`1+1 = 2`"
## 🎉 恭喜！
你已经成功创建了第一个MaiBot插件！插件包含：
✅ **一个Action组件** - 智能响应用户问候
✅ **一个Command组件** - 提供计算器功能
✅ **完整的配置** - 支持自定义行为
✅ **错误处理** - 优雅处理异常情况
## 🔍 代码解析
### Action组件关键点
```python
# 声明激活条件
focus_activation_type = ActionActivationType.KEYWORD
activation_keywords = ["hello", "Hello", "HELLO"]
# 执行逻辑
async def execute(self) -> Tuple[bool, str]:
# 处理逻辑
return True, response # (成功状态, 回复内容)
```
### Command组件关键点
```python
# 声明命令模式（正则表达式）
command_pattern = r"^/calc\s+(?P<expression>[\d\+\-\*/\(\)\s\.]+)$"
# 执行逻辑
async def execute(self) -> Tuple[bool, Optional[str]]:
expression = self.matched_groups.get("expression") # 获取匹配参数
await self.send_reply(response) # 发送回复
return True, response
```
### 插件注册
```python
@register_plugin # 装饰器注册插件
class MyFirstPlugin(BasePlugin):
# 基本信息
plugin_name = "my_first_plugin"
plugin_description = "插件描述"
# 返回组件列表
def get_plugin_components(self):
return [(组件信息, 组件类), ...]
```
## 🎯 下一步学习
现在你已经掌握了基础，可以继续学习：
1. **深入API** - 探索[详细解析](plugin_detailed_guide.md)了解更多API功能
2. **参考示例** - 查看 `simple_plugin` 了解更复杂的功能
3. **自定义扩展** - 尝试添加更多组件和功能
## 🛠️ 常见问题
### Q: 插件没有被加载？
A: 检查：
- 插件目录是否在 `src/plugins/` 下
- 文件名是否为 `plugin.py`
- 类是否有 `@register_plugin` 装饰器
- 是否有语法错误
### Q: Action组件没有触发？
A: 检查：
- `activation_keywords` 是否正确设置
- `focus_activation_type` 和 `normal_activation_type` 是否设置
- 消息内容是否包含关键词
### Q: Command组件不响应？
A: 检查：
- `command_pattern` 正则表达式是否正确
- 用户输入是否完全匹配模式
- 是否有语法错误
## 📚 相关文档
- [系统总览](plugin_guide_overview.md) - 了解整体架构
- [详细解析](plugin_detailed_guide.md) - 深入学习各个组件
- [示例插件](../src/plugins/examples/simple_plugin/) - 完整功能示例
---
> 🎉 **恭喜完成快速开始！** 现在你已经是MaiBot插件开发者了！

10
scripts/log_viewer.py
View File

@@ -860,7 +860,8 @@ class LogViewer:
while self.running:
if log_file.exists():
try:
with open(log_file, "r", encoding="utf-8") as f:
# 使用共享读取模式，避免文件锁定
with open(log_file, "r", encoding="utf-8", buffering=1) as f:
f.seek(last_position)
new_lines = f.readlines()
last_position = f.tell()
@@ -881,8 +882,13 @@ class LogViewer:
except json.JSONDecodeError:
continue
except (FileNotFoundError, PermissionError) as e:
# 文件被占用或不存在时，等待更长时间
print(f"日志文件访问受限: {e}")
time.sleep(1)
continue
except Exception as e:
print(f"Error reading log file: {e}")
print(f"读取日志文件时出错: {e}")
time.sleep(0.1)

2
src/chat/normal_chat/normal_chat.py
View File

@@ -328,8 +328,6 @@ class NormalChat:
thinking_id = await self._create_thinking_message(message)
logger.debug(f"[{self.stream_name}] 创建捕捉器，thinking_id:{thinking_id}")
# 如果启用planner，预先修改可用actions（避免在并行任务中重复调用）
available_actions = None
if self.enable_planner:

280
src/common/logger.py
View File

@@ -11,6 +11,47 @@ import toml
LOG_DIR = Path("logs")
LOG_DIR.mkdir(exist_ok=True)
# 全局handler实例，避免重复创建
_file_handler = None
_console_handler = None
def get_file_handler():
"""获取文件handler单例"""
global _file_handler
if _file_handler is None:
_file_handler = logging.handlers.RotatingFileHandler(
LOG_DIR / "app.log.jsonl",
maxBytes=10 * 1024 * 1024, # 10MB
backupCount=5,
encoding="utf-8",
)
# 设置文件handler的日志级别
file_level = LOG_CONFIG.get("file_log_level", LOG_CONFIG.get("log_level", "INFO"))
_file_handler.setLevel(getattr(logging, file_level.upper(), logging.INFO))
return _file_handler
def get_console_handler():
"""获取控制台handler单例"""
global _console_handler
if _console_handler is None:
_console_handler = logging.StreamHandler()
# 设置控制台handler的日志级别
console_level = LOG_CONFIG.get("console_log_level", LOG_CONFIG.get("log_level", "INFO"))
_console_handler.setLevel(getattr(logging, console_level.upper(), logging.INFO))
return _console_handler
def close_handlers():
"""安全关闭所有handler"""
global _file_handler, _console_handler
if _file_handler:
_file_handler.close()
_file_handler = None
if _console_handler:
_console_handler.close()
_console_handler = None
# 读取日志配置
def load_log_config():
@@ -20,7 +61,9 @@ def load_log_config():
"date_style": "Y-m-d H:i:s",
"log_level_style": "lite",
"color_text": "title",
"log_level": "INFO",
"log_level": "INFO", # 全局日志级别（向下兼容）
"console_log_level": "INFO", # 控制台日志级别
"file_log_level": "DEBUG", # 文件日志级别
"suppress_libraries": [],
"library_log_levels": {},
}
@@ -61,10 +104,17 @@ def get_timestamp_format():
def configure_third_party_loggers():
"""配置第三方库的日志级别"""
# 设置全局日志级别
global_log_level = LOG_CONFIG.get("log_level", "INFO")
# 设置根logger级别为所有handler中最低的级别，确保所有日志都能被捕获
console_level = LOG_CONFIG.get("console_log_level", LOG_CONFIG.get("log_level", "INFO"))
file_level = LOG_CONFIG.get("file_log_level", LOG_CONFIG.get("log_level", "INFO"))
# 获取最低级别（DEBUG < INFO < WARNING < ERROR < CRITICAL）
console_level_num = getattr(logging, console_level.upper(), logging.INFO)
file_level_num = getattr(logging, file_level.upper(), logging.INFO)
min_level = min(console_level_num, file_level_num)
root_logger = logging.getLogger()
root_logger.setLevel(getattr(logging, global_log_level.upper(), logging.INFO))
root_logger.setLevel(min_level)
# 完全屏蔽的库
suppress_libraries = LOG_CONFIG.get("suppress_libraries", [])
@@ -117,19 +167,23 @@ def reconfigure_existing_loggers():
# 强制清除并重新设置所有handler
original_handlers = logger_obj.handlers[:]
for handler in original_handlers:
# 安全关闭handler
if hasattr(handler, 'close'):
handler.close()
logger_obj.removeHandler(handler)
# 如果logger没有handler，让它使用根logger的handler（propagate=True）
if not logger_obj.handlers:
logger_obj.propagate = True
# 如果logger有自己的handler，重新配置它们
# 如果logger有自己的handler，重新配置它们（避免重复创建文件handler）
for handler in original_handlers:
if isinstance(handler, logging.handlers.RotatingFileHandler):
handler.setFormatter(file_formatter)
# 不重新添加，让它使用根logger的文件handler
continue
elif isinstance(handler, logging.StreamHandler):
handler.setFormatter(console_formatter)
logger_obj.addHandler(handler)
logger_obj.addHandler(handler)
# 定义模块颜色映射
@@ -377,20 +431,14 @@ class ModuleColoredConsoleRenderer:
# 配置标准logging以支持文件输出和压缩
# 使用单例handler避免重复创建
file_handler = get_file_handler()
console_handler = get_console_handler()
logging.basicConfig(
level=logging.INFO,
format="%(message)s",
handlers=[
# 带压缩的轮转文件处理器
logging.handlers.RotatingFileHandler(
LOG_DIR / "app.log.jsonl",
maxBytes=10 * 1024 * 1024, # 10MB
backupCount=5,
encoding="utf-8",
),
# 控制台处理器
logging.StreamHandler(),
],
handlers=[file_handler, console_handler],
)
@@ -462,23 +510,17 @@ def _immediate_setup():
for handler in root_logger.handlers[:]:
root_logger.removeHandler(handler)
# 使用单例handler避免重复创建
file_handler = get_file_handler()
console_handler = get_console_handler()
# 重新添加配置好的handler
root_logger.addHandler(
logging.handlers.RotatingFileHandler(
LOG_DIR / "app.log.jsonl",
maxBytes=10 * 1024 * 1024,
backupCount=5,
encoding="utf-8",
)
)
root_logger.addHandler(logging.StreamHandler())
root_logger.addHandler(file_handler)
root_logger.addHandler(console_handler)
# 设置格式化器
for handler in root_logger.handlers:
if isinstance(handler, logging.handlers.RotatingFileHandler):
handler.setFormatter(file_formatter)
else:
handler.setFormatter(console_formatter)
file_handler.setFormatter(file_formatter)
console_handler.setFormatter(console_formatter)
# 配置第三方库日志
configure_third_party_loggers()
@@ -508,6 +550,8 @@ def get_logger(name: Optional[str]) -> structlog.stdlib.BoundLogger:
def configure_logging(
level: str = "INFO",
console_level: str = None,
file_level: str = None,
max_bytes: int = 10 * 1024 * 1024,
backup_count: int = 5,
log_dir: str = "logs",
@@ -517,15 +561,31 @@ def configure_logging(
log_path.mkdir(exist_ok=True)
# 更新文件handler配置
root_logger = logging.getLogger()
for handler in root_logger.handlers:
if isinstance(handler, logging.handlers.RotatingFileHandler):
handler.maxBytes = max_bytes
handler.backupCount = backup_count
handler.baseFilename = str(log_path / "app.log.jsonl")
file_handler = get_file_handler()
if file_handler:
file_handler.maxBytes = max_bytes
file_handler.backupCount = backup_count
file_handler.baseFilename = str(log_path / "app.log.jsonl")
# 更新文件handler日志级别
if file_level:
file_handler.setLevel(getattr(logging, file_level.upper(), logging.INFO))
# 设置日志级别
root_logger.setLevel(getattr(logging, level.upper()))
# 更新控制台handler日志级别
console_handler = get_console_handler()
if console_handler and console_level:
console_handler.setLevel(getattr(logging, console_level.upper(), logging.INFO))
# 设置根logger日志级别为最低级别
if console_level or file_level:
console_level_num = getattr(logging, (console_level or level).upper(), logging.INFO)
file_level_num = getattr(logging, (file_level or level).upper(), logging.INFO)
min_level = min(console_level_num, file_level_num)
root_logger = logging.getLogger()
root_logger.setLevel(min_level)
else:
root_logger = logging.getLogger()
root_logger.setLevel(getattr(logging, level.upper()))
def set_module_color(module_name: str, color_code: str):
@@ -548,6 +608,17 @@ def reload_log_config():
global LOG_CONFIG
LOG_CONFIG = load_log_config()
# 重新设置handler的日志级别
file_handler = get_file_handler()
if file_handler:
file_level = LOG_CONFIG.get("file_log_level", LOG_CONFIG.get("log_level", "INFO"))
file_handler.setLevel(getattr(logging, file_level.upper(), logging.INFO))
console_handler = get_console_handler()
if console_handler:
console_level = LOG_CONFIG.get("console_log_level", LOG_CONFIG.get("log_level", "INFO"))
console_handler.setLevel(getattr(logging, console_level.upper(), logging.INFO))
# 重新配置console渲染器
root_logger = logging.getLogger()
for handler in root_logger.handlers:
@@ -579,8 +650,66 @@ def get_log_config():
return LOG_CONFIG.copy()
def set_console_log_level(level: str):
"""设置控制台日志级别
Args:
level: 日志级别 ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")
"""
global LOG_CONFIG
LOG_CONFIG["console_log_level"] = level.upper()
console_handler = get_console_handler()
if console_handler:
console_handler.setLevel(getattr(logging, level.upper(), logging.INFO))
# 重新设置root logger级别
configure_third_party_loggers()
logger = get_logger("logger")
logger.info(f"控制台日志级别已设置为: {level.upper()}")
def set_file_log_level(level: str):
"""设置文件日志级别
Args:
level: 日志级别 ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")
"""
global LOG_CONFIG
LOG_CONFIG["file_log_level"] = level.upper()
file_handler = get_file_handler()
if file_handler:
file_handler.setLevel(getattr(logging, level.upper(), logging.INFO))
# 重新设置root logger级别
configure_third_party_loggers()
logger = get_logger("logger")
logger.info(f"文件日志级别已设置为: {level.upper()}")
def get_current_log_levels():
"""获取当前的日志级别设置"""
file_handler = get_file_handler()
console_handler = get_console_handler()
file_level = logging.getLevelName(file_handler.level) if file_handler else "UNKNOWN"
console_level = logging.getLevelName(console_handler.level) if console_handler else "UNKNOWN"
return {
"console_level": console_level,
"file_level": file_level,
"root_level": logging.getLevelName(logging.getLogger().level)
}
def force_reset_all_loggers():
"""强制重置所有logger，解决格式不一致问题"""
# 先关闭现有的handler
close_handlers()
# 清除所有现有的logger配置
logging.getLogger().manager.loggerDict.clear()
@@ -588,27 +717,27 @@ def force_reset_all_loggers():
root_logger = logging.getLogger()
root_logger.handlers.clear()
# 使用单例handler避免重复创建
file_handler = get_file_handler()
console_handler = get_console_handler()
# 重新添加我们的handler
root_logger.addHandler(
logging.handlers.RotatingFileHandler(
LOG_DIR / "app.log.jsonl",
maxBytes=10 * 1024 * 1024,
backupCount=5,
encoding="utf-8",
)
)
root_logger.addHandler(logging.StreamHandler())
root_logger.addHandler(file_handler)
root_logger.addHandler(console_handler)
# 设置格式化器
for handler in root_logger.handlers:
if isinstance(handler, logging.handlers.RotatingFileHandler):
handler.setFormatter(file_formatter)
else:
handler.setFormatter(console_formatter)
file_handler.setFormatter(file_formatter)
console_handler.setFormatter(console_formatter)
# 设置级别
global_log_level = LOG_CONFIG.get("log_level", "INFO")
root_logger.setLevel(getattr(logging, global_log_level.upper(), logging.INFO))
# 设置根logger级别为所有handler中最低的级别
console_level = LOG_CONFIG.get("console_log_level", LOG_CONFIG.get("log_level", "INFO"))
file_level = LOG_CONFIG.get("file_log_level", LOG_CONFIG.get("log_level", "INFO"))
console_level_num = getattr(logging, console_level.upper(), logging.INFO)
file_level_num = getattr(logging, file_level.upper(), logging.INFO)
min_level = min(console_level_num, file_level_num)
root_logger.setLevel(min_level)
def initialize_logging():
@@ -623,8 +752,9 @@ def initialize_logging():
# 输出初始化信息
logger = get_logger("logger")
log_level = LOG_CONFIG.get("log_level", "INFO")
logger.info(f"日志系统已重新初始化，日志级别: {log_level}，所有logger已统一配置")
console_level = LOG_CONFIG.get("console_log_level", LOG_CONFIG.get("log_level", "INFO"))
file_level = LOG_CONFIG.get("file_log_level", LOG_CONFIG.get("log_level", "INFO"))
logger.info(f"日志系统已重新初始化，控制台级别: {console_level}，文件级别: {file_level}，所有logger已统一配置")
def force_initialize_logging():
@@ -643,8 +773,9 @@ def force_initialize_logging():
# 输出初始化信息
logger = get_logger("logger")
log_level = LOG_CONFIG.get("log_level", "INFO")
logger.info(f"日志系统已强制重新初始化，日志级别: {log_level}，所有logger格式已统一")
console_level = LOG_CONFIG.get("console_log_level", LOG_CONFIG.get("log_level", "INFO"))
file_level = LOG_CONFIG.get("file_log_level", LOG_CONFIG.get("log_level", "INFO"))
logger.info(f"日志系统已强制重新初始化，控制台级别: {console_level}，文件级别: {file_level}，所有logger格式已统一")
def show_module_colors():
@@ -678,3 +809,30 @@ def format_json_for_logging(data, indent=2, ensure_ascii=False):
else:
# 如果是对象，直接格式化
return json.dumps(data, indent=indent, ensure_ascii=ensure_ascii)
def shutdown_logging():
"""优雅关闭日志系统，释放所有文件句柄"""
logger = get_logger("logger")
logger.info("正在关闭日志系统...")
# 关闭所有handler
root_logger = logging.getLogger()
for handler in root_logger.handlers[:]:
if hasattr(handler, 'close'):
handler.close()
root_logger.removeHandler(handler)
# 关闭全局handler
close_handlers()
# 关闭所有其他logger的handler
logger_dict = logging.getLogger().manager.loggerDict
for name, logger_obj in logger_dict.items():
if isinstance(logger_obj, logging.Logger):
for handler in logger_obj.handlers[:]:
if hasattr(handler, 'close'):
handler.close()
logger_obj.removeHandler(handler)
logger.info("日志系统已关闭")

4
template/bot_config_template.toml
View File

@@ -188,7 +188,9 @@ enable_kaomoji_protection = false # 是否启用颜文字保护
date_style = "Y-m-d H:i:s" # 日期格式
log_level_style = "lite" # 日志级别样式,可选FULL，compact，lite
color_text = "title" # 日志文本颜色，可选none，title，full
log_level = "INFO"
log_level = "INFO" # 全局日志级别（向下兼容，优先级低于下面的分别设置）
console_log_level = "INFO" # 控制台日志级别，可选: DEBUG, INFO, WARNING, ERROR, CRITICAL
file_log_level = "DEBUG" # 文件日志级别，可选: DEBUG, INFO, WARNING, ERROR, CRITICAL
# 第三方库日志控制
suppress_libraries = ["faiss","httpx", "urllib3", "asyncio", "websockets", "httpcore", "requests", "peewee", "openai","uvicorn"] # 完全屏蔽的库

129
消息发送API使用说明.md
View File

@@ -1,129 +0,0 @@
# 消息发送API使用说明
## 概述
新的消息发送API允许插件直接向指定的平台和ID发送消息，无需依赖当前聊天上下文。API会自动从数据库中匹配chat_stream并构建相应的发送消息对象。
## 可用方法
### 1. `send_message_to_target()`
最通用的消息发送方法，支持各种类型的消息。
```python
async def send_message_to_target(
self,
message_type: str, # 消息类型：text, image, emoji等
content: str, # 消息内容
platform: str, # 目标平台：qq等
target_id: str, # 目标ID（群ID或用户ID）
is_group: bool = True, # 是否为群聊
display_message: str = "", # 显示消息（可选）
) -> bool:
```
**示例用法：**
```python
# 发送文本消息到群聊
success = await self.send_message_to_target(
message_type="text",
content="Hello, 这是一条测试消息！",
platform="qq",
target_id="123456789",
is_group=True
)
# 发送图片到私聊
success = await self.send_message_to_target(
message_type="image",
content="https://example.com/image.jpg",
platform="qq",
target_id="987654321",
is_group=False
)
# 发送表情包
success = await self.send_message_to_target(
message_type="emoji",
content="😄",
platform="qq",
target_id="123456789",
is_group=True
)
```
### 2. `send_text_to_group()`
便捷方法，专门用于向群聊发送文本消息。
```python
async def send_text_to_group(
self,
text: str, # 文本内容
group_id: str, # 群聊ID
platform: str = "qq" # 平台，默认为qq
) -> bool:
```
**示例用法：**
```python
success = await self.send_text_to_group(
text="群聊测试消息",
group_id="123456789"
)
```
### 3. `send_text_to_user()`
便捷方法，专门用于向用户发送私聊文本消息。
```python
async def send_text_to_user(
self,
text: str, # 文本内容
user_id: str, # 用户ID
platform: str = "qq" # 平台，默认为qq
) -> bool:
```
**示例用法：**
```python
success = await self.send_text_to_user(
text="私聊测试消息",
user_id="987654321"
)
```
## 支持的消息类型
- `"text"` - 文本消息
- `"image"` - 图片消息（需要提供图片URL或路径）
- `"emoji"` - 表情消息
- `"voice"` - 语音消息
- `"video"` - 视频消息
- 其他类型根据平台支持情况
## 注意事项
1. **前提条件**：目标群聊或用户必须已经在数据库中存在对应的chat_stream记录
2. **权限要求**：机器人必须在目标群聊中有发言权限
3. **错误处理**：所有方法都会返回bool值表示发送成功与否，同时会在日志中记录详细错误信息
4. **异步调用**：所有方法都是异步的，需要使用`await`调用
## 完整示例插件
参考 `example_send_message_plugin.py` 文件，该文件展示了如何在插件中使用新的消息发送API。
## 配置文件支持
可以通过TOML配置文件管理目标ID、默认平台等设置。参考 `example_config.toml` 文件。
## 错误排查
如果消息发送失败，请检查：
1. 目标ID是否正确
2. chat_stream是否已加载到ChatManager中
3. 机器人是否有相应权限
4. 网络连接是否正常
5. 查看日志中的详细错误信息