AI Agent Memory: Manual, Mem0, LangMem, & AWS AgentCore

import json
import os
import datetime
from typing import List, Dict, Any, Optional

class AdvancedAgentMemory:
    """Complete memory system with Procedural and Working memory"""

    def __init__(self, storage_dir: str = "./agent_memory"):
        self.storage_dir = storage_dir
        os.makedirs(storage_dir, exist_ok=True)

        # File paths
        self.files = {
            "facts": os.path.join(storage_dir, "facts_semantic.json"),
            "procedures": os.path.join(storage_dir, "procedures.json"),
            "episodes": os.path.join(storage_dir, "conversations_episodic.json")
        }

        # Load persistent memories
        self.facts = self._load_json(self.files["facts"], [])
        self.procedures = self._load_json(self.files["procedures"], {})
        self.episodes = self._load_json(self.files["episodes"], [])

        # Working memory (RAM only, resets on restart)
        # Stores items as: {"content": str, "importance": float, "timestamp": str}
        self.working_memory = []
        self.working_memory_capacity = 10

    def _load_json(self, path, default):
        if os.path.exists(path):
            with open(path, 'r') as f: return json.load(f)
        return default

    def _save_json(self, data, path):
        with open(path, 'w') as f: json.dump(data, f, indent=2)

    # --- Working Memory (Short-term / Attention) ---
    def update_working_memory(self, content: str, importance: float = 1.0):
        """Add item to working memory, keeping only high-importance items"""
        item = {
            "content": content,
            "importance": importance,
            "timestamp": datetime.datetime.now().isoformat()
        }
        self.working_memory.append(item)

        # Sort by importance (descending) and keep top N
        self.working_memory.sort(key=lambda x: x["importance"], reverse=True)
        if len(self.working_memory) > self.working_memory_capacity:
            self.working_memory = self.working_memory[:self.working_memory_capacity]

    # --- Procedural Memory (How-to knowledge) ---
    def learn_procedure(self, name: str, steps: List[str], description: str = ""):
        """Store a multi-step procedure"""
        self.procedures[name] = {
            "steps": steps,
            "description": description,
            "timestamp": datetime.datetime.now().isoformat()
        }
        self._save_json(self.procedures, self.files["procedures"])

    def get_procedure(self, query: str):
        """Find relevant procedure"""
        # Simple keyword matching
        for name, proc in self.procedures.items():
            if name.lower() in query.lower():
                return proc
        return None

    # --- Semantic & Episodic Memory (Facts & History) ---
    def add_fact(self, content: str):
        self.facts.append({"content": content, "timestamp": datetime.datetime.now().isoformat()})
        self._save_json(self.facts, self.files["facts"])

    def add_episode(self, user_msg: str, agent_msg: str):
        self.episodes.append({
            "user": user_msg, 
            "agent": agent_msg, 
            "timestamp": datetime.datetime.now().isoformat()
        })
        self._save_json(self.episodes, self.files["episodes"])

        # Update working memory with context
        self.update_working_memory(f"User asked: {user_msg}", importance=0.8)

    # --- Context Construction ---
    def build_context(self, query: str) -> str:
        """Assemble all memory types into a context string for the LLM"""

        # 1. Get high-importance working memory
        working_context = "\n".join([f"- {i['content']}" for i in self.working_memory])

        # 2. Search facts
        relevant_facts = [f["content"] for f in self.facts if any(w in f["content"].lower() for w in query.lower().split())]

        # 3. Check for procedures
        proc = self.get_procedure(query)
        proc_text = ""
        if proc:
            steps = "\n".join([f"  {i+1}. {s}" for i, s in enumerate(proc["steps"])])
            proc_text = f"Procedure '{query}':\n{steps}"

        return f"""
=== Working Memory (Current Focus) ===
{working_context}

=== Relevant Facts ===
{chr(10).join(relevant_facts[:3])}

=== Relevant Procedures ===
{proc_text}

=== Recent History ===
{chr(10).join([f"U: {e['user']}\nA: {e['agent']}" for e in self.episodes[-3:]])}
"""

