建议收藏！Supervisor 动态 Skills 知识库架构设计方案

supervisor_init/
├── skills/ # [新增] 知识库根目录
│ ├── DC/ # LOB
│ │ └── supervisor_sop.md # 专属SOP
│ └── ATA/
│ └── supervisor_sop.md

# 全局 Skills 配置
skills_config:
# 按 LOB 配置 Skills 文件路径
lob_skills:
DC: "./skills/DC/supervisor_sop.md"
ATA: "./skills/ATA/supervisor_sop.md"
5. 知识库内容规范 (Per-LOB Markdown Template)
每个 LOB 的 supervisor_sop.md 内容独立于能力卡片，作为该业务线 Supervisor 的"大脑补充知识
区"。
文件结构示例
# 缓存策略
cache_enabled: true
cache_ttl: 300
# 格式校验开关
validate_format: true# 是否在加载时强制校验 Markdown 格式

# Agent 引用配置
# 说明：各 Agent 所属的 LOB 决定其使用哪个 supervisor_sop.md
remote_agents:
DC:
limo_agent:
name: "limo_agent"
description: "机场接送车辆服务"
# LOB 自动决定：使用 skills/DC/supervisor_sop.md
ATA:
hotel_agent:
name: "hotel_agent"
description: "酒店查询与预订服务"
# LOB 自动决定：使用 skills/ATA/supervisor_sop.md

# Supervisor Routing SOP

## [GLOBAL_RULES]
> 跨 Agent 的通用判决逻辑
1. **单一职责**：优先匹配功能最专一的 Agent。
2. **上下文优先**：如果 Query 是对上文的补充，必须延续上轮的 Agent。

## [CONFLICT_RESOLUTION]
> 常见冲突仲裁与典型歧义场景 Few-Shot
> 注意：此处不按 Agent 分类，而是按"冲突场景"分类，辅助消除歧义
### 场景1: 航班号相关歧义 (Limo vs Flight)
- **判定逻辑**：
- 询问包含航班号比如QR944等，并没有直接表明预订/查询机票 → `limo_agent`
- 询问非航班号机票搜索/预订信息 → `flight_agent`
- **Few-Shot Examples**:
- Query: "CA1234，明天下午帮我安排一下" → `limo_agent` (语境暗示用车)
- Query: "CA1234 几点落地？" → `flight_agent` (纯信息查询)

### 场景2: 纯包含xxx酒店地址的歧义 (Limo vs Hotel)
- **判定逻辑**：
- 上下文涉及接送机 → `limo_agent` (作为目的地/出发地)
- 上下文为空或明确"订房" → `hotel_agent`
- **Few-Shot Examples**:
校验规则 (Validation Rules)
1. 必需区块：必须包含 [GLOBAL_RULES] 和 [CONFLICT_RESOLUTION]，且目前改造只支持这俩个
区块标签。
2. 内容导向：内容必须关注"歧义消除"和"冲突仲裁"，而非简单的功能描述。
6. 核心模块设计 (SkillsManager )
SkillsManager 是连接配置与运行时的桥梁。
主要职责：
1. 加载 (Loader)：统一处理 file:// , oss:// , http:// 协议。
2. 解析 (Parser)：提取 Markdown 中的关键块（如 ## [GLOBAL_RULES] , ##
[CONFLICT_RESOLUTION] ）。
3. 校验 (Validator)：严格校验文档格式，确保包含必需区块。
4. 缓存 (Caching)：将解析后的内容序列化为 JSON 格式，独立缓存，与 Capability Cards 完全隔
离。
5. 监控 (Watcher)：计算文件内容 MD5 Hash，配合 Supervisor 的热重载机制。
关键逻辑代码框架：
- Query: "去希尔顿" (History: "派个车") → `limo_agent`
- Query: "帮我查一下希尔顿" (No History) → `hotel_agent`

