LangChain - Maximem Synap

Build LangChain applications that remember context across sessions. Synap handles memory ingestion, retrieval, and per-user scoping; LangChain handles your chain or agent.

Requires Python 3.11+.

Overview

This guide shows how to add Synap to a LangChain application to build agents that:

Maintain conversation history across sessions and processes
Retrieve user- and organization-scoped context as part of a RAG pipeline
Decide for themselves when to search or store memories

The Synap LangChain integration ships four drop-in components. Each one slots into a native LangChain interface so you do not have to wrap or re-implement anything.

Component	LangChain interface	Purpose
`SynapChatMessageHistory`	`BaseChatMessageHistory`	Persistent chat history per `conversation_id`
`SynapCallbackHandler`	`BaseCallbackHandler`	Auto-records every LLM turn, no application changes
`SynapRetriever`	`BaseRetriever`	Semantic retriever that returns Synap memories as `Document`s
`SynapSearchTool`, `SynapStoreTool`	`BaseTool`	Agent-callable tools for explicit memory read/write

Setup

Install the package alongside LangChain and your model provider:

pip install maximem-synap-langchain langchain langchain-openai

uv add maximem-synap-langchain langchain langchain-openai
# pip-compatible (existing venv): uv pip install maximem-synap-langchain langchain langchain-openai

The pip package is maximem-synap-langchain, but the import drops the maximem- prefix and uses underscores: from synap_langchain import ....

Configure your API key. Generate one from the Synap Dashboard.

.env

SYNAP_API_KEY=synap_your_key_here
OPENAI_API_KEY=your-openai-api-key

Then initialize the SDK once at application startup:

from maximem_synap import MaximemSynapSDK

sdk = MaximemSynapSDK()  # picks up SYNAP_API_KEY from env
await sdk.initialize()

See SDK Initialization for the full lifecycle and configuration options.

Basic integration

The smallest useful integration is a chain with automatic memory via SynapCallbackHandler. Every turn is ingested without changing your chain logic:

# pip install maximem-synap-langchain langchain langchain-openai
import uuid
from maximem_synap import MaximemSynapSDK
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from synap_langchain import SynapCallbackHandler

sdk = MaximemSynapSDK()
await sdk.initialize()

llm = ChatOpenAI(model="gpt-4o")

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant with long-term memory."),
    ("human", "{question}"),
])

chain = prompt | llm

# conversation_id (and every other id) must be a valid UUID.
conversation_id = str(uuid.uuid4())

handler = SynapCallbackHandler(
    sdk=sdk,
    conversation_id=conversation_id,
    user_id="alice",
    customer_id="acme",   # optional, required for B2B instances
)

response = await chain.ainvoke(
    {"question": "Remind me what we agreed on for the Q2 roadmap."},
    config={"callbacks": [handler]},
)

The handler observes the chain via LangChain’s callback system, captures the user/assistant pair, and ingests it into Synap asynchronously. Ingestion failures are logged at ERROR level and never propagate to your chain. Your application keeps running even if Synap is unreachable. This is the smallest viable setup. To make the model aware of memory at inference time, combine the callback with SynapChatMessageHistory and/or SynapRetriever below.

Core concepts

Persistent chat history

SynapChatMessageHistory implements LangChain’s BaseChatMessageHistory interface. Wrap any chain with RunnableWithMessageHistory to give it conversation memory that survives restarts:

from langchain_core.runnables.history import RunnableWithMessageHistory
from synap_langchain import SynapChatMessageHistory

def get_history(session_id: str) -> SynapChatMessageHistory:
    return SynapChatMessageHistory(
        sdk=sdk,
        conversation_id=session_id,
        user_id="alice",
        customer_id="acme",   # optional
    )

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history=get_history,
    input_messages_key="question",
    history_messages_key="history",
)

response = await chain_with_history.ainvoke(
    {"question": "What did we discuss last time?"},
    config={"configurable": {"session_id": str(uuid.uuid4())}},
)

Each session_id maps one-to-one to a Synap conversation_id. Messages are persisted on every turn and replayed on the next invocation.

Automatic ingestion

SynapCallbackHandler covers a different need: it does not feed history into the model, it records every turn for long-term memory. Use it whenever you want a chain’s output to enrich the user’s profile and become searchable later.

