enne2/pi-qmem
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(hierarchy): supporto metadati strutturati e gerarchici su Qdrant (parent_id, level, topic, links)

Browse Source
This commit is contained in:
Matteo Benedetto
2026-08-23 13:07:16 +02:00
parent fc11f878f6
commit 369a2c2d9e
4 changed files with 224 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/main.py
+45 -5
View File
@@ -66,7 +66,7 @@ GUARDRAIL_VERSION = "similarity-v1"
# Versione del codice: hash del commit Git da cui è stato costruito il container
# (iniettato come build arg nel Dockerfile: ARG GIT_COMMIT / ENV GIT_COMMIT)
GIT_COMMIT = os.environ.get("GIT_COMMIT", "unknown").strip()
GATEWAY_VERSION = os.environ.get("GATEWAY_VERSION", "2.7.0").strip()
GATEWAY_VERSION = os.environ.get("GATEWAY_VERSION", "2.8.0").strip()
# Metriche: push a VictoriaMetrics (stesso pattern dell'energy engine domotics)
VM_PUSH_URL = os.environ.get("VM_PUSH_URL", "http://host.docker.internal:8428/api/v1/import/prometheus")
@@ -110,7 +110,7 @@ async def lifespan(_app: FastAPI):
SPARSE_VECTOR_NAME: qm.SparseVectorParams(modifier=qm.Modifier.IDF),
},
)
for field in ("agent_id", "project_id", "scope", "kind", "supersedes_id", "superseded_by", "text_hash"):
for field in ("agent_id", "project_id", "scope", "kind", "supersedes_id", "superseded_by", "text_hash", "parent_id", "level", "topic"):
qdrant.create_payload_index(
collection_name=COLLECTION,
field_name=field,
@@ -124,6 +124,16 @@ async def lifespan(_app: FastAPI):
log.info("collection %s creata con indici (dense + sparse %s)", COLLECTION, SPARSE_VECTOR_NAME)
else:
log.info("collection %s già esistente", COLLECTION)
# Migrazione indici gerarchici: crea se mancanti
for field in ("parent_id", "level", "topic"):
try:
qdrant.create_payload_index(
collection_name=COLLECTION,
field_name=field,
field_schema=qm.PayloadSchemaType.KEYWORD,
)
except Exception: # noqa: BLE001
pass
# Migrazione: aggiunge lo sparse vector se manca (collection pre-ibrida)
info = qdrant.get_collection(COLLECTION)
sparse_vectors = (info.config.params.sparse_vectors or {}) if info.config and info.config.params else {}
@@ -158,7 +168,7 @@ async def lifespan(_app: FastAPI):
_http = None
app = FastAPI(title="Memory Gateway", version="2.7.0", lifespan=lifespan)
app = FastAPI(title="Memory Gateway", version="2.8.0", lifespan=lifespan)
qdrant = QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY)
# Request ID: generato per richiesta, loggato nell'audit e restituito in header
@@ -226,6 +236,12 @@ def _invalidate_meta() -> None:
# ---------------------------------------------------------------------------
# Modelli
# ---------------------------------------------------------------------------
class MemoryLink(BaseModel):
target_id: str = Field(..., description="UUID del record target collegato")
predicate: str = Field(default="part_of", max_length=64, description="Tipo di relazione: parent_of, part_of, relates_to, supersedes...")
weight: float = Field(default=1.0, ge=0.0, le=1.0)
class MemoryIn(BaseModel):
text: str = Field(min_length=1, max_length=MAX_TEXT_LEN)
kind: Literal["decision", "fact", "episode", "preference"] = "fact"
@@ -237,6 +253,10 @@ class MemoryIn(BaseModel):
expires_at: Optional[str] = None # ISO 8601
supersedes_id: Optional[str] = None
supersede_reason: Optional[str] = Field(default=None, max_length=512)
parent_id: Optional[str] = Field(default=None, description="UUID del record genitore per gerarchia/subtopic")
level: Optional[Literal["L1_ROOT", "L2_SUBTOPIC", "L3_DETAIL"]] = Field(default=None, description="Livello gerarchico del record")
topic: Optional[str] = Field(default=None, max_length=128, description="Topic gerarchico (es. ALFA-ROMEO-GT-1300-JUNIOR/SPECS)")
links: Optional[list[MemoryLink]] = Field(default=None, description="Collegamenti semantici e relazionali verso altri record")
@field_validator("expires_at")
@classmethod
@@ -260,6 +280,9 @@ class SearchIn(BaseModel):
min_score: Optional[float] = Field(default=None, ge=0.0, le=1.0)
top_k: int = Field(default=5, ge=1, le=20)
hybrid: bool = Field(default=False, description="True = hybrid retrieval (BM25 + vettoriale, RRF). I punteggi risultanti sono RRF, non cosine.")
parent_id: Optional[str] = Field(default=None, description="Filtra per UUID del record genitore")
level: Optional[Literal["L1_ROOT", "L2_SUBTOPIC", "L3_DETAIL"]] = Field(default=None, description="Filtra per livello gerarchico")
topic: Optional[str] = Field(default=None, description="Filtra per topic esatto")
# ---------------------------------------------------------------------------
@@ -334,7 +357,7 @@ def _find_similar(text: str, vector: list[float], top_k: int = 3) -> list[dict]:
]
def _decide_guardrail(text: str, vector: list[float]) -> dict:
def _decide_guardrail(text: str, vector: list[float], topic: Optional[str] = None, parent_id: Optional[str] = None) -> dict:
"""Applica il guardrail a 2 strati. Ritorna {decision, reason, matches}."""
# Strato 1 — hash esatto (duplicato identico)
text_hash = _text_hash(text)
@@ -365,6 +388,9 @@ def _decide_guardrail(text: str, vector: list[float]) -> dict:
top1 = matches[0]["score"]
if top1 >= GUARDRAIL_BLOCK_THRESHOLD:
# Se il nuovo record ha un topic o parent_id esplicito che lo differenzia, permetti con WARN
if (topic or parent_id) and any(m.get("memory_id") != parent_id for m in matches):
return {"decision": "WARN", "reason": "HIERARCHICAL_SUBTOPIC", "matches": matches}
return {"decision": "BLOCK", "reason": "KNOWN_SOLUTION", "matches": matches}
if top1 >= GUARDRAIL_WARN_THRESHOLD:
return {"decision": "WARN", "reason": "MODERATE_SIMILARITY", "matches": matches}
@@ -512,7 +538,7 @@ async def add_memory(
# Il supersede esplicito è una correzione intenzionale: bypassa il guardrail.
guardrail: Optional[dict] = None
if GUARDRAIL_ENABLED and not body.supersedes_id:
guardrail = _decide_guardrail(body.text, vector)
guardrail = _decide_guardrail(body.text, vector, topic=body.topic, parent_id=body.parent_id)
if guardrail["decision"] == "BLOCK":
_audit(
key,
@@ -544,6 +570,10 @@ async def add_memory(
"expires_at": _parse_ts(body.expires_at),
"supersedes_id": superseded_id,
"supersede_reason": body.supersede_reason,
"parent_id": body.parent_id,
"level": body.level,
"topic": body.topic,
"links": [link.model_dump() for link in body.links] if body.links else None,
"embedding_model": EMBED_MODEL,
"text_hash": _text_hash(body.text),
}
@@ -600,6 +630,12 @@ async def search_memories(body: SearchIn, key: str = Depends(require_auth)) -> d
must.append(qm.FieldCondition(key="project_id", match=qm.MatchValue(value=body.project_id)))
if body.scope:
must.append(qm.FieldCondition(key="scope", match=qm.MatchValue(value=body.scope)))
if body.parent_id:
must.append(qm.FieldCondition(key="parent_id", match=qm.MatchValue(value=body.parent_id)))
if body.level:
must.append(qm.FieldCondition(key="level", match=qm.MatchValue(value=body.level)))
if body.topic:
must.append(qm.FieldCondition(key="topic", match=qm.MatchValue(value=body.topic)))
if not body.include_superseded:
# default: esclude i record già corretti (superseded_by presente)
must.append(qm.IsEmptyCondition(is_empty=qm.PayloadField(key="superseded_by")))
@@ -655,6 +691,10 @@ async def search_memories(body: SearchIn, key: str = Depends(require_auth)) -> d
"supersedes_id": h.payload.get("supersedes_id"),
"superseded_by": h.payload.get("superseded_by"),
"supersede_reason": h.payload.get("supersede_reason"),
"parent_id": h.payload.get("parent_id"),
"level": h.payload.get("level"),
"topic": h.payload.get("topic"),
"links": h.payload.get("links"),
}
for h in hits
]
gateway/tests/test_api.py
+94
View File
@@ -216,3 +216,97 @@ def test_status_espone_git_commit(client):
assert "git_commit" in data
assert "version" in data
assert "guardrail_version" in data
# ---------------------------------------------------------------------------
# Struttura Gerarchica e Relazionale
# ---------------------------------------------------------------------------
def test_create_and_retrieve_hierarchical_record(client):
"""Crea un nodo Root L1 e un nodo Figlio L2 con links, parent_id, level, topic."""
# 1. Crea Root L1
r_root = client.post(
"/v1/memories",
json=make_record(
text="Master Topic Alfa Romeo",
level="L1_ROOT",
topic="ALFA-ROMEO/ROOT",
),
headers=auth_headers(),
)
assert r_root.status_code == 200
root_id = r_root.json()["memory_id"]
# 2. Crea Figlio L2 collegato
r_child = client.post(
"/v1/memories",
json=make_record(
text="Scheda Tecnica Bialbero 1.3",
parent_id=root_id,
level="L2_SUBTOPIC",
topic="ALFA-ROMEO/SPECS",
links=[{"target_id": root_id, "predicate": "part_of", "weight": 1.0}],
),
headers=auth_headers(),
)
assert r_child.status_code == 200
child_id = r_child.json()["memory_id"]
# 3. Recupera e verifica payload strutturato
g = client.get(f"/v1/memories/{child_id}", headers=auth_headers())
assert g.status_code == 200
data = g.json()
assert data["parent_id"] == root_id
assert data["level"] == "L2_SUBTOPIC"
assert data["topic"] == "ALFA-ROMEO/SPECS"
assert len(data["links"]) == 1
assert data["links"][0]["target_id"] == root_id
def test_search_filters_hierarchical(client):
"""Filtra per parent_id, level e topic."""
r_root = client.post(
"/v1/memories",
json=make_record(text="Root doc", level="L1_ROOT", topic="TOPIC/ROOT"),
headers=auth_headers(),
)
root_id = r_root.json()["memory_id"]
client.post(
"/v1/memories",
json=make_record(text="Child A", parent_id=root_id, level="L2_SUBTOPIC", topic="TOPIC/A"),
headers=auth_headers(),
)
client.post(
"/v1/memories",
json=make_record(text="Child B", parent_id=root_id, level="L2_SUBTOPIC", topic="TOPIC/B"),
headers=auth_headers(),
)
# Cerca solo L1_ROOT
s1 = client.post(
"/v1/memories:search",
json={"query": "doc", "level": "L1_ROOT", "top_k": 5, "min_score": 0.0},
headers=auth_headers(),
)
assert s1.status_code == 200
assert len(s1.json()["results"]) == 1
assert s1.json()["results"][0]["level"] == "L1_ROOT"
# Cerca per parent_id
s2 = client.post(
"/v1/memories:search",
json={"query": "Child", "parent_id": root_id, "top_k": 5, "min_score": 0.0},
headers=auth_headers(),
)
assert s2.status_code == 200
assert len(s2.json()["results"]) == 2
# Cerca per topic specifico
s3 = client.post(
"/v1/memories:search",
json={"query": "Child", "topic": "TOPIC/A", "top_k": 5, "min_score": 0.0},
headers=auth_headers(),
)
assert s3.status_code == 200
assert len(s3.json()["results"]) == 1
assert s3.json()["results"][0]["topic"] == "TOPIC/A"