210 lines
6.9 KiB
Python
210 lines
6.9 KiB
Python
"""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
|
||
|
||
|
||
@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,
|
||
):
|
||
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._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"),
|
||
)
|
||
|
||
@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,
|
||
}
|
||
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],
|
||
}
|