from synap_langchain import SynapCallbackHandler

handler = SynapCallbackHandler(
    sdk=sdk,
    conversation_id=str(uuid.uuid4()),
    user_id="alice",
)

await chain.ainvoke(
    {"question": "Book me a flight to Tokyo next month."},
    config={"callbacks": [handler]},
)

You can attach the handler to a single invocation (as above) or globally with chain.with_config(callbacks=[handler]).

Semantic retrieval

SynapRetriever implements BaseRetriever. Use it inside ConversationalRetrievalChain, create_retrieval_chain, or any RAG pipeline that expects a retriever:

from synap_langchain import SynapRetriever

retriever = SynapRetriever(
    sdk=sdk,
    user_id="alice",
    customer_id="acme",
    max_results=8,
    mode="fast",   # "fast" (vector + graph, no LLM decomposition) or "accurate" (adds LLM subquery decomposition + reranking)
)

docs = await retriever.aget_relevant_documents("project deadlines")
# Document.page_content = memory text
# Document.metadata = {"confidence": 0.92, "type": "fact", ...}

The two retrieval modes trade latency against comprehensiveness:

	`fast`	`accurate`
Search	Vector + graph (no LLM subquery decomposition)	Vector + graph + LLM subquery decomposition + reranking
Best for	Real-time chat	Multi-entity queries

fast is lower-latency and suited to the hot path; accurate adds LLM-driven query decomposition and reranking for relationship-aware queries at a higher latency cost. See Context Fetch for the full retrieval contract.

Agent-callable memory

For agent-style chains where the model decides when memory is relevant, expose SynapSearchTool and SynapStoreTool:

from langchain.agents import AgentExecutor, create_tool_calling_agent
from synap_langchain import SynapSearchTool, SynapStoreTool

tools = [
    SynapSearchTool(sdk=sdk, user_id="alice", customer_id="acme"),
    SynapStoreTool(sdk=sdk, user_id="alice", customer_id="acme"),
]

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools)

result = await executor.ainvoke({"input": "What did I say about my dietary preferences?"})

The tools surface as synap_search and synap_store in the model’s tool-calling schema, with descriptions that nudge the model toward calling them when context is needed.

Complete example: support assistant with memory

The following class assembles all four components into a single agent. It remembers conversation history, retrieves user-specific context, and lets the model store new facts explicitly.

import uuid
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain.agents import AgentExecutor, create_tool_calling_agent
from synap_langchain import (
    SynapChatMessageHistory,
    SynapCallbackHandler,
    SynapRetriever,
    SynapSearchTool,
    SynapStoreTool,
)


class SupportAssistant:
    def __init__(self, sdk, user_id: str, customer_id: str | None = None):
        self.sdk = sdk
        self.user_id = user_id
        self.customer_id = customer_id

        self.llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
        self.retriever = SynapRetriever(
            sdk=sdk, user_id=user_id, customer_id=customer_id,
            max_results=6, mode="fast",
        )
        self.tools = [
            SynapSearchTool(sdk=sdk, user_id=user_id, customer_id=customer_id),
            SynapStoreTool(sdk=sdk, user_id=user_id, customer_id=customer_id),
        ]

        prompt = ChatPromptTemplate.from_messages([
            ("system",
             "You are a support assistant. Use known context when relevant. "
             "Call synap_search if you need older context, and synap_store "
             "to remember new facts the user shares.\n\n"
             "Known context:\n{context}"),
            MessagesPlaceholder("history"),
            ("human", "{input}"),
            MessagesPlaceholder("agent_scratchpad"),
        ])

        agent = create_tool_calling_agent(self.llm, self.tools, prompt)
        executor = AgentExecutor(agent=agent, tools=self.tools)

        self.chain = RunnableWithMessageHistory(
            executor,
            get_session_history=self._history,
            input_messages_key="input",
            history_messages_key="history",
        )

    def _history(self, session_id: str) -> SynapChatMessageHistory:
        return SynapChatMessageHistory(
            sdk=self.sdk,
            conversation_id=session_id,
            user_id=self.user_id,
            customer_id=self.customer_id,
        )

    async def ask(self, session_id: str, message: str) -> str:
        # Retrieve relevant context for this turn
        docs = await self.retriever.aget_relevant_documents(message)
        context = "\n".join(f"- {d.page_content}" for d in docs) or "No prior context."

        # Auto-record this turn for future memory
        callback = SynapCallbackHandler(
            sdk=self.sdk,
            conversation_id=session_id,
            user_id=self.user_id,
            customer_id=self.customer_id,
        )

        result = await self.chain.ainvoke(
            {"input": message, "context": context},
            config={
                "configurable": {"session_id": session_id},
                "callbacks": [callback],
            },
        )
        return result["output"]


