sylvain/agent1
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Ajouter skill prompt_memory (ChromaDB Phase 1) + loader générique multi-triggers

Browse Source
This commit is contained in:
sylvain
2026-03-07 10:56:53 +00:00
parent 9f45e4a2d6
commit 30f3d00bec
3 changed files with 128 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/loader.py
+22 -16
View File
@@ -7,21 +7,28 @@ Format attendu dans la réponse du LLM :
READ: <url>
REMEMBER: <clé> | <valeur>
RECALL: <clé>
PROMPT_SAVE: <nom> | <texte>
PROMPT_GET: <nom>
PROMPT_LIST:
PROMPT_DEL: <nom>
Interface d'un skill :
- Trigger unique → TRIGGER = "CMD:" + execute(args) -> str
- Multi-triggers → TRIGGERS = {"CMD1:": "fn1", "CMD2:": "fn2"}
"""
import importlib
import re
from pathlib import Path
SKILLS_DIR = Path(__file__).parent
# Map trigger -> fonction d'exécution
# Map trigger (uppercase) -> callable
_REGISTRY: dict = {}
def load_skills():
"""Charge tous les skills disponibles dans le dossier skills/."""
_REGISTRY.clear()
for py_file in SKILLS_DIR.glob("*.py"):
for py_file in sorted(SKILLS_DIR.glob("*.py")):
if py_file.name.startswith("_") or py_file.name == "loader.py":
continue
module_name = "skills.{}".format(py_file.stem)
@@ -31,19 +38,18 @@ def load_skills():
print("[Skills] Impossible de charger {} : {}".format(py_file.name, e))
continue
# Skill avec un seul trigger (ex: SEARCH:, READ:)
# Skill avec trigger unique : TRIGGER = "CMD:" + execute()
if hasattr(mod, "TRIGGER") and mod.TRIGGER and hasattr(mod, "execute"):
_REGISTRY[mod.TRIGGER] = mod.execute
_REGISTRY[mod.TRIGGER.upper()] = 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:")
# Skill avec plusieurs triggers : TRIGGERS = {"CMD1:": "fn_name", ...}
if hasattr(mod, "TRIGGERS") and isinstance(mod.TRIGGERS, dict):
for trigger, fn_name in mod.TRIGGERS.items():
fn = getattr(mod, fn_name, None)
if fn:
_REGISTRY[trigger.upper()] = fn
print("[Skills] Chargé : {}".format(trigger))
def run_skills(llm_response: str) -> tuple[bool, str]:
"""
@@ -52,10 +58,10 @@ def run_skills(llm_response: str) -> tuple[bool, str]:
Sinon retourne (False, réponse originale).
"""
for line in llm_response.splitlines():
line = line.strip()
stripped = line.strip()
for trigger, fn in _REGISTRY.items():
if line.upper().startswith(trigger):
args = line[len(trigger):].strip()
if stripped.upper().startswith(trigger):
args = stripped[len(trigger):].strip()
result = fn(args)
return True, result
return False, llm_response
skills/memory.py
+2 -1
View File
@@ -6,7 +6,8 @@ import sqlite3
from pathlib import Path
SKILL_NAME = "memory"
TRIGGER = None # Géré via REMEMBER: et RECALL: séparément dans le loader
TRIGGER = None
TRIGGERS = {"REMEMBER:": "remember", "RECALL:": "recall"}
DB_PATH = Path("/opt/agent/memory.db")
skills/prompt_memory.py
+104
View File
@@ -0,0 +1,104 @@
"""
Skill : PROMPT_SAVE / PROMPT_GET / PROMPT_LIST / PROMPT_DEL
Mémoire de prompts persistante via ChromaDB.
Prête pour la recherche vectorielle (Phase 2).
Commandes :
PROMPT_SAVE: <nom> | <texte>
PROMPT_GET: <nom>
PROMPT_LIST:
PROMPT_DEL: <nom>
"""
import chromadb
from chromadb import EmbeddingFunction, Documents, Embeddings
from pathlib import Path
import hashlib
SKILL_NAME = "prompt_memory"
TRIGGER = None
TRIGGERS = {
"PROMPT_SAVE:": "prompt_save",
"PROMPT_GET:": "prompt_get",
"PROMPT_LIST:": "prompt_list",
"PROMPT_DEL:": "prompt_del",
}
DB_PATH = Path("/opt/agent/chroma_db")
# Phase 1 : embedding factice (hash MD5 → vecteur 16 dims)
# Phase 2 : remplacer par un vrai modèle (ex: sentence-transformers)
class HashEmbeddingFunction(EmbeddingFunction):
def __call__(self, input: Documents) -> Embeddings:
embeddings = []
for text in input:
h = hashlib.md5(text.encode()).digest()
vec = [b / 255.0 for b in h]
embeddings.append(vec)
return embeddings
def _get_collection():
client = chromadb.PersistentClient(path=str(DB_PATH))
return client.get_or_create_collection(
name="prompts",
embedding_function=HashEmbeddingFunction(),
metadata={"description": "Mémoire de prompts de l'agent"}
)
def prompt_save(args: str) -> str:
if "|" not in args:
return "Erreur : format attendu → PROMPT_SAVE: <nom> | <texte>"
name, _, text = args.partition("|")
name, text = name.strip(), text.strip()
if not name or not text:
return "Erreur : nom ou texte vide."
try:
col = _get_collection()
col.upsert(
ids=[name],
documents=[text],
metadatas=[{"name": name}]
)
return "Prompt «{}» sauvegardé ({} caractères).".format(name, len(text))
except Exception as e:
return "Erreur PROMPT_SAVE : {}".format(e)
def prompt_get(args: str) -> str:
name = args.strip()
if not name:
return "Erreur : nom vide."
try:
col = _get_collection()
result = col.get(ids=[name])
if result["documents"]:
return "Prompt «{}» :\n{}".format(name, result["documents"][0])
return "Aucun prompt trouvé avec le nom «{}».".format(name)
except Exception as e:
return "Erreur PROMPT_GET : {}".format(e)
def prompt_list(args: str) -> str:
try:
col = _get_collection()
result = col.get()
if not result["ids"]:
return "Aucun prompt en mémoire."
lines = ["Prompts disponibles ({}) :".format(len(result["ids"]))]
for id_, doc in zip(result["ids"], result["documents"]):
preview = doc[:80].replace("\n", " ")
lines.append("- **{}** : {}{}".format(id_, preview, "…" if len(doc) > 80 else ""))
return "\n".join(lines)
except Exception as e:
return "Erreur PROMPT_LIST : {}".format(e)
def prompt_del(args: str) -> str:
name = args.strip()
if not name:
return "Erreur : nom vide."
try:
col = _get_collection()
existing = col.get(ids=[name])
if not existing["ids"]:
return "Aucun prompt trouvé avec le nom «{}».".format(name)
col.delete(ids=[name])
return "Prompt «{}» supprimé.".format(name)
except Exception as e:
return "Erreur PROMPT_DEL : {}".format(e)