feat(vwb): expose competence yaml catalog

This commit is contained in:
Dom
2026-05-29 11:28:25 +02:00
parent e8a0fb0e42
commit c1a144c673
4 changed files with 302 additions and 3 deletions

View File

@@ -0,0 +1,14 @@
"""Competence catalogue helpers."""
from .catalog import (
CompetenceSummary,
load_competence_catalog_actions,
load_competences,
)
__all__ = [
"CompetenceSummary",
"load_competence_catalog_actions",
"load_competences",
]

201
core/competences/catalog.py Normal file
View File

@@ -0,0 +1,201 @@
"""Load Lea competence YAML files as runtime catalogue entries."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable
import yaml
REPO_ROOT = Path(__file__).resolve().parents[2]
DEFAULT_COMPETENCE_ROOT = REPO_ROOT / "data" / "competences"
KNOWN_STATES = ("candidate", "supervised", "stable", "observed")
@dataclass(frozen=True)
class CompetenceSummary:
"""Small, UI-safe projection of a persisted competence YAML."""
id: str
name: str
learning_state: str
intent_fr: str
source_path: str
methods: tuple[dict[str, Any], ...]
success_marker: dict[str, Any]
failure_message_template: dict[str, Any]
t2_known_gaps: tuple[dict[str, Any], ...]
def to_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"name": self.name,
"learning_state": self.learning_state,
"intent_fr": self.intent_fr,
"source_path": self.source_path,
"methods": list(self.methods),
"success_marker": self.success_marker,
"failure_message_template": self.failure_message_template,
"t2_known_gaps": list(self.t2_known_gaps),
}
def load_competences(
*,
root: Path | str = DEFAULT_COMPETENCE_ROOT,
states: Iterable[str] | None = None,
) -> list[CompetenceSummary]:
"""Load all competence YAML files under ``data/competences``.
``states`` filters by directory/``learning_state`` value. Returned entries
are sorted by state maturity first, then by id, to make catalogue output
deterministic.
"""
competence_root = Path(root)
state_filter = set(states or KNOWN_STATES)
summaries: list[CompetenceSummary] = []
for state in KNOWN_STATES:
if state not in state_filter:
continue
state_dir = competence_root / state
if not state_dir.exists():
continue
for path in sorted(state_dir.glob("*.yaml")):
summary = load_competence_file(path, repo_root=REPO_ROOT)
if summary.learning_state in state_filter:
summaries.append(summary)
return sorted(summaries, key=lambda item: (KNOWN_STATES.index(item.learning_state), item.id))
def load_competence_file(path: Path | str, *, repo_root: Path = REPO_ROOT) -> CompetenceSummary:
competence_path = Path(path)
with competence_path.open("r", encoding="utf-8") as handle:
data = yaml.safe_load(handle) or {}
if not isinstance(data, dict):
raise ValueError(f"{competence_path} must contain a YAML mapping")
competence_id = _required_text(data, "id", competence_path)
learning_state = _required_text(data, "learning_state", competence_path)
name = str(data.get("name") or competence_id)
intent = data.get("intent") if isinstance(data.get("intent"), dict) else {}
intent_fr = str(intent.get("fr") or name)
methods = _method_summaries(data.get("methods"))
success_marker = data.get("success_marker") if isinstance(data.get("success_marker"), dict) else {}
failure_template = (
data.get("failure_message_template")
if isinstance(data.get("failure_message_template"), dict)
else {}
)
promotion = data.get("promotion") if isinstance(data.get("promotion"), dict) else {}
gaps = promotion.get("t2_known_gaps") if isinstance(promotion.get("t2_known_gaps"), list) else []
try:
source_path = str(competence_path.resolve().relative_to(repo_root.resolve()))
except ValueError:
source_path = str(competence_path)
return CompetenceSummary(
id=competence_id,
name=name,
learning_state=learning_state,
intent_fr=intent_fr,
source_path=source_path,
methods=tuple(methods),
success_marker=success_marker,
failure_message_template=failure_template,
t2_known_gaps=tuple(gap for gap in gaps if isinstance(gap, dict)),
)
def load_competence_catalog_actions(
*,
root: Path | str = DEFAULT_COMPETENCE_ROOT,
states: Iterable[str] | None = ("candidate", "supervised", "stable"),
) -> list[dict[str, Any]]:
"""Expose competences in the VWB action-catalogue shape."""
return [competence_to_catalog_action(item) for item in load_competences(root=root, states=states)]
def competence_to_catalog_action(summary: CompetenceSummary) -> dict[str, Any]:
method_labels = ", ".join(
str(method.get("kind") or method.get("primitive_ref") or method.get("id"))
for method in summary.methods
)
description = f"Compétence Léa {summary.learning_state}: {summary.intent_fr}"
if method_labels:
description = f"{description} ({method_labels})"
return {
"id": f"lea_competence_{summary.id}",
"name": summary.intent_fr,
"description": description,
"category": "lea_competence",
"icon": "🧠",
"source": "competence_yaml",
"competence_id": summary.id,
"learning_state": summary.learning_state,
"source_path": summary.source_path,
"parameters": {
"competence_id": {
"type": "string",
"required": True,
"default": summary.id,
"description": "Identifiant de la compétence Léa à tester ou rejouer",
},
"supervised": {
"type": "boolean",
"required": False,
"default": True,
"description": "Exécuter en mode supervisé humain",
},
},
"methods": list(summary.methods),
"success_marker": summary.success_marker,
"failure_message_template": summary.failure_message_template,
"t2_known_gaps": list(summary.t2_known_gaps),
"examples": [
{
"name": "Tester en supervision",
"description": f"Rejouer la compétence {summary.id} avec validation humaine",
"parameters": {
"competence_id": summary.id,
"supervised": True,
},
}
],
}
def _required_text(data: dict[str, Any], key: str, path: Path) -> str:
value = data.get(key)
if not isinstance(value, str) or not value.strip():
raise ValueError(f"{path} missing required text field {key!r}")
return value.strip()
def _method_summaries(methods: Any) -> list[dict[str, Any]]:
if not isinstance(methods, list):
return []
summaries: list[dict[str, Any]] = []
for method in methods:
if not isinstance(method, dict):
continue
summaries.append(
{
"id": method.get("id"),
"kind": method.get("kind"),
"primitive_ref": method.get("primitive_ref"),
"description": method.get("description"),
"parameters": method.get("parameters") if isinstance(method.get("parameters"), dict) else {},
}
)
return summaries