# Usage
assistant = SupportAssistant(sdk, user_id="alice", customer_id="acme")
reply = await assistant.ask(str(uuid.uuid4()), "I just upgraded to the Pro plan.")

Three things to notice in this pattern:

SynapChatMessageHistory keeps the chain itself stateful turn-to-turn.
SynapRetriever injects per-user long-term context as a system message, independent of the chat history.
SynapCallbackHandler persists each turn back to Synap so future invocations can retrieve it.

You can drop any one of these and still have a working agent. Adding all three gives you a memory loop that improves over time.

Advanced patterns

Multi-tenant scoping

Every component accepts the same scoping triple: user_id, optional customer_id, and optional conversation_id:

retriever = SynapRetriever(
    sdk=sdk,
    user_id="alice",
    customer_id="acme",                  # scope retrievals to acme's tenancy
    conversation_id=str(uuid.uuid4()),   # bias ranking toward this conversation
)

customer_id is required for B2B Synap instances and ignored on single-tenant instances. See Memory Scopes for the full hierarchy.

Combining automatic and explicit memory

The four components compose freely. A common production setup:

SynapChatMessageHistory for the running conversation buffer
SynapCallbackHandler for long-term ingestion (runs in parallel with history)
SynapRetriever injected as a system message before each turn
SynapSearchTool exposed to the model for explicit lookups when retrieval misses

The callback handler and the retriever do not conflict: the handler writes, the retriever reads. The chat-history component is orthogonal to both: it stores raw messages for replay, while the handler ingests structured memories.

Tuning retrieval mode per call

SynapRetriever accepts a default mode at construction time, but the underlying sdk.conversation.context.fetch() call accepts a mode override too. For a single high-recall query inside an otherwise low-latency agent, swap the retriever’s mode temporarily:

retriever.mode = "accurate"
docs = await retriever.aget_relevant_documents("Summarize everything about the Acme account.")
retriever.mode = "fast"

Failure semantics

The Synap integration follows the Synap-wide contract:

Retrieval failures degrade gracefully: SynapRetriever returns [] and logs an error
Callback failures degrade gracefully: SynapCallbackHandler logs an ERROR and your chain continues
Explicit tool calls surface failures: SynapSearchTool / SynapStoreTool raise SynapIntegrationError so the model can react

This is by design: the read path should never break a user-facing turn, while the write path must surface errors so callers know when persistence failed.

Going further

Patterns overview: reusable memory patterns across frameworks.
Cookbook overview: end-to-end worked examples.

Next steps

LangGraph

Checkpointer and cross-thread store for LangGraph state graphs.

Context Fetch

The retrieval API that powers SynapRetriever: fast vs accurate, scopes, and response shapes.

Ingestion

Direct ingestion API for custom pipelines that need finer control than the callback handler.

Memory Scopes

How user_id, customer_id, and conversation_id interact across retrievals.

​Overview

​Setup

​Basic integration

​Core concepts

​Persistent chat history

​Automatic ingestion

​Semantic retrieval

​Agent-callable memory

​Complete example: support assistant with memory

​Advanced patterns

​Multi-tenant scoping

​Combining automatic and explicit memory

​Tuning retrieval mode per call

​Failure semantics

​Going further

​Next steps

LangGraph

Context Fetch

Ingestion

Memory Scopes

Overview

Setup

Basic integration

Core concepts

Persistent chat history

Automatic ingestion

Semantic retrieval

Agent-callable memory

Complete example: support assistant with memory

Advanced patterns

Multi-tenant scoping

Combining automatic and explicit memory

Tuning retrieval mode per call

Failure semantics

Going further

Next steps