"""Skill 触发检测器 - Phase 9.5 检测消息中的 Skill 触发条件。 """ import re from typing import Any from app.agents.skills.metadata import SkillMetadata class SkillTriggerDetector: """Skill 触发检测器 检测用户消息中是否触发了某个 Skill。 """ def __init__(self): self._skills: dict[str, SkillMetadata] = {} def register_skill(self, skill: SkillMetadata) -> None: """注册 Skill Args: skill: Skill 元数据 """ self._skills[skill.name] = skill def unregister_skill(self, name: str) -> bool: """注销 Skill Args: name: Skill 名称 Returns: 是否成功 """ if name in self._skills: del self._skills[name] return True return False def detect_triggered_skills(self, message: str) -> list[str]: """检测触发的 Skills Args: message: 用户消息 Returns: 触发的 Skill 名称列表 """ triggered = [] message_lower = message.lower() for skill in self._skills.values(): if not skill.enabled: continue if self._matches_triggers(message, message_lower, skill): triggered.append(skill.name) return triggered def _matches_triggers(self, message: str, message_lower: str, skill: SkillMetadata) -> bool: """检查消息是否匹配 Skill 触发条件 Args: message: 原始消息 message_lower: 小写消息 skill: Skill 元数据 Returns: 是否匹配 """ for trigger in skill.triggers: trigger_lower = trigger.lower() # 前缀匹配,如 "/code" 或 "@git" if trigger_lower.startswith("/") or trigger_lower.startswith("@"): if message_lower.startswith(trigger_lower): return True # 命令格式,如 "//analyze" if trigger_lower.startswith("//"): pattern = trigger_lower[2:] if re.search(rf"\b{re.escape(pattern)}\b", message_lower): return True # 关键词匹配 if trigger_lower in message_lower: return True return False def get_skill_prompt(self, skill_name: str) -> str | None: """获取 Skill 的提示词 Args: skill_name: Skill 名称 Returns: Skill 内容或 None """ skill = self._skills.get(skill_name) if skill: return skill.content return None def get_triggered_skill_context(self, message: str) -> str: """获取触发的 Skills 上下文 Args: message: 用户消息 Returns: 拼接的 Skill 上下文 """ triggered = self.detect_triggered_skills(message) if not triggered: return "" contexts = [] for skill_name in triggered: skill = self._skills.get(skill_name) if skill: contexts.append(f"# {skill.name}\n\n{skill.content}") return "\n\n---\n\n".join(contexts) # 全局检测器 _detector: SkillTriggerDetector | None = None def get_skill_trigger_detector() -> SkillTriggerDetector: """获取全局 Skill 触发检测器""" global _detector if _detector is None: _detector = SkillTriggerDetector() return _detector