Skip to main content
This document covers practical usage of memory modules in AgentScope, including short-term memory backends and long-term memory integrations. The memory module in AgentScope is responsible for:
  • storing messages, and
  • managing messages with marks across different storage implementations.
A mark is a string label associated with each message. It is commonly used to categorize, filter, and retrieve messages based on context or purpose. This mechanism supports high-level memory management in agents. For example, in ReActAgent, hint messages are typically stored with mark hint, and memory-aware workflows (such as compression pipelines) can be organized around marks.
The memory module focuses on storage and management. Algorithmic logic (for example, compression strategy) is implemented at the agent layer.
For conceptual background, see Context and Memory.
In AgentScope, memory can be viewed in two layers:
  1. Short-term memory (MemoryBase implementations) for current conversation/session state
  2. Long-term memory (LongTermMemoryBase implementations) for cross-session persistence and retrieval

​
Short-Term Memory

Short-term memory stores Msg objects and supports optional marks (for example: hint, summary, tool_result) for filtering and lifecycle management. Built-in short-term memory implementations:
Memory ClassDescription
InMemoryMemoryIn-process message storage, ideal for local development and lightweight workloads.
AsyncSQLAlchemyMemoryAsync SQLAlchemy-backed storage, supporting SQLite/PostgreSQL/MySQL and more.
RedisMemoryRedis-backed storage for distributed deployments and shared session state.

​
Common API

All short-term memory classes inherit from MemoryBase and expose a unified async API:
MethodDescription
add(memories, marks=None)Add one or multiple Msg objects with optional marks.
delete(msg_ids)Delete messages by IDs.
delete_by_mark(mark)Delete messages by mark(s).
size()Return message count in memory.
clear()Remove all messages from memory.
get_memory(mark=None, exclude_mark=None)Retrieve messages with optional include/exclude mark filters.
update_messages_mark(new_mark, old_mark=None, msg_ids=None)Add/remove/replace marks on selected messages.
update_compressed_summary(summary)Update compressed summary data used by memory-aware agents.
state_dict() / load_state_dict(...)Export/import memory state if supported by backend.

​
InMemoryMemory: Basic Usage

InMemoryMemory is the easiest option for quick testing and local prototyping. 
import asyncio
import json
from agentscope.memory import InMemoryMemory
from agentscope.message import Msg

async def in_memory_example():
    memory = InMemoryMemory()

    await memory.add(
        Msg("Alice", "Generate a report about AgentScope", "user"),
    )

    await memory.add(
        Msg(
            "system",
            "<system-hint>Create a plan first and then proceed step by step.</system-hint>",
            "system",
        ),
        marks="hint",
    )

    hint_msgs = await memory.get_memory(mark="hint")
    print("Messages with mark 'hint':")
    for msg in hint_msgs:
        print("-", msg)

    state = memory.state_dict()
    print("Current memory state:")
    print(json.dumps(state, indent=2))

    deleted_count = await memory.delete_by_mark("hint")
    print(f"Deleted {deleted_count} message(s).")

asyncio.run(in_memory_example())

​
AsyncSQLAlchemyMemory: Basic Usage

AsyncSQLAlchemyMemory works with either an async engine or an async session. 
import asyncio
from sqlalchemy.ext.asyncio import create_async_engine
from agentscope.memory import AsyncSQLAlchemyMemory
from agentscope.message import Msg

async def sqlalchemy_example():
    engine = create_async_engine("sqlite+aiosqlite:///./test_memory.db")

    memory = AsyncSQLAlchemyMemory(
        engine_or_session=engine,
        user_id="user_1",
        session_id="session_1",
    )

    await memory.add(Msg("Alice", "Generate a report about AgentScope", "user"))
    await memory.add(
        Msg(
            "system",
            "<system-hint>Draft a plan before writing the report.</system-hint>",
            "system",
        ),
        marks="hint",
    )

    msgs = await memory.get_memory(mark="hint")
    for msg in msgs:
        print(msg)

    await memory.close()

