core/engine/src/primitives/skill.py - main - alexefan136/flowstack

core/engine/src/primitives/skill.py
575 строк19 KB
Alexander Efanov
upd fix
31 июл 2026, 19:17
31 июл 2026, 19:17d146d86
"""Навыки агентов — обёртки над tools и переиспользуемые prompt templates.

Использует:
- ``StrEnum`` (Python 3.11+) для типов параметров и категорий
- Pydantic v2 (``ConfigDict``) для ``Skill`` (tool wrapper)
"""

from __future__ import annotations

import json
import re
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from enum import StrEnum
from typing import Any

from pydantic import BaseModel, ConfigDict, Field

# ============================================================================
# Skill Parameter Types
# ============================================================================


class SkillParameterType(StrEnum):
    """Типы параметров скилла."""

    TEXT = "text"
    TEXTAREA = "textarea"
    SELECT = "select"
    NUMBER = "number"
    BOOLEAN = "boolean"
    DATE = "date"
    FILE = "file"


class SkillCategory(StrEnum):
    """Категории скиллов."""

    WRITING = "writing"
    CODING = "coding"
    ANALYSIS = "analysis"
    RESEARCH = "research"
    CREATIVE = "creative"
    PRODUCTIVITY = "productivity"
    COMMUNICATION = "communication"


# ============================================================================
# Skill Parameter
# ============================================================================


