141 lines
3.5 KiB
Python
141 lines
3.5 KiB
Python
"""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
|