Comparing InMemoryMemory, RedisMemory, and AsyncSQLAlchemyMemory in AgentScope

InMemoryMemory stores conversation history in a volatile Python list for single-process prototyping, RedisMemory persists data to a Redis server with automatic TTL for distributed multi-agent systems, and AsyncSQLAlchemyMemory leverages relational databases via SQLAlchemy for ACID-compliant persistence in production environments.

AgentScope provides a pluggable working memory system that abstracts storage operations behind the MemoryBase interface in src/agentscope/memory/_working_memory/_base.py. The library includes three concrete implementations—InMemoryMemory, RedisMemory, and AsyncSQLAlchemyMemory—each designed for distinct scalability, durability, and concurrency requirements. All three classes expose an identical async API (get_memory, add, delete, delete_by_mark, update_messages_mark, size, clear, close), allowing developers to swap backends without modifying agent logic.

InMemoryMemory: Volatile Single-Process Storage

InMemoryMemory is the default implementation for local development and unit testing, defined in src/agentscope/memory/_working_memory/_in_memory_memory.py. It stores messages in a simple Python list (self.content) where each entry is a tuple[Msg, list[str]] containing the message object and its associated marks.

Because data lives entirely within the interpreter process, this backend offers the lowest latency for read/write operations. However, persistence is volatile—all data disappears when the process exits. Concurrency is handled by wrapping synchronous list operations in async methods, though no actual I/O occurs. The class supports deduplication via the allow_duplicates flag, which filters messages by their id attribute using simple list comprehension in memory.

RedisMemory: Distributed Caching with TTL

RedisMemory connects to a Redis server using the redis.asyncio client, enabling durable storage that survives process restarts as long as Redis persists its data (RDB/AOF). The implementation in src/agentscope/memory/_working_memory/_redis_memory.py uses a sophisticated key structure:

• A Redis List for the session history
• Per-mark Lists for tagged message retrieval (user_id:{uid}:session:{sid}:mark:{mark})
• A Hash for message payloads
• A Set index (marks_index) for fast enumeration of available marks

This backend supports horizontal scaling—multiple agents across different processes can connect to the same Redis cluster. It offers optional time-to-live (TTL) support via the key_ttl parameter, which creates a sliding expiration window refreshed on every operation through the _refresh_session_ttl method. Deduplication is controlled by skip_duplicated (default True), implemented by scanning the session list before insertion. Redis pipelines group commands atomically using await pipe.execute() to ensure transaction safety without full ACID overhead.

AsyncSQLAlchemyMemory: Relational Database Persistence

AsyncSQLAlchemyMemory provides enterprise-grade persistence using SQLAlchemy's async ORM, defined in src/agentscope/memory/_working_memory/_sqlalchemy_memory.py. It supports SQLite, PostgreSQL, MySQL, and other relational databases through async engines like create_async_engine("sqlite+aiosqlite:///agent_memory.db").

The implementation uses three core tables: message for content storage, message_mark for tag associations with a composite primary key (msg_id, mark), and session for metadata isolation. Unlike Redis, this backend offers full ACID guarantees and complex query capabilities, handling deduplication via SELECT statements on the message table before insertion. Session and user isolation is enforced at the database level through user_id and session_id column filters in every query. While it lacks built-in TTL, administrators can implement expiration through database-native scheduled jobs or cleanup procedures.

Key Technical Differences

When selecting a backend, consider these architectural distinctions:

• Data Structure: InMemoryMemory uses a flat Python list; RedisMemory uses separate keys for sessions, marks, and indexes; AsyncSQLAlchemyMemory uses normalized relational tables with foreign key relationships.
• Persistence: InMemoryMemory is transient; RedisMemory and AsyncSQLAlchemyMemory provide durable storage.
• Concurrency: InMemoryMemory is single-process only; RedisMemory supports distributed multi-process agents; AsyncSQLAlchemyMemory scales with the underlying database's connection pooling.
• Duplicate Handling: InMemoryMemory filters duplicates in-process by message ID; RedisMemory scans the session list; AsyncSQLAlchemyMemory queries the message table.
• Transaction Safety: InMemoryMemory requires no transactions; RedisMemory uses pipelines for atomic command batches; AsyncSQLAlchemyMemory uses await self.session.commit() for full transactions.

Usage Examples

All three implementations share the same async interface, requiring only different constructor arguments.

Common Imports

from agentscope.message import Msg
from agentscope.memory._working_memory._in_memory_memory import InMemoryMemory
from agentscope.memory._working_memory._redis_memory import RedisMemory
from agentscope.memory._working_memory._sqlalchemy_memory import AsyncSQLAlchemyMemory

InMemoryMemory

mem = InMemoryMemory()

await mem.add(Msg("assistant", "Hello world!"))
msgs = await mem.get_memory()
print([m.content for m in msgs])  # Output: ['Hello world!']

RedisMemory

mem = RedisMemory(
    session_id="sess-123",
    user_id="user-abc",
    host="localhost",
    port=6379,
    key_prefix="agentscope:",
    key_ttl=3600,  # 1-hour sliding TTL

)

await mem.add(Msg("assistant", "Stored in Redis"))
await mem.add(Msg("assistant", "Important note"), marks="important")

# Retrieve only marked messages

important_msgs = await mem.get_memory(mark="important")

AsyncSQLAlchemyMemory

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine("sqlite+aiosqlite:///agent_memory.db")
mem = AsyncSQLAlchemyMemory(engine)

await mem.add(Msg("assistant", "Persisted in SQLite"))
await mem.add(Msg("assistant", "Review needed"), marks=["todo", "review"])

todo_msgs = await mem.get_memory(mark="todo")

Summary

• InMemoryMemory (_in_memory_memory.py) provides zero-configuration, volatile storage ideal for unit tests and single-process prototyping, storing data in a Python list with synchronous operations wrapped in async methods.
• RedisMemory (_redis_memory.py) offers distributed, durable caching with automatic TTL expiration, using Redis pipelines for atomic operations and supporting multi-user sessions through prefixed key namespaces.
• AsyncSQLAlchemyMemory (_sqlalchemy_memory.py) delivers full relational database persistence with ACID compliance, supporting complex queries and bulk operations through SQLAlchemy's async ORM.

Frequently Asked Questions

Can I switch between memory backends without changing agent code?

Yes. All three implementations inherit from MemoryBase and expose identical async methods including add, get_memory, delete, and clear. You only need to change the instantiation call—passing either InMemoryMemory(), RedisMemory(...) with connection details, or AsyncSQLAlchemyMemory(engine)—while the agent logic remains unchanged.

Does InMemoryMemory support multi-process deployments?

No. InMemoryMemory stores data in a local Python list (self.content) within the interpreter process. It cannot share state between processes or survive restarts. For distributed agents, use RedisMemory or AsyncSQLAlchemyMemory with a shared database or Redis cluster.

How does RedisMemory handle message expiration?

RedisMemory supports optional TTL through the key_ttl constructor parameter. When set, the implementation calls _refresh_session_ttl after every operation to update the expiration window on all keys associated with that session. Once the TTL elapses, Redis automatically removes the session data, marks, and indexes.

Which backend is best for production multi-agent systems?

Choose RedisMemory when you need fast random access, automatic eviction policies, and simple key-value semantics across distributed agents. Select AsyncSQLAlchemyMemory when you require complex relational queries, strict ACID guarantees, or integration with existing database infrastructure. Use InMemoryMemory only for testing or single-process applications where persistence is unnecessary.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}