@dataclass
class SkillParameter:
    """
    Параметр скилла.

    Определяет входной параметр для skill template.
    """

    name: str
    type: str  # text, textarea, select, number, boolean, date, file
    label: str
    required: bool = False
    default: Any = None
    placeholder: str | None = None
    description: str | None = None
    help_text: str | None = None
    order: int | None = None

    # Type-specific fields
    options: list[dict[str, str]] | None = None  # для select
    min: int | float | None = None  # для number
    max: int | float | None = None  # для number
    step: int | float | None = None  # для number
    rows: int | None = None  # для textarea
    min_length: int | None = None  # для text
    max_length: int | None = None  # для text
    pattern: str | None = None  # regex для text
    accept: str | None = None  # для file
    max_size: int | None = None  # для file в bytes
    multiple: bool | None = None  # для file/select

    def validate_value(self, value: Any) -> tuple[bool, str | None]:
        """
        Валидировать значение параметра.

        Returns:
            (is_valid, error_message)
        """
        if value is None:
            if self.required:
                return False, f"Parameter '{self.name}' is required"
            return True, None

        # Type validation
        if self.type == SkillParameterType.NUMBER:
            if not isinstance(value, (int, float)):
                return False, f"Parameter '{self.name}' must be a number"
            if self.min is not None and value < self.min:
                return False, f"Parameter '{self.name}' must be >= {self.min}"
            if self.max is not None and value > self.max:
                return False, f"Parameter '{self.name}' must be <= {self.max}"

        elif self.type in (SkillParameterType.TEXT, SkillParameterType.TEXTAREA):
            if not isinstance(value, str):
                return False, f"Parameter '{self.name}' must be a string"
            if self.min_length and len(value) < self.min_length:
                return False, (
                    f"Parameter '{self.name}' must be at least {self.min_length} characters"
                )
            if self.max_length and len(value) > self.max_length:
                return False, (
                    f"Parameter '{self.name}' must be at most {self.max_length} characters"
                )

        elif self.type == SkillParameterType.SELECT:
            if self.options:
                valid_values = [opt["value"] for opt in self.options]
                if value not in valid_values:
                    return False, (
                        f"Parameter '{self.name}' must be one of: {', '.join(valid_values)}"
                    )

        return True, None

    def to_dict(self) -> dict[str, Any]:
        """Преобразовать в словарь."""
        return {
            "name": self.name,
            "type": self.type,
            "label": self.label,
            "required": self.required,
            "default": self.default,
            "placeholder": self.placeholder,
            "description": self.description,
            "help_text": self.help_text,
            "order": self.order,
            "options": self.options,
            "min": self.min,
            "max": self.max,
            "step": self.step,
            "rows": self.rows,
            "min_length": self.min_length,
            "max_length": self.max_length,
            "pattern": self.pattern,
            "accept": self.accept,
            "max_size": self.max_size,
            "multiple": self.multiple,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> SkillParameter:
        """Создать из словаря."""
        return cls(
            name=data["name"],
            type=data["type"],
            label=data["label"],
            required=data.get("required", False),
            default=data.get("default"),
            placeholder=data.get("placeholder"),
            description=data.get("description"),
            help_text=data.get("help_text"),
            order=data.get("order"),
            options=data.get("options"),
            min=data.get("min"),
            max=data.get("max"),
            step=data.get("step"),
            rows=data.get("rows"),
            min_length=data.get("min_length"),
            max_length=data.get("max_length"),
            pattern=data.get("pattern"),
            accept=data.get("accept"),
            max_size=data.get("max_size"),
            multiple=data.get("multiple"),
        )


# ============================================================================
# Skill Example
# ============================================================================


@dataclass
class SkillExample:
    """
    Пример использования скилла.

    Содержит параметры и ожидаемый output для демонстрации.
    """

    title: str
    parameters: dict[str, Any]
    output_preview: str | None = None

    def to_dict(self) -> dict[str, Any]:
        """Преобразовать в словарь."""
        return {
            "title": self.title,
            "parameters": self.parameters,
            "output_preview": self.output_preview,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> SkillExample:
        """Создать из словаря."""
        return cls(
            title=data["title"],
            parameters=data["parameters"],
            output_preview=data.get("output_preview"),
        )


# ============================================================================
# Skill (Tool Wrapper)
# ============================================================================


class Skill(BaseModel):
    """
    Навык, доступный агенту.

    Простая обёртка над tool с handler функцией.
    Используется для прямого вызова из агентов.

    Для prompt templates используйте SkillTemplate.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    name: str
    description: str
    parameters: dict[str, Any] = Field(default_factory=dict)
    handler: Callable[..., Awaitable[Any]] | None = Field(default=None, exclude=True)

    async def execute(self, **kwargs: Any) -> Any:
        """Выполнить навык с переданными параметрами."""
        if self.handler is None:
            raise RuntimeError(f"Skill '{self.name}' has no handler")
        return await self.handler(**kwargs)


# ============================================================================
# Skill Template (Prompt Template)
# ============================================================================


@dataclass
class SkillTemplate:
    """
    Шаблон скилла — переиспользуемый prompt template с параметрами.

    Используется для создания задач, запуска через UI, агентами.
    Параметры подставляются в шаблон через {{variable_name}}.
    """

    id: str
    name: str
    prompt_template: str
    description: str = ""
    icon: str = "Sparkles"
    category: str = SkillCategory.PRODUCTIVITY

    # Параметры
    parameters: list[SkillParameter] = field(default_factory=list)

    # Конфигурация
    default_model: str = "deepseek-ai/DeepSeek-V4-Pro"
    agent_compatible: bool = True

    # Метаданные
    tags: list[str] = field(default_factory=list)
    is_public: bool = False
    version: int = 1
    complexity: str | None = None  # simple, intermediate, advanced
    expected_duration_seconds: int | None = None

    # Примеры
    examples: list[SkillExample] = field(default_factory=list)

    # Статистика
    usage_count: int = 0
    average_rating: float | None = None

    def render(self, parameters: dict[str, Any], strict: bool = False) -> str:
        """
        Отрендерить шаблон с параметрами.

        Заменяет {{variable_name}} на значения из parameters.

        Args:
            parameters: значения параметров
            strict: если True — бросает ValueError при отсутствии
                обязательного (required) параметра. Если False —
                оставляет {{var}} как есть.

        Raises:
            ValueError: в strict-режиме при отсутствии required-параметра.
        """

        def replace_var(match: re.Match[str]) -> str:
            var_name = match.group(1)
            value = parameters.get(var_name)

            if value is None:
                if strict:
                    param = self.get_parameter(var_name)
                    if param and param.required:
                        raise ValueError(f"Required parameter '{var_name}' is missing")
                return match.group(0)  # Оставляем как есть

            # Обработка разных типов
            if isinstance(value, list):
                return ", ".join(str(v) for v in value)
            if isinstance(value, dict):
                return json.dumps(value, indent=2, ensure_ascii=False)
            return str(value)

        return re.sub(r"\{\{(\w+)\}\}", replace_var, self.prompt_template)

    def validate_parameters(self, parameters: dict[str, Any]) -> list[str]:
        """
        Валидировать параметры против определения.

        Returns:
            Список ошибок (пустой если всё ок).
        """
        errors: list[str] = []
        for param in self.parameters:
            value = parameters.get(param.name)
            is_valid, error = param.validate_value(value)
            if not is_valid and error:
                errors.append(error)
        return errors

    def get_parameter(self, name: str) -> SkillParameter | None:
        """Получить параметр по имени."""
        return next((p for p in self.parameters if p.name == name), None)

    def get_required_parameters(self) -> list[SkillParameter]:
        """Получить список обязательных параметров."""
        return [p for p in self.parameters if p.required]

    def get_optional_parameters(self) -> list[SkillParameter]:
        """Получить список опциональных параметров."""
        return [p for p in self.parameters if not p.required]

    def add_parameter(self, param: SkillParameter) -> None:
        """Добавить параметр."""
        self.parameters.append(param)

    def remove_parameter(self, name: str) -> bool:
        """Удалить параметр по имени."""
        param = self.get_parameter(name)
        if param:
            self.parameters.remove(param)
            return True
        return False

    def add_example(self, example: SkillExample) -> None:
        """Добавить пример."""
        self.examples.append(example)

    def remove_example(self, title: str) -> bool:
        """Удалить пример по заголовку."""
        example = next((e for e in self.examples if e.title == title), None)
        if example:
            self.examples.remove(example)
            return True
        return False

    def add_tag(self, tag: str) -> None:
        """Добавить тег."""
        if tag not in self.tags:
            self.tags.append(tag)

    def remove_tag(self, tag: str) -> None:
        """Удалить тег."""
        if tag in self.tags:
            self.tags.remove(tag)

    def has_tag(self, tag: str) -> bool:
        """Проверить наличие тега."""
        return tag in self.tags

    def increment_usage(self) -> None:
        """Инкрементировать счётчик использования."""
        self.usage_count += 1

    def update_rating(self, new_rating: float) -> None:
        """Обновить средний рейтинг (экспоненциальное сглаживание)."""
        if self.average_rating is None:
            self.average_rating = new_rating
        else:
            alpha = 0.1
            self.average_rating = alpha * new_rating + (1 - alpha) * self.average_rating

    def to_dict(self) -> dict[str, Any]:
        """Преобразовать в словарь."""
        return {
            "id": self.id,
            "name": self.name,
            "prompt_template": self.prompt_template,
            "description": self.description,
            "icon": self.icon,
            "category": self.category,
            "parameters": [p.to_dict() for p in self.parameters],
            "default_model": self.default_model,
            "agent_compatible": self.agent_compatible,
            "tags": self.tags,
            "is_public": self.is_public,
            "version": self.version,
            "complexity": self.complexity,
            "expected_duration_seconds": self.expected_duration_seconds,
            "examples": [e.to_dict() for e in self.examples],
            "usage_count": self.usage_count,
            "average_rating": self.average_rating,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> SkillTemplate:
        """Создать из словаря."""
        return cls(
            id=data["id"],
            name=data["name"],
            prompt_template=data["prompt_template"],
            description=data.get("description", ""),
            icon=data.get("icon", "Sparkles"),
            category=data.get("category", SkillCategory.PRODUCTIVITY),
            parameters=[SkillParameter.from_dict(p) for p in data.get("parameters", [])],
            default_model=data.get("default_model", "deepseek-ai/DeepSeek-V4-Pro"),
            agent_compatible=data.get("agent_compatible", True),
            tags=data.get("tags", []),
            is_public=data.get("is_public", False),
            version=data.get("version", 1),
            complexity=data.get("complexity"),
            expected_duration_seconds=data.get("expected_duration_seconds"),
            examples=[SkillExample.from_dict(e) for e in data.get("examples", [])],
            usage_count=data.get("usage_count", 0),
            average_rating=data.get("average_rating"),
        )


# ============================================================================
# Helper Functions
# ============================================================================


def create_skill_parameter(
    name: str,
    type: str,
    label: str,
    required: bool = False,
    **kwargs: Any,
) -> SkillParameter:
    """Создать параметр скилла."""
    return SkillParameter(
        name=name,
        type=type,
        label=label,
        required=required,
        **kwargs,
    )


def create_text_parameter(
    name: str,
    label: str,
    required: bool = False,
    placeholder: str | None = None,
    **kwargs: Any,
) -> SkillParameter:
    """Создать текстовый параметр."""
    return SkillParameter(
        name=name,
        type=SkillParameterType.TEXT,
        label=label,
        required=required,
        placeholder=placeholder,
        **kwargs,
    )


def create_textarea_parameter(
    name: str,
    label: str,
    required: bool = False,
    rows: int = 4,
    **kwargs: Any,
) -> SkillParameter:
    """Создать многострочный текстовый параметр."""
    return SkillParameter(
        name=name,
        type=SkillParameterType.TEXTAREA,
        label=label,
        required=required,
        rows=rows,
        **kwargs,
    )


def create_select_parameter(
    name: str,
    label: str,
    options: list[dict[str, str]],
    required: bool = False,
    **kwargs: Any,
) -> SkillParameter:
    """Создать параметр-селектор."""
    return SkillParameter(
        name=name,
        type=SkillParameterType.SELECT,
        label=label,
        required=required,
        options=options,
        **kwargs,
    )


def create_number_parameter(
    name: str,
    label: str,
    required: bool = False,
    min: int | float | None = None,
    max: int | float | None = None,
    **kwargs: Any,
) -> SkillParameter:
    """Создать числовой параметр."""
    return SkillParameter(
        name=name,
        type=SkillParameterType.NUMBER,
        label=label,
        required=required,
        min=min,
        max=max,
        **kwargs,
    )


def create_boolean_parameter(
    name: str,
    label: str,
    required: bool = False,
    default: bool = False,
    **kwargs: Any,
) -> SkillParameter:
    """Создать булевый параметр."""
    return SkillParameter(
        name=name,
        type=SkillParameterType.BOOLEAN,
        label=label,
        required=required,
        default=default,
        **kwargs,
    )


# ============================================================================
# Exports
# ============================================================================

__all__ = [
    "Skill",
    "SkillCategory",
    "SkillExample",
    "SkillParameter",
    "SkillParameterType",
    "SkillTemplate",
    "create_boolean_parameter",
    "create_number_parameter",
    "create_select_parameter",
    "create_skill_parameter",
    "create_text_parameter",
    "create_textarea_parameter",
]