MemLayer integrates naturally with LangGraph. The pattern is simple: add two nodes to your graph — one before your agent that loads relevant memories, and one after that saves what was learned. Your agent node doesn’t need to change at all; it just receives enriched state and returns updated state.
Prerequisites
Install both packages and make sure you have a MemLayer API key:
pip install memlayer-py langgraph
ml_live_xxx.
The Pattern
Here is a complete LangGraph graph with MemLayer memory nodes wired around a central agent node.
import os
from memlayer import AsyncMemLayerClient
from langgraph.graph import StateGraph, MessagesState
# Initialise the async client once at module level
maas = AsyncMemLayerClient(api_key=os.environ["MEMLAYER_API_KEY"])
async def load_memory(state: MessagesState):
"""Load relevant memories before the agent responds."""
last_message = state["messages"][-1].content
memories = await maas.recall(
query=last_message,
user_id=state["user_id"],
agent_id="my_agent",
top_k=5,
)
memory_context = "\n".join(f"- {m.content}" for m in memories)
state["memory_context"] = memory_context
return state
async def save_memory(state: MessagesState):
"""Save what the agent learned after responding."""
last_message = state["messages"][-1].content
await maas.remember(
content=last_message,
user_id=state["user_id"],
agent_id="my_agent",
memory_type="episodic",
)
return state
# Wire into your graph
builder = StateGraph(MessagesState)
builder.add_node("load_memory", load_memory)
builder.add_node("agent", your_agent_node)
builder.add_node("save_memory", save_memory)
builder.add_edge("load_memory", "agent")
builder.add_edge("agent", "save_memory")
graph = builder.compile()
load_memory node writes a memory_context string into the state. Your agent node reads it and uses it to build a context-aware system prompt. The save_memory node runs after the agent responds and persists the last message as an episodic memory.
Using context() at Session Start
For the very first message in a session, consider using context() instead of recall() in your load node. context() returns a broad snapshot of the user’s most important memories ranked by importance and recency — without needing a query.
async def load_memory(state: MessagesState):
"""Use context() for the first turn, recall() for subsequent turns."""
messages = state["messages"]
last_message = messages[-1].content
# First turn: load broad context
# Later turns: use the actual message as the search query
if len(messages) == 1:
result = await maas.context(
user_id=state["user_id"],
agent_id="my_agent",
top_k=10,
current_message=last_message, # pass first message for smarter retrieval
)
memories = result.memories
else:
memories = await maas.recall(
query=last_message,
user_id=state["user_id"],
agent_id="my_agent",
top_k=5,
)
state["memory_context"] = "\n".join(f"- {m.content}" for m in memories)
return state
Tips
Always use AsyncMemLayerClient with LangGraph. LangGraph executes nodes as coroutines. The sync MemLayerClient will block the event loop — use AsyncMemLayerClient and await every call.
Pass current_message to context() to enable semantic ranking instead of importance/recency ordering. For the first message in a session this dramatically improves the relevance of loaded memories.
Use importance scores to mark high-value memories. When saving a memory in save_memory, set importance=0.9 for critical facts (user escalations, confirmed account details) and leave the default 0.5 for routine conversation turns. High-importance memories rank higher in future retrievals.# Example: detect a complaint and save it with elevated importance
last_message = state["messages"][-1].content
is_complaint = "frustrated" in last_message.lower() or "complaint" in last_message.lower()
await maas.remember(
content=last_message,
user_id=state["user_id"],
agent_id="my_agent",
memory_type="episodic",
importance=0.9 if is_complaint else 0.5,
)
