sylvain
a8727654af
- agent1.py : bot XMPP connecté à Ollama avec boucle agentique - skills/web_search.py : recherche DuckDuckGo (ddgs) - skills/web_read.py : lecture et extraction de pages web - skills/memory.py : mémoire persistante SQLite (REMEMBER/RECALL) - skills/loader.py : chargement dynamique des skills
62 lines
2.0 KiB
Python
62 lines
2.0 KiB
Python
"""
|
|
Chargeur de skills.
|
|
Détecte les commandes dans une réponse LLM et exécute le skill correspondant.
|
|
|
|
Format attendu dans la réponse du LLM :
|
|
SEARCH: <requête>
|
|
READ: <url>
|
|
REMEMBER: <clé> | <valeur>
|
|
RECALL: <clé>
|
|
"""
|
|
import importlib
|
|
import re
|
|
from pathlib import Path
|
|
|
|
SKILLS_DIR = Path(__file__).parent
|
|
|
|
# Map trigger -> fonction d'exécution
|
|
_REGISTRY: dict = {}
|
|
|
|
def load_skills():
|
|
"""Charge tous les skills disponibles dans le dossier skills/."""
|
|
_REGISTRY.clear()
|
|
|
|
for py_file in SKILLS_DIR.glob("*.py"):
|
|
if py_file.name.startswith("_") or py_file.name == "loader.py":
|
|
continue
|
|
module_name = "skills.{}".format(py_file.stem)
|
|
try:
|
|
mod = importlib.import_module(module_name)
|
|
except Exception as e:
|
|
print("[Skills] Impossible de charger {} : {}".format(py_file.name, e))
|
|
continue
|
|
|
|
# Skill avec un seul trigger (ex: SEARCH:, READ:)
|
|
if hasattr(mod, "TRIGGER") and mod.TRIGGER and hasattr(mod, "execute"):
|
|
_REGISTRY[mod.TRIGGER] = mod.execute
|
|
print("[Skills] Chargé : {}".format(mod.TRIGGER))
|
|
|
|
# Skill memory : deux triggers
|
|
if py_file.stem == "memory":
|
|
if hasattr(mod, "remember"):
|
|
_REGISTRY["REMEMBER:"] = mod.remember
|
|
print("[Skills] Chargé : REMEMBER:")
|
|
if hasattr(mod, "recall"):
|
|
_REGISTRY["RECALL:"] = mod.recall
|
|
print("[Skills] Chargé : RECALL:")
|
|
|
|
def run_skills(llm_response: str) -> tuple[bool, str]:
|
|
"""
|
|
Parcourt la réponse du LLM ligne par ligne.
|
|
Si une commande est détectée, exécute le skill et retourne (True, résultat).
|
|
Sinon retourne (False, réponse originale).
|
|
"""
|
|
for line in llm_response.splitlines():
|
|
line = line.strip()
|
|
for trigger, fn in _REGISTRY.items():
|
|
if line.upper().startswith(trigger):
|
|
args = line[len(trigger):].strip()
|
|
result = fn(args)
|
|
return True, result
|
|
return False, llm_response
|