synap-langgraph

Synap memory integration for LangGraph.

Install

pip install synap-langgraph

Requires langgraph>=1.0, maximem-synap>=0.2.0.

What's in the box

• SynapStore — implements LangGraph's BaseStore for cross-thread long-term memory. Semantic search via store.search(namespace, query=...) routes to sdk.fetch(...), so your graph nodes get Synap-powered recall out of the box.

  • User or customer scope. Pass user_id for private per-user memory, or just a customer_id (no user_id) for a customer-wide shared pool visible to every user in the deployment.
  • All memory types. Reads surface facts and preferences (plus episodes / emotions / temporal events), so stated preferences aren't dropped.
  • Anticipation (optional). Construct with include_conversation_context=True and feed turns via store.record_message(conversation_id, role, content) so just-stated context is in play on the next read. The turn goes out on the live stream when one is open and over REST only when there is none, never both. (This lives alongside the BaseStore API — anticipation has no key/value analogue.)

• SynapCheckpointSaver — implements BaseCheckpointSaver with best-effort fuzzy retrieval. Checkpoint writes succeed durably; reads use sdk.fetch which is semantic-search shaped rather than exact KV. Use for observability/audit and demo flows. For production checkpoint durability, pair with SqliteSaver / PostgresSaver.

• create_synap_node — re-exported from synap-langchain for users who discovered our LangGraph support through the LangChain package. This is the canonical home.

• synap_st_prompt — short-term conversation context as a prompt callable for create_react_agent. Prepends Synap's compacted summary + recent turns above your system prompt at every LLM step.

• create_synap_st_node — same short-term context, exposed as a StateGraph node that writes the ST string into state for your LLM node to consume.

• SynapLangGraphCallbackHandler — reports the whole turn on Synap's live gRPC stream so the anticipation agent can see it: the user turn, the reasoning, every tool call and result, and the assistant turn. Pass it like any LangChain callback.

Reporting a run on the live stream

LangGraph runs on LangChain's callback machinery, so a Synap handler in the run config reaches the chat model and the tool nodes inside your compiled graph.

from langchain_core.messages import HumanMessage
from synap_langgraph import SynapLangGraphCallbackHandler

handler = SynapLangGraphCallbackHandler(
    sdk=sdk,
    conversation_id="conv_abc123",
    user_id="user_456",
    customer_id="cust_789",
)

await app.ainvoke(
    {"messages": [HumanMessage("where is my order QX41RT")]},
    config={"callbacks": [handler]},
)

Build one handler per conversation. What it reports:

Stream first, REST only as a fallback, never both. The server persists the user and assistant turns from the stream itself, so the handler falls back to sdk.conversation.record_message only when no sdk.instance.listen() is running. The same rule applies to SynapStore.record_message. Every one of these calls is silent when there is no stream and none of them can raise into your graph.

Short-term context (compacted conversation, on every LLM step)

LangGraph's built-in memory truncates recent turns to a token budget. Synap's short-term context is the compacted summary + recent turns maintained per conversation by the Synap server — a richer, more token-efficient view of "what happened so far." Drop it into a prebuilt agent or a custom graph; both helpers serve from the SDK's local cache when warm (near-zero overhead) and fall back to the Synap server when cold.

The SDK helper they both wrap is sdk.conversation.context.get_context_for_prompt(conv_id, style=...), which is cache-first whenever the SYNAP_SDK_ST_AUTHORITATIVE flag is on.

A) Prebuilt agent — one-line drop-in

from langgraph.prebuilt import create_react_agent  # or langchain.agents.create_agent
from maximem_synap import MaximemSynapSDK
from synap_langgraph import synap_st_prompt

sdk = MaximemSynapSDK(api_key="sk-...")

agent = create_react_agent(
    model="anthropic:claude-3-5-sonnet-20241022",
    tools=[...],
    prompt=synap_st_prompt(
        sdk,
        conversation_id="conv_abc123",       # required, explicit
        system="You are a helpful agent.",   # your own instructions
        style="narrative",                   # default; also "structured" | "bullet_points"
    ),
)

What the model sees at every step (system message content):

<synap_short_term_context>
... compacted summary + recent turns from Synap ...
</synap_short_term_context>

You are a helpful agent.

B) Custom graph — write ST into state, consume in your own LLM node

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from synap_langgraph import create_synap_st_node

class State(TypedDict):
    messages: Annotated[list, add_messages]
    synap_st: str   # populated by the node

async def my_llm_node(state: State):
    system_text = f"You are a helpful agent.\n\n{state['synap_st']}".strip()
    ...   # build your prompt template with system_text and invoke the model

graph = StateGraph(State)
graph.add_node("st", create_synap_st_node(sdk, conversation_id="conv_abc123"))
graph.add_node("llm", my_llm_node)
graph.add_edge(START, "st")
graph.add_edge("st", "llm")
graph.add_edge("llm", END)

Which to use

Error policy

• SDK failures never crash the graph by default (on_error="fallback"): logged at ERROR via SynapIntegrationError's log path, then the helper degrades to your bare system prompt (or empty state slot).
• Pass on_error="raise" for strict environments that want the failure surfaced as SynapIntegrationError.
• An empty short-term result (no compaction yet and no recent turns) is a legitimate empty case, not a failure — the user's system prompt is preserved as-is.

Conversation ID

Always explicit. We deliberately do not infer it from LangGraph's thread_id because the two namespaces can diverge — your thread might span multiple Synap conversations, or vice versa. For multi-conversation agents, construct one prompt callable per conversation inside your per-run setup.

Quickstart

from langgraph.graph import StateGraph, START, END
from maximem_synap import MaximemSynapSDK
from synap_langgraph import SynapStore, SynapCheckpointSaver

sdk = MaximemSynapSDK(api_key="sk-...")

store = SynapStore(sdk, user_id="alice", customer_id="acme")
saver = SynapCheckpointSaver(sdk, user_id="alice", customer_id="acme")

graph = StateGraph(MyState)
# ... add nodes / edges ...
app = graph.compile(checkpointer=saver, store=store)

# Store usage inside a node:
async def remember(state, runtime):
    await runtime.store.aput(
        ("alice", "preferences"),
        "language",
        {"value": "English"},
    )

Error policy

• Writes (SynapStore.put, SynapCheckpointSaver.put, put_writes) surface SDK failures as SynapIntegrationError. Silent drops would hide ingestion outages.
• Reads (get, search, get_tuple, list) degrade gracefully — they log at ERROR and return None/[] rather than crashing the graph.
• Deletes (SynapStore.delete, SynapCheckpointSaver.delete_thread) warn once and no-op — Synap has no public delete API.

Note on metadata-stripping backends. get/search identify items by custom metadata markers. On instances that atomize content during extraction (e.g. MACA), those markers are stripped, so:

• search falls back to returning the scope-filtered results Synap ranked (semantic retrieval still works). Scope (user_id/customer_id) is enforced at the fetch layer, but sub-namespace isolation within a scope is not. A one-time warning is logged. Set SynapStore(..., semantic_fallback=False) for strict namespace semantics (search returns [] when markers are absent).
• get (exact key) returns None — there's no reliable way to resolve an exact key without the markers. Use search as the read path.

Job/document-level attribution (mapping fragments back to a source id) is not done in the store; build it in app code.

When to use which checkpointer

Cross-thread long-term memory (BaseStore) maps cleanly to Synap — use SynapStore as your default.