from __future__ import annotations import re from dataclasses import dataclass, field from typing import Dict, List QUESTION_PATTERNS = [ r"\?$", r"?$", r"怎么", r"如何", r"咋弄", r"为啥", r"为什么", r"有人知道", r"谁知道", r"能不能", r"可以吗", r"报错", r"怎么解决", ] SOCIAL_PATTERNS = [r"小牛", r"在吗", r"出来", r"帮忙看", r"看看"] LIGHT_SOCIAL_PATTERNS = [ r"哈哈", r"笑死", r"绷不住", r"离谱", r"逆天", r"牛逼", r"卧槽", r"我去", r"确实", r"也是", r"6$", r"可以啊", r"有点东西", r"抽象", ] @dataclass class TriggerResult: trigger_type: str = "none" priority: float = 0.0 is_question: bool = False is_directed: bool = False is_followup: bool = False is_social_call: bool = False is_returning_member: bool = False should_respond: bool = False topic: str = "" reasons: List[str] = field(default_factory=list) class TriggerRouter: def __init__(self, config: Dict): self.config = config or {} self.topic_keywords = [str(item).lower() for item in self.config.get("focus", [])] def route(self, message: Dict, memory_hints: Dict, conversation_hints: Dict | None = None) -> TriggerResult: conversation_hints = conversation_hints or {} content = str(message.get("content", "")).strip() content_lower = content.lower() result = TriggerResult() if message.get("is_at"): result.trigger_type = "at_trigger" result.priority = float(self.config.get("at_bot", 1.0)) result.should_respond = True result.is_directed = True result.reasons.append("is_at") if self._is_question(content): if result.priority < float(self.config.get("explicit_question", 0.95)): result.trigger_type = "question_trigger" result.priority = float(self.config.get("explicit_question", 0.95)) result.is_question = True result.reasons.append("question") if memory_hints.get("is_followup"): if result.priority < float(self.config.get("followup", 0.90)): result.trigger_type = "followup_trigger" result.priority = float(self.config.get("followup", 0.90)) result.is_followup = True result.should_respond = True result.is_directed = True result.reasons.append("followup") topic = self._detect_topic(content_lower) if topic: result.topic = topic result.reasons.append("topic") if result.priority < float(self.config.get("casual_topic", 0.35)): result.trigger_type = result.trigger_type if result.trigger_type != "none" else "topic_trigger" result.priority = max(result.priority, float(self.config.get("casual_topic", 0.35))) if self._is_social_call(content_lower): if result.priority < float(self.config.get("social_call", 0.65)): result.trigger_type = result.trigger_type if result.trigger_type != "none" else "social_trigger" result.priority = max(result.priority, float(self.config.get("social_call", 0.65))) result.is_social_call = True result.is_directed = True result.should_respond = True result.reasons.append("social_call") elif self._is_light_social_moment(content_lower): if result.priority < float(self.config.get("light_social", 0.45)): result.trigger_type = result.trigger_type if result.trigger_type != "none" else "light_social_trigger" result.priority = max(result.priority, float(self.config.get("light_social", 0.45))) result.is_social_call = True result.reasons.append("light_social") if conversation_hints.get("quote_targets_bot"): result.is_directed = True result.should_respond = True result.reasons.append("quote_targets_bot") if result.trigger_type == "none": result.trigger_type = "quote_followup_trigger" result.priority = max(result.priority, float(self.config.get("followup", 0.90))) if conversation_hints.get("previous_same_sender_directed") and result.is_question: result.is_directed = True result.should_respond = True result.reasons.append("same_sender_directed") if result.is_question and result.is_directed: result.should_respond = True if memory_hints.get("returning_member_state") in {"returning_member", "long_absent_member"}: result.is_returning_member = True result.reasons.append(memory_hints.get("returning_member_state")) if result.trigger_type == "none": result.trigger_type = "returning_member" result.priority = max(result.priority, float(self.config.get("casual_topic", 0.35))) return result def _is_question(self, content: str) -> bool: return any(re.search(pattern, content, flags=re.IGNORECASE) for pattern in QUESTION_PATTERNS) def _is_social_call(self, content: str) -> bool: return any(re.search(pattern, content, flags=re.IGNORECASE) for pattern in SOCIAL_PATTERNS) def _is_light_social_moment(self, content: str) -> bool: if len(content) > 24: return False return any(re.search(pattern, content, flags=re.IGNORECASE) for pattern in LIGHT_SOCIAL_PATTERNS) def _detect_topic(self, content_lower: str) -> str: for keyword in self.topic_keywords: if keyword and keyword in content_lower: return keyword return ""