cortexm/plugins/structured.py
Rama main · réplica de lectura del repositorio original.
"""Structured Tier plugin — wraps the existing Memory class.
This is NOT a re-implementation. The structured tier (bi-temporal
Trace + VSA Palace + Bridge reader/writer) already exists and has
400+ tests. This plugin is just an ADAPTER so the kernel can mount
it alongside the verbatim plugin.
The plugin takes a "memory" service (a fully-constructed
``cortexm.api.memory.Memory`` instance) and exposes its read/write
APIs as a kernel service.
Mount order matters: the ``memory`` plugin must mount BEFORE this
plugin because this plugin injects it. The default ``Context``
config (in cortexm/__init__.py) handles this.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
@dataclass
class StructuredHit:
"""One structured-tier retrieval result (a Fact).
Mirrors the verbatim tier's VerbatimHit shape so the fusion
bridge can rerank across both tiers uniformly.
"""
fact_id: str
subject: str
relation: str
value: str
user_id: str
source_tx_id: int | None
valid_from: str
valid_to: str | None
confidence: float
score: float # retrieval score from the structured reader
def to_dict(self) -> dict:
return {
"fact_id": self.fact_id,
"subject": self.subject,
"relation": self.relation,
"value": self.value,
"user_id": self.user_id,
"source_tx_id": self.source_tx_id,
"valid_from": self.valid_from,
"valid_to": self.valid_to,
"confidence": self.confidence,
"score": round(self.score, 4),
}
class StructuredPlugin:
"""Adapter for the existing Memory class — kernel-mountable.
Mount it on a kernel Context that already has a "memory" service
(a fully-constructed Memory instance). The plugin does NOT own
the Memory — it just forwards calls. The kernel's effect/cleanup
contract still applies: on dispose, we close the Memory's DB
connection if it owns it.
Usage::
from cortexm.kernel import Context
from cortexm.api.memory import Memory
from cortexm.plugins.structured import StructuredPlugin
ctx = Context()
mem = Memory()
ctx.service("memory", mem)
ctx.mount(StructuredPlugin())
s = ctx.inject("structured")["structured"]
s.add("Alice works at Google", user_id="alice")
hits = s.search("Where does Alice work?", user_id="alice")
"""
name = "structured"
inject = ["memory"]
def __init__(self, dispose_memory: bool = False) -> None:
# If True, we own the Memory instance and will close its DB
# on dispose. If False (default), the caller owns it.
# Most callers construct the Memory themselves and pass it
# in; they want to keep ownership. Set True for tests.
self.dispose_memory = dispose_memory
self._mem = None
def apply(self, ctx) -> None:
deps = ctx.inject("memory")
self._mem = deps["memory"]
ctx.service("structured", self)
if self.dispose_memory:
ctx.effect(self._close_memory)
def _close_memory(self) -> None:
if self._mem is None:
return
try:
# Close the underlying TraceStore — safe even if Memory
# doesn't expose it (the kernel is best-effort on dispose).
store = getattr(self._mem, "store", None)
if store is not None and hasattr(store, "close"):
store.close()
except Exception:
pass
# ---------------------------- write path ------------------------
def add(self, text: str, *, user_id: str, agent_id: str | None = None,
run_id: str | None = None, metadata: dict | None = None,
timestamp=None) -> dict:
"""μ=0 ingest through the existing MemoryWriter.
The Memory class already routes through PII guard, pattern
extraction, contradiction resolution, palace indexing, and
BLAKE3-chained audit log. This is a thin forwarder.
"""
return self._mem.add(text, user_id=user_id, agent_id=agent_id,
run_id=run_id, metadata=metadata,
timestamp=timestamp)
def edit(self, fact_id: str, new_text: str, *,
edited_by: str = "user", reason: str = "") -> dict:
"""Human-in-the-loop correction with provenance stamping."""
return self._mem.edit(fact_id, new_text, edited_by=edited_by,
reason=reason)
# ---------------------------- read path -------------------------
def search(self, query: str, *, user_id: str, k: int = 10,
agent_id: str | None = None, run_id: str | None = None
) -> list[StructuredHit]:
"""Run the structured reader and adapt results to StructuredHit.
The reader returns RetrievalResult objects with .facts and
.score. We flatten to a list of StructuredHit for fusion
with the verbatim tier.
"""
raw = self._mem.search(query, user_id=user_id, limit=k,
agent_id=agent_id, run_id=run_id)
# mem.search returns a DICT with a 'results' key (the list
# of fact dicts) + extra context_block/provenance/relations.
# Adapt both this shape and the legacy list shape.
if isinstance(raw, dict):
results = raw.get("results", [])
elif isinstance(raw, (list, tuple)):
results = raw
else:
results = []
out: list[StructuredHit] = []
for r in results:
if isinstance(r, dict):
out.append(StructuredHit(
fact_id=r.get("id", ""),
subject=r.get("subject", "") or r.get("memory", "")[:60],
relation=r.get("relation", "memory") or "memory",
value=r.get("memory", "") or r.get("value", ""),
user_id=r.get("user_id", user_id),
source_tx_id=r.get("source_tx_id") or r.get("source_id"),
valid_from=r.get("valid_from", "") or "",
valid_to=r.get("valid_to"),
confidence=float(r.get("confidence", 1.0) or 1.0),
score=float(r.get("score", 0.5) or 0.5),
))
else:
# RetrievalResult-like: has .facts list + .score
facts = getattr(r, "facts", None) or []
score = getattr(r, "score", 0.5) or 0.5
for f in facts:
out.append(StructuredHit(
fact_id=getattr(f, "id", ""),
subject=getattr(f, "subject", ""),
relation=getattr(f, "relation", ""),
value=getattr(f, "value", ""),
user_id=getattr(f, "user_id", user_id),
source_tx_id=getattr(f, "source_tx_id", None)
or getattr(f, "source_id", None),
valid_from=str(getattr(f, "valid_from", "") or ""),
valid_to=str(getattr(f, "valid_to", None) or "") or None,
confidence=float(getattr(f, "confidence", 1.0) or 1.0),
score=score,
))
# Sort by score desc, take top-k
out.sort(key=lambda h: h.score, reverse=True)
return out[:k]
def structural_query(self, *, user_id: str, subject: str | None = None,
relation: str | None = None,
valid_at: str | None = None) -> list[StructuredHit]:
"""Direct Trace query bypassing VSA — for temporal/multi-hop.
Wraps mem.store.find_facts / lifecycle.at helpers.
"""
# The Memory class exposes structural queries via .search()
# with temporal kwargs; we use those if available. Otherwise
# fall back to a search query built from the subject+relation.
if hasattr(self._mem, "structural_query"):
return [StructuredHit(
fact_id=r.get("id", ""), subject=r.get("subject", ""),
relation=r.get("relation", ""), value=r.get("value", ""),
user_id=r.get("user_id", user_id),
source_tx_id=r.get("source_tx_id") or r.get("source_id"),
valid_from=str(r.get("valid_from", "")),
valid_to=r.get("valid_to"),
confidence=float(r.get("confidence", 1.0) or 1.0),
score=float(r.get("score", 1.0) or 1.0),
) for r in self._mem.structural_query(
user_id=user_id, subject=subject, relation=relation,
valid_at=valid_at)]
# Fallback: synthesize a query and use search()
synth_q = " ".join(filter(None, [subject or "", relation or ""]))
return self.search(synth_q or subject or relation or "",
user_id=user_id, k=50)
__all__ = ["StructuredPlugin", "StructuredHit"]