### 场景3: 隐晦意图识别
- Query: "我想出去玩" → `trip_planner_agent`
- Query: "这就出发" (History: "订好票了吗") → `limo_agent` (立即叫车)

class SkillsManager:
    def __init__(self, config):
        self.config = config.get('skills_config', {})
        self.skills_cache_by_lob = {} # 按 LOB 分组的 JSON 缓存: {LOB: parsed_data}

    def load_all_lob_sops(self):
        """加载所有 LOB 的 SOP 文件"""
        lob_skills = self.config.get('lob_skills', {})
        for lob, file_path in lob_skills.items():
            self.skills_cache_by_lob[lob] = self._load_sop(file_path, lob)

    def _load_sop(self, file_path, lob):
        # 1. 读取原始内容
        content = self._read_source(file_path)

        # 2. 格式校验
        if not self._validate_structure(content):
            raise ValueError(f"[{lob}] SOP 文件格式校验失败: 缺少必需区块")

        # 3. 解析为结构化 JSON
        parsed_data = {
            "lob": lob,
            "global_rules": self._extract_section(content, "GLOBAL_RULES"),
            "conflict_resolution": self._extract_section(content, "CONFLICT_RESOLUTION"),
            "raw_content": content,  # 保留原始内容用于直接注入
            "file_hash": self._calculate_md5(content),
            "last_updated": datetime.now().isoformat()
        }

        logger.info(f"已加载 [{lob}] Skills, Hash: {parsed_data['file_hash'][:8]}")
        return parsed_data

    def _validate_structure(self, content):
        # 检查必需区块
        required = ["## [GLOBAL_RULES]", "## [CONFLICT_RESOLUTION]"]
        return all(tag in content for tag in required)

    def get_skills_for_prompt(self, lob: str):
        """获取指定 LOB 的 Skills 内容 (独立模块) """
        lob_upper = lob.upper()
        if lob_upper not in self.skills_cache_by_lob:
            logger.warning(f"未找到 [{lob_upper}] 的 Skills 配置, 返回空字符串")
            return""
        return self.skills_cache_by_lob[lob_upper].get("raw_content", "")

    def get_skills_json(self, lob: str = None):
        """获取结构化的 Skills 数据 (用于调试或 API 返回) """
        if lob:
            return self.skills_cache_by_lob.get(lob.upper(), {})
        return self.skills_cache_by_lob

class AgentRegistry:
    def __init__(self, config_path: str = "agents_config.yaml"):
        # ... 原有代码 ...
        # 注意：此处假设 self.config 已在省略的代码中被加载
        # self.config = load_config(config_path) 
        
        self.skills_manager = SkillsManager(self.config)
        self.skills_manager.load_all_lob_sops()  # 加载所有 LOB 的 SOP

    def build_capability_cards(self, format_type: str = "agent", lob: Optional[str] = None) -> str:
        """构建能力卡片 (不包含 Skills 内容) """
        agent_cards = []
        agents_to_build = {}

        if lob:
            # 只构建指定 LOB 的 Agents
            agents_to_build = self.capabilities_by_lob.get(lob.upper(), {})
        else:
            agents_to_build = self.capabilities

        for agent_id, capability in agents_to_build.items():
            if capability.is_builtin:
                continue
            agent_cards.append(capability.format_for_prompt())

        # 返回纯 Capability Cards, 不包含 Skills
        return"\n\n".join(agent_cards)

    def get_supervisor_prompt_components(self, lob: str):
        """获取指定 LOB 的 Supervisor Prompt 的两个独立组件"""
        return {
            "skills": self.skills_manager.get_skills_for_prompt(lob), # 该 LOB 的 Skills
            "capability_cards": self.build_capability_cards(lob=lob)  # 该 LOB 的 Capability Cards
        }

# 假设当前 LOB 为 "DC"
components = registry.get_supervisor_prompt_components(lob="DC")
prompt = f"""
{components['skills']} # DC 业务线的 Skills
---
{components['capability_cards']} # DC 业务线的 Capability Cards
"""