300 lines
11 KiB
Python
300 lines
11 KiB
Python
"""Skill 基础类 - SkillConfig, IntentConfig, QualityGateConfig, Skill"""
|
||
|
||
from __future__ import annotations
|
||
|
||
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.skills.schema import CapabilityTag, DependencyDecl
|
||
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
|
||
optimizer_type: str = "auto" # "llm" / "bootstrap" / "auto"
|
||
strategy_tuning_enabled: bool = False # Whether to enable strategy tuning
|
||
ab_test_min_samples: int = 10 # Minimum samples for A/B test significance
|
||
|
||
|
||
@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", "rewoo", "plan_exec", "reflexion"}
|
||
|
||
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,
|
||
# v4 新增字段：依赖声明、能力标签
|
||
dependencies: list[dict[str, Any] | DependencyDecl] | None = None,
|
||
capabilities: list[str | dict[str, Any] | CapabilityTag] | None = None,
|
||
):
|
||
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
|
||
# v4: 解析依赖和能力标签
|
||
self.dependencies = self._parse_dependencies(dependencies or [])
|
||
self.capabilities = self._parse_capabilities(capabilities or [])
|
||
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}"
|
||
),
|
||
)
|
||
|
||
@staticmethod
|
||
def _parse_dependencies(
|
||
raw: list[dict[str, Any] | DependencyDecl],
|
||
) -> list[DependencyDecl]:
|
||
"""解析依赖声明列表，支持 dict 或 DependencyDecl 实例"""
|
||
result: list[DependencyDecl] = []
|
||
for item in raw:
|
||
if isinstance(item, DependencyDecl):
|
||
result.append(item)
|
||
elif isinstance(item, dict):
|
||
result.append(DependencyDecl(**item))
|
||
else:
|
||
logger.warning(f"Skipping invalid dependency declaration: {item}")
|
||
return result
|
||
|
||
@staticmethod
|
||
def _parse_capabilities(
|
||
raw: list[str | dict[str, Any] | CapabilityTag],
|
||
) -> list[CapabilityTag]:
|
||
"""解析能力标签列表，支持 str / dict / CapabilityTag 实例"""
|
||
result: list[CapabilityTag] = []
|
||
for item in raw:
|
||
if isinstance(item, CapabilityTag):
|
||
result.append(item)
|
||
elif isinstance(item, str):
|
||
result.append(CapabilityTag(tag=item))
|
||
elif isinstance(item, dict):
|
||
result.append(CapabilityTag(**item))
|
||
else:
|
||
logger.warning(f"Skipping invalid capability declaration: {item}")
|
||
return result
|
||
|
||
@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),
|
||
dependencies=data.get("dependencies"),
|
||
capabilities=data.get("capabilities"),
|
||
)
|
||
|
||
@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,
|
||
"optimizer_type": self.evolution.optimizer_type,
|
||
"strategy_tuning_enabled": self.evolution.strategy_tuning_enabled,
|
||
"ab_test_min_samples": self.evolution.ab_test_min_samples,
|
||
}
|
||
d["skill_md_path"] = self.skill_md_path
|
||
d["disclosure_level"] = self.disclosure_level
|
||
# v4: 序列化依赖和能力标签
|
||
d["dependencies"] = [
|
||
{
|
||
"name": dep.name,
|
||
"version_constraint": dep.version_constraint,
|
||
"type": dep.type,
|
||
"required": dep.required,
|
||
}
|
||
for dep in self.dependencies
|
||
]
|
||
d["capabilities"] = [
|
||
{"tag": cap.tag, "description": cap.description}
|
||
for cap in self.capabilities
|
||
]
|
||
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 version(self) -> str:
|
||
return self._config.version
|
||
|
||
@property
|
||
def config(self) -> SkillConfig:
|
||
return self._config
|
||
|
||
@property
|
||
def tools(self) -> list[Tool]:
|
||
return self._tools
|
||
|
||
@property
|
||
def capabilities(self) -> list:
|
||
"""返回 Skill 的能力标签列表"""
|
||
return self._config.capabilities
|
||
|
||
@property
|
||
def dependencies(self) -> list:
|
||
"""返回 Skill 的依赖声明列表"""
|
||
return self._config.dependencies
|
||
|
||
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],
|
||
}
|