# Usage
agent_mem = AdvancedAgentMemory()

# 1. Teach a procedure
agent_mem.learn_procedure(
    "deploy website", 
    ["npm run build", "docker build .", "kubectl apply -f deployment.yaml"]
)

# 2. Add to working memory (urgent task)
agent_mem.update_working_memory("User needs to fix production bug ASAP", importance=5.0)

# 3. Generate context for a query
context = agent_mem.build_context("How do I deploy website?")
print(context)

from langgraph.prebuilt import create_react_agent
from langgraph.store.memory import InMemoryStore
from langmem import create_manage_memory_tool, create_search_memory_tool
from langchain_openai import ChatOpenAI

# 1. Setup Storage
store = InMemoryStore(
    index={
        "dims": 1536,
        "embed": "openai:text-embedding-3-small",
    }
)

# 2. Create Agent with Memory Tools
# These tools let the agent decide WHEN to save or search memory
agent = create_react_agent(
    ChatOpenAI(model="gpt-4"),
    tools=[
        create_manage_memory_tool(namespace=("user_memories",)),
        create_search_memory_tool(namespace=("user_memories",)),
    ],
    store=store,
)

# 3. Usage
# The agent will "subconsciously" use the tools to save important info
agent.invoke({"messages": [{"role": "user", "content": "I'm allergic to peanuts."}]})

# Later, it searches memory to answer
response = agent.invoke({"messages": [{"role": "user", "content": "Can I eat Satay sauce?"}]})
# Agent: "No, Satay sauce typically contains peanuts, and you mentioned you are allergic."

import chromadb
from chromadb.utils import embedding_functions

class ChromaSemanticMemory:
    """Semantic memory using ChromaDB"""

    def __init__(self, collection_name="knowledge"):
        self.client = chromadb.PersistentClient(path="./chroma_db")

        # Use OpenAI embeddings
        self.embedding_fn = embedding_functions.OpenAIEmbeddingFunction(
            model_name="text-embedding-3-small"
        )

        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            embedding_function=self.embedding_fn
        )

    def add_knowledge(self, content, metadata=None):
        """Add knowledge to the database"""
        self.collection.add(
            documents=[content],
            metadatas=[metadata or {}]
        )

    def search(self, query, n_results=3):
        """Search for semantically similar knowledge"""
        results = self.collection.query(
            query_texts=[query],
            n_results=n_results
        )
        return results['documents'][0]  # Returns list of relevant content

# Usage
memory = ChromaSemanticMemory()

# Add knowledge
memory.add_knowledge("Alice prefers Python over JavaScript")
memory.add_knowledge("Alice is building a recommendation system")

# Search semantically
results = memory.search("What programming language does Alice like?")
# Returns: ["Alice prefers Python over JavaScript"]
# Even though query doesn't match exactly, semantic search finds it

import os
from mem0 import MemoryClient

# 1. Initialize Client
# Get API Key from: https://app.mem0.ai/
os.environ["MEM0_API_KEY"] = "m0-xxx-your-api-key"
client = MemoryClient()

def chat_managed(user_id, message):
    # 2. Add Memory (Platform handles extraction & storage)
    # The platform automatically extracts facts like "User likes Python"
    client.add(message, user_id=user_id)

    # 3. Search Memory
    # The platform performs semantic search on its hosted vector DB
    memories = client.search(message, user_id=user_id)

    # Format for LLM
    context = "\n".join([m['memory'] for m in memories['results']])
    return context

# Usage
chat_managed("alice_123", "I am a vegan and I love spicy food.")
print(chat_managed("alice_123", "What should I eat for dinner?"))
# Returns: "User is vegan. User loves spicy food."

import os
from mem0 import Memory
import supabase
from datetime import datetime

