68 lines
2.7 KiB
Python
68 lines
2.7 KiB
Python
"""L1: arricchimento metadata memory — subject/detail/rationale nel jsonb (decisione spec 6).
|
|
|
|
Una volta che il registry e' droppato (decisione spec 5), il vectordb e' l'unica
|
|
fonte delle memory. L'hit di search_similar deve bastare per ricostruire la decisione
|
|
completa. Per questo memory_vector_records mette subject/detail/rationale nel metadata
|
|
del VectorRecord (pack_metadata li serializza nel jsonb via **record.metadata).
|
|
"""
|
|
from datetime import datetime
|
|
|
|
from tht.memory import MemoryRecord, memory_vector_records
|
|
|
|
|
|
def _record(**kw) -> MemoryRecord:
|
|
base = dict(
|
|
id="mem-x", ts=datetime(2025, 1, 1), session_id="s", decision_seq=1,
|
|
type="concept_clarified", subject="paziente attivo", detail="flag_attivo = TRUE",
|
|
rationale="perche' serve", question_context="dammi pazienti",
|
|
tables=[], concepts=["paziente attivo"],
|
|
)
|
|
base.update(kw)
|
|
return MemoryRecord(**base)
|
|
|
|
|
|
def test_memory_vector_record_has_subject_detail_rationale_in_metadata():
|
|
vr = memory_vector_records([_record()])[0]
|
|
assert vr.metadata["subject"] == "paziente attivo"
|
|
assert vr.metadata["detail"] == "flag_attivo = TRUE"
|
|
assert vr.metadata["rationale"] == "perche' serve"
|
|
|
|
|
|
def test_memory_vector_record_metadata_keeps_existing_fields():
|
|
vr = memory_vector_records([_record()])[0]
|
|
# i campi che gia' c'erano restano (backward compat)
|
|
assert vr.metadata["type"] == "concept_clarified"
|
|
assert vr.metadata["tables"] == []
|
|
assert vr.metadata["concepts"] == ["paziente attivo"]
|
|
assert vr.metadata["session_id"] == "s"
|
|
|
|
|
|
def test_each_record_carries_its_own_subject():
|
|
# piu' record, ciascuno con il proprio subject (no cross-contamination)
|
|
recs = [
|
|
_record(id="m1", subject="t_a", detail="d_a", rationale="r_a"),
|
|
_record(id="m2", subject="t_b", detail="d_b", rationale="r_b"),
|
|
]
|
|
out = memory_vector_records(recs)
|
|
assert {v.metadata["subject"] for v in out} == {"t_a", "t_b"}
|
|
assert {v.metadata["detail"] for v in out} == {"d_a", "d_b"}
|
|
|
|
|
|
def test_save_one_memory_preserves_subject_through_upsert_row():
|
|
# end-to-end: save_one_memory costruisce la riga upsert; il metadata deve arrivare
|
|
# con subject/detail/rationale (pack_metadata lo prende da record.metadata).
|
|
from unittest.mock import MagicMock
|
|
|
|
from tht.memory import save_one_memory
|
|
|
|
writer = MagicMock()
|
|
writer.upsert.return_value = 1
|
|
embedder = MagicMock()
|
|
embedder.embed_documents.return_value = [[0.0] * 8]
|
|
save_one_memory([_record(decision_seq=1)], decision_seq=1, store=writer, embedder=embedder)
|
|
row = writer.upsert.call_args[0][1][0]
|
|
md = row.record.metadata
|
|
assert md["subject"] == "paziente attivo"
|
|
assert md["detail"] == "flag_attivo = TRUE"
|
|
assert md["rationale"] == "perche' serve"
|