feat: auto-create edges on memory_store in MCP server

Automatically find and link related memories after every store operation,
removing reliance on Claude manually creating edges. Uses recall + type
heuristics to choose edge types (SOLVES, BUILDS_ON, RELATED_TO) and
returns created edge IDs in the store response for optional review.

Key design choices:
- Keyword-only fallback requires tag overlap to prevent spurious edges
- Similarity threshold (0.4) filters before max-edge cap (3)
- Edge description arrow matches actual from/to direction
- Failures never break the store operation

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Cal Corum 2026-02-28 22:07:05 -06:00
parent e87c7598d1
commit 194990d424

View File

@ -20,6 +20,23 @@ from common import resolve_graph_path, list_graphs
SYNC_SCRIPT = Path(__file__).parent / "scripts" / "memory-git-sync.sh"
# Auto-edge heuristics: (new_type, match_type) -> (rel_type, direction)
# direction "ab" = new->match, "ba" = match->new
AUTO_EDGE_HEURISTICS = {
("fix", "problem"): ("SOLVES", "ab"),
("solution", "problem"): ("SOLVES", "ab"),
("solution", "error"): ("SOLVES", "ab"),
("fix", "error"): ("SOLVES", "ab"),
("insight", "solution"): ("BUILDS_ON", "ab"),
("insight", "decision"): ("BUILDS_ON", "ab"),
("decision", "solution"): ("BUILDS_ON", "ab"),
("fix", "solution"): ("BUILDS_ON", "ab"),
("code_pattern", "solution"): ("BUILDS_ON", "ab"),
("procedure", "workflow"): ("BUILDS_ON", "ab"),
}
AUTO_EDGE_SIMILARITY_THRESHOLD = 0.4
AUTO_EDGE_MAX = 3
_clients: Dict[str, CognitiveMemoryClient] = {}
@ -527,6 +544,97 @@ def create_tools() -> list:
]
def _auto_create_edges(
client: CognitiveMemoryClient,
memory_id: str,
title: str,
mem_type: str,
tags: Optional[list] = None,
) -> list:
"""Auto-create edges between a newly stored memory and related existing memories.
Uses recall to find similar memories and heuristic type-pairs to choose
relationship types. Returns list of created edge info dicts.
Never raises failures are silently swallowed so store always succeeds.
"""
try:
# Build recall query from title + tags for better precision (#4)
query = title
if tags:
query = f"{title} {' '.join(tags)}"
results = client.recall(query, limit=5, semantic=True)
# Filter out the just-stored memory
results = [r for r in results if r.get("id") != memory_id]
# Filter by similarity threshold before slicing to AUTO_EDGE_MAX (#5)
# When embeddings are absent, keyword-only results lack "similarity" —
# require at least a tag overlap to avoid spurious edges (#1)
filtered = []
new_tags = set(tags or [])
for result in results:
similarity = result.get("similarity")
if similarity is not None:
# Semantic path: use similarity threshold
if similarity < AUTO_EDGE_SIMILARITY_THRESHOLD:
continue
else:
# Keyword-only path: require at least one shared tag
match_tags = set(result.get("tags") or [])
if not new_tags or not (new_tags & match_tags):
continue
filtered.append(result)
created_edges = []
for result in filtered[:AUTO_EDGE_MAX]:
match_type = result.get("type", "")
match_id = result["id"]
match_title = result.get("title", "")
# Look up heuristic for (new_type, match_type) in both orderings
key_ab = (mem_type, match_type)
key_ba = (match_type, mem_type)
if key_ab in AUTO_EDGE_HEURISTICS:
rel_type, direction = AUTO_EDGE_HEURISTICS[key_ab]
from_id = memory_id if direction == "ab" else match_id
to_id = match_id if direction == "ab" else memory_id
elif key_ba in AUTO_EDGE_HEURISTICS:
rel_type, direction = AUTO_EDGE_HEURISTICS[key_ba]
# Reverse: if heuristic says ab for (match, new), then match->new
from_id = match_id if direction == "ab" else memory_id
to_id = memory_id if direction == "ab" else match_id
else:
rel_type = "RELATED_TO"
from_id = memory_id
to_id = match_id
# Build description matching actual edge direction (#2)
from_title = title if from_id == memory_id else match_title
to_title = match_title if to_id == match_id else title
desc = f"Auto-edge: {from_title}{to_title}"
edge_id = client.relate(
from_id=from_id,
to_id=to_id,
rel_type=rel_type,
description=desc,
)
if edge_id: # Empty string means duplicate
created_edges.append(
{
"edge_id": edge_id,
"rel_type": rel_type,
"linked_memory_id": match_id,
"linked_title": match_title,
}
)
return created_edges
except Exception:
return []
def handle_tool_call(tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Dispatch MCP tool calls to the CognitiveMemoryClient."""
@ -561,12 +669,14 @@ def handle_tool_call(tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any
tags=tags,
)
episode_logged = True
auto_edges = _auto_create_edges(client, memory_id, title, mem_type, tags)
_trigger_git_sync()
return ok(
{
"success": True,
"memory_id": memory_id,
"episode_logged": episode_logged,
"auto_edges": auto_edges,
}
)