asyncio.run(sqlalchemy_example())

​
AsyncSQLAlchemyMemory as Context Manager

When used as an async context manager, session cleanup is handled automatically. 
import asyncio
from sqlalchemy.ext.asyncio import create_async_engine
from agentscope.memory import AsyncSQLAlchemyMemory
from agentscope.message import Msg

async def sqlalchemy_context_example():
    engine = create_async_engine("sqlite+aiosqlite:///./test_memory.db")

    async with AsyncSQLAlchemyMemory(
        engine_or_session=engine,
        user_id="user_1",
        session_id="session_1",
    ) as memory:
        await memory.add(Msg("Alice", "Hello memory", "user"))
        msgs = await memory.get_memory()
        print(f"Total messages: {len(msgs)}")

asyncio.run(sqlalchemy_context_example())

​
SQLAlchemy Memory with FastAPI Pooling

In production, create and reuse an async engine/session maker with pooling: 
from typing import AsyncGenerator
from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from agentscope.memory import AsyncSQLAlchemyMemory

app = FastAPI()

engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost:5432/agentscope",
    pool_size=10,
    max_overflow=20,
    pool_timeout=30,
)

async_session_maker = async_sessionmaker(
    engine,
    expire_on_commit=False,
    autocommit=False,
    autoflush=False,
)

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

@app.post("/chat")
async def chat_endpoint(
    user_id: str,
    session_id: str,
    db_session: AsyncSession = Depends(get_db),
):
    memory = AsyncSQLAlchemyMemory(
        engine_or_session=db_session,
        user_id=user_id,
        session_id=session_id,
    )
    # Use `memory` in your agent pipeline.
    return {"ok": True}

​
RedisMemory: Basic Usage

RedisMemory is suitable for distributed services and horizontally scaled workers. 
import asyncio
import fakeredis
from agentscope.memory import RedisMemory
from agentscope.message import Msg

async def redis_memory_example():
    fake_redis = fakeredis.aioredis.FakeRedis(decode_responses=True)

    memory = RedisMemory(
        connection_pool=fake_redis.connection_pool,
        user_id="user_1",
        session_id="session_1",
    )

    await memory.add(Msg("Alice", "Generate a report about AgentScope", "user"))
    await memory.add(
        Msg(
            "system",
            "<system-hint>Create a plan first.</system-hint>",
            "system",
        ),
        marks="hint",
    )

    hint_msgs = await memory.get_memory(mark="hint")
    print(hint_msgs)

    client = memory.get_client()
    await client.aclose()

asyncio.run(redis_memory_example())

​
Redis Memory with FastAPI Pooling

Use a global Redis connection pool and create RedisMemory per request: 
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from redis.asyncio import ConnectionPool
from agentscope.memory import RedisMemory

redis_pool: ConnectionPool | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global redis_pool
    redis_pool = ConnectionPool(
        host="localhost",
        port=6379,
        db=0,
        decode_responses=True,
        max_connections=20,
    )
    yield
    await redis_pool.disconnect()

app = FastAPI(lifespan=lifespan)

@app.post("/chat")
async def chat_endpoint(user_id: str, session_id: str):
    if redis_pool is None:
        raise HTTPException(status_code=500, detail="Redis pool not initialized")

    memory = RedisMemory(
        connection_pool=redis_pool,
        user_id=user_id,
        session_id=session_id,
    )
    # Use `memory` in your agent pipeline.
    return {"ok": True}

​
Customizing Short-Term Memory

To build a custom short-term memory backend, inherit from MemoryBase and implement the required methods:
MethodDescription
addAdd Msg objects into storage.
deleteDelete Msg objects by IDs.
delete_by_markDelete by mark(s).
sizeReturn storage size.
clearRemove all content.
get_memoryReturn memory content as list[Msg].
update_messages_markUpdate message marks.
state_dict / load_state_dictExport/import backend state.

​
Long-Term Memory

