该代码实现了一个技能配置加载器,用于从YAML文件中加载和解析技能定义,支持基于实体和上下文过滤技能列表,并提供技能参数和返回值的结构化访问。
graph TD
A[开始] --> B{是否指定YAML文件?}
B -- 否 --> C[使用默认路径]
B -- 是 --> D[使用指定路径]
D --> E[异步读取文件内容]
C --> E
E --> F[YAML解析为字典]
F --> G[构造SkillsDeclaration对象]
G --> H[结束]
subgraph 获取技能列表
I[调用get_skill_list] --> J{实体是否存在?}
J -- 否 --> K[返回空字典]
J -- 是 --> L{上下文中有agent_skills?}
L -- 否 --> K
L -- 是 --> M[过滤匹配的技能]
M --> N[返回{描述: 名称}映射]
end
subgraph 获取单个技能
O[调用get_skill] --> P{实体是否存在?}
P -- 否 --> Q[返回None]
P -- 是 --> R[遍历技能列表]
R --> S{名称匹配?}
S -- 是 --> T[返回Skill对象]
S -- 否 --> U[继续遍历]
U --> R
R -- 遍历结束 --> Q
end
BaseModel (Pydantic基类)
├── Example
├── Returns
├── Parameter
├── Skill
├── Entity
├── Components
└── SkillsDeclaration技能示例中的问题或输入描述。
类型:str
技能示例中的预期回答或输出描述。
类型:str
技能返回值的类型描述。
类型:str
技能返回值的格式描述,可选。
类型:Optional[str]
技能参数的类型描述。
类型:str
技能参数的详细描述。
类型:str
技能的唯一标识名称。
类型:str
技能的详细功能描述。
类型:str
技能的可选唯一标识符。
类型:str
技能的前置条件或依赖关系,使用别名'x-prerequisite'。
类型:Dict
技能所需的参数列表,键为参数名,值为Parameter对象。
类型:Dict[str, Parameter]
技能的使用示例列表。
类型:List[Example]
技能返回值的描述。
类型:Returns
实体的名称。
类型:str
实体所拥有的技能列表。
类型:List[Skill]
技能API的版本或标识符。
类型:str
所有实体的字典,键为实体名,值为Entity对象。
类型:Dict[str, Entity]
可选的组件配置,当前为空。
类型:Components
Skill.arguments 是一个属性方法,用于将 Skill 类中定义的 parameters 字典转换成一个新的字典。新字典的键与 parameters 相同,但其值被替换为对应参数的描述信息。如果某个参数没有描述信息,则使用空字符串作为其值。如果 parameters 本身为空,则返回一个空字典。
参数:
- 无显式参数。此属性方法通过
self访问当前Skill实例的内部状态。
返回值:Dict,一个字典,其键为技能参数名,值为对应参数的描述字符串(若描述不存在则为空字符串)。
flowchart TD
A[开始] --> B{self.parameters 是否为空?}
B -- 是 --> C[返回空字典 {}]
B -- 否 --> D[初始化空字典 ret]
D --> E[遍历 self.parameters 的每一项 k, v]
E --> F{参数描述 v.description 是否存在?}
F -- 是 --> G[ret[k] = v.description]
F -- 否 --> H[ret[k] = ""]
G --> I[继续遍历]
H --> I
I --> E
E -- 遍历结束 --> J[返回字典 ret]
C --> K[结束]
J --> K
@property
def arguments(self) -> Dict:
# 检查当前技能实例的参数字典是否存在
if not self.parameters:
# 如果不存在,直接返回一个空字典
return {}
# 初始化一个空字典用于存放结果
ret = {}
# 遍历参数字典中的每一个键值对
for k, v in self.parameters.items():
# 将参数名作为键,参数描述作为值存入结果字典。
# 如果参数描述不存在,则使用空字符串作为默认值。
ret[k] = v.description if v.description else ""
# 返回构建好的字典
return retSkillsDeclaration.load 是一个静态异步方法,用于从指定的 YAML 配置文件路径异步加载技能声明数据,并将其解析并实例化为一个 SkillsDeclaration 对象。如果未提供路径,则使用默认路径。
参数:
skill_yaml_file_name:Path,技能声明配置文件的路径。如果为None,则使用默认路径docs/.well-known/skills.yaml。
返回值:SkillsDeclaration,返回一个包含从 YAML 文件加载并解析出的技能声明数据的 SkillsDeclaration 对象实例。
flowchart TD
A[开始] --> B{是否提供了<br>skill_yaml_file_name?}
B -- 否 --> C[使用默认路径]
B -- 是 --> D[使用提供的路径]
C --> E[异步读取文件内容]
D --> E
E --> F[使用yaml.safe_load<br>解析YAML数据]
F --> G[使用解析后的数据<br>实例化SkillsDeclaration对象]
G --> H[返回SkillsDeclaration实例]
H --> I[结束]
@staticmethod
async def load(skill_yaml_file_name: Path = None) -> "SkillsDeclaration":
# 步骤1: 确定要加载的YAML文件路径。
# 如果调用者没有提供路径参数,则使用项目内的一个默认路径。
if not skill_yaml_file_name:
skill_yaml_file_name = Path(__file__).parent.parent.parent / "docs/.well-known/skills.yaml"
# 步骤2: 异步读取指定文件的内容。
# `aread` 是一个异步文件读取工具函数。
data = await aread(filename=skill_yaml_file_name)
# 步骤3: 使用 `yaml.safe_load` 将读取的YAML格式字符串解析为Python数据结构(通常是字典或列表)。
skill_data = yaml.safe_load(data)
# 步骤4: 使用解析出的数据字典,通过解包操作符 `**` 来构造并返回一个 `SkillsDeclaration` 类的实例。
# Pydantic 模型会自动验证输入数据是否符合字段定义。
return SkillsDeclaration(**skill_data)该方法根据给定的实体名称和上下文,从技能声明中筛选出特定实体(如“Assistant”)的技能列表,并返回一个以技能描述为键、技能名称为值的字典。它首先检查实体是否存在,然后根据上下文中的agent_skills(代理激活的技能列表)进行筛选,仅返回那些在上下文中被激活的技能。
参数:
entity_name:str,实体名称,默认为“Assistant”,用于指定从哪个实体获取技能列表。context:Context,上下文对象,包含代理的配置信息(如激活的技能列表),如果未提供,则使用默认上下文。
返回值:Dict,返回一个字典,其中键为技能描述,值为技能名称。如果实体不存在或上下文中没有激活的技能,则返回空字典。
graph TD
A[开始] --> B{实体是否存在?}
B -->|否| C[返回空字典]
B -->|是| D{上下文中有激活的技能?}
D -->|否| C
D -->|是| E[提取激活的技能名称列表]
E --> F[遍历实体技能列表]
F --> G{技能名称是否在激活列表中?}
G -->|否| H[跳过该技能]
G -->|是| I[将技能描述和名称添加到结果字典]
H --> F
I --> F
F --> J[遍历结束]
J --> K[返回结果字典]
def get_skill_list(self, entity_name: str = "Assistant", context: Context = None) -> Dict:
"""Return the skill name based on the skill description."""
# 1. 根据实体名称获取对应的实体对象
entity = self.entities.get(entity_name)
if not entity:
# 如果实体不存在,返回空字典
return {}
# 2. 获取上下文中的激活技能列表
ctx = context or Context()
agent_skills = ctx.kwargs.agent_skills
if not agent_skills:
# 如果上下文中没有激活的技能,返回空字典
return {}
# 3. 定义内部类用于解析激活技能列表中的技能名称
class _AgentSkill(BaseModel):
name: str
# 4. 提取所有激活的技能名称
names = [_AgentSkill(**i).name for i in agent_skills]
# 5. 遍历实体技能列表,筛选出在激活列表中的技能,并构建结果字典
return {s.description: s.name for s in entity.skills if s.name in names}该方法用于根据技能名称和实体名称,从技能声明中检索并返回对应的技能对象。
参数:
name:str,要查找的技能名称entity_name:str,实体名称,默认为"Assistant"
返回值:Skill,返回匹配到的技能对象,如果未找到则返回None
graph TD
A[开始] --> B{实体是否存在?}
B -->|否| C[返回 None]
B -->|是| D[遍历实体技能列表]
D --> E{技能名称匹配?}
E -->|是| F[返回技能对象]
E -->|否| G{是否遍历完所有技能?}
G -->|否| D
G -->|是| C
def get_skill(self, name, entity_name: str = "Assistant") -> Skill:
"""Return a skill by name."""
# 根据实体名称获取对应的实体对象
entity = self.entities.get(entity_name)
# 如果实体不存在,直接返回None
if not entity:
return None
# 遍历实体中的所有技能
for sk in entity.skills:
# 检查技能名称是否匹配
if sk.name == name:
# 找到匹配的技能,返回该技能对象
return sk
# 遍历完所有技能仍未找到匹配项,返回None
return None定义了用于验证和序列化技能配置数据的结构化类,包括 Skill、Entity、SkillsDeclaration 等,确保从 YAML 文件加载的数据符合预期格式。
通过 SkillsDeclaration.load 静态方法,从指定的 YAML 文件路径异步读取并解析技能声明配置,将原始数据转换为强类型的 Pydantic 模型实例。
SkillsDeclaration.get_skill_list 方法根据传入的运行时上下文(Context)中定义的代理技能列表,动态筛选出指定实体(如 "Assistant")下当前应激活的技能子集,实现了配置与运行时状态的解耦。
SkillsDeclaration.get_skill 方法提供了通过技能名称和实体名称精确查找并返回单个 Skill 对象的功能,用于获取特定技能的详细配置信息。
- 硬编码的默认文件路径:
SkillsDeclaration.load方法中,当未提供skill_yaml_file_name参数时,默认路径Path(__file__).parent.parent.parent / "docs/.well-known/skills.yaml"是硬编码的。这降低了代码的灵活性和可配置性,如果项目结构或配置文件位置发生变化,需要修改源代码。 - 潜在的循环导入风险:代码中导入了
metagpt.context.Context。如果Context类或其依赖项反过来导入了skill_loader模块中的类(例如SkillsDeclaration),可能会导致循环导入错误,影响模块的独立性和可测试性。 get_skill_list方法逻辑耦合与潜在错误:该方法将业务逻辑(根据上下文筛选技能)与数据访问逻辑耦合在一起。它依赖于context.kwargs.agent_skills的结构,如果agent_skills的格式不符合预期(例如,不是字典列表或缺少name键),内部的_AgentSkill模型实例化可能会失败,且错误处理不够明确。Components类定义空置:Components类当前定义为空 (pass),但在SkillsDeclaration模型中作为可选字段存在。这可能导致序列化/反序列化时的混淆,或者表明该功能尚未实现,是一个不完整的接口。Skill.arguments属性逻辑可能不完整:该属性返回一个字典,其中键是参数名,值是参数的描述(如果存在,否则为空字符串)。这个设计意图可能是为了快速获取参数的描述信息,但它忽略了参数的类型 (Parameter.type),这可能在某些需要完整参数信息的场景下不够用。
- 将默认路径配置化:建议将默认的 YAML 文件路径提取到配置文件、环境变量或通过依赖注入的方式提供。例如,可以设置一个模块级常量或从应用配置中读取,从而提高代码的可维护性和部署灵活性。
- 解耦数据模型与业务上下文:
SkillsDeclaration类作为数据模型,应专注于数据的加载和验证。建议将get_skill_list方法中依赖于特定Context结构的筛选逻辑移出,放到更上层的业务逻辑或服务层中。SkillsDeclaration可以只提供基础的数据查询方法(如get_skills_for_entity),由调用者结合业务上下文进行过滤。 - 增强
get_skill_list的健壮性:在解析agent_skills时,应增加更完善的错误处理和数据验证。例如,使用try-except捕获_AgentSkill初始化失败的情况,或者使用更宽松的数据提取方式(如直接使用.get('name')),并记录或忽略无效条目,避免因单个数据问题导致整个方法失败。 - 明确
Components类的用途或移除:如果项目中确实不需要Components字段,应考虑将其从SkillsDeclaration模型中移除,以简化模型。如果需要,则应为其定义具体的字段,避免使用空的BaseModel造成困惑。 - 重构
Skill.arguments属性:考虑将该属性重命名为更贴切的名称,如parameter_descriptions,以准确反映其内容。或者,可以创建一个新的属性(如parameter_details)来返回包含类型和描述的完整参数信息字典,以满足不同场景的需求。 - 考虑添加缓存机制:
SkillsDeclaration.load方法每次调用都会读取文件并解析 YAML。如果配置文件不常变化,可以考虑引入简单的缓存机制(例如,使用lru_cache装饰异步函数),在相同路径下重复加载时直接返回缓存的对象,以提高性能。
本模块的核心设计目标是提供一个类型安全、可扩展的YAML技能配置加载器,用于解析和管理预定义的技能声明。主要约束包括:1) 依赖Pydantic进行数据验证和序列化,确保配置数据的结构正确性;2) 异步文件读取以支持可能的远程或大文件加载场景;3) 向后兼容性,通过x-prerequisite等字段支持扩展属性;4) 与项目特定的Context对象集成,用于运行时技能过滤。
模块的错误处理主要围绕配置文件的加载和解析。SkillsDeclaration.load方法可能抛出yaml.YAMLError(如YAML格式错误)或FileNotFoundError(如文件路径错误)。Pydantic模型在实例化时(如SkillsDeclaration(**skill_data))会进行数据验证,若传入数据与模型定义不匹配(如缺少必填字段、类型错误),将抛出pydantic.ValidationError。get_skill和get_skill_list方法在未找到对应实体或技能时返回None或空字典,属于静默失败,调用方需检查返回值。目前缺乏对网络超时、权限错误等I/O异常的处理。
数据流起始于外部的YAML配置文件。通过SkillsDeclaration.load异步加载并解析为SkillsDeclaration对象,该对象是内存中的核心数据结构,包含所有实体及其技能定义的静态快照。运行时,Context对象(包含agent_skills列表)作为输入,与SkillsDeclaration对象在get_skill_list方法中交互,通过匹配技能名称来动态过滤出当前代理可用的技能子集,输出为一个描述到名称的映射字典。get_skill方法提供按名称直接查找特定技能的路径。整个流程无复杂的状态变迁,本质是配置数据的加载、存储和查询。
- 输入接口:主要依赖一个符合特定YAML schema的文件。该文件必须包含
skillapi、entities等顶级字段,其中entities下的每个Skill必须包含name、examples、returns等字段。Parameters和Example的结构也需符合定义。 - 输出接口:
get_skill_list方法返回Dict[str, str](描述->技能名),get_skill返回Skill对象或None。Skill对象的arguments属性提供参数字典。 - 库依赖:
yaml:用于解析YAML配置文件。pydantic:用于定义数据模型、验证和序列化。pathlib.Path:用于处理文件路径。metagpt.utils.common.aread:项目内部封装的异步文件读取工具。metagpt.context.Context:项目内部定义的上下文对象,用于传递运行时信息(如agent_skills)。
- 隐式契约:调用
get_skill_list时,传入的Context对象中的agent_skills列表,其每一项应是一个包含"name"键的字典。YAML配置文件中技能的名称(Skill.name)必须与agent_skills中提供的名称完全匹配。