refactor: SkillConfig 三级隔离 (global/org/user)

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"),
 )
 
 

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/          ← 所有角色共用
+    global/              ← 全局技能（所有人可见）
-      code-style/SKILL.md
+      skill-name/
-    sdlc/            ← SDLC 专用
+        SKILL.md
-      git-workflow/SKILL.md
+    orgs/                ← 组织技能
-    requirement/     ← 角色专用
+      {org_id}/
-    design/
+        skill-name/
-    develop/
+          SKILL.md
-    test/
+    users/               ← 用户个人技能
-    deploy/
+      {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 完整内容
+    content: str = ""
-    category: str = "common"         # common / sdlc / requirement / design / ...
+    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,
+            name=name, path=filepath, description=description,
-            path=filepath,
+            content=content, scope=scope, scope_id=scope_id,
-            description=description,
+            trigger_keywords=trigger_keywords, version=version,
-            content=content,
-            category=category,
-            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):
+    def __init__(self, base_dir: str = ""):
-        self._skills: Dict[str, Dict[str, Skill]] = {}  # {category: {name: Skill}}
+        self.base_dir = base_dir
-        self._base_dirs: List[str] = []
+        # 三级存储: {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):
+    def set_base_dir(self, base_dir: str):
-        """添加技能目录。目录下每个子目录是一个技能。"""
+        """设置技能根目录并扫描"""
-        if base_dir not in self._base_dirs:
+        self.base_dir = base_dir
-            self._base_dirs.append(base_dir)
+        if base_dir:
-        self._scan_dir(base_dir)
+            self.reload()
 
-    def _scan_dir(self, base_dir: str):
+    # ── 目录扫描 ──
-        """扫描目录下的所有 SKILL.md"""
+
-        if not os.path.isdir(base_dir):
+    def reload(self):
-            logger.debug(f"Skill dir not found: {base_dir}")
+        """重新扫描所有技能目录"""
+        self._global = {}
+        self._orgs = {}
+        self._users = {}
+
+        if not self.base_dir or not os.path.isdir(self.base_dir):
             return
 
-        # base_dir 下的每个子目录是一个技能分类
+        # 全局技能: skills/global/
-        for category in os.listdir(base_dir):
+        global_dir = os.path.join(self.base_dir, "global")
-            cat_path = os.path.join(base_dir, category)
+        if os.path.isdir(global_dir):
-            if not os.path.isdir(cat_path):
+            self._global = self._scan_scope_dir(global_dir, "global")
-                continue
 
-            # 直接是 SKILL.md
+        # 组织技能: skills/orgs/{org_id}/
-            skill_file = os.path.join(cat_path, "SKILL.md")
+        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)
+                skill = Skill.from_file(skill_file, scope, scope_id)
-                self._register(skill)
+                skills[skill.name] = skill
-                continue
+            # 也支持子目录嵌套
-
+            for sub in os.listdir(entry_path):
-            # 子目录下可能有多个技能
+                sub_path = os.path.join(entry_path, sub)
-            for sub in os.listdir(cat_path):
-                sub_path = os.path.join(cat_path, sub)
                 if os.path.isdir(sub_path):
-                    skill_file = os.path.join(sub_path, "SKILL.md")
+                    sf = os.path.join(sub_path, "SKILL.md")
-                    if os.path.isfile(skill_file):
+                    if os.path.isfile(sf):
-                        skill = Skill.from_file(skill_file, category)
+                        skill = Skill.from_file(sf, scope, scope_id)
-                        self._register(skill)
+                        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]:
+    def get_merged(self, org_id: str = "", user_id: str = "") -> Dict[str, Skill]:
-        """获取指定分类的所有技能"""
+        """按优先级合并：全局 + 组织 + 用户。同名技能后者覆盖前者。"""
-        result = []
+        merged = dict(self._global)  # 全局为基础
-        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_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 all_skills:
-            for skill in cat_skills.values():
+            score = 0
-                score = 0
+            for kw in skill.trigger_keywords:
-                for kw in skill.trigger_keywords:
+                if kw.lower() in user_lower:
-                    if kw.lower() in user_lower:
+                    score += 1
-                        score += 1
+            if skill.name.lower() in user_lower:
-                # 标题匹配加分
+                score += 2
-                if skill.name.lower() in user_lower:
+            # scope 加分：用户 > 组织 > 全局
-                    score += 2
+            if skill.scope == "user":
-                if score > 0:
+                score += 3
-                    scored.append((score, skill))
+            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:
+    def get_skill_count(self, org_id: str = "", user_id: str = "") -> int:
-        """已加载技能总数"""
+        """可见技能总数"""
-        return sum(len(skills) for skills in self._skills.values())
+        return len(self.get_merged(org_id, user_id))
 
-    def reload(self):
+    def list_by_scope(self) -> Dict[str, int]:
-        """重新扫描所有技能目录（热更新）"""
+        """各 scope 的技能数量"""
-        self._skills = {}
+        return {
-        for d in self._base_dirs:
+            "global": len(self._global),
-            self._scan_dir(d)
+            "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,
+    # ── Prompt 构建 ──
-                           max_skills: int = 5) -> str:
+
+    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.scope}] {s.name}**: {s.description[:200]}")
-            blocks.append(f"- **{s.category}/{s.name}**: {s.description[:200]}")
         blocks.append("")
         return "\n".join(blocks)
 
-    def build_full_prompt_block(self, skill_names: List[str]) -> str:
+    def build_full_prompt_block(self, skill_names: List[str],
-        """注入指定技能的完整内容到 system prompt"""
+                                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 in skill_names:
-            for name, skill in cat_skills.items():
+            if name in merged:
-                if name in skill_names:
+                blocks.append(merged[name].to_prompt_block())
-                    blocks.append(skill.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