"""Skill 基础类 - SkillConfig, IntentConfig, QualityGateConfig, Skill""" import logging from dataclasses import dataclass, field from typing import Any from agentkit.core.config_driven import AgentConfig from agentkit.core.exceptions import ConfigValidationError from agentkit.tools.base import Tool logger = logging.getLogger(__name__) @dataclass class EvolutionConfig: """Evolution configuration""" enabled: bool = False reflect_on_failure: bool = True # Whether to reflect on failed tasks auto_apply: bool = False # Whether to auto-apply optimizations (without AB test) min_quality_threshold: float = 0.5 # Minimum quality score to trigger optimization reflector_type: str = "auto" # "llm" / "rule" / "auto" auxiliary_model: str | None = None # Model name for LLM reflection @dataclass class IntentConfig: """意图配置""" keywords: list[str] = field(default_factory=list) description: str = "" examples: list[str] = field(default_factory=list) @dataclass class QualityGateConfig: """质量门控配置""" required_fields: list[str] = field(default_factory=list) min_word_count: int = 0 max_retries: int = 0 custom_validator: str | None = None class SkillConfig(AgentConfig): """扩展 AgentConfig，新增 intent、quality_gate、execution_mode 等 v2 字段 完全向后兼容：旧 YAML 无 intent/quality_gate/execution_mode 字段时自动填充默认值。 """ VALID_EXECUTION_MODES = {"react", "direct", "custom"} def __init__( self, name: str, agent_type: str, version: str = "1.0.0", description: str = "", task_mode: str = "llm_generate", supported_tasks: list[str] | None = None, max_concurrency: int = 1, input_schema: dict[str, Any] | None = None, output_schema: dict[str, Any] | None = None, prompt: dict[str, str] | None = None, llm: dict[str, Any] | None = None, tools: list[str] | None = None, memory: dict[str, Any] | None = None, custom_handler: str | None = None, # v2 新增字段 intent: dict[str, Any] | None = None, quality_gate: dict[str, Any] | None = None, execution_mode: str = "react", max_steps: int = 5, evolution: dict[str, Any] | None = None, # v3 新增字段：SKILL.md 支持 skill_md_path: str | None = None, disclosure_level: int = 0, ): super().__init__( name=name, agent_type=agent_type, version=version, description=description, task_mode=task_mode, supported_tasks=supported_tasks, max_concurrency=max_concurrency, input_schema=input_schema, output_schema=output_schema, prompt=prompt, llm=llm, tools=tools, memory=memory, custom_handler=custom_handler, ) self.intent = IntentConfig(**(intent or {})) self.quality_gate = QualityGateConfig(**(quality_gate or {})) self.execution_mode = execution_mode self.max_steps = max_steps self.evolution = EvolutionConfig(**(evolution or {})) self.skill_md_path = skill_md_path self.disclosure_level = disclosure_level self._validate_v2() def _validate_v2(self) -> None: """校验 v2 新增字段""" if self.execution_mode not in self.VALID_EXECUTION_MODES: raise ConfigValidationError( agent_name=self.name, key="execution_mode", reason=( f"Invalid execution_mode '{self.execution_mode}', " f"must be one of {self.VALID_EXECUTION_MODES}" ), ) @classmethod def from_dict(cls, data: dict[str, Any]) -> "SkillConfig": """从字典创建配置""" return cls( name=data["name"], agent_type=data["agent_type"], version=data.get("version", "1.0.0"), description=data.get("description", ""), task_mode=data.get("task_mode", "llm_generate"), supported_tasks=data.get("supported_tasks"), max_concurrency=data.get("max_concurrency", 1), input_schema=data.get("input_schema"), output_schema=data.get("output_schema"), prompt=data.get("prompt"), llm=data.get("llm"), tools=data.get("tools"), memory=data.get("memory"), custom_handler=data.get("custom_handler"), intent=data.get("intent"), quality_gate=data.get("quality_gate"), execution_mode=data.get("execution_mode", "react"), max_steps=data.get("max_steps", 5), evolution=data.get("evolution"), skill_md_path=data.get("skill_md_path"), disclosure_level=data.get("disclosure_level", 0), ) @classmethod def from_yaml(cls, path: str) -> "SkillConfig": """从 YAML 文件加载配置""" import yaml with open(path, "r", encoding="utf-8") as f: data = yaml.safe_load(f) if not isinstance(data, dict): raise ConfigValidationError( agent_name="unknown", key="config", reason=f"YAML config must be a mapping, got {type(data)}", ) return cls.from_dict(data) def to_dict(self) -> dict[str, Any]: """序列化为字典，包含 v2 字段""" d = super().to_dict() d["intent"] = { "keywords": self.intent.keywords, "description": self.intent.description, "examples": self.intent.examples, } d["quality_gate"] = { "required_fields": self.quality_gate.required_fields, "min_word_count": self.quality_gate.min_word_count, "max_retries": self.quality_gate.max_retries, "custom_validator": self.quality_gate.custom_validator, } d["execution_mode"] = self.execution_mode d["max_steps"] = self.max_steps d["evolution"] = { "enabled": self.evolution.enabled, "reflect_on_failure": self.evolution.reflect_on_failure, "auto_apply": self.evolution.auto_apply, "min_quality_threshold": self.evolution.min_quality_threshold, "reflector_type": self.evolution.reflector_type, "auxiliary_model": self.evolution.auxiliary_model, } d["skill_md_path"] = self.skill_md_path d["disclosure_level"] = self.disclosure_level return d class Skill: """Skill 封装 SkillConfig + 绑定 Tools 一个 Skill 代表一个可执行的技能，包含配置和绑定的工具。 """ def __init__(self, config: SkillConfig, tools: list[Tool] | None = None): self._config = config self._tools: list[Tool] = tools or [] @property def name(self) -> str: return self._config.name @property def config(self) -> SkillConfig: return self._config @property def tools(self) -> list[Tool]: return self._tools def bind_tool(self, tool: Tool) -> None: """绑定工具到 Skill""" self._tools.append(tool) def unbind_tool(self, tool_name: str) -> None: """解绑工具""" self._tools = [t for t in self._tools if t.name != tool_name] def to_dict(self) -> dict: """序列化为字典""" return { "config": self._config.to_dict(), "tools": [t.to_dict() for t in self._tools], }