""" 技能路由器 (Skill Router) ======================== 根据用户消息的关键词，自动匹配最合适的 Skill，并推荐给 Agent 加载。 设计原则: - 每个 Skill 在 SKILL.md 的 frontmatter 里声明 triggers 字段（逗号分隔的关键词） - 路由器读取所有 skill 的 triggers，与用户消息做匹配 - 匹配度超过阈值的 skill 会被推荐给 Agent - 用户新添加的 skill 只要声明了 triggers，自动参与路由，无需修改代码 匹配算法: - 关键词命中数（加权）+ 描述相似度 + 场景上下文 - 返回按置信度排序的推荐列表 调试: - 设置环境变量 SKILL_ROUTER_LOG=1 可输出调试日志到 workspace/skill_router.log """ from __future__ import annotations import os import re from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Optional @dataclass class SkillMatch: """技能匹配结果""" name: str confidence: float # 0.0 - 1.0 matched_keywords: list[str] reason: str = "" class SkillRouter: """技能路由器""" # 置信度阈值：超过这个值才会推荐 RECOMMEND_THRESHOLD = 0.18 # 强推荐阈值：超过这个值会强烈建议加载 STRONG_RECOMMEND_THRESHOLD = 0.35 # 自动预加载阈值：超过这个值直接把技能指令注入上下文 # 模型会认为这个技能已经加载了，直接按照指令执行 PRELOAD_THRESHOLD = 0.35 def __init__(self, loader) -> None: """ Args: loader: SkillLoader 实例 """ self._loader = loader def match(self, user_message: str) -> list[SkillMatch]: """ 匹配用户消息，返回推荐的技能列表（按置信度降序） Args: user_message: 用户的最新消息文本 Returns: 匹配结果列表，每个元素是 SkillMatch """ if not user_message or len(user_message.strip()) == 0: return [] msg = user_message.lower() results: list[SkillMatch] = [] for skill_name in self._loader.names(): skill = self._loader.get(skill_name) if skill is None: continue score = 0.0 matched = [] # 1. 触发词匹配（权重最高） triggers = getattr(skill, "triggers", []) if triggers: for trigger in triggers: trigger_lower = trigger.lower() if len(trigger_lower) < 2: continue # 精确包含匹配（权重高） if trigger_lower in msg: score += 0.20 matched.append(trigger) continue # 子序列匹配：所有字都在消息里且顺序一致（不一定连续） # 中文里插入少量字是常见的（如"做个图表"≈"做图表"），权重较高 if self._is_subsequence(trigger_lower, msg): # 触发词越长且插入字越少，权重越高 insert_ratio = len(trigger_lower) / max(len(msg), 1) weight = 0.10 + 0.08 * insert_ratio score += weight matched.append(trigger + "(顺序)") continue # 部分匹配：触发词有 2 个字以上连续出现在消息里 hit_len = self._max_common_substring_len(trigger_lower, msg) if hit_len >= 3: weight = 0.05 + 0.02 * (hit_len - 3) score += weight matched.append(trigger + f"(部分{hit_len}字)") # 2. 描述匹配（次要权重） desc = skill.description.lower() desc_keywords = self._extract_keywords(desc) for kw in desc_keywords: if len(kw) >= 2 and kw in msg: score += 0.05 if kw not in matched: matched.append(kw) # 3. 名称匹配（小权重） if skill_name.lower() in msg: score += 0.1 matched.append(skill_name) # 归一化到 0-1 confidence = min(score, 1.0) if confidence >= self.RECOMMEND_THRESHOLD: reason = self._build_reason(skill_name, confidence, matched) results.append(SkillMatch( name=skill_name, confidence=confidence, matched_keywords=matched, reason=reason, )) # 按置信度降序 results.sort(key=lambda x: x.confidence, reverse=True) return results def build_context_text(self, user_message: str) -> str: """ 生成注入到系统提示的路由推荐文本 如果匹配度足够高（超过 PRELOAD_THRESHOLD），直接把技能的完整指令 注入上下文，模型会认为技能已加载，直接按指令执行。 Args: user_message: 用户最新消息 Returns: 推荐文本，如果没有匹配则返回空字符串 """ if not user_message or len(user_message.strip()) == 0: return "" matches = self.match(user_message) # 调试日志 self._log_match(user_message, matches) if not matches: return "" strong = [m for m in matches if m.confidence >= self.STRONG_RECOMMEND_THRESHOLD] preload = [m for m in matches if m.confidence >= self.PRELOAD_THRESHOLD] normal = [m for m in matches if self.RECOMMEND_THRESHOLD <= m.confidence < self.STRONG_RECOMMEND_THRESHOLD] lines = [] # 自动预加载：高置信度时直接注入技能完整指令 if preload: top = preload[0] skill = self._loader.get(top.name) if skill and skill.instructions: lines.append(f"🎯 自动技能路由: 已为你预加载「{skill.name}」技能") lines.append(f" 匹配原因: {top.reason}") lines.append(f" 置信度: {top.confidence:.0%}") lines.append("") lines.append(f"=== 技能: {skill.name} ===") lines.append("以下是该技能的完整说明，请直接按照说明执行：") lines.append("") lines.append(skill.instructions) lines.append("") lines.append("=== 技能说明结束 ===") lines.append("") lines.append( "📋 执行提示: 请使用上面预加载的技能来完成用户需求。" "如果该技能有工具调用方式，请按说明调用。" "不要使用 Python 脚本或其他方式重新实现该技能已有的功能。" ) return "\n".join(lines) # 低于预加载阈值时，只做推荐 lines.append("💡 Skill 推荐（根据用户消息自动匹配）:") if strong: lines.append(" 🌟 强烈推荐:") for m in strong: lines.append(f" - {m.name} → {m.reason}") lines.append(f" ⚡ 请立即调用 load_skill(\"{m.name}\") 加载此技能") lines.append(f" 不要用基础工具手动实现，直接使用该技能") if normal: lines.append(" 📌 可能相关:") for m in normal[:3]: lines.append(f" - {m.name} → {m.reason}") lines.append(f" 如需使用可调用 load_skill(\"{m.name}\")") if strong: lines.append( "\n ⚡ 重要提示: 用户需求明确匹配上面的强烈推荐技能，" "请立即调用 load_skill 加载该技能，然后按照技能说明执行。" "不要尝试用 Python 脚本或其他方式手动实现。" ) return "\n".join(lines) def _extract_keywords(self, text: str) -> list[str]: """从文本中提取有意义的关键词（简单实现：提取中文词组和英文单词）""" # 提取英文单词（长度 >= 3） english_words = re.findall(r'[a-zA-Z]{3,}', text.lower()) # 提取中文词组（2-6 字的连续中文字符） chinese_phrases = re.findall(r'[\u4e00-\u9fa5]{2,6}', text) # 去掉常见停用词 stop_words = { "的", "了", "和", "是", "在", "有", "与", "等", "将", "及", "可以", "使用", "支持", "进行", "生成", "创建", "处理", "数据", "文件", "功能", "工具", "技能", "输出", "输入", "多种", "各种", "一个", "一些", "什么", "如何", "怎么", } keywords = [] for w in english_words + chinese_phrases: if w not in stop_words and len(w) >= 2: keywords.append(w) # 去重并返回 return list(dict.fromkeys(keywords)) def _build_reason(self, name: str, confidence: float, matched: list[str]) -> str: """构建推荐理由文本""" if matched: show = matched[:4] kw_text = "、".join(show) if len(matched) > 4: kw_text += " 等" return f"匹配关键词「{kw_text}」" return f"置信度 {confidence:.0%}" def _log_match(self, user_message: str, matches: list[SkillMatch]) -> None: """输出调试日志（环境变量 SKILL_ROUTER_LOG=1 时启用）""" if not os.environ.get("SKILL_ROUTER_LOG"): return try: log_path = Path(os.environ.get("SKILL_ROUTER_LOG_PATH", r"D:\project\senmeshworker-main\dist\openmesh\workspace\skill_router.log")) log_path.parent.mkdir(parents=True, exist_ok=True) ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with open(log_path, "a", encoding="utf-8") as f: f.write(f"\n[{ts}] 用户消息: {user_message[:100]}\n") if matches: f.write(f" 匹配到 {len(matches)} 个技能:\n") for m in matches: f.write(f" - {m.name}: {m.confidence:.1%} (关键词: {m.matched_keywords[:5]})\n") if m.confidence >= self.PRELOAD_THRESHOLD: f.write(f" → 自动预加载\n") else: f.write(" 无匹配\n") except Exception: pass # 日志失败不影响主流程 @staticmethod def _is_subsequence(needle: str, haystack: str) -> bool: """检查 needle 是否是 haystack 的子序列（字符顺序一致，可不连续）""" it = iter(haystack) return all(c in it for c in needle) @staticmethod def _max_common_substring_len(s1: str, s2: str) -> int: """返回两个字符串的最长公共子串长度""" if not s1 or not s2: return 0 m, n = len(s1), len(s2) max_len = 0 # 用滚动数组，O(n) 空间 prev = [0] * (n + 1) for i in range(1, m + 1): curr = [0] * (n + 1) for j in range(1, n + 1): if s1[i - 1] == s2[j - 1]: curr[j] = prev[j - 1] + 1 max_len = max(max_len, curr[j]) else: curr[j] = 0 prev = curr return max_len