refactor: migrate workflow to v2 core and archive legacy modules
This commit is contained in:
56
core/skill_manager.py
Normal file
56
core/skill_manager.py
Normal file
@@ -0,0 +1,56 @@
|
||||
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) -> str:
|
||||
"""获取所有技能的合集(用于全能大脑模式)"""
|
||||
return self.compose_skills(list(self._skill_cache.keys()))
|
||||
|
||||
# 全局单例
|
||||
skill_manager = SkillManager()
|
||||
Reference in New Issue
Block a user