Cesar Adrian Cantero
Feat: Huge intelligence upgrade. Switched both Library (FAISS) and Memory (ChromaDB) embeddings to paraphrase-multilingual-MiniLM-L12-v2 to enable perfect Portuguese comprehension. Bumped directories to library_v2 and memoria_v4 to start fresh. Forced Brain to prioritize library context over web search hallucination.
8293b2e | # core/memory.py | |
| import chromadb | |
| from chromadb.utils import embedding_functions | |
| import uuid, datetime | |
| class Memory: | |
| def __init__(self, persist_directory="/tmp/lum3n_data/memoria_v4"): | |
| self.client = chromadb.PersistentClient(path=persist_directory) | |
| self.ef = embedding_functions.SentenceTransformerEmbeddingFunction(model_name="paraphrase-multilingual-MiniLM-L12-v2") | |
| self.collection = self.client.get_or_create_collection( | |
| name="experiencias", | |
| embedding_function=self.ef, | |
| metadata={"hnsw:space": "cosine"} | |
| ) | |
| def memorizar(self, texto, tipo="neutro", valencia=None, correto=None, tags=None): | |
| id = str(uuid.uuid4()) | |
| # Filtra valores None para evitar o erro de conversão do ChromaDB | |
| metadata = { | |
| "tipo": tipo, | |
| "timestamp": datetime.datetime.now().isoformat(), | |
| "tags": ",".join(tags) if tags else "" | |
| } | |
| if valencia is not None: | |
| metadata["valencia"] = valencia | |
| if correto is not None: | |
| metadata["correto"] = correto | |
| self.collection.add( | |
| documents=[texto], | |
| metadatas=[metadata], | |
| ids=[id] | |
| ) | |
| # Sincroniza a memória com a nuvem (background) | |
| try: | |
| from core.hf_sync import sync_up | |
| sync_up() | |
| except Exception: | |
| pass | |
| return id | |
| def recordar(self, consulta, top_k=5): | |
| resultados = self.collection.query(query_texts=[consulta], n_results=top_k) | |
| memorias = [] | |
| if resultados['ids'] and resultados['ids'][0]: | |
| for i in range(len(resultados['ids'][0])): | |
| memorias.append({ | |
| "id": resultados['ids'][0][i], | |
| "texto": resultados['documents'][0][i], | |
| "metadata": resultados['metadatas'][0][i] if resultados['metadatas'] else {} | |
| }) | |
| return memorias | |
| def recordar_licoes(self, consulta, top_k=3): | |
| """ | |
| Retorna um resumo das experiências com erro ou correção, | |
| ordenadas por relevância e priorizando valência negativa. | |
| """ | |
| resultados = self.collection.query(query_texts=[consulta], n_results=top_k * 3) | |
| licoes = [] | |
| if resultados['ids'] and resultados['ids'][0]: | |
| for i in range(len(resultados['ids'][0])): | |
| meta = resultados['metadatas'][0][i] | |
| if meta.get('correto') is not None: # se houve avaliação | |
| licoes.append({ | |
| 'texto': resultados['documents'][0][i], | |
| 'valencia': meta.get('valencia'), | |
| 'correto': meta.get('correto'), | |
| 'tags': meta.get('tags', '') | |
| }) | |
| # Ordena: erros primeiro (valencia negativa, correto=False) | |
| licoes.sort(key=lambda x: (x['correto'] is not False, x['valencia'] if x['valencia'] is not None else 0)) | |
| return licoes[:top_k] | |