class SelfHostedMemoryAgent:
    def __init__(self):
        # 1. Initialize Supabase (SQL)
        # We use this for raw logs and audit trails
        self.db = supabase.create_client(
            os.getenv("SUPABASE_URL"), 
            os.getenv("SUPABASE_KEY")
        )

        # 2. Initialize Mem0 (Vector Memory)
        # We tell Mem0 OSS to store its vectors inside OUR Supabase instance
        # NOTE: 'Memory' class is used for OSS, 'MemoryClient' for Platform
        self.memory = Memory.from_config({
            "vector_store": {
                "provider": "supabase",
                "config": {
                    "connection_string": os.getenv("DATABASE_URL"),
                    "collection_name": "mem0_vectors"
                }
            }
        })

    def chat(self, user_id: str, message: str):
        # A. Retrieve context from Vector Memory (Semantic)
        # Searches your Supabase vector table
        relevant = self.memory.search(message, user_id=user_id, limit=3)
        context = "\n".join([r['memory'] for r in relevant['results']])

        # B. Get recent history from SQL (Linear)
        # Queries your Supabase 'chat_logs' table
        history = self.db.table("chat_logs")\
            .select("*")\
            .eq("user_id", user_id)\
            .order("created_at", desc=True)\
            .limit(5)\
            .execute()

        # C. Generate Response (Pseudo-code)
        response = f"Simulated response based on: {context}"

        # D. Save to Both Systems
        # 1. Save raw log to SQL (The "Truth")
        self.db.table("chat_logs").insert({
            "user_id": user_id,
            "message": message,
            "response": response,
            "created_at": datetime.now().isoformat()
        }).execute()

        # 2. Save to Mem0 (The "Knowledge")
        # Mem0 OSS will analyze this, extract facts, and insert vectors into Supabase
        self.memory.add([
            {"role": "user", "content": message},
            {"role": "assistant", "content": response}
        ], user_id=user_id)

        return response

from bedrock_agentcore.memory import MemoryClient
from bedrock_agentcore.memory.session import MemorySessionManager
from bedrock_agentcore.memory.constants import ConversationalMessage, MessageRole
from typing import List, Dict, Optional
from datetime import datetime

class AWSAgentCOREMemory:
    """Agent using AWS Bedrock AgentCORE Memory"""

    def __init__(self, region_name="us-east-1", memory_name="AgentMemory"):
        self.memory_client = MemoryClient(region_name=region_name)

        # 1. Create Memory Resource
        # We define strategies here: "Watch for User Preferences" and "Watch for Facts"
        self.memory = self._get_or_create_memory(memory_name)
        self.memory_id = self.memory['id']

        self.session_manager = MemorySessionManager(
            memory_id=self.memory_id,
            region_name=region_name
        )

    def _get_or_create_memory(self, name: str) -> Dict:
        try:
            # Check if exists...
            memories = self.memory_client.list_memories()
            for mem in memories.get('memories', []):
                if mem.get('name') == name: return mem

            # Create new with STRATEGIES
            # These strategies tell AWS what to look for in conversations
            return self.memory_client.create_memory(
                name=name,
                description="Memory store for AI agent",
                eventExpiryDuration=30,
                memoryStrategies=[
                    {
                        "userPreferenceMemoryStrategy": {
                            "name": "UserPreferences",
                            "namespaces": ["agent/{actorId}/preferences"]
                        }
                    },
                    {
                        "semanticMemoryStrategy": {
                            "name": "SemanticKnowledge",
                            "namespaces": ["agent/{actorId}/knowledge"]
                        }
                    }
                ]
            )
        except Exception as e:
            print(f"Error: {e}")
            raise

    def chat(self, user_input: str, user_id: str, session_id: str, llm_callback):
        # 1. Retrieve Context
        # AWS automatically indexed previous chats into these "records"
        long_term_memories = self.retrieve_long_term_memories(user_id, user_input)

        # 2. Generate Response
        context_str = "\n".join([m['content']['text'] for m in long_term_memories])
        response = llm_callback(user_input, context_str)

        # 3. Store Interaction
        # We just dump the raw text here.
        # AWS's background process will wake up, read this, and extract new facts automatically.
        self.store_interaction(user_id, session_id, user_input, response)

        return response

    # ... (Helper methods for storage/retrieval omitted for brevity) ...

