refactor(reaction): 重构关键词反应系统为通用反应规则系统

将原有的 `keyword_reaction` 系统重构为一个更通用、更强大的 `reaction` 系统。新系统统一了关键词和正则表达式规则，并增加了按聊天流ID（`chat_stream_id`）进行规则作用域限定的功能。主要变更包括： - **统一配置模型**：将 `KeywordReactionConfig` 和 `KeywordRuleConfig` 合并重构为 `ReactionConfig` 和 `ReactionRuleConfig`，提供了更清晰、统一的规则定义方式。 - **增加作用域控制**：新的 `ReactionRuleConfig` 增加了 `chat_stream_id` 字段，允许规则被限定在全局范围或特定的聊天会话中。 - **简化逻辑实现**：`DefaultReplyer` 中的实现被重构，以适应新的配置模型，能够筛选并应用适用范围内的规则。 - **更新配置文件模板**：`bot_config_template.toml` 已更新，以反映新的 `[[reaction.rules]]` 配置结构，并提供了详细的注释说明。 - **新增辅助方法**：在 `ChatStream` 中添加了 `get_raw_id()` 方法，用于获取未哈希的原始聊天流ID，以支持新系统的作用域匹配。 BREAKING CHANGE: 关键词反应功能的配置结构已完全改变。旧的 `[keyword_reaction]` 配置不再兼容。用户需要将原有的 `keyword_rules` 和 `regex_rules` 迁移到新的 `[[reaction.rules]]` 格式。
2025-10-26 12:16:37 +08:00
parent 0ed35ff33b
commit ccbac891b6
5 changed files with 105 additions and 72 deletions
--- a/src/chat/message_receive/chat_stream.py
+++ b/src/chat/message_receive/chat_stream.py
@@ -156,6 +156,13 @@ class ChatStream:

        return instance

+    def get_raw_id(self) -> str:
+        """获取原始的、未哈希的聊天流ID字符串"""
+        if self.group_info:
+            return f"{self.platform}:{self.group_info.group_id}:group"
+        else:
+            return f"{self.platform}:{self.user_info.user_id}:private"
+
    def update_active_time(self):
        """更新最后活跃时间"""
        self.last_active_time = time.time()
--- a/src/chat/replyer/default_generator.py
+++ b/src/chat/replyer/default_generator.py
@@ -796,44 +796,63 @@ class DefaultReplyer:
    async def build_keywords_reaction_prompt(self, target: str | None) -> str:
        """构建关键词反应提示

+        该方法根据配置的关键词和正则表达式规则，
+        检查目标消息内容是否触发了任何反应。
+        如果匹配成功，它会生成一个包含所有触发反应的提示字符串，
+        用于指导LLM的回复。
+
        Args:
            target: 目标消息内容

        Returns:
-            str: 关键词反应提示字符串
+            str: 关键词反应提示字符串，如果没有触发任何反应则为空字符串
        """
-        # 关键词检测与反应
-        keywords_reaction_prompt = ""
+        if target is None:
+            return ""
+
+        reaction_prompt = ""
        try:
-            # 添加None检查，防止NoneType错误
-            if target is None:
-                return keywords_reaction_prompt
+            current_chat_stream_id_str = self.chat_stream.get_raw_id()
+            # 2. 筛选适用的规则（全局规则 + 特定于当前聊天的规则）
+            applicable_rules = []
+            for rule in global_config.reaction.rules:
+                if rule.chat_stream_id == "" or rule.chat_stream_id == current_chat_stream_id_str:
+                    applicable_rules.append(rule)  # noqa: PERF401

-            # 处理关键词规则
-            for rule in global_config.keyword_reaction.keyword_rules:
-                if any(keyword in target for keyword in rule.keywords):
-                    logger.info(f"检测到关键词规则：{rule.keywords}，触发反应：{rule.reaction}")
-                    keywords_reaction_prompt += f"{rule.reaction}，"
+            # 3. 遍历适用规则并执行匹配
+            for rule in applicable_rules:
+                matched = False
+                if rule.rule_type == "keyword":
+                    if any(keyword in target for keyword in rule.patterns):
+                        logger.info(f"检测到关键词规则：{rule.patterns}，触发反应：{rule.reaction}")
+                        reaction_prompt += f"{rule.reaction}，"
+                        matched = True
+
+                elif rule.rule_type == "regex":
+                    for pattern_str in rule.patterns:
+                        try:
+                            pattern = re.compile(pattern_str)
+                            if result := pattern.search(target):
+                                reaction = rule.reaction
+                                # 替换命名捕获组
+                                for name, content in result.groupdict().items():
+                                    reaction = reaction.replace(f"[{name}]", content)
+                                logger.info(f"匹配到正则表达式：{pattern_str}，触发反应：{reaction}")
+                                reaction_prompt += f"{reaction}，"
+                                matched = True
+                                break  # 一个正则规则里只要有一个 pattern 匹配成功即可
+                        except re.error as e:
+                            logger.error(f"正则表达式编译错误: {pattern_str}, 错误信息: {e!s}")
+                            continue
+
+                if matched:
+                    # 如果需要每条消息只触发一个反应规则，可以在这里 break
+                    pass