AgentScope provides long-term memory abstractions for cross-session persistence and retrieval.
ClassDescription
LongTermMemoryBaseAbstract base class for long-term memory interfaces.
Mem0LongTermMemoryLong-term memory implementation powered by mem0.
ReMePersonalLongTermMemoryPersonal long-term memory implementation based on ReMe.

​
Long-Term Memory Modes in ReActAgent

ReActAgent supports three long-term memory modes:
ModeBehavior
agent_controlAgent autonomously uses memory tools.
static_controlDeveloper explicitly calls memory APIs in code.
bothEnables both patterns.
When mode is agent_control or both, tool functions such as record_to_memory and retrieve_from_memory are registered in the toolkit.

​
Mem0LongTermMemory: Basic Usage

import asyncio
import os
from agentscope.memory import Mem0LongTermMemory
from agentscope.model import DashScopeChatModel
from agentscope.embedding import DashScopeTextEmbedding
from agentscope.message import Msg

long_term_memory = Mem0LongTermMemory(
    agent_name="Friday",
    user_name="user_123",
    model=DashScopeChatModel(
        model_name="qwen-max-latest",
        api_key=os.environ["DASHSCOPE_API_KEY"],
        stream=False,
    ),
    embedding_model=DashScopeTextEmbedding(
        model_name="text-embedding-v2",
        api_key=os.environ["DASHSCOPE_API_KEY"],
    ),
    on_disk=False,
)

async def mem0_basic_usage():
    await long_term_memory.record(
        [Msg("user", "I like staying in homestays when traveling.", "user")]
    )
    results = await long_term_memory.retrieve(
        [Msg("user", "What are my accommodation preferences?", "user")]
    )
    print(results)

asyncio.run(mem0_basic_usage())

​
ReMePersonalLongTermMemory: Usage Patterns

ReMePersonalLongTermMemory supports both tool-style APIs and direct APIs:
APITypical Usage
record_to_memory(...) / retrieve_from_memory(...)Tool-call style for agent_control workflows
record(...) / retrieve(...)Direct calls for static_control workflows
import os
from agentscope.memory import ReMePersonalLongTermMemory
from agentscope.embedding import DashScopeTextEmbedding
from agentscope.model import DashScopeChatModel

reme_long_term_memory = ReMePersonalLongTermMemory(
    agent_name="Friday",
    user_name="user_123",
    model=DashScopeChatModel(
        model_name="qwen3-max",
        api_key=os.environ["DASHSCOPE_API_KEY"],
        stream=False,
    ),
    embedding_model=DashScopeTextEmbedding(
        model_name="text-embedding-v4",
        api_key=os.environ["DASHSCOPE_API_KEY"],
        dimensions=1024,
    ),
)

​
Integrating Long-Term Memory with ReActAgent

import os
from agentscope.agent import ReActAgent
from agentscope.formatter import DashScopeChatFormatter
from agentscope.memory import InMemoryMemory
from agentscope.model import DashScopeChatModel
from agentscope.tool import Toolkit

agent = ReActAgent(
    name="Friday",
    sys_prompt="You are a helpful assistant with long-term memory.",
    model=DashScopeChatModel(
        model_name="qwen3-max",
        api_key=os.environ["DASHSCOPE_API_KEY"],
        stream=False,
    ),
    formatter=DashScopeChatFormatter(),
    toolkit=Toolkit(),
    memory=InMemoryMemory(),
    long_term_memory=long_term_memory,
    long_term_memory_mode="static_control",
)
For agent_control mode, add explicit instructions in the agent’s system prompt specifying when to record and retrieve memory. Without clear instructions, the agent may not use memory tools optimally.

​
Customizing Long-Term Memory

To implement your own long-term memory backend, inherit from LongTermMemoryBase and implement methods according to your target mode:
MethodRequired For
record / retrievestatic_control
record_to_memory / retrieve_from_memoryagent_control
If your backend supports all methods, it can be used in both mode.

​
Further Reading

Agent

Understand the agent’s core methods and the ReAct paradigm.

Tool

Learn how to extend agents with native functions, MCP, and skills.