59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
import os
|
|
import glob
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional
|
|
|
|
logger = logging.getLogger("cs_agent")
|
|
|
|
class SkillManager:
|
|
"""
|
|
技能包管理器：
|
|
1. 自动扫描 skills/ 目录下的 SKILL.md 文件。
|
|
2. 提供按需加载和组合技能的能力。
|
|
3. 支持热加载（无需重启即可更新 AI 知识）。
|
|
"""
|
|
def __init__(self, skills_dir: str = "skills"):
|
|
self.skills_dir = Path(skills_dir)
|
|
self._skill_cache: Dict[str, str] = {}
|
|
self.reload_skills()
|
|
|
|
def reload_skills(self):
|
|
"""扫描并加载所有技能文件"""
|
|
new_cache = {}
|
|
skill_files = glob.glob(str(self.skills_dir / "**/SKILL.md"), recursive=True)
|
|
|
|
for file_path in skill_files:
|
|
try:
|
|
path = Path(file_path)
|
|
skill_name = path.parent.name.lower()
|
|
content = path.read_text(encoding="utf-8")
|
|
new_cache[skill_name] = content
|
|
except Exception as e:
|
|
logger.error(f"[SkillManager] 加载技能失败 {file_path}: {e}")
|
|
|
|
self._skill_cache = new_cache
|
|
logger.info(f"[SkillManager] 成功加载 {len(self._skill_cache)} 个技能包: {list(self._skill_cache.keys())}")
|
|
|
|
def get_skill(self, name: str) -> str:
|
|
"""获取单个技能内容"""
|
|
return self._skill_cache.get(name.lower(), "")
|
|
|
|
def compose_skills(self, names: List[str]) -> str:
|
|
"""组合多个技能内容，用于注入 System Prompt"""
|
|
parts = []
|
|
for name in names:
|
|
content = self.get_skill(name)
|
|
if content:
|
|
parts.append(f"### 技能：{name}\n{content}")
|
|
return "\n\n".join(parts)
|
|
|
|
def get_all_skills_text(self, exclude: Optional[List[str]] = None) -> str:
|
|
"""获取所有技能的合集（用于全能大脑模式）"""
|
|
exclude_set = {n.lower() for n in (exclude or [])}
|
|
names = [n for n in self._skill_cache.keys() if n not in exclude_set]
|
|
return self.compose_skills(names)
|
|
|
|
# 全局单例
|
|
skill_manager = SkillManager()
|