Agent Skill 编排架构技能注册、发现、组合的设计原则一、个性化深度引言Agent 接了五个技能网页搜索、代码执行、图片生成、数据分析、邮件发送。每个技能独立测试都正常。但当用户说帮我搜一下最近的 AI 新闻整理成表格发到团队邮箱Agent 调用了搜索 → 代码执行 → 邮件发送——跳过了数据分析直接把搜索结果当表格发了出去。问题出在哪里Agent 知道有哪些技能可用但不知道技能之间怎么串联。Skill 编排不是简单的按需调用而是理解技能之间的输入输出依赖、组合约束、执行顺序。见证奇迹的时刻引入了技能图谱Skill Graph后Agent 在调用前会先检查技能间的拓扑关系。这不是让 Agent 变聪明了而是把编排逻辑从模型推理转移到了显式知识表示中。二、个性化原理剖析Agent Skill 编排的核心架构分为三层注册层的设计原则技能元信息结构化每个技能必须声明输入类型、输出类型、前置依赖、预估耗时、置信度评分。没有元信息的技能不可编排。版本兼容性标记技能接口变更时标注 Breaking Change。编排器发现不兼容的版本组合时直接拒绝编排并给出修复建议。发现层的设计原则意图→能力映射不是简单的关键词匹配而是语义理解。用 Embedding 相似度 规则匹配的混合策略。多候选排序同时返回前 3 个候选技能让编排层做最终选择。单一推荐可能因理解偏差而选错。编排层的设计原则DAG 拓扑排序所有技能编排最终表达为一个有向无环图拓扑排序确定执行顺序。并行机会检测DAG 中多个入度为 0 的节点可并行执行减少总等待时间。三、个性化代码实践 Agent Skill 编排系统。 设计理念显式的技能元信息 DAG 编排 可预测的 Agent 行为。 from abc import ABC, abstractmethod from dataclasses import dataclass, field from enum import Enum from typing import Any, Dict, List, Optional, Set, Tuple from collections import deque class SkillStatus(Enum): 设计原因技能执行状态驱动编排引擎的状态机。 PENDING pending RUNNING running COMPLETED completed FAILED failed SKIPPED skipped dataclass class SkillSignature: 设计原因技能的输入输出签名。 明确的接口契约是技能可编排的前提。 inputs: Dict[str, str] # {param_name: type} outputs: Dict[str, str] # {param_name: type} description: str # 自然语言描述 estimated_cost_ms: int # 预估耗时 confidence: float 0.9 # 成功率估计 dataclass class SkillNode: 设计原因编排 DAG 中的节点。 记录技能、依赖关系、执行状态。 skill_id: str skill_name: str signature: SkillSignature dependencies: Set[str] field(default_factoryset) status: SkillStatus SkillStatus.PENDING result: Any None error: Optional[str] None class SkillGraph: 设计原因技能 DAG。 维护技能间的依赖关系支持拓扑排序和并行检测。 def __init__(self): # 设计原因邻接表存储 DAG 结构。 self._nodes: Dict[str, SkillNode] {} # 设计原因反向边——快速查找某个技能被谁依赖。 self._reverse_edges: Dict[str, Set[str]] {} def add_node(self, node: SkillNode) - None: 设计原因注册节点到 DAG。 self._nodes[node.skill_id] node if node.skill_id not in self._reverse_edges: self._reverse_edges[node.skill_id] set() def add_dependency( self, skill_id: str, depends_on: str ) - bool: 设计原因添加有向边 (depends_on → skill_id)。 skill_id 依赖 depends_on 的输出。 返回 False 表示添加后会形成环非法操作。 if skill_id not in self._nodes or depends_on not in self._nodes: return False # 设计原因临时加边 环检测。 self._nodes[skill_id].dependencies.add(depends_on) self._reverse_edges[depends_on].add(skill_id) if self._has_cycle(): # 设计原因有环就回滚。 self._nodes[skill_id].dependencies.remove(depends_on) self._reverse_edges[depends_on].remove(skill_id) return False return True def _has_cycle(self) - bool: 设计原因Kahn 算法检测环。 DAG 中的环意味着编排不可执行——必须拒绝。 in_degree { nid: len(node.dependencies) for nid, node in self._nodes.items() } queue deque([ nid for nid, deg in in_degree.items() if deg 0 ]) visited 0 while queue: nid queue.popleft() visited 1 for neighbor in self._reverse_edges.get(nid, set()): in_degree[neighbor] - 1 if in_degree[neighbor] 0: queue.append(neighbor) return visited ! len(self._nodes) def get_execution_order(self) - List[List[str]]: 设计原因生成分层执行顺序。 返回 List[List[skill_id]] - 内层 List可并行的技能集合 - 外层 List必须顺序执行的分层 in_degree { nid: len(node.dependencies) for nid, node in self._nodes.items() } # 设计原因入度为 0 的节点是第一层无依赖可立即执行。 current_layer [ nid for nid, deg in in_degree.items() if deg 0 ] layers [] while current_layer: layers.append(current_layer) next_layer [] for nid in current_layer: for neighbor in self._reverse_edges.get(nid, set()): in_degree[neighbor] - 1 if in_degree[neighbor] 0: next_layer.append(neighbor) current_layer next_layer return layers def get_parallel_groups(self) - List[Set[str]]: 设计原因识别可并行的技能组。 同一层中的所有技能互不依赖可并行执行。 order self.get_execution_order() return [set(layer) for layer in order] class SkillOrchestrator: 设计原因技能编排的执行引擎。 根据 SkillGraph 的分层执行顺序逐层调度技能。 def __init__(self): self._skills: Dict[str, SkillNode] {} self._graph SkillGraph() def register_skill( self, skill_id: str, name: str, signature: SkillSignature, dependencies: List[str] None, ) - None: 设计原因注册技能到编排器。 dependencies 是此技能依赖的其他技能 ID 列表。 node SkillNode( skill_idskill_id, skill_namename, signaturesignature, dependenciesset(dependencies or []), ) self._skills[skill_id] node self._graph.add_node(node) for dep in (dependencies or []): self._graph.add_dependency(skill_id, dep) async def execute_plan( self, skill_ids: List[str], ) - Dict[str, Any]: 设计原因执行编排计划。 按 DAG 分层顺序执行同层技能并行执行。 # 设计原因只取需要的技能构建子图。 sub_graph self._build_sub_graph(skill_ids) execution_order sub_graph.get_execution_order() results: Dict[str, Any] {} for layer_idx, layer in enumerate(execution_order): # 设计原因同层技能可并行执行。 import asyncio tasks [] for skill_id in layer: node self._skills[skill_id] tasks.append(self._execute_skill(node, results)) layer_results await asyncio.gather( *tasks, return_exceptionsTrue ) for skill_id, result in zip(layer, layer_results): if isinstance(result, Exception): node self._skills[skill_id] node.status SkillStatus.FAILED node.error str(result) # 设计原因技能失败时检查依赖它的技能。 self._mark_dependents_failed(skill_id) else: results[skill_id] result return results def _build_sub_graph(self, skill_ids: List[str]) - SkillGraph: 设计原因从完整 DAG 中提取子图。 只包含指定技能及其传递依赖。 sub_graph SkillGraph() # 设计原因BFS 收集所有需要的技能含传递依赖。 to_include: Set[str] set(skill_ids) queue deque(skill_ids) while queue: nid queue.popleft() node self._skills.get(nid) if node: for dep in node.dependencies: if dep not in to_include: to_include.add(dep) queue.append(dep) # 设计原因重建子图的节点和边。 for nid in to_include: node self._skills[nid] sub_graph.add_node(SkillNode( skill_idnode.skill_id, skill_namenode.skill_name, signaturenode.signature, dependencies( node.dependencies to_include ), )) return sub_graph async def _execute_skill( self, node: SkillNode, previous_results: Dict[str, Any] ) - Any: 设计原因执行单个技能。 从 previous_results 获取依赖技能的输出。 node.status SkillStatus.RUNNING # 设计原因收集依赖技能的输出作为当前技能的输入。 skill_inputs {} for dep_id in node.dependencies: if dep_id in previous_results: skill_inputs[dep_id] previous_results[dep_id] try: # 设计原因实际技能调用此处为占位。 result f{node.skill_name} executed with {skill_inputs} node.status SkillStatus.COMPLETED node.result result return result except Exception as e: node.status SkillStatus.FAILED node.error str(e) raise def _mark_dependents_failed(self, failed_skill_id: str) - None: 设计原因传播失败——依赖已失败技能的其他技能自动跳过。 避免无效的等待和执行。 dependents self._graph._reverse_edges.get( failed_skill_id, set() ) queue deque(dependents) while queue: nid queue.popleft() node self._skills.get(nid) if node and node.status SkillStatus.PENDING: node.status SkillStatus.SKIPPED node.error ( fSkipped: dependency {failed_skill_id} failed ) # 设计原因递归传播跳过间接依赖。 for dep in self._graph._reverse_edges.get(nid, set()): queue.append(dep) # ── 使用示例 ── async def demo_skill_orchestration(): 设计原因演示完整的技能编排流程。 orchestrator SkillOrchestrator() # 注册技能网页搜索 orchestrator.register_skill( web_search, 网页搜索, SkillSignature( inputs{query: string}, outputs{results: list[dict]}, description搜索网络信息, estimated_cost_ms2000, ), ) # 注册技能数据分析依赖搜索结果 orchestrator.register_skill( data_analysis, 数据分析, SkillSignature( inputs{data: list[dict]}, outputs{table: dataframe}, description分析数据并生成表格, estimated_cost_ms500, ), dependencies[web_search], ) # 注册技能邮件发送依赖数据分析结果 orchestrator.register_skill( email_sender, 邮件发送, SkillSignature( inputs{content: dataframe}, outputs{status: bool}, description发送邮件, estimated_cost_ms1000, ), dependencies[data_analysis], ) # 执行编排 results await orchestrator.execute_plan([ web_search, data_analysis, email_sender ]) print(f编排完成: {results})四、个性化边界权衡1. 显式依赖 vs 隐式推断显式依赖手动声明输入输出准确可靠但维护成本随技能数量线性增长。隐式推断模型根据输入输出类型自动匹配灵活但可能误判。推荐核心业务流程用显式依赖探索性场景用隐式推断。2. 静态 DAG vs 动态规划静态 DAG执行前一次性规划效率高但无法处理执行中的条件分支。动态规划每步执行后重新评估灵活但增加编排延迟。建议确定性流程用静态 DAG用户交互场景用动态规划。3. 并行 vs 顺序DAG 中同层的并行执行能减少总耗时但并行执行增加资源竞争如多个技能同时读写同一文件。推荐I/O 密集型技能放开并行状态修改类技能保持串行。4. 失败策略整体重试 vs 局部重试整体重试整个编排计划从头执行简单但浪费已完成的计算。局部重试只重试失败节点及其下游高效但需要幂等性保证。建议设计技能时遵循幂等原则支持局部重试。5. 技能版本向前兼容 vs 破坏性升级向前兼容新版本接受旧版本的输入格式降低编排风险但限制了接口优化。破坏性升级灵活但需要全局协调升级。推荐主版本号变更才允许 Breaking Change次版本号保持向前兼容。五、总结Agent Skill 编排架构通过三层设计注册层、发现层、编排层实现了技能的系统化管理。注册层提供结构化的技能元信息发现层做意图到能力的映射编排层基于 DAG 拓扑排序执行依赖解析和并行调度。工程实践中需要在依赖声明显式化、编排策略静动态、失败处理粒度上做权衡核心原则是技能的输入输出契约必须明确技能间的依赖关系必须有向无环编排执行的结果必须可审计。