atypique-api / core /skill_schema.py
RDS777's picture
Backend v3.0 complet
8e3a425
Raw
History Blame Contribute Delete
6.79 kB
# core/skill_schema.py
"""
Skill Store – Structure de données pour les compétences persistantes.
Chaque skill est un morceau de code validé, versionné, et réutilisable.
Utilisé par :
- VectorMemory (indexation et recherche)
- FableEngine (enregistrement automatique)
- App (commandes @skill)
"""
import uuid
from datetime import datetime
from typing import List, Optional, Dict, Any
from pydantic import BaseModel, Field
# ─── Dépendances ────────────────────────────────────────────────────────────
class SkillDependency(BaseModel):
"""Dépendance d'un skill (bibliothèque Python)."""
name: str
version: Optional[str] = None # ex: ">=1.0.0", "==2.1.0"
def __str__(self) -> str:
return f"{self.name}{self.version or ''}"
def to_dict(self) -> Dict[str, Any]:
return {"name": self.name, "version": self.version}
# ─── Métadonnées ────────────────────────────────────────────────────────────
class SkillMetadata(BaseModel):
"""Métadonnées descriptives d'un skill."""
name: str
description: str
category: str # "web", "algorithm", "data", "ml", "utility"
version: str = "1.0.0"
author: str = "VORTEX"
created_at: datetime = Field(default_factory=datetime.now)
updated_at: datetime = Field(default_factory=datetime.now)
tags: List[str] = []
dependencies: List[SkillDependency] = []
test_status: str = "passed" # "passed", "failed", "untested"
def to_dict(self) -> Dict[str, Any]:
return {
"name": self.name,
"description": self.description,
"category": self.category,
"version": self.version,
"author": self.author,
"created_at": self.created_at.isoformat(),
"updated_at": self.updated_at.isoformat(),
"tags": self.tags,
"dependencies": [d.to_dict() for d in self.dependencies],
"test_status": self.test_status,
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "SkillMetadata":
"""Reconstruit un SkillMetadata à partir d'un dictionnaire."""
return cls(
name=data.get("name", "unknown"),
description=data.get("description", ""),
category=data.get("category", "general"),
version=data.get("version", "1.0.0"),
author=data.get("author", "VORTEX"),
created_at=datetime.fromisoformat(data["created_at"]) if data.get("created_at") else datetime.now(),
updated_at=datetime.fromisoformat(data["updated_at"]) if data.get("updated_at") else datetime.now(),
tags=data.get("tags", []),
dependencies=[SkillDependency(**d) for d in data.get("dependencies", [])],
test_status=data.get("test_status", "untested"),
)
# ─── Skill complet ─────────────────────────────────────────────────────────
class Skill(BaseModel):
"""Un skill complet : métadonnées + code + tests + score."""
id: str = Field(default_factory=lambda: str(uuid.uuid4())[:8])
metadata: SkillMetadata
code: str
tests: Optional[str] = None
score: float = 0.0 # Qualité (0-1)
def to_text(self) -> str:
"""Construit le texte indexé dans le RAG."""
lines = [
f"Skill: {self.metadata.name}",
f"Version: {self.metadata.version}",
f"Catégorie: {self.metadata.category}",
f"Description: {self.metadata.description}",
]
if self.metadata.tags:
lines.append(f"Tags: {', '.join(self.metadata.tags)}")
if self.metadata.dependencies:
lines.append(f"Dépendances: {', '.join(str(d) for d in self.metadata.dependencies)}")
if self.tests:
lines.append("--- Tests ---")
lines.append(self.tests[:300])
lines.append("--- Code ---")
lines.append(self.code[:500])
return "\n".join(lines)
def to_metadata(self) -> Dict[str, Any]:
"""Construit les métadonnées pour le RAG (ChromaDB)."""
base = self.metadata.to_dict()
return {
"type": "skill",
"name": base["name"],
"version": base["version"],
"category": base["category"],
"tags": ",".join(base["tags"]),
"dependencies": ",".join([str(d) for d in self.metadata.dependencies]),
"score": self.score,
"test_status": base["test_status"],
"created_at": base["created_at"],
"author": base["author"],
}
@classmethod
def from_rag(cls, doc_id: str, text: str, metadata: Dict[str, Any]) -> "Skill":
"""
Reconstruit un Skill depuis les données d'un document RAG.
"""
# Métadonnées de base
meta = SkillMetadata(
name=metadata.get("name", "unknown"),
description=metadata.get("description", ""),
category=metadata.get("category", "general"),
version=metadata.get("version", "1.0.0"),
author=metadata.get("author", "VORTEX"),
tags=metadata.get("tags", "").split(",") if metadata.get("tags") else [],
dependencies=[
SkillDependency(name=d.split(">=")[0].strip())
for d in metadata.get("dependencies", "").split(",") if d
],
test_status=metadata.get("test_status", "unknown"),
)
return cls(
id=doc_id,
metadata=meta,
code=text,
score=float(metadata.get("score", 0.0)),
tests=None,
)
def to_dict(self) -> Dict[str, Any]:
"""Sérialisation complète en dictionnaire."""
return {
"id": self.id,
"metadata": self.metadata.to_dict(),
"code": self.code,
"tests": self.tests,
"score": self.score,
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "Skill":
"""Reconstruit un Skill à partir d'un dictionnaire."""
return cls(
id=data.get("id", str(uuid.uuid4())[:8]),
metadata=SkillMetadata.from_dict(data.get("metadata", {})),
code=data.get("code", ""),
tests=data.get("tests"),
score=data.get("score", 0.0),
)
def __repr__(self) -> str:
return f"<Skill id={self.id} name={self.metadata.name} score={self.score:.2f}>"