from typing import TypedDict, List, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI
from mem0 import MemoryClient

# 1. Define State
class State(TypedDict):
    # 'messages' tracks the conversation history (Short-term memory)
    messages: Annotated[List[HumanMessage | AIMessage], add_messages]
    # 'user_id' is needed to look up the correct memories
    user_id: str
    # 'memory_context' holds the retrieved facts from Mem0
    memory_context: str

# Initialize clients
mem0 = MemoryClient(api_key="your-mem0-key")
llm = ChatOpenAI(model="gpt-4o")

# 2. Define Nodes

def retrieve_memory_node(state: State):
    """
    Step 1: Look up relevant long-term memories based on the last user message.
    """
    last_user_msg = state["messages"][-1].content

    # Search Mem0 Platform
    results = mem0.search(last_user_msg, user_id=state["user_id"])

    # Format results into a string
    context_str = ""
    if results and "results" in results:
        context_str = "\n".join([f"- {m['memory']}" for m in results["results"]])

    return {"memory_context": context_str}

def generate_response_node(state: State):
    """
    Step 2: Generate a response using both conversation history AND long-term memory.
    """
    system_prompt = f"""You are a helpful assistant.

    Here is what we know about this user from past conversations:
    {state['memory_context']}

    Use this context to personalize your answer.
    """

    messages = [SystemMessage(content=system_prompt)] + state["messages"]
    response = llm.invoke(messages)

    return {"messages": [response]}

def save_memory_node(state: State):
    """
    Step 3: Save the new interaction to Mem0 so we remember it next time.
    """
    last_user_msg = state["messages"][-2].content
    last_ai_msg = state["messages"][-1].content

    # Add to Mem0 Platform (It will extract facts automatically)
    mem0.add([
        {"role": "user", "content": last_user_msg},
        {"role": "assistant", "content": last_ai_msg}
    ], user_id=state["user_id"])

    # This node doesn't return state updates, it's just a side-effect
    return {}

# 3. Build Graph
workflow = StateGraph(State)

workflow.add_node("retrieve", retrieve_memory_node)
workflow.add_node("generate", generate_response_node)
workflow.add_node("save", save_memory_node)

workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "save")
workflow.add_edge("save", END)

app = workflow.compile()

# 4. Run
result = app.invoke({
    "messages": [HumanMessage(content="I'm moving to Tokyo next month!")],
    "user_id": "user_123"
})

from langgraph.prebuilt import create_react_agent
from langgraph.store.memory import InMemoryStore
from langmem import create_manage_memory_tool, create_search_memory_tool
from langchain_openai import ChatOpenAI

# 1. Setup Storage (Using In-Memory for demo, use Postgres for prod)
store = InMemoryStore(
    index={
        "dims": 1536,
        "embed": "openai:text-embedding-3-small",
    }
)

# 2. Define Model
model = ChatOpenAI(model="gpt-4o")

# 3. Create Tools
# These tools allow the agent to actively "manage" its memory
tools = [
    create_manage_memory_tool(namespace=("user_memories",)),
    create_search_memory_tool(namespace=("user_memories",))
]

# 4. Create Agent
# The 'store' is automatically injected into the tools
agent = create_react_agent(model, tools=tools, store=store)

# 5. Run
# The agent will decide to call 'manage_memory' to save this fact
agent.invoke({
    "messages": [{"role": "user", "content": "My favorite color is green."}]
})

import os
from crewai import Agent, Task, Crew, Process

# 1. Setup Environment
os.environ["MEM0_API_KEY"] = "your-mem0-key"
os.environ["OPENAI_API_KEY"] = "your-openai-key"

# 2. Define Agent
# Note: We set memory=True here to enable short-term memory
analyst = Agent(
    role="Tech Analyst",
    goal="Analyze technology trends",
    backstory="You are an expert analyst who remembers past insights.",
    verbose=True,
    memory=True
)

# 3. Define Task
task = Task(
    description="What did we discuss about Generative AI last week?",
    expected_output="A summary of past discussions.",
    agent=analyst
)

