How to Build Multi-Agent Systems with Shared Persistent Memory in Python (LangGraph + MemorySync)

Building multi-agent systems using frameworks like LangGraph, CrewAI, or AutoGen is one of the most exciting patterns in modern AI engineering.

Instead of a single massive prompt, you break tasks down into specialized agents:

• Supervisor Agent: Coordinates requirements, plans architecture, and establishes rules.
• Researcher Agent: Gathers documentation, API specs, and citations.
• Coder Agent: Implements and tests code based on architecture decisions.

---

The Problem: Context Drift and Amnesia Across Agents

Most multi-agent frameworks use short-term execution graphs or in-memory state objects (like LangGraph's StateGraph or thread checkpointers). While this passes messages during a single run, it creates two major production bottlenecks:

1. Context Drift: Feeding a 20-page web dump from a Researcher agent directly into the Coder agent wastes tokens, degrades reasoning, and causes hallucinations.
2. Cross-Session Amnesia: When the workflow finishes and the developer returns tomorrow, the agents start with a blank slate. The Coder forgets the architecture decisions made by the Supervisor yesterday.

Below, we build a production multi-agent workflow using LangGraph where agents share a scoped, persistent memory layer via MemorySync.

---

Architecture: Shared Scoped Memory vs. State Bloat

Rather than passing monolithic chat histories between nodes in your state graph, agents read and write to an external scoped memory store:

Dimension	Native LangGraph StateGraph	MemorySync Shared Memory Layer
Persistence	Ephemeral (resets on process restart)	Durable across sessions and restarts
Context Window Cost	Linear growth with every intermediate agent output	Constant: queries retrieve only top-k relevant facts
Agent Isolation	Monolithic state visible to all nodes	Scoped recall (tenant_id, project_id, agent_role)
Deduplication	None (duplicate facts bloat context)	Automatic semantic deduplication and compaction
Latency	In-memory serialization	Sub-50ms hybrid vector retrieval

---

Prerequisites and Installation

Install the required Python packages:

pip install langgraph langchain-openai memorysync-python

Set your API keys:

export OPENAI_API_KEY="sk-..."
export MEMORYSYNC_API_KEY="ms_..."

---

Step 1: Initialize the Scoped Memory Client

MemorySync organizes memories using Scopes (tenant_id, project_id, and user_id). This ensures multi-agent workflows in one company or project never pollute another:

from memorysync import MemorySync
import os

memory = MemorySync(
    api_key=os.environ["MEMORYSYNC_API_KEY"],
    endpoint="https://api.memorysync.io"
)

PROJECT_SCOPE = {
    "tenant_id": "org_acme_corp",
    "project_id": "ai_agent_swarm_v1",
    "user_id": "lead_dev_01"
}

---

Step 2: Define Shared Memory Tools for Agents

We equip our agents with two tools: store_shared_memory and recall_shared_memory:

from langchain_core.tools import tool

@tool
def store_shared_memory(content: str, category: str = "architecture"):
    """Store an architectural decision or constraint into shared memory."""
    result = memory.add(
        content=content,
        tenant_id=PROJECT_SCOPE["tenant_id"],
        project_id=PROJECT_SCOPE["project_id"],
        metadata={
            "category": category,
            "recorded_by": "agent_worker"
        }
    )
    return f"Memory stored successfully with ID: {result.id}"

@tool
def recall_shared_memory(query: str, top_k: int = 3):
    """Search and recall relevant past architectural decisions or constraints."""
    memories = memory.query(
        query=query,
        tenant_id=PROJECT_SCOPE["tenant_id"],
        project_id=PROJECT_SCOPE["project_id"],
        top_k=top_k
    )

    if not memories:
        return "No relevant memories found in project scope."

    formatted = []
    for m in memories:
        formatted.append(f"- [Score: {m.score:.2f}] {m.content}")
    return "\n".join(formatted)

---

Step 3: Build the Multi-Agent Workflow in LangGraph

Now we connect our agents using LangGraph's StateGraph:

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import operator

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    task: str
    current_agent: str

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)

def supervisor_node(state: AgentState):
    task = state["task"]
    past_context = recall_shared_memory.invoke({"query": task, "top_k": 3})

    system_prompt = f"""You are the System Architect.
Past Project Memory:
{past_context}

Break down the user's task into clear architectural constraints."""

    response = llm.invoke([
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": task}
    ])

    store_shared_memory.invoke({
        "content": response.content,
        "category": "architecture_constraint"
    })

    return {
        "messages": [response],
        "current_agent": "coder"
    }

def coder_node(state: AgentState):
    task = state["task"]
    recalled_rules = recall_shared_memory.invoke({
        "query": f"architecture constraints for {task}",
        "top_k": 2
    })

    system_prompt = f"""You are the Senior Implementation Engineer.
Follow these recalled project constraints strictly:
{recalled_rules}

Write clean, robust code that adheres to all project rules."""

    response = llm.invoke([
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": task}
    ])

    return {
        "messages": [response],
        "current_agent": "finished"
    }

workflow = StateGraph(AgentState)
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("coder", coder_node)

workflow.set_entry_point("supervisor")
workflow.add_edge("supervisor", "coder")
workflow.add_edge("coder", END)

app = workflow.compile()

---

Step 4: Run the Cross-Session Verification Test

First, run Session 1 where the Supervisor establishes a strict constraint:

# Session 1: Establish project convention
inputs = {
    "task": "Design our user auth endpoints. Constraint: All timestamps must be ISO-8601 UTC and tokens expire in 15 minutes.",
    "messages": [],
    "current_agent": "start"
}

print("=== Running Session 1 ===")
for output in app.stream(inputs):
    for key, value in output.items():
        print(f"[{key.upper()} Finished]: {value['messages'][-1].content[:150]}...\n")

Now, simulate a fresh process restart tomorrow. The Coder is asked to write a new profile endpoint without re-explaining the timestamp rule:

# Session 2: Fresh session, different prompt
fresh_inputs = {
    "task": "Write the GET /user/profile endpoint response handler.",
    "messages": [],
    "current_agent": "start"
}

print("=== Running Session 2 (Fresh Restart) ===")
for output in app.stream(fresh_inputs):
    for key, value in output.items():
        print(f"[{key.upper()} Output]:\n{value['messages'][-1].content}\n")

Result

The Coder agent automatically recalls:

"Constraint from past session: All timestamps must be ISO-8601 UTC and tokens expire in 15 minutes."

The generated endpoint includes datetime.now(timezone.utc).isoformat() without the developer ever re-typing the requirement.

---

Key Takeaways

1. Zero Prompt Bloat: Instead of passing 50,000-token histories across agents, workers query 2ΓÇô3 relevant facts into context.
2. Full Auditability: Every recalled fact carries an explicit memory ID and score, making agent behavior reproducible.
3. Multi-Tenant Isolation: tenant_id prevents data leakage across client accounts or disparate agent swarms.

---

Resources

• Interactive Documentation: docs.memorysync.io/guides/langgraph
• Model Context Protocol (MCP) Server: docs.memorysync.io/mcp
• Quickstart Guide: docs.memorysync.io/quickstart