Surfacing API

📚 Repository: smartmemory-ai/smart-memory 🐛 Issues: Report bugs or request features

API reference for SmartMemory's retrieval-time surfacing layer: the opt-in router, cross-type context assembly, and session pins.

Conceptual background lives in concepts/memory-dynamics.md. This page is the developer reference.

Constructor parameter

`surfacing_mode`

SmartMemory(..., surfacing_mode: Literal["legacy", "router"] = "legacy")

Gates the entire M3 retrieval layer. "legacy" (default) preserves the M1a behavior — lexical × recency × activation-with-floor, no spreading, no PPR, session_pin_boost/freshness_boost/centrality stay zero in responses. "router" opts into the full dynamics model.

from smartmemory import SmartMemory

# Default — unchanged from pre-M3 behavior.
memory = SmartMemory()

# Opt in to router-mode surfacing.
memory = SmartMemory(surfacing_mode="router")

Invalid values raise ValueError at construction time.

`get_working_context()`

The primary retrieval API. Returns a contract-shaped dict with ranked items for the current turn.

Signature

def get_working_context(
    self,
    session_id: str,
    query: str,
    k: int = 20,
    max_tokens: Optional[int] = None,
    strategy: Optional[str] = None,
) -> Dict[str, Any]

Parameters

Parameter	Type	Default	Description
`session_id`	`str`	required	Session scope for anchors and pins.
`query`	`str`	required	Natural-language query string.
`k`	`int`	`20`	Max non-mandatory items returned (anchors + pins are additive). Must be `1..100`.
`max_tokens`	`Optional[int]`	`None`	Soft token budget for returned content (≈ `len(content) // 4`). Mandatory items (anchors + pins) always included even when over budget.
`strategy`	`Optional[str]`	`None`	Override the router's classification. Valid values in router mode: `"fast:recency"`, `"deep:multi-hop"`, `"factual:point-query"`. Legacy mode only accepts `None` or `"fast:recency"` (others raise).

Response shape

{
  "decision_id":     "uuid-hex",                   # unique per call
  "strategy_used":   "deep:multi-hop",             # resolved strategy
  "items": [
    {
      "item_id":     "abc...",
      "content":     "...",
      "memory_type": "semantic",
      "metadata":    {...},
      "score_breakdown": {
        "activation":         0.73,
        "relevance":          0.92,
        "recency":            0.88,
        "centrality":         0.65,   # 0 in legacy mode and Slice A
        "anchor_forced":      False,
        "session_pin_boost":  0.0,
        "freshness_boost":    0.15,
      },
    },
    ...
  ],
  "drift_warnings":  [],                           # reserved, empty in current release
  "tokens_used":     420,                          # estimated tokens of returned content
  "tokens_budget":   null,                         # echo of max_tokens
  "deprecation":     null,                         # populated via the memory_recall shim
}

Strategy precedence

Caller-explicit strategy wins. In router mode, whatever the caller passes is honored (reserved strategies soft-degrade to fast:recency with a reason flag).

Else, router classifies from the query text. Heuristic rules:

Strategy	When it fires
`deep:multi-hop`	Query contains `why`, `how`, `what caused`, or `explain`; OR 2+ detected entities; OR query > 80 chars.
`factual:point-query`	Short (≤ 6 words), not a question, starts with `list`/`show`/`get`/`find`/`all` OR is a bare noun. Recency signals (`recent`, `latest`, `newest`, `today`, `yesterday`) hard-skip to `fast:recency`.
`fast:recency`	Default fallback.

Legacy mode — any non-fast:recency explicit strategy raises ValueError.

Observability

After each call, memory.last_surfacing_decision holds the SurfacingDecision that drove the dispatch:

memory.get_working_context("s1", "Why did we pick JWT?")

memory.last_surfacing_decision
# SurfacingDecision(
#   strategy=SurfacingStrategy.DEEP_MULTI_HOP,
#   confidence=0.75,
#   reasons=["keyword:why"],
#   signals={"query_length": 27, "word_count": 6, "entity_count": 0},
# )

Examples

# Simple turn-level retrieval.
resp = memory.get_working_context(
    session_id="chat-abc",
    query="What was our decision on JWT?",
)
for item in resp["items"]:
    print(item["item_id"], item["score_breakdown"]["activation"])

# Token-budgeted for an LLM context window.
resp = memory.get_working_context(
    session_id="chat-abc",
    query="walk me through the auth flow",
    max_tokens=4000,
)

# Force a specific strategy (router mode).
resp = memory.get_working_context(
    session_id="chat-abc",
    query="user list",
    strategy="factual:point-query",
)

Session pins

Pins let you force-include specific items in get_working_context() results for a session, regardless of lexical match.

`pin_item()`

