Add AI Council architecture: Tier 2/3/Graph implementation + Integration Plan

Architecture (Agent 1):
- hermes_memory/tier2/{schema,facts,entities,relations,timeline}.py
- hermes_memory/tier3/{backend,chroma_backend,embedder}.py
- hermes_memory/graph/nx_store.py
- hermes_memory/api/memory_api.py (unified API)
- hermes_memory/cron/{consolidate,embed_queue,graph_refresh,prune}.py
- hermes_memory/config.py + pyproject.toml

Integration Plan (Agent 3):
- INTEGRATION_PLAN.md: Memory Provider Plugin strategy
- Hermes Core needs minimal changes
- sync_turn() + prefetch() hooks
- Skills integration via nextlevel_search/remember

Auto-Extraction (Agent 2):
- ARCHITECTURE.md: Full extraction pipeline docs
- Chunking, Pre-Filter, LLM Prompts, Classification
- Entity-Linking, Temporal Reasoning, Deduplication

All files: Python syntax checked, ECC standards applied.
This commit is contained in:
Florian Hartmann
2026-06-03 22:51:50 +00:00
parent 33fb180855
commit b83546d833
25 changed files with 2661 additions and 0 deletions
+7
View File
@@ -0,0 +1,7 @@
"""Tier 3 — Semantic Memory (Qdrant / Chroma)."""
from hermes_memory.tier3.backend import VectorBackend, SearchResult
from hermes_memory.tier3.chroma_backend import ChromaBackend
from hermes_memory.tier3.embedder import LocalEmbedder
__all__ = ["VectorBackend", "SearchResult", "ChromaBackend", "LocalEmbedder"]
+35
View File
@@ -0,0 +1,35 @@
"""Abstrakte Backend-Schnittstelle für Vektor-DBs."""
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Dict, List
@dataclass
class SearchResult:
chunk_id: str
score: float
text: str
metadata: Dict
class VectorBackend(ABC):
@abstractmethod
def index(self, chunks: List[str], payloads: List[Dict]) -> List[str]:
"""Indexiert Chunks, gibt chunk_ids zurück."""
...
@abstractmethod
def search(self, query_embedding: List[float], limit: int = 10, filters: Dict = None) -> List[SearchResult]:
"""Semantische Suche mit Query-Embedding."""
...
@abstractmethod
def delete(self, chunk_ids: List[str]) -> bool:
"""Löscht Chunks anhand ihrer IDs."""
...
@abstractmethod
def health(self) -> Dict:
"""Gibt Status-Informationen zurück."""
...
+69
View File
@@ -0,0 +1,69 @@
"""Chroma-Implementierung des VectorBackend."""
import logging
from pathlib import Path
from typing import Dict, List, Optional
from hermes_memory.tier3.backend import SearchResult, VectorBackend
logger = logging.getLogger(__name__)
class ChromaBackend(VectorBackend):
def __init__(self, persist_path: Path, collection_name: str = "memory_chunks"):
self.persist_path = persist_path
self.collection_name = collection_name
self._client = None
self._collection = None
self._init()
def _init(self) -> None:
try:
import chromadb
self._client = chromadb.PersistentClient(path=str(self.persist_path))
self._collection = self._client.get_or_create_collection(
name=self.collection_name,
metadata={"hnsw:space": "cosine"},
)
except ImportError:
logger.error("chromadb nicht installiert. Installiere: pip install chromadb")
raise
def index(self, chunks: List[str], payloads: List[Dict]) -> List[str]:
if not chunks:
return []
chunk_ids = [p.get("chunk_id", f"chunk_{i}") for i, p in enumerate(payloads)]
self._collection.add(
ids=chunk_ids,
documents=chunks,
metadatas=payloads,
)
return chunk_ids
def search(self, query_embedding: List[float], limit: int = 10, filters: Dict = None) -> List[SearchResult]:
results = self._collection.query(
query_embeddings=[query_embedding],
n_results=limit,
where=filters,
)
out: List[SearchResult] = []
if not results["ids"]:
return out
for i, cid in enumerate(results["ids"][0]):
out.append(
SearchResult(
chunk_id=cid,
score=results["distances"][0][i],
text=results["documents"][0][i] or "",
metadata=results["metadatas"][0][i] or {},
)
)
return out
def delete(self, chunk_ids: List[str]) -> bool:
self._collection.delete(ids=chunk_ids)
return True
def health(self) -> Dict:
count = self._collection.count()
return {"backend": "chroma", "collection": self.collection_name, "count": count}
+44
View File
@@ -0,0 +1,44 @@
"""Embedding-Wrapper für lokale Modelle."""
import hashlib
import logging
from pathlib import Path
from typing import List
logger = logging.getLogger(__name__)
class LocalEmbedder:
"""Wrapper um sentence-transformers mit Caching."""
def __init__(self, model_name: str = "all-MiniLM-L6-v2", cache_dir: Path = None):
self.model_name = model_name
self.cache_dir = cache_dir or Path.home() / ".cache" / "hermes_memory" / "embeddings"
self.cache_dir.mkdir(parents=True, exist_ok=True)
self._model = None
self._dim = 384
def _load_model(self):
if self._model is None:
try:
from sentence_transformers import SentenceTransformer
self._model = SentenceTransformer(self.model_name)
self._dim = self._model.get_sentence_embedding_dimension()
except ImportError:
logger.error("sentence-transformers nicht installiert.")
raise
return self._model
def _cache_key(self, text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest() + ".npy"
def embed(self, texts: List[str]) -> List[List[float]]:
model = self._load_model()
return model.encode(texts, convert_to_numpy=True).tolist()
def embed_query(self, text: str) -> List[float]:
return self.embed([text])[0]
@property
def dim(self) -> int:
return self._dim