diff --git a/pipeline_core/__pycache__/__init__.cpython-310.pyc b/pipeline_core/__pycache__/__init__.cpython-310.pyc new file mode 100644 index 0000000..4d0e725 Binary files /dev/null and b/pipeline_core/__pycache__/__init__.cpython-310.pyc differ diff --git a/pipeline_core/__pycache__/init.cpython-310.pyc b/pipeline_core/__pycache__/init.cpython-310.pyc new file mode 100644 index 0000000..3e43a94 Binary files /dev/null and b/pipeline_core/__pycache__/init.cpython-310.pyc differ diff --git a/pipeline_core/agent_config.py b/pipeline_core/agent_config.py index efefedc..0cc1601 100644 --- a/pipeline_core/agent_config.py +++ b/pipeline_core/agent_config.py @@ -49,10 +49,13 @@ class MemoryConfig: @dataclass class SkillConfig: - """技能配置""" + """技能配置 — 三级隔离：全局 + 组织 + 用户""" enabled: bool = True - dirs: List[str] = field(default_factory=lambda: ["skills/common", "skills/sdlc"]) + base_dir: str = "skills" # 技能根目录，其下为 global/orgs/users/ max_skills_per_turn: int = 5 + enable_global: bool = True # 启用全局技能 + enable_org: bool = True # 启用组织技能 + enable_user: bool = True # 启用用户技能 @dataclass @@ -110,8 +113,11 @@ class AgentConfig: }, "skills": { "enabled": self.skills.enabled, - "dirs": self.skills.dirs, "max_skills_per_turn": self.skills.max_skills_per_turn, + "base_dir": self.skills.base_dir, + "enable_global": self.skills.enable_global, + "enable_org": self.skills.enable_org, + "enable_user": self.skills.enable_user, }, "tools": [ { @@ -155,8 +161,11 @@ class AgentConfig: ), skills=SkillConfig( enabled=sk.get("enabled", True), - dirs=sk.get("dirs", ["skills/common", "skills/sdlc"]), + base_dir=sk.get("base_dir", "skills"), max_skills_per_turn=sk.get("max_skills_per_turn", 5), + enable_global=sk.get("enable_global", True), + enable_org=sk.get("enable_org", True), + enable_user=sk.get("enable_user", True), ), tools=[ToolDefinition(**t) if isinstance(t, dict) else t for t in data.get("tools", [])], session_isolation=data.get("session_isolation", "project"), @@ -317,7 +326,7 @@ SDLC_DEFAULT_CONFIG = AgentConfig( history_limit=50, compression=CompressionConfig(enabled=True, threshold=0.50, target_ratio=0.20, keep_recent=6), memory=MemoryConfig(enabled=True), - skills=SkillConfig(enabled=True, dirs=["skills/common", "skills/sdlc"]), + skills=SkillConfig(enabled=True, base_dir="skills"), ) diff --git a/pipeline_core/skill_loader.py b/pipeline_core/skill_loader.py index 1ede6f1..f76ba3c 100644 --- a/pipeline_core/skill_loader.py +++ b/pipeline_core/skill_loader.py @@ -1,23 +1,22 @@ """ -pipeline-core: Skill Loader — 技能加载与注入 +pipeline-core: Skill Loader v2 — 三级技能隔离 -对照 Hermes Agent 的 skills 系统： -- 从目录扫描 SKILL.md 文件 -- 解析 YAML frontmatter -- 按产线/角色注入到 system prompt -- 支持热更新（agent 执行时可重新扫描） - -目录结构（与 HA 一致）： +技能目录结构： skills/ - common/ ← 所有角色共用 - code-style/SKILL.md - sdlc/ ← SDLC 专用 - git-workflow/SKILL.md - requirement/ ← 角色专用 - design/ - develop/ - test/ - deploy/ + global/ ← 全局技能（所有人可见） + skill-name/ + SKILL.md + orgs/ ← 组织技能 + {org_id}/ + skill-name/ + SKILL.md + users/ ← 用户个人技能 + {user_id}/ + skill-name/ + SKILL.md + +加载优先级：用户 > 组织 > 全局 +同名技能：用户级覆盖组织级覆盖全局级 """ import json @@ -29,6 +28,7 @@ from typing import Dict, List, Optional logger = logging.getLogger("pipeline.skill_loader") +# ── 技能数据模型 ── @dataclass class Skill: @@ -36,13 +36,15 @@ class Skill: name: str path: str # SKILL.md 绝对路径 description: str = "" - content: str = "" # SKILL.md 完整内容 - category: str = "common" # common / sdlc / requirement / design / ... + content: str = "" + scope: str = "global" # global / org / user + scope_id: str = "" # org_id 或 user_id trigger_keywords: List[str] = field(default_factory=list) version: str = "1.0.0" @classmethod - def from_file(cls, filepath: str, category: str = "common") -> "Skill": + def from_file(cls, filepath: str, scope: str = "global", + scope_id: str = "") -> "Skill": """从 SKILL.md 文件加载技能""" name = os.path.basename(os.path.dirname(filepath)) content = "" @@ -55,16 +57,17 @@ class Skill: content = f.read() except Exception as e: logger.warning(f"Skill read failed: {filepath} err={e}") - return cls(name=name, path=filepath, category=category) + return cls(name=name, path=filepath, scope=scope, scope_id=scope_id) - # 解析 YAML frontmatter（如果存在） + # 解析 YAML frontmatter fm = _parse_frontmatter(content) if fm: description = fm.get("description", description) trigger_keywords = fm.get("trigger_keywords", trigger_keywords) version = fm.get("version", version) + # frontmatter 中的 scope 优先 + scope = fm.get("scope", scope) - # 无 frontmatter 时从内容首行提取描述 if not description: for line in content.split("\n"): line = line.strip() @@ -73,18 +76,15 @@ class Skill: break return cls( - name=name, - path=filepath, - description=description, - content=content, - category=category, - trigger_keywords=trigger_keywords, - version=version, + name=name, path=filepath, description=description, + content=content, scope=scope, scope_id=scope_id, + trigger_keywords=trigger_keywords, version=version, ) def to_prompt_block(self) -> str: """生成注入到 system prompt 的技能文本块""" - return f"""## 技能: {self.name} + scope_tag = {"global": "全局", "org": "组织", "user": "个人"}.get(self.scope, "") + return f"""## [{scope_tag}] {self.name} {self.description} {self.content} @@ -92,145 +92,189 @@ class Skill: def _parse_frontmatter(content: str) -> Optional[dict]: - """解析 YAML frontmatter（--- 之间的内容）""" + """解析 YAML frontmatter""" if not content.startswith("---"): return None - end = content.find("---", 3) if end == -1: return None - fm_text = content[3:end].strip() result = {} - - # 简单的 YAML 解析（不依赖 PyYAML） for line in fm_text.split("\n"): line = line.strip() if ":" in line: key, _, val = line.partition(":") key = key.strip() val = val.strip().strip('"').strip("'") - # 处理列表 if val.startswith("[") and val.endswith("]"): - val = [v.strip().strip('"').strip("'") for v in val[1:-1].split(",")] + val = [v.strip().strip('"').strip("'") + for v in val[1:-1].split(",") if v.strip()] result[key] = val - return result +# ── 三级技能加载器 ── + class SkillLoader: - """技能加载器 — 管理技能目录并注入到 prompt""" + """三级技能加载器：global → org → user""" - def __init__(self): - self._skills: Dict[str, Dict[str, Skill]] = {} # {category: {name: Skill}} - self._base_dirs: List[str] = [] + def __init__(self, base_dir: str = ""): + self.base_dir = base_dir + # 三级存储: {scope: {name: Skill}} + self._global: Dict[str, Skill] = {} + self._orgs: Dict[str, Dict[str, Skill]] = {} # {org_id: {name: Skill}} + self._users: Dict[str, Dict[str, Skill]] = {} # {user_id: {name: Skill}} - def add_skill_dir(self, base_dir: str): - """添加技能目录。目录下每个子目录是一个技能。""" - if base_dir not in self._base_dirs: - self._base_dirs.append(base_dir) - self._scan_dir(base_dir) + def set_base_dir(self, base_dir: str): + """设置技能根目录并扫描""" + self.base_dir = base_dir + if base_dir: + self.reload() - def _scan_dir(self, base_dir: str): - """扫描目录下的所有 SKILL.md""" - if not os.path.isdir(base_dir): - logger.debug(f"Skill dir not found: {base_dir}") + # ── 目录扫描 ── + + def reload(self): + """重新扫描所有技能目录""" + self._global = {} + self._orgs = {} + self._users = {} + + if not self.base_dir or not os.path.isdir(self.base_dir): return - # base_dir 下的每个子目录是一个技能分类 - for category in os.listdir(base_dir): - cat_path = os.path.join(base_dir, category) - if not os.path.isdir(cat_path): - continue + # 全局技能: skills/global/ + global_dir = os.path.join(self.base_dir, "global") + if os.path.isdir(global_dir): + self._global = self._scan_scope_dir(global_dir, "global") - # 直接是 SKILL.md - skill_file = os.path.join(cat_path, "SKILL.md") + # 组织技能: skills/orgs/{org_id}/ + orgs_dir = os.path.join(self.base_dir, "orgs") + if os.path.isdir(orgs_dir): + for org_id in os.listdir(orgs_dir): + org_path = os.path.join(orgs_dir, org_id) + if os.path.isdir(org_path): + self._orgs[org_id] = self._scan_scope_dir(org_path, "org", org_id) + + # 用户技能: skills/users/{user_id}/ + users_dir = os.path.join(self.base_dir, "users") + if os.path.isdir(users_dir): + for user_id in os.listdir(users_dir): + user_path = os.path.join(users_dir, user_id) + if os.path.isdir(user_path): + self._users[user_id] = self._scan_scope_dir(user_path, "user", user_id) + + logger.debug(f"SkillLoader reloaded: global={len(self._global)}, " + f"orgs={len(self._orgs)}, users={len(self._users)}") + + def _scan_scope_dir(self, scope_dir: str, scope: str, + scope_id: str = "") -> Dict[str, Skill]: + """扫描单个 scope 目录下的所有技能""" + skills = {} + for entry in os.listdir(scope_dir): + entry_path = os.path.join(scope_dir, entry) + if not os.path.isdir(entry_path): + continue + skill_file = os.path.join(entry_path, "SKILL.md") if os.path.isfile(skill_file): - skill = Skill.from_file(skill_file, category) - self._register(skill) - continue - - # 子目录下可能有多个技能 - for sub in os.listdir(cat_path): - sub_path = os.path.join(cat_path, sub) + skill = Skill.from_file(skill_file, scope, scope_id) + skills[skill.name] = skill + # 也支持子目录嵌套 + for sub in os.listdir(entry_path): + sub_path = os.path.join(entry_path, sub) if os.path.isdir(sub_path): - skill_file = os.path.join(sub_path, "SKILL.md") - if os.path.isfile(skill_file): - skill = Skill.from_file(skill_file, category) - self._register(skill) + sf = os.path.join(sub_path, "SKILL.md") + if os.path.isfile(sf): + skill = Skill.from_file(sf, scope, scope_id) + skills[skill.name] = skill + return skills - def _register(self, skill: Skill): - """注册技能到内存""" - if skill.category not in self._skills: - self._skills[skill.category] = {} - self._skills[skill.category][skill.name] = skill - logger.debug(f"SkillLoader: loaded {skill.category}/{skill.name}") + # ── 查询 ── - def get_all(self, categories: List[str] = None) -> List[Skill]: - """获取指定分类的所有技能""" - result = [] - cats = categories or list(self._skills.keys()) - for cat in cats: - if cat in self._skills: - result.extend(self._skills[cat].values()) - return result + def get_merged(self, org_id: str = "", user_id: str = "") -> Dict[str, Skill]: + """按优先级合并：全局 + 组织 + 用户。同名技能后者覆盖前者。""" + merged = dict(self._global) # 全局为基础 - def get_by_trigger(self, user_input: str, max_skills: int = 5) -> List[Skill]: + # 组织技能覆盖 + if org_id and org_id in self._orgs: + merged.update(self._orgs[org_id]) + + # 用户技能覆盖（最高优先级） + if user_id and user_id in self._users: + merged.update(self._users[user_id]) + + return merged + + def get_by_trigger(self, user_input: str, org_id: str = "", + user_id: str = "", max_skills: int = 5) -> List[Skill]: """根据用户输入匹配相关技能（基于 trigger_keywords）""" + all_skills = list(self.get_merged(org_id, user_id).values()) scored = [] user_lower = user_input.lower() - for cat_skills in self._skills.values(): - for skill in cat_skills.values(): - score = 0 - for kw in skill.trigger_keywords: - if kw.lower() in user_lower: - score += 1 - # 标题匹配加分 - if skill.name.lower() in user_lower: - score += 2 - if score > 0: - scored.append((score, skill)) + for skill in all_skills: + score = 0 + for kw in skill.trigger_keywords: + if kw.lower() in user_lower: + score += 1 + if skill.name.lower() in user_lower: + score += 2 + # scope 加分：用户 > 组织 > 全局 + if skill.scope == "user": + score += 3 + elif skill.scope == "org": + score += 2 + elif skill.scope == "global": + score += 1 + if score > 0: + scored.append((score, skill)) scored.sort(key=lambda x: x[0], reverse=True) return [s[1] for s in scored[:max_skills]] - def get_skill_count(self) -> int: - """已加载技能总数""" - return sum(len(skills) for skills in self._skills.values()) + def get_skill_count(self, org_id: str = "", user_id: str = "") -> int: + """可见技能总数""" + return len(self.get_merged(org_id, user_id)) - def reload(self): - """重新扫描所有技能目录（热更新）""" - self._skills = {} - for d in self._base_dirs: - self._scan_dir(d) + def list_by_scope(self) -> Dict[str, int]: + """各 scope 的技能数量""" + return { + "global": len(self._global), + "orgs": sum(len(s) for s in self._orgs.values()), + "users": sum(len(s) for s in self._users.values()), + } - def build_prompt_block(self, categories: List[str] = None, user_input: str = None, - max_skills: int = 5) -> str: + # ── Prompt 构建 ── + + def build_prompt_block(self, org_id: str = "", user_id: str = "", + user_input: str = None, max_skills: int = 5) -> str: """构建注入 system prompt 的技能段落。 - 如果有 user_input，优先匹配相关技能。 + 有 user_input 时按 trigger 匹配，否则取全部（受限 max_skills）。 """ if user_input: - skills = self.get_by_trigger(user_input, max_skills) + skills = self.get_by_trigger(user_input, org_id, user_id, max_skills) else: - skills = self.get_all(categories)[:max_skills] + merged = self.get_merged(org_id, user_id) + # 优先用户级 + skills = sorted(merged.values(), + key=lambda s: {"user": 0, "org": 1, "global": 2}.get(s.scope, 9)) + skills = skills[:max_skills] if not skills: return "" blocks = ["## 可用技能\n"] for s in skills: - # 只注入技能摘要，不注入完整内容（节省 token） - blocks.append(f"- **{s.category}/{s.name}**: {s.description[:200]}") + blocks.append(f"- **[{s.scope}] {s.name}**: {s.description[:200]}") blocks.append("") return "\n".join(blocks) - def build_full_prompt_block(self, skill_names: List[str]) -> str: - """注入指定技能的完整内容到 system prompt""" + def build_full_prompt_block(self, skill_names: List[str], + org_id: str = "", user_id: str = "") -> str: + """注入指定技能的完整内容""" + merged = self.get_merged(org_id, user_id) blocks = [] - for cat_skills in self._skills.values(): - for name, skill in cat_skills.items(): - if name in skill_names: - blocks.append(skill.to_prompt_block()) + for name in skill_names: + if name in merged: + blocks.append(merged[name].to_prompt_block()) return "\n".join(blocks) @@ -239,8 +283,11 @@ class SkillLoader: _loader: Optional[SkillLoader] = None -def get_skill_loader() -> SkillLoader: +def get_skill_loader(base_dir: str = "") -> SkillLoader: + """获取全局技能加载器单例。首次调用可传入 base_dir。""" global _loader if _loader is None: - _loader = SkillLoader() + _loader = SkillLoader(base_dir) + elif base_dir and base_dir != _loader.base_dir: + _loader.set_base_dir(base_dir) return _loader