Krystie
dc5dc94d79
Multi-channel retrieval sidecar over Universal Memory Protocol: - 3-channel RRF (UMP FTS5 + Qdrant vector + knowledge graph) - ACT-R re-ranking (Anderson 1983) with access tracking - Co-occurrence graph edges (Phase 6) for dense traversal - Memory lifecycle decay (Phase 4) with per-kind confidence - MCP shim routes recall through sidecar, falls back to canonical UMP Architecture: - src/server.js HTTP sidecar on port 4380 - src/graph.js 2592-node / 111-edge graph from UMP (or +cooccur: 13k+) - src/actr.js A_i = -d*ln(age) + beta*log1p(freq) + epsilon*conf - src/access_log.js per-URN counter + last_accessed_at - src/ump-recall-mcp.js MCP shim (recall via sidecar, others passthrough) Eval results (851-record UMP corpus): - 2ch RRF over baseline: +50pp recall@10 - 3ch RRF (+graph): +60pp, 12 unique wins - ACT-R re-rank: 4/20 #1 changes, 84% top-5 retention Tests: 76/76 passing across graph (27), actr (27), access_log (28), decay (20), mcp-shim (sidecar + fallback). Run with: npm test Inspired by AIAppsAPI/adaptive-recall but built from scratch against existing DNS2 infrastructure (UMP at :4317, Qdrant at :6333, Ollama at :11434). No paid SaaS, MIT-licensed.
37 lines
1.4 KiB
Python
37 lines
1.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Smoke-test the 3-channel RRF sidecar."""
|
|
import json
|
|
import urllib.request
|
|
|
|
def call(query, limit=5):
|
|
req = urllib.request.Request(
|
|
"http://127.0.0.1:4380/recall",
|
|
data=json.dumps({"query": query, "limit": limit}).encode(),
|
|
headers={"content-type": "application/json"},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=60) as r:
|
|
return json.loads(r.read())
|
|
|
|
def show(label, query):
|
|
print(f"\n=== {label}: query={query!r} ===")
|
|
try:
|
|
r = call(query, 5)
|
|
except Exception as e:
|
|
print(f"ERROR: {e}")
|
|
return
|
|
print(f"phase={r.get('phase')} fused_count={r.get('fused_count')} returned={r.get('returned')}")
|
|
print(f"channels={r.get('channels')}")
|
|
if r.get('channel_errors'):
|
|
print(f"channel_errors={r.get('channel_errors')}")
|
|
for i, h in enumerate(r.get('hits', [])[:5]):
|
|
urn = h['urn'][:60]
|
|
bc = h.get('by_channel', {})
|
|
ranks = {k: v.get('rank') for k, v in bc.items()}
|
|
scores = {k: round(v.get('score', 0), 3) if v.get('score') is not None else None for k, v in bc.items()}
|
|
score = round(h['score'], 4)
|
|
print(f" #{h['rrf_rank']} score={score} urn={urn}... via={ranks} scores={scores}")
|
|
|
|
show("Test 1 (entity-rich)", "DNS2 ollama")
|
|
show("Test 2 (graph-hostile)", "supercalifragilistic")
|
|
show("Test 3 (single entity)", "Triangles")
|
|
show("Test 4 (zero entity)", "the and of") |