105 lines
3.4 KiB
Python
105 lines
3.4 KiB
Python
"""
|
|
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)
|