shen/Bubbles
1
0
Fork 0
mirror of https://github.com/Zippland/Bubbles.git synced 2026-01-19 01:21:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

删掉 command 实现

Browse Source
This commit is contained in:
zihanjian
2025-09-25 14:17:54 +08:00
parent 173dca3f37
commit e92473d9f4
8 changed files with 81 additions and 482 deletions
Download Patch File Download Diff File Expand all files Collapse all files

191
README.MD
View File

@@ -33,31 +33,16 @@
Bubbles 是一个功能丰富的微信机器人框架，基于 [wcferry](https://github.com/lich0821/wcferry) 和 [WechatRobot](https://github.com/lich0821/wechatrobot) 开发，支持接入多种LLM，提供丰富的交互功能和定时任务。该项目旨在将微信客户端转变为一个智能的个人助手，可以执行多种实用功能，带来便捷的用户体验。
和一般机器人框架不同的是，Bubbles 设计了两套灵活的路由系统：
和一般机器人框架不同的是，Bubbles 构建了一套基于 AI 的智能路由系统：AI 会读取所有已注册的功能描述，自动判断用户想要执行的操作并调用对应的处理函数。添加新功能时，只需要编写处理逻辑并通过装饰器注册到 AI 路由，无需在消息处理流程里额外写分支逻辑。
1. **命令路由系统** - 基于正则表达式的精确命令匹配，适合有明确触发词的功能
2. **AI智能路由系统** - 基于AI的自然语言理解，自动识别用户意图并调用相应功能
通过这两套路由系统，同一个功能函数，可以让 ai 有两种方式进行调用。这不但使得添加新功能变得极其简单，且不需要改动原有代码。相当于给一个主线 Hub 添加插件，让海量的、不同种类的工具都能集成到 AI 里。
路由系统是本项目的核心，通过它，理论上可以实现任何操作。
已实现的操作详见 **如何添加新功能** 章节。
这套路由机制是项目的核心，使得机器人可以在保持主线稳定的同时，快速接入数量众多、类型各异的工具。已实现的功能示例见 **如何添加新功能** 章节。
## 案例演示
#### 案例演示其一：使用自然语言设置提醒（命令路由）
#### 案例演示：使用自然语言设置提醒（AI 智能路由）
结构：
用户输入 -> 击中命令 -> 调用命令函数 -> agent分析 -> agent格式化输出 -> 选择函数 -> 格式解析 -> 函数循环调用 -> 数据库持久化 -> 结果回调
<img src="img_1.jpg" width="400"/>
#### 案例演示其二：使用自然语言设置提醒（AI 智能路由）
结构：
用户输入 -> 未击中命令 -> agent路由选择 -> 满足功能要求 -> agent格式化输出 -> 格式解析 -> 函数调用 -> 接口访问 -> 查询数据 -> 数据库持久化 -> 结果回调
用户输入 -> AI 路由决策 -> 满足功能要求 -> agent格式化输出 -> 格式解析 -> 函数调用 -> 接口访问 -> 查询数据 -> 数据库持久化 -> 结果回调
<img src="img_3.png" width="400"/>
@@ -69,47 +54,37 @@ Bubbles 是一个功能丰富的微信机器人框架，基于 [wcferry](https:/
- DeepSeek
- Gemini
#### 🛠️ 双重路由系统
- **命令路由系统**：基于正则表达式的精确匹配，高效处理特定命令
- **AI智能路由**：自然语言理解，无需记住特定命令格式
- 支持自定义命令及参数
- 预设 [多种实用和娱乐命令](#可用命令)
#### 🛠️ 智能路由系统
- 基于 AI 的意图识别，无需记住特定命令格式
- 支持为功能提供描述、示例和参数说明，帮助模型精准调用
- 通过装饰器注册新功能即可扩展机器人能力
#### 路由系统架构图
```mermaid
flowchart TD
A[User Message] --> B[Message Preprocessing]
B --> C{At Bot or Private Chat?}
C -->|Yes| D[Command Router]
C -->|No| E[Ignore Message]
D --> F{Regex Match?}
F -->|Yes| G[Execute Command Handler]
F -->|No| H[AI Router]
H --> I[AI Analyze Intent]
I --> J{Function Match?}
J -->|Yes| K[Call Function]
J -->|No| L[Chat Mode]
G --> M[Return Result]
K --> M
L --> N[AI Conversation]
N --> M
B --> C{私聊或@机器人?}
C -->|No| D[忽略消息]
C -->|Yes| E[AI Router]
E --> F[AI Analyze Intent]
F --> G{Function Decision?}
G -->|Function| H[Call Registered Handler]
G -->|Chat| I[AI Conversation]
H --> J[Return Result]
I --> J
style A fill:#f9f,stroke:#333,stroke-width:2px
style D fill:#bbf,stroke:#333,stroke-width:2px
style H fill:#bfb,stroke:#333,stroke-width:2px
style M fill:#fbb,stroke:#333,stroke-width:2px
style E fill:#bfb,stroke:#333,stroke-width:2px
style H fill:#bbf,stroke:#333,stroke-width:2px
style J fill:#fbb,stroke:#333,stroke-width:2px
```
消息处理流程说明：
1. **消息预处理**：系统接收用户消息，判断是否需要响应
2. **命令路由优先**：首先尝试使用正则表达式匹配已注册的命令
3. **AI路由兜底**：如果没有匹配到命令，则使用AI分析用户意图
4. **智能分发**：AI可以理解自然语言并调用相应功能，或进入聊天模式
1. **消息预处理**：系统接收用户消息，提取纯文本、发送者等上下文
2. **权限判断**：仅在私聊或群聊中被 @ 时才进入 AI 路由，避免打扰其他群成员
3. **AI 决策**：大模型根据功能描述判断是否调用某个功能或直接聊天
4. **结果输出**：调用成功的功能会返回处理结果，否则使用默认闲聊响应
### ⏰ 定时任务与提醒功能
- 每日天气预报推送
@@ -225,30 +200,23 @@ GROUP_MODELS:
python main.py
```
#### 可用命令（命令路由系统）
#### 可用功能示例（AI 路由）
机器人支持多种命令，按功能分类如下：
以下是已经注册到 AI 路由的典型意图，机器人会尽力理解类似的自然语言表述：
##### 提醒功能
- `..提醒我..` - 用自然语言设置一个或多个提醒
- `查看提醒`、`我的提醒`、`提醒列表` - 查看您设置的所有提醒
- `..删..提醒..` - 用自然语言删除指定的（或所有）提醒
- “提醒我明天下午三点开会” —— 设置一次性提醒
- “每天早上 8 点提醒我跑步” —— 支持循环提醒
- “查看我的提醒” / “我有哪些提醒” —— 查看当前所有提醒
- “把明天的开会提醒删掉” —— 删除指定提醒
##### 基础系统命令
- `info`、`帮助`、`指令` - 显示机器人的帮助信息
- `骂一下 @用户名` - 让机器人骂指定用户（仅群聊）
##### 搜索与资讯
- “帮我查查最近的 AI 新闻” —— 使用 Perplexity 进行深度检索
- “看看今天的新闻” —— 推送最新要闻
##### Perplexity AI 命令
- `ask 问题内容` - 使用 Perplexity AI 进行深度查询（需@机器人）
##### 消息管理命令
- `summary`、`/总结` - 总结群聊最近的消息（仅群聊）
- `clearmessages`、`/清除历史` - 从数据库中清除群聊的历史消息记录（仅群聊）
##### 天气和新闻工具
- `天气预报 城市名`、`预报 城市名` - 查询指定城市未来几天的天气预报
- `天气 城市名`、`温度 城市名` - 查询指定城市的当前天气
- `新闻` - 获取最新新闻
##### 天气工具
- “北京天气怎么样”
- “查一下上海未来几天的天气”
## 📋 项目结构
@@ -257,9 +225,8 @@ Bubbles-WechatAI/
├── ai_providers/ # AI 模块
│ ├── ai_name.py # AI 模型接口实现
│ └── ...
├── commands/ # 命令系统
│ ├── registry.py # 正则命令注册
│ ├── handlers.py # 命令处理函数
├── commands/ # 消息上下文与 AI 路由
│ ├── handlers.py # 功能处理函数
│ ├── ai_router.py # AI智能路由器
│ ├── ai_functions.py # AI路由功能注册
│ └── ...
@@ -274,74 +241,18 @@ Bubbles-WechatAI/
### ✨ 如何添加新功能
本项目提供两种方式添加新功能：
目前所有新功能均通过 AI 智能路由接入，开发流程如下：
```mermaid
graph LR
A[新功能] --> B{选择路由方式}
B --> C[命令路由]
B --> D[AI路由]
C --> E[精确匹配]
C --> F[固定格式命令]
C --> G[例如:天气北京]
D --> H[自然语言理解]
D --> I[灵活表达]
D --> J[例如:北京天气怎么样]
style A fill:#f9f,stroke:#333,stroke-width:2px
style C fill:#bbf,stroke:#333,stroke-width:2px
style D fill:#bfb,stroke:#333,stroke-width:2px
```
1. **实现功能逻辑**
* 在 `function/` 目录下创建或复用功能模块，封装核心业务逻辑，便于测试和复用。
#### 方式一：使用命令路由系统（适合有明确触发词的功能）
1. **定义功能逻辑 (可选但推荐)**:
* 如果你的功能逻辑比较复杂，建议在 `function/` 目录下创建一个新的 Python 文件 (例如 `func_your_feature.py`)。
* 在这个文件中实现你的核心功能代码，例如定义类或函数。这有助于保持代码结构清晰。
2. **创建命令处理器**:
* 打开 `commands/handlers.py` 文件。
* 添加一个新的处理函数，例如 `handle_your_feature(ctx: 'MessageContext', match: Optional[Match]) -> bool:`。
* 这个函数接收 `MessageContext` (包含消息上下文信息) 和 `match` (正则表达式匹配结果) 作为参数。
* 在函数内部，你可以：
* 调用你在 `function/` 目录下创建的功能模块。
* 使用 `ctx.send_text()` 发送回复消息。
* 根据需要处理 `match` 对象提取用户输入的参数。
* 函数应返回 `True` 表示命令已被处理，`False` 则表示未处理 (会继续尝试匹配后续命令或进行闲聊)。
* 确保从 `function` 目录导入必要的模块。
3. **注册命令**:
* 打开 `commands/registry.py` 文件。
* 在 `COMMANDS` 列表中，按照优先级顺序添加一个新的 `Command` 对象。
* 配置 `Command` 参数：
* `name`: 命令的唯一标识名 (小写下划线)。
* `pattern`: 用于匹配用户输入的正则表达式 (`re.compile`)。注意捕获用户参数。
* `scope`: 命令适用范围 (`"group"`, `"private"`, `"both"`)。
* `need_at`: 在群聊中是否需要 `@` 机器人才能触发 (`True`/`False`)。
* `priority`: 命令的优先级 (数字越小越优先匹配)。
* `handler`: 指向你在 `handlers.py` 中创建的处理函数 (例如 `handle_your_feature`)。
* `description`: 命令的简短描述，用于帮助信息。
* 确保从 `handlers.py` 导入你的新处理函数。
4. **更新帮助信息 (可选)**:
* 如果希望用户能在 `帮助` 命令中看到你的新功能，可以更新 `commands/handlers.py` 中的 `handle_help` 函数，将新命令的用法添加到帮助文本中。
#### 方式二：使用AI智能路由（适合自然语言交互的功能）
AI路由系统让用户可以用自然语言触发功能，无需记住特定命令格式：
1. **实现功能逻辑**:
* 在 `function/` 目录下创建功能模块（如已有则跳过）
2. **注册AI路由功能**:
2. **注册 AI 路由功能**
* 打开 `commands/ai_functions.py`
* 使用装饰器注册你的功能：
* 使用装饰器注册功能，提供给模型判定意图所需的描述信息：
```python
@ai_router.register(
name="your_function_name",
description="功能描述（AI会根据这个判断用户意图）",
description="功能描述（AI 会根据这个判断用户意图）",
examples=[
"示例用法1",
"示例用法2",
@@ -350,23 +261,23 @@ AI路由系统让用户可以用自然语言触发功能，无需记住特定命
params_description="参数说明"
)
def ai_handle_your_function(ctx: MessageContext, params: str) -> bool:
# params 是AI从用户输入中提取的参数
# params 是 AI 从用户输入中提取的参数
# 调用你的功能逻辑
# 使用 ctx.send_text() 发送回复
return True
```
3. **工作原理**:
* 用户发送消息时，如果正则路由未匹配，AI会分析用户意图
* AI根据功能描述和示例，判断应该调用哪个功能
* AI会自动提取参数并传递给功能处理函数
3. **理解路由机制**
* 用户消息会在私聊或被 @ 时进入 AI 路由
* AI 根据 `description`、`examples` 和 `params_description` 选择最合适的功能
* `params` 字段由模型提取，可根据需要进一步解析或校验
例如，注册了天气查询功能后，用户可以说：
- "北京天气怎么样"
- "查一下上海的天气"
- "明天深圳会下雨吗"
AI都能理解并调用天气查询功能。
AI 都能理解并调用天气查询功能。
完成以上步骤后，重启机器人即可测试你的新功能！

13
commands/__init__.py
View File

@@ -1,11 +1,10 @@
# commands package
"""
命令路由系统包
消息处理组件包
此包包含了命令路由系统的所有组件:
当前模块提供基于 AI 的功能派发能力:
- context: 消息上下文类
- models: 命令数据模型
- router: 命令路由器
- registry: 命令注册表
- handlers: 命令处理函数
"""
- handlers: 功能处理函数
- ai_router: AI 智能路由核心
- ai_functions: 面向 AI 路由的功能注册
"""

23
commands/handlers.py
View File

@@ -755,20 +755,11 @@ def handle_list_reminders(ctx: 'MessageContext', match: Optional[Match]) -> bool
return True
def handle_delete_reminder(ctx: 'MessageContext', match: Optional[Match]) -> bool:
"""
处理删除提醒命令（支持群聊和私聊）。
检查消息是否包含"提醒"和"删"相关字眼，然后使用 AI 理解具体意图。
"""
"""处理删除提醒命令（支持群聊和私聊），通过 AI 理解用户意图并执行操作。"""
# 1. 获取用户输入的完整内容
raw_text = ctx.msg.content.strip()
# 2. 检查是否包含删除提醒的两个核心要素："提醒"和"删/删除/取消"
# Regex 已经保证了后者，这里只需检查前者
if "提醒" not in raw_text:
# 如果消息匹配了 "删" 但没有 "提醒"，说明不是删除提醒的意图，不处理
return False # 返回 False，让命令路由器可以尝试匹配其他命令
# 3. 检查 ReminderManager 是否存在
# 2. 检查 ReminderManager 是否存在
if not hasattr(ctx.robot, 'reminder_manager'):
# 这个检查需要保留，是内部依赖
ctx.send_text("❌ 内部错误：提醒管理器未初始化。", ctx.msg.sender if ctx.is_group else "")
@@ -779,7 +770,7 @@ def handle_delete_reminder(ctx: 'MessageContext', match: Optional[Match]) -> boo
# --- 核心流程：直接使用 AI 分析 ---
# 4. 获取用户的所有提醒作为 AI 的上下文
# 3. 获取用户的所有提醒作为 AI 的上下文
reminders = ctx.robot.reminder_manager.list_reminders(ctx.msg.sender)
if not reminders:
# 如果用户没有任何提醒，直接告知
@@ -794,7 +785,7 @@ def handle_delete_reminder(ctx: 'MessageContext', match: Optional[Match]) -> boo
if ctx.logger: ctx.logger.error(f"序列化提醒列表失败: {e}", exc_info=True)
return True
# 5. 构造 AI Prompt (与之前相同，AI 需要能处理所有情况)
# 4. 构造 AI Prompt (与之前相同，AI 需要能处理所有情况)
# 注意：确保 prompt 中的 {{ 和 }} 转义正确
sys_prompt = """
你是提醒删除助手。用户会提出删除提醒的请求。我会提供用户的**完整请求原文**，以及一个包含该用户所有当前提醒的 JSON 列表。
@@ -870,7 +861,7 @@ def handle_delete_reminder(ctx: 'MessageContext', match: Optional[Match]) -> boo
return True
# 6. 调用 AI (使用完整的用户原始输入)
# 5. 调用 AI (使用完整的用户原始输入)
q_for_ai = f"请根据以下用户完整请求，分析需要删除哪个提醒：\n{raw_text}" # 使用 raw_text
try:
if not hasattr(ctx, 'chat') or not ctx.chat:
@@ -903,7 +894,7 @@ def handle_delete_reminder(ctx: 'MessageContext', match: Optional[Match]) -> boo
# 获取AI回答
ai_response = ctx.chat.get_answer(q_for_ai, ctx.get_receiver(), system_prompt_override=formatted_prompt)
# 7. 解析 AI 的 JSON 回复
# 6. 解析 AI 的 JSON 回复
json_str = None
json_match_obj = re.search(r'\{.*\}', ai_response, re.DOTALL)
if json_match_obj:
@@ -932,7 +923,7 @@ def handle_delete_reminder(ctx: 'MessageContext', match: Optional[Match]) -> boo
return True
# 否则继续下一次循环重试
# 8. 根据 AI 指令执行操作 (与之前相同)
# 7. 根据 AI 指令执行操作 (与之前相同)
action = parsed_ai_response.get("action")
if action == "delete_specific":

38
commands/models.py
View File

@@ -1,38 +0,0 @@
import re
from dataclasses import dataclass
from typing import Pattern, Callable, Literal, Optional, Any, Union, Match
# 导入 MessageContext，使用前向引用避免循环导入
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .context import MessageContext
@dataclass
class Command:
"""
命令定义类，封装命令的匹配条件和处理函数
"""
name: str # 命令名称，用于日志和调试
pattern: Union[Pattern, Callable[['MessageContext'], Optional[Match]]] # 匹配规则：正则表达式或自定义匹配函数
scope: Literal["group", "private", "both"] # 生效范围: "group"-仅群聊, "private"-仅私聊, "both"-两者都可
handler: Callable[['MessageContext', Optional[Match]], bool] # 处理函数
need_at: bool = False # 在群聊中是否必须@机器人才能触发
priority: int = 100 # 优先级，数字越小越先匹配
description: str = "" # 命令的描述，用于生成帮助信息
def __post_init__(self):
"""验证命令配置的有效性"""
if self.scope not in ["group", "private", "both"]:
raise ValueError(f"无效的作用域: {self.scope}，必须是 'group', 'private' 或 'both'")
# 检查pattern是否为正则表达式或可调用对象
if not isinstance(self.pattern, (Pattern, Callable)):
# 如果是字符串，尝试转换为正则表达式
if isinstance(self.pattern, str):
try:
self.pattern = re.compile(self.pattern)
except re.error:
raise ValueError(f"无效的正则表达式: {self.pattern}")
else:
raise TypeError(f"pattern 必须是正则表达式或可调用对象，而不是 {type(self.pattern)}")

134
commands/registry.py
View File

@@ -1,134 +0,0 @@
import re
from .models import Command
from .handlers import (
handle_help,
handle_summary, handle_clear_messages, handle_news_request,
handle_chitchat, handle_insult,
handle_perplexity_ask, handle_reminder, handle_list_reminders, handle_delete_reminder,
handle_weather_forecast
)
# 命令列表，按优先级排序
# 优先级越小越先匹配
COMMANDS = [
# ======== 基础系统命令 ========
Command(
name="help",
pattern=re.compile(r"^(info|帮助|指令)$", re.IGNORECASE),
scope="both", # 群聊和私聊都支持
need_at=False, # 不需要@机器人
priority=10, # 优先级较高
handler=handle_help,
description="显示机器人的帮助信息"
),
# ======== Perplexity AI 命令 ========
Command(
name="perplexity_ask",
pattern=re.compile(r"^ask\s*(.+)", re.IGNORECASE | re.DOTALL),
scope="both", # 群聊和私聊都支持
need_at=True, # 需要@机器人
priority=25, # 较高优先级，确保在闲聊之前处理
handler=handle_perplexity_ask,
description="使用 Perplexity AI 进行深度查询"
),
# ======== 消息管理命令 ========
Command(
name="summary",
pattern=re.compile(r"^(summary|总结)$", re.IGNORECASE),
scope="group", # 仅群聊支持
need_at=True, # 需要@机器人
priority=30, # 优先级一般
handler=handle_summary,
description="总结群聊最近的消息"
),
Command(
name="clear_messages",
pattern=re.compile(r"^(clearmessages|清除历史)$", re.IGNORECASE),
scope="group", # 仅群聊支持
need_at=True, # 需要@机器人
priority=31, # 优先级一般
handler=handle_clear_messages,
description="从数据库中清除群聊的历史消息记录"
),
# ======== 提醒功能 ========
Command(
name="reminder",
pattern=re.compile(r"提醒我", re.IGNORECASE),
scope="both", # 支持群聊和私聊
need_at=True, # 在群聊中需要@机器人
priority=35, # 优先级适中，在基础命令后，复杂功能或闲聊前
handler=handle_reminder,
description="设置一个提醒 (包含 '提醒我' 关键字即可, 例如：提醒我明天下午3点开会)"
),
Command(
name="list_reminders",
pattern=re.compile(r"^(查看提醒|我的提醒|提醒列表)$", re.IGNORECASE),
scope="both", # 支持群聊和私聊
need_at=True, # 在群聊中需要@机器人
priority=36, # 优先级略低于设置提醒
handler=handle_list_reminders,
description="查看您设置的所有提醒"
),
Command(
name="delete_reminder",
# 修改为只匹配包含"删"、"删除"或"取消"的消息，不再要求特定格式
pattern=re.compile(r"(?:删|删除|取消)", re.IGNORECASE),
scope="both", # 支持群聊和私聊
need_at=True, # 在群聊中需要@机器人
priority=37,
handler=handle_delete_reminder,
description="删除提醒 (包含'删'和'提醒'关键字即可，如: 把开会的提醒删了)"
),
# ======== 新闻和实用工具 ========
Command(
name="weather_forecast",
pattern=re.compile(r"^(?:天气预报|天气)\s+(.+)$"), # 匹配 天气预报/预报 城市名
scope="both", # 群聊和私聊都支持
need_at=True, # 需要@机器人
priority=38, # 优先级比天气高一点
handler=handle_weather_forecast,
description="查询指定城市未来几天的天气预报 (例如：天气预报 北京)"
),
Command(
name="news",
pattern=re.compile(r"^新闻$"),
scope="both", # 群聊和私聊都支持
need_at=True, # 需要@机器人
priority=40, # 优先级一般
handler=handle_news_request,
description="获取最新新闻"
),
# ======== 骂人命令 ========
Command(
name="insult",
pattern=re.compile(r"骂一下\s*@([^\s@]+)"),
scope="group", # 仅群聊支持
need_at=True, # 需要@机器人
priority=100, # 优先级较高
handler=handle_insult,
description="骂指定用户"
),
]
# 可以添加一个函数，获取命令列表的简单描述
def get_commands_info():
"""获取所有命令的简要信息，用于调试"""
info = []
for i, cmd in enumerate(COMMANDS):
scope_str = {"group": "仅群聊", "private": "仅私聊", "both": "群聊私聊"}[cmd.scope]
at_str = "需要@" if cmd.need_at else "不需@"
info.append(f"{i+1}. [{cmd.priority}] {cmd.name} ({scope_str},{at_str}) - {cmd.description or '无描述'}")
return "\n".join(info)
# 导出所有命令
__all__ = ["COMMANDS", "get_commands_info"]

117
commands/router.py
View File

@@ -1,117 +0,0 @@
import re
import logging
from typing import List, Optional, Any, Dict, Match
import traceback
from .models import Command
from .context import MessageContext
# 获取模块级 logger
logger = logging.getLogger(__name__)
class CommandRouter:
"""
命令路由器，负责将消息路由到对应的命令处理函数
"""
def __init__(self, commands: List[Command], robot_instance: Optional[Any] = None):
# 按优先级排序命令列表，数字越小优先级越高
self.commands = sorted(commands, key=lambda cmd: cmd.priority)
self.robot_instance = robot_instance
# 分析并输出命令注册信息，便于调试
scope_count = {"group": 0, "private": 0, "both": 0}
for cmd in commands:
scope_count[cmd.scope] += 1
logger.info(f"命令路由器初始化成功，共加载 {len(commands)} 个命令")
logger.info(f"命令作用域分布: 仅群聊 {scope_count['group']}，仅私聊 {scope_count['private']}，两者均可 {scope_count['both']}")
# 按优先级输出命令信息
for i, cmd in enumerate(self.commands[:10]): # 只输出前10个
logger.info(f"{i+1}. [{cmd.priority}] {cmd.name} - {cmd.description or '无描述'}")
if len(self.commands) > 10:
logger.info(f"... 共 {len(self.commands)} 个命令")
def dispatch(self, ctx: MessageContext) -> bool:
"""
根据消息上下文分发命令
:param ctx: 消息上下文对象
:return: 是否有命令成功处理
"""
# 确保context可以访问到robot实例
if self.robot_instance and not ctx.robot:
ctx.robot = self.robot_instance
# 如果robot有logger属性且ctx没有logger，则使用robot的logger
if hasattr(self.robot_instance, 'LOG') and not ctx.logger:
ctx.logger = self.robot_instance.LOG
# 记录日志，便于调试
if ctx.logger:
ctx.logger.debug(f"开始路由消息: '{ctx.text}', 来自: {ctx.sender_name}, 群聊: {ctx.is_group}, @机器人: {ctx.is_at_bot}")
# 遍历命令列表，按优先级顺序匹配
for cmd in self.commands:
# 1. 检查作用域 (scope)
if cmd.scope != "both":
if (cmd.scope == "group" and not ctx.is_group) or \
(cmd.scope == "private" and ctx.is_group):
continue # 作用域不匹配，跳过
# 2. 检查是否需要 @ (need_at) - 仅在群聊中有效
if ctx.is_group and cmd.need_at and not ctx.is_at_bot:
continue # 需要@机器人但未被@，跳过
# 3. 执行匹配逻辑
match_result = None
try:
# 根据pattern类型执行匹配
if callable(cmd.pattern):
# 自定义匹配函数
match_result = cmd.pattern(ctx)
else:
# 正则表达式匹配
match_obj = cmd.pattern.search(ctx.text)
match_result = match_obj
# 匹配失败，尝试下一个命令
if match_result is None:
continue
# 匹配成功，记录日志
if ctx.logger:
ctx.logger.info(f"命令 '{cmd.name}' 匹配成功，准备处理")
# 4. 执行命令处理函数
try:
result = cmd.handler(ctx, match_result)
if result:
if ctx.logger:
ctx.logger.info(f"命令 '{cmd.name}' 处理成功")
return True
else:
if ctx.logger:
ctx.logger.warning(f"命令 '{cmd.name}' 处理返回False，尝试下一个命令")
except Exception as e:
if ctx.logger:
ctx.logger.error(f"执行命令 '{cmd.name}' 处理函数时出错: {e}")
ctx.logger.error(traceback.format_exc())
else:
logger.error(f"执行命令 '{cmd.name}' 处理函数时出错: {e}", exc_info=True)
# 出错后继续尝试下一个命令
except Exception as e:
# 匹配过程出错，记录并继续
if ctx.logger:
ctx.logger.error(f"匹配命令 '{cmd.name}' 时出错: {e}")
else:
logger.error(f"匹配命令 '{cmd.name}' 时出错: {e}", exc_info=True)
continue
# 所有命令都未匹配或处理失败
if ctx.logger:
ctx.logger.debug("所有命令匹配失败或处理失败")
return False
def get_command_descriptions(self) -> Dict[str, str]:
"""获取所有命令的描述，用于生成帮助信息"""
return {cmd.name: cmd.description for cmd in self.commands if cmd.description}

1
function/func_summary.py
View File

@@ -8,7 +8,6 @@ from collections import deque
import sqlite3 # 添加sqlite3模块
import os # 用于处理文件路径
from function.func_xml_process import XmlProcessor # 导入XmlProcessor
# from commands.registry import COMMANDS # 不再需要导入命令列表
class MessageSummary:
"""消息总结功能类 (使用SQLite持久化)

46
robot.py
View File

@@ -27,10 +27,8 @@ from constants import ChatType
from job_mgmt import Job
from function.func_xml_process import XmlProcessor
# 导入命令路由系统
# 导入上下文及常用处理函数
from commands.context import MessageContext
from commands.router import CommandRouter
from commands.registry import COMMANDS, get_commands_info
from commands.handlers import handle_chitchat # 导入闲聊处理函数
# 导入AI路由系统
@@ -163,10 +161,6 @@ class Robot(Job):
# 初始化图像生成管理器
self.image_manager = ImageGenerationManager(self.config, self.wcf, self.LOG, self.sendTextMsg)
# 初始化命令路由器
self.command_router = CommandRouter(COMMANDS, robot_instance=self)
self.LOG.info(f"命令路由系统初始化完成，共加载 {len(COMMANDS)} 条命令")
# 初始化AI路由器
self.LOG.info(f"AI路由系统初始化完成，共加载 {len(ai_router.functions)} 个AI功能")
@@ -179,9 +173,6 @@ class Robot(Job):
except Exception as e:
self.LOG.error(f"初始化提醒管理器失败: {e}", exc_info=True)
# 输出命令列表信息，便于调试
# self.LOG.debug(get_commands_info()) # 如果需要在日志中输出所有命令信息，取消本行注释
@staticmethod
def value_check(args: dict) -> bool:
if args:
@@ -210,24 +201,21 @@ class Robot(Job):
setattr(ctx, 'chat', self.chat)
setattr(ctx, 'specific_max_history', specific_limit)
# 5. 使用命令路由器分发处理消息
handled = self.command_router.dispatch(ctx)
# 6. 如果正则路由器没有处理，尝试AI路由器
if not handled:
# 只在被@或私聊时才使用AI路由
if (msg.from_group() and msg.is_at(self.wxid)) or not msg.from_group():
print(f"[AI路由调试] 准备调用AI路由器处理消息: {msg.content}")
ai_handled = ai_router.dispatch(ctx)
print(f"[AI路由调试] AI路由器处理结果: {ai_handled}")
if ai_handled:
self.LOG.info("消息已由AI路由器处理")
print("[AI路由调试] 消息已成功由AI路由器处理")
return
else:
print("[AI路由调试] AI路由器未处理该消息")
# 7. 如果没有命令处理器处理，则进行特殊逻辑处理
handled = False
# 5. 优先尝试使用AI路由器处理消息（仅限私聊或@机器人）
if (msg.from_group() and msg.is_at(self.wxid)) or not msg.from_group():
print(f"[AI路由调试] 准备调用AI路由器处理消息: {msg.content}")
handled = ai_router.dispatch(ctx)
print(f"[AI路由调试] AI路由器处理结果: {handled}")
if handled:
self.LOG.info("消息已由AI路由器处理")
print("[AI路由调试] 消息已成功由AI路由器处理")
return
else:
print("[AI路由调试] AI路由器未处理该消息")
# 6. 如果AI路由器未处理，则进行特殊逻辑处理
if not handled:
# 7.1 好友请求自动处理
if msg.type == 37: # 好友请求
@@ -254,7 +242,7 @@ class Robot(Job):
# 7.3 群聊消息，且配置了响应该群
if msg.from_group() and msg.roomid in self.config.GROUPS:
# 如果在群里被@了，但命令路由器没有处理，则进行闲聊
# 如果在群里被@了，但AI路由器未处理，则进行闲聊
if msg.is_at(self.wxid):
# 调用handle_chitchat函数处理闲聊，传递完整的上下文
handle_chitchat(ctx, None)