def pin_item(self, session_id: str, item_id: str, weight: float = 1.0) -> None

Adds a pin for item_id in session_id with the given weight. The item is force-included in all subsequent get_working_context(session_id=..., ...) calls and receives an additive session_pin_boost = weight × 0.3 in its score breakdown.

Workspace binding. The first pin in a session binds the session to the current workspace. Subsequent pins that disagree raise SessionPinWorkspaceConflict (a ValueError subclass).

`unpin_item()`

def unpin_item(self, session_id: str, item_id: str) -> None

Removes the pin. No-op if the pin doesn't exist.

`get_pins()`

def get_pins(self, session_id: str) -> Dict[str, float]

Returns the current pin set for session_id as {item_id: weight}. Returns {} for unknown sessions, expired sessions, or sessions bound to a different workspace than the current SmartMemory instance (cross-workspace leak guard).

Example

from smartmemory import SmartMemory
from smartmemory.session.pins import SessionPinWorkspaceConflict

memory = SmartMemory(surfacing_mode="router")

memory.pin_item("session-1", "project-requirements", weight=1.0)
memory.pin_item("session-1", "style-guide", weight=0.5)
memory.get_pins("session-1")
# {"project-requirements": 1.0, "style-guide": 0.5}

# Pinned items force-include regardless of query.
resp = memory.get_working_context("session-1", "anything")
# Both pinned items appear with session_pin_boost > 0.

memory.unpin_item("session-1", "style-guide")

# Cross-workspace pin raises.
other_memory = SmartMemory(surfacing_mode="router", scope_provider=other_ws)
try:
    other_memory.pin_item("session-1", "some-item", weight=1.0)
except SessionPinWorkspaceConflict:
    # Session-1 is bound to the first workspace.
    pass

TTL and cleanup

Pins and the session's workspace binding both have a 24-hour Redis TTL. Callers should unpin explicitly at session close when possible; the TTL is a safety net for forgotten cleanups, not a primary expiry mechanism.

Strategy enum

from smartmemory.search.surfacing_router import SurfacingStrategy

SurfacingStrategy.FAST_RECENCY         # "fast:recency"       — shipped
SurfacingStrategy.DEEP_MULTI_HOP       # "deep:multi-hop"     — shipped
SurfacingStrategy.FACTUAL_POINT        # "factual:point-query" — shipped
SurfacingStrategy.PROCEDURAL_LOOKUP    # "procedural:lookup"  — reserved, soft-degrades to fast:recency
SurfacingStrategy.TEMPORAL_WINDOWED    # "temporal:windowed"  — reserved, soft-degrades to fast:recency

Spreading and PPR internals

These are called automatically by get_working_context() in router mode — most callers don't need them directly. Exposed for advanced use (custom retrieval pipelines, benchmarking).

`spread()`

from smartmemory.search.spreading_activation import spread

result = spread(
    memory,
    seed_item_ids=["item-a", "item-b"],
    depth=2,
    max_fan_out=20,
    damping=0.5,
    edge_types=None,       # None → default allowlist (see below)
    budget_ms=150,
)
# {item_id: propagated_score, ...}

Default edge allowlist: RELATED, RELATES_TO, CO_RETRIEVED, ANSWERED_BY, PART_OF, MENTIONS, MENTIONED_IN, CONTAINS_ENTITY, LINKS_TO, LINKED_TO, DERIVED_FROM, INFERRED_FROM, HAS_SKILL, HAS_CONTEXT, TAGGED_WITH, ALIAS_OF.

Excluded by default: CONTRADICTS, SUPERSEDES, GROUNDED_IN (design rationale: contradictions shouldn't propagate boosts; supersession is a replacement relation; grounding edges lead to global Wikidata entities the workspace filter would skip anyway).

Returns {} on empty seeds, no graph hits, or backend timeout. Never raises for budget exhaustion.

`personalized_pagerank()`

from smartmemory.search.ppr import personalized_pagerank

scores = personalized_pagerank(
    memory,
    seed_item_ids=["item-a"],
    alpha=0.85,
    top_k=100,
    budget_ms=200,
    cache=memory._ppr_cache(),   # optional — enables the Redis cache
    workspace_id="ws-a",         # required when cache is provided
)

Runs NetworkX's exact pagerank against the materialized workspace subgraph. Bounded-concurrency: at most 2 concurrent tasks per process (threading.BoundedSemaphore(2)); additional callers short-circuit to {} with a WARN log.

Workspaces exceeding max_nodes=10000 return {} (PPR skipped + INFO log); the returned cache is empty and centrality stays 0 in the composite.

concepts/memory-dynamics.md — conceptual overview
guides/tuning-surfacing.md — task-oriented configuration guide
api/smart-memory.md — the rest of the SmartMemory class

Surfacing API

On this page