# 4. Define Crew with Mem0 Provider
# This tells CrewAI to use Mem0 for Long-Term Memory
my_crew = Crew(
    agents=[analyst],
    tasks=[task],
    process=Process.sequential,
    memory=True,
    memory_config={
        "provider": "mem0",
        "config": {
            "user_id": "user_123",
            "org_id": "my_org"  # Optional
        }
    }
)

# 5. Run
result = my_crew.kickoff()

from crewai import Agent, Task, Crew
from crewai.tools import tool
from bedrock_agentcore.memory import MemoryClient

# 1. Create the Tool
class AWSMemoryTools:
    def __init__(self):
        self.client = MemoryClient(region_name="us-east-1")
        self.memory_id = "your-memory-id" # Created previously

    @tool("Save to AWS Memory")
    def save_memory(self, content: str, user_id: str):
        """Useful for saving important facts or preferences to long-term memory."""
        # This creates an event that AWS will process asynchronously
        self.client.create_event(
            memory_id=self.memory_id,
            actor_id=user_id,
            messages=[("content", content)]
        )
        return "Memory saved to AWS."

    @tool("Search AWS Memory")
    def search_memory(self, query: str, user_id: str):
        """Useful for searching past interactions and facts."""
        results = self.client.retrieve_memory_records(
            memory_id=self.memory_id,
            namespace=f"agent/{user_id}/preferences",
            searchCriteria={"searchQuery": query}
        )
        return str(results)

# 2. Assign Tools to Agent
aws_tools = AWSMemoryTools()
agent = Agent(
    role="AWS Assistant",
    goal="Assist user while remembering context",
    backstory="You are powered by AWS Bedrock Memory.",
    tools=[aws_tools.save_memory, aws_tools.search_memory]
)

# 3. Create Crew
crew = Crew(agents=[agent], tasks=[...])

import autogen
from mem0 import MemoryClient

# 1. Setup
mem0 = MemoryClient(api_key="your-key")
user_id = "user_123"

config_list = [{"model": "gpt-4", "api_key": "your-openai-key"}]

# 2. Define Agent
assistant = autogen.AssistantAgent(
    name="assistant",
    llm_config={"config_list": config_list}
)

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="ALWAYS",
    code_execution_config=False
)

# 3. Define the Hook
def memory_reply_hook(recipient, messages, sender, config):
    """
    This function runs every time the assistant receives a message.
    We hijack it to inject memory context.
    """
    last_msg = messages[-1]['content']

    # A. Search Memory
    results = mem0.search(last_msg, user_id=user_id)
    memories = "\n".join([m['memory'] for m in results.get('results', [])])

    if memories:
        # B. Inject Context (Invisible to user, visible to LLM)
        # We append the memory to the user's message in the agent's view
        print(f"\n[System] Found relevant memories: {memories}")

        # Modify the last message in the list
        messages[-1]['content'] = f"""
        User Message: {last_msg}

        [Relevant Memories from Past Conversations]
        {memories}
        """

    return False, None  # Return False to let the agent generate the reply normally

# 4. Register the Hook
assistant.register_reply(autogen.Agent, memory_reply_hook, position=1)

# 5. Start Chat
user_proxy.initiate_chat(assistant, message="What should I buy for my dog?")
# If memory has "User has a Golden Retriever", the agent will know!

# Assuming 'agent_mem' is your instance of AdvancedAgentMemory

def manual_memory_hook(recipient, messages, sender, config):
    last_msg = messages[-1]['content']

    # 1. Generate Context using our manual class
    # This pulls from Working Memory, Procedures, and Facts
    context = agent_mem.build_context(last_msg)

    # 2. Inject
    messages[-1]['content'] = f"{last_msg}\n\n[Internal Memory]\n{context}"

    # 3. Update Memory (Save the interaction)
    # We do this after the reply is generated usually, but for simplicity:
    agent_mem.update_working_memory(f"User asking about: {last_msg}", importance=0.5)

    return False, None

assistant.register_reply(autogen.Agent, manual_memory_hook, position=1)