Memory

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:

Short-term memory (MemoryBase implementations) for current conversation/session state
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 Class	Description
`InMemoryMemory`	In-process message storage, ideal for local development and lightweight workloads.
`AsyncSQLAlchemyMemory`	Async SQLAlchemy-backed storage, supporting SQLite/PostgreSQL/MySQL and more.
`RedisMemory`	Redis-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:

Method	Description
`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:

Method	Description
`add`	Add `Msg` objects into storage.
`delete`	Delete `Msg` objects by IDs.
`delete_by_mark`	Delete by mark(s).
`size`	Return storage size.
`clear`	Remove all content.
`get_memory`	Return memory content as `list[Msg]`.
`update_messages_mark`	Update message marks.
`state_dict` / `load_state_dict`	Export/import backend state.

Long-Term Memory

AgentScope provides long-term memory abstractions for cross-session persistence and retrieval.

Class	Description
`LongTermMemoryBase`	Abstract base class for long-term memory interfaces.
`Mem0LongTermMemory`	Long-term memory implementation powered by mem0.
`ReMePersonalLongTermMemory`	Personal long-term memory implementation based on ReMe.

Long-Term Memory Modes in ReActAgent

ReActAgent supports three long-term memory modes:

Mode	Behavior
`agent_control`	Agent autonomously uses memory tools.
`static_control`	Developer explicitly calls memory APIs in code.
`both`	Enables 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:

API	Typical 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:

Method	Required For
`record` / `retrieve`	`static_control`
`record_to_memory` / `retrieve_from_memory`	`agent_control`

If your backend supports all methods, it can be used in both mode.

Agent

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

Tool

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

Get Started

Tutorial

Out-of-box Agents

Basic Concepts

Building Blocks

Observe & Evaluate

Tune Agent

Deploy & Serve

Others

Short-Term Memory

Common API

InMemoryMemory: Basic Usage

AsyncSQLAlchemyMemory: Basic Usage

AsyncSQLAlchemyMemory as Context Manager

SQLAlchemy Memory with FastAPI Pooling

RedisMemory: Basic Usage

Redis Memory with FastAPI Pooling

Customizing Short-Term Memory

Long-Term Memory

Long-Term Memory Modes in ReActAgent

Mem0LongTermMemory: Basic Usage

ReMePersonalLongTermMemory: Usage Patterns

Integrating Long-Term Memory with ReActAgent

Customizing Long-Term Memory

Further Reading

Agent

Tool

​Short-Term Memory

​Common API

​InMemoryMemory: Basic Usage

​AsyncSQLAlchemyMemory: Basic Usage

​AsyncSQLAlchemyMemory as Context Manager

​SQLAlchemy Memory with FastAPI Pooling

​RedisMemory: Basic Usage

​Redis Memory with FastAPI Pooling

​Customizing Short-Term Memory

​Long-Term Memory

​Long-Term Memory Modes in ReActAgent

​Mem0LongTermMemory: Basic Usage

​ReMePersonalLongTermMemory: Usage Patterns

​Integrating Long-Term Memory with ReActAgent

​Customizing Long-Term Memory

​Further Reading

Agent

Tool

Short-Term Memory

Common API

InMemoryMemory: Basic Usage

AsyncSQLAlchemyMemory: Basic Usage

AsyncSQLAlchemyMemory as Context Manager

SQLAlchemy Memory with FastAPI Pooling

RedisMemory: Basic Usage

Redis Memory with FastAPI Pooling

Customizing Short-Term Memory

Long-Term Memory

Long-Term Memory Modes in ReActAgent

Mem0LongTermMemory: Basic Usage

ReMePersonalLongTermMemory: Usage Patterns

Integrating Long-Term Memory with ReActAgent

Customizing Long-Term Memory

Further Reading