How to Give AutoGen Agents Persistent Memory Across Conversations

# AutoGen + Synap — persistent agent memory
import autogen
from synap import SynapClient

synap = SynapClient(api_key="syn_your_key")

# Before agent run: inject user memory
user_context = synap.recall("user_456")
system_message = f"User context: {user_context}\n\nYou are a helpful assistant."

assistant = autogen.AssistantAgent(
    name="assistant",
    system_message=system_message,
    llm_config={"model": "gpt-4"}
)

# After agent run: store new memories
synap.remember("user_456", "User is building an e-commerce recommendation engine")

pip install synap-memory pyautogen

import autogen
from synap import SynapClient

synap = SynapClient(api_key="syn_your_key")

def run_agent_session(user_id: str, user_message: str) -> str:
    # 1. Recall relevant context from past sessions
    user_context = synap.recall(user_id, query=user_message)

    # 2. Inject memory into the system prompt
    system_message = f"""You are a helpful software engineering assistant.

Context from previous sessions with this user:
{user_context if user_context else "No previous context yet."}

Use this context to give personalized, continuous assistance.
Do not ask for information the user has already provided."""

    # 3. Set up AutoGen agents with injected memory
    assistant = autogen.AssistantAgent(
        name="assistant",
        system_message=system_message,
        llm_config={"model": "gpt-4", "temperature": 0},
    )
    user_proxy = autogen.UserProxyAgent(
        name="user_proxy",
        human_input_mode="NEVER",
        max_consecutive_auto_reply=1,
    )

    # 4. Run the agent conversation
    user_proxy.initiate_chat(assistant, message=user_message)
    response = user_proxy.last_message()["content"]

    # 5. Store what was learned for next time
    synap.remember(user_id, f"User asked: {user_message[:300]}")
    synap.remember(user_id, f"Assistant answered: {response[:300]}")

    return response


# Session 1 — user shares project context
run_agent_session("user_456", "I'm building a recommendation engine using collaborative filtering.")
run_agent_session("user_456", "We're using PostgreSQL for storage and Python + FastAPI for the backend.")

# --- Next day, new Python process ---

# Session 2 — agents already know the project
response = run_agent_session("user_456", "What database should I use for storing user interaction logs?")
# Agent knows: PostgreSQL is already in the stack, recommends it without asking
print(response)

project_id = "proj_ecommerce_rec_engine"

# Planner agent stores architectural decisions
synap.remember(project_id, "Architecture decision: use Redis for session caching")
synap.remember(project_id, "Decided to use collaborative filtering over content-based")

# Executor agent recalls the same decisions — no sync code needed
context = synap.recall(project_id, query="what caching layer should I use?")
# Returns: "Architecture decision: use Redis for session caching"

executor_system = f"Project context:\n{context}\n\nYou implement decisions already made."

from crewai import Agent, Task, Crew
from synap import SynapClient

synap = SynapClient(api_key="syn_your_key")

def run_crew_session(user_id: str, goal: str):
    # Recall user context before spinning up the crew
    user_context = synap.recall(user_id, query=goal)

    researcher = Agent(
        role="Senior Researcher",
        goal=f"Research {goal} thoroughly",
        backstory=f"""You are an expert researcher with deep domain knowledge.
Context about this user from previous sessions:
{user_context if user_context else "New user — no prior context."}
Use this to tailor your research to their specific situation.""",
        verbose=False,
    )

    writer = Agent(
        role="Technical Writer",
        goal="Write clear, actionable summaries",
        backstory=f"""You produce concise technical documentation.
User context: {user_context}""",
        verbose=False,
    )

    research_task = Task(
        description=f"Research: {goal}",
        agent=researcher,
        expected_output="Detailed research findings",
    )
    write_task = Task(
        description="Summarize findings for the user",
        agent=writer,
        expected_output="Concise summary with action items",
    )

    crew = Crew(agents=[researcher, writer], tasks=[research_task, write_task])
    result = crew.kickoff()

    # Store what the crew learned
    synap.remember(user_id, f"Research completed: {goal}")
    synap.remember(user_id, f"Key findings: {str(result)[:400]}")

    return result