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

core/engine/src/schemas/skill.py
168 строк5 KB
Alexander Efanov
upd fix
31 июл 2026, 19:17
31 июл 2026, 19:17d146d86
"""Pydantic schemas для скиллов."""

from __future__ import annotations

from datetime import datetime
from typing import Any, Literal
from uuid import UUID

from pydantic import BaseModel, ConfigDict, Field

# ============================================================================
# Types
# ============================================================================

SkillParameterType = Literal[
    "text",
    "textarea",
    "select",
    "number",
    "boolean",
    "date",
    "file",
]
SkillCategory = Literal[
    "writing",
    "coding",
    "analysis",
    "research",
    "creative",
    "productivity",
    "communication",
]
SkillComplexity = Literal["simple", "intermediate", "advanced"]


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


class SkillParameterSchema(BaseModel):
    """Параметр скилла (элемент JSONB ``Skill.parameters``)."""

    name: str = Field(..., min_length=1)
    type: SkillParameterType
    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
    options: list[dict[str, str]] | None = None  # для select
    min: float | None = None  # для number
    max: float | None = None  # для number
    step: float | None = None  # для number
    rows: int | None = None  # для textarea
    min_length: int | None = None  # для text
    max_length: int | None = None  # для text


class SkillExampleSchema(BaseModel):
    """Пример использования скилла (элемент JSONB ``Skill.examples``)."""

    title: str = Field(..., min_length=1)
    parameters: dict[str, Any] = Field(default_factory=dict)
    output_preview: str | None = None


# ============================================================================
# Skill
# ============================================================================


class SkillCreate(BaseModel):
    """
    Запрос на создание скилла.

    Note: ``prompt_template`` обязателен (в модели ``nullable=False``).
    """

    name: str = Field(..., min_length=1, max_length=200)
    prompt_template: str = Field(..., min_length=1)
    description: str | None = Field(default=None, max_length=2000)
    icon: str = Field(default="Sparkles", max_length=50)
    category: SkillCategory = "productivity"
    parameters: list[SkillParameterSchema] = Field(default_factory=list)
    default_model: str = Field(default="deepseek-ai/DeepSeek-V4-Pro", max_length=100)
    agent_compatible: bool = True
    tags: list[str] = Field(default_factory=list)
    is_public: bool = False
    complexity: SkillComplexity | None = None
    expected_duration_seconds: int | None = Field(default=None, ge=0)
    examples: list[SkillExampleSchema] = Field(default_factory=list)


class SkillUpdate(BaseModel):
    """Запрос на обновление скилла (все поля опциональны)."""

    name: str | None = Field(default=None, min_length=1, max_length=200)
    prompt_template: str | None = Field(default=None, min_length=1)
    description: str | None = Field(default=None, max_length=2000)
    icon: str | None = Field(default=None, max_length=50)
    category: SkillCategory | None = None
    parameters: list[SkillParameterSchema] | None = None
    default_model: str | None = Field(default=None, max_length=100)
    agent_compatible: bool | None = None
    tags: list[str] | None = None
    is_public: bool | None = None
    complexity: SkillComplexity | None = None
    expected_duration_seconds: int | None = Field(default=None, ge=0)
    examples: list[SkillExampleSchema] | None = None
    is_archived: bool | None = None


class SkillResponse(BaseModel):
    """Ответ с информацией о скилле."""

    model_config = ConfigDict(from_attributes=True)

    id: UUID
    workspace_id: str
    name: str
    prompt_template: str
    description: str | None
    icon: str
    category: str
    parameters: list[dict[str, Any]] = Field(default_factory=list)
    default_model: str
    agent_compatible: bool
    tags: list[str]
    is_public: bool
    version: int = 1
    complexity: str | None = None
    expected_duration_seconds: int | None = None
    examples: list[dict[str, Any]] = Field(default_factory=list)

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

    created_at: datetime
    updated_at: datetime


# ============================================================================
# Skill Run
# ============================================================================


class SkillRunRequest(BaseModel):
    """Запрос на запуск скилла."""

    parameters: dict[str, Any] = Field(default_factory=dict)
    model: str | None = Field(default=None, max_length=100)
    stream: bool = Field(default=False)


class SkillRunResponse(BaseModel):
    """Ответ с результатом запуска скилла (non-streaming)."""

    skill_id: UUID
    output: str
    rendered_prompt: str | None = None
    tokens_used: int = 0
    duration_ms: float = 0.0