-            # 处理正则表达式规则
-            for rule in global_config.keyword_reaction.regex_rules:
-                for pattern_str in rule.regex:
-                    try:
-                        pattern = re.compile(pattern_str)
-                        if result := pattern.search(target):
-                            reaction = rule.reaction
-                            for name, content in result.groupdict().items():
-                                reaction = reaction.replace(f"[{name}]", content)
-                            logger.info(f"匹配到正则表达式：{pattern_str}，触发反应：{reaction}")
-                            keywords_reaction_prompt += f"{reaction}，"
-                            break
-                    except re.error as e:
-                        logger.error(f"正则表达式编译错误: {pattern_str}, 错误信息: {e!s}")
-                        continue
        except Exception as e:
            logger.error(f"关键词检测与反应时发生异常: {e!s}", exc_info=True)

-        return keywords_reaction_prompt
+        return reaction_prompt

    async def build_notice_block(self, chat_id: str) -> str:
        """构建notice信息块
--- a/src/config/config.py
+++ b/src/config/config.py
@@ -26,7 +26,7 @@ from src.config.official_configs import (
    EmojiConfig,
    ExperimentalConfig,
    ExpressionConfig,
-    KeywordReactionConfig,
+    ReactionConfig,
    LPMMKnowledgeConfig,
    MaimMessageConfig,
    MemoryConfig,
@@ -384,7 +384,7 @@ class Config(ValidatedConfigBase):
    expression: ExpressionConfig = Field(..., description="表达配置")
    memory: MemoryConfig = Field(..., description="记忆配置")
    mood: MoodConfig = Field(..., description="情绪配置")
-    keyword_reaction: KeywordReactionConfig = Field(..., description="关键词反应配置")
+    reaction: ReactionConfig = Field(default_factory=ReactionConfig, description="反应规则配置")
    chinese_typo: ChineseTypoConfig = Field(..., description="中文错别字配置")
    response_post_process: ResponsePostProcessConfig = Field(..., description="响应后处理配置")
    response_splitter: ResponseSplitterConfig = Field(..., description="响应分割配置")
--- a/src/config/official_configs.py
+++ b/src/config/official_configs.py
@@ -401,32 +401,31 @@ class MoodConfig(ValidatedConfigBase):
    mood_update_threshold: float = Field(default=1.0, description="情绪更新阈值")


-class KeywordRuleConfig(ValidatedConfigBase):
-    """关键词规则配置类"""
+class ReactionRuleConfig(ValidatedConfigBase):
+    """反应规则配置类"""

-    keywords: list[str] = Field(default_factory=lambda: [], description="关键词列表")
-    regex: list[str] = Field(default_factory=lambda: [], description="正则表达式列表")
-    reaction: str = Field(default="", description="反应内容")
+    chat_stream_id: str = Field(default="", description='聊天流ID，格式为 "platform:id:type"，空字符串表示全局')
+    rule_type: Literal["keyword", "regex"] = Field(..., description='规则类型，必须是 "keyword" 或 "regex"')
+    patterns: list[str] = Field(..., description="关键词或正则表达式列表")
+    reaction: str = Field(..., description="触发后的回复内容")

    def __post_init__(self):
        import re

-        if not self.keywords and not self.regex:
-            raise ValueError("关键词规则必须至少包含keywords或regex中的一个")
-        if not self.reaction:
-            raise ValueError("关键词规则必须包含reaction")
-        for pattern in self.regex:
-            try:
-                re.compile(pattern)
-            except re.error as e:
-                raise ValueError(f"无效的正则表达式 '{pattern}': {e!s}") from e
+        if not self.patterns:
+            raise ValueError("patterns 列表不能为空")
+        if self.rule_type == "regex":
+            for pattern in self.patterns:
+                try:
+                    re.compile(pattern)
+                except re.error as e:
+                    raise ValueError(f"无效的正则表达式 '{pattern}': {e!s}") from e


-class KeywordReactionConfig(ValidatedConfigBase):
-    """关键词配置类"""
+class ReactionConfig(ValidatedConfigBase):
+    """反应规则系统配置"""

-    keyword_rules: list[KeywordRuleConfig] = Field(default_factory=lambda: [], description="关键词规则列表")
-    regex_rules: list[KeywordRuleConfig] = Field(default_factory=lambda: [], description="正则表达式规则列表")
+    rules: list[ReactionRuleConfig] = Field(default_factory=list, description="反应规则列表")


 class CustomPromptConfig(ValidatedConfigBase):
--- a/template/bot_config_template.toml
+++ b/template/bot_config_template.toml
@@ -279,12 +279,6 @@ max_frequency_bonus = 10.0 # 最大激活频率奖励天数
 # 休眠机制
 dormant_threshold_days = 90 # 休眠状态判定天数（超过此天数未访问的记忆进入休眠状态）

-# 统一存储配置 (已弃用 - 请使用Vector DB配置)
-# DEPRECATED: unified_storage_path = "data/unified_memory"
-# DEPRECATED: unified_storage_cache_limit = 10000
-# DEPRECATED: unified_storage_auto_save_interval = 50
-# DEPRECATED: unified_storage_enable_compression = true
-
 # Vector DB存储配置 (新增 - 替代JSON存储)
 enable_vector_memory_storage = true # 启用Vector DB存储
 enable_llm_instant_memory = true # 启用基于LLM的瞬时记忆
@@ -336,22 +330,36 @@ qa_ppr_damping = 0.8 # PPR阻尼系数
 qa_res_top_k = 3 # 最终提供的文段TopK
 embedding_dimension = 1024 # 嵌入向量维度,应该与模型的输出维度一致

-# keyword_rules 用于设置关键词触发的额外回复知识
-# 添加新规则方法：在 keyword_rules 数组中增加一项，格式如下：
-# { keywords = ["关键词1", "关键词2"], reaction = "触发这些关键词时的回复内容" }
-# 例如，添加一个新规则：当检测到“你好”或“hello”时回复“你好，有什么可以帮你？”
-# { keywords = ["你好", "hello"], reaction = "你好，有什么可以帮你？" }
-[keyword_reaction]
-keyword_rules = [
-    { keywords = ["人机", "bot", "机器", "入机", "robot", "机器人", "ai", "AI"], reaction = "有人说你是人机或者机器人，否定这一事实，攻击或反问他。注意，只有认为你是机器人才回复，否则不要否认" },
-    { keywords = ["测试关键词回复", "test"], reaction = "回答测试成功" },
-    #{ keywords = ["你好", "hello"], reaction = "你好，有什么可以帮你？" }    
-    # 在此处添加更多规则，格式同上
-]
+# --- 反应规则系统 ---
+# 在这里，您可以定义一系列基于关键词或正则表达式的自动回复规则。
+# 每条规则都是一个独立的 [[reaction.rules]] 块。

-regex_rules = [
-    { regex = ["^(?P<n>\\S{1,20})是这样的$"], reaction = "请按照以下模板造句：[n]是这样的，xx只要xx就可以，可是[n]要考虑的事情就很多了，比如什么时候xx，什么时候xx，什么时候xx。（请自由发挥替换xx部分，只需保持句式结构，同时表达一种将[n]过度重视的反讽意味）" }
-]
+# chat_stream_id: 聊天流ID (格式 "platform:id:type")。
+#                 用于指定此规则仅在哪个聊天中生效。
+#                 如果留空 ("")，则为全局规则，对所有聊天生效。
+# rule_type:      规则类型，必须是 "keyword" 或 "regex"。
+#                 "keyword": 表示本条规则使用关键词匹配。
+#                 "regex":   表示本条规则使用正则表达式匹配。
+# patterns:       一个字符串列表，根据 rule_type 的不同，这里填写关键词或正则表达式。
+# reaction:       触发规则后，机器人发送的回复内容。
+
+[[reaction.rules]]
+chat_stream_id = ""
+rule_type = "keyword"
+patterns = ["人机", "bot", "机器", "入机", "robot", "机器人", "ai", "AI"]
+reaction = "有人说你是人机或者机器人，否定这一事实，攻击或反问他。注意，只有认为你是机器人才回复，否则不要否认"
+
+[[reaction.rules]]
+chat_stream_id = ""
+rule_type = "keyword"
+patterns = ["测试关键词回复", "test"]
+reaction = "回答测试成功"
+
+[[reaction.rules]]
+chat_stream_id = ""
+rule_type = "regex"
+patterns = ["^(?P<n>\\S{1,20})是这样的$"]
+reaction = "请按照以下模板造句：[n]是这样的，xx只要xx就可以，可是[n]要考虑的事情就很多了，比如什么时候xx，什么时候xx，什么时候xx。（请自由发挥替换xx部分，只需保持句式结构，同时表达一种将[n]过度重视的反讽意味）"

 # 可以自定义部分提示词
 [custom_prompt]