Memory

Memory primitives for conversation history and long-term recall.

Memory

Memory

Memory(redis_client: Redis | None = None, namespace: str | None = None, *, enabled: bool = True)

Memory interface for agent execution.

Provides working memory (conversation history) and long-term memory (persistent facts, preferences, etc.).

Memory is always available on TaskContext. Sessions are created lazily on first write - no data exists until something is added.

Example

async def my_agent(ctx: TaskContext) -> dict: # Get conversation history messages = await ctx.memory.get_messages()

# Search relevant memories
memories = await ctx.memory.search("user preferences")

# Add response (creates session if needed)
await ctx.memory.add_message("assistant", response)

return {"response": response}

Source code in redis_agent_kit/memory.py

def __init__(
    self,
    redis_client: Redis | None = None,
    namespace: str | None = None,
    *,
    enabled: bool = True,
):
    self._redis = redis_client
    self._namespace = namespace
    self._enabled = enabled and _check_ams()
    self._session_id: str | None = None
    self._user_id: str | None = None
    self._session_exists: bool = False
    self._ams_configured: bool = False

session_id property

session_id: str | None

Current session ID (may not exist in Redis yet).

user_id property

user_id: str | None

Current user ID.

enabled property

enabled: bool

Whether memory features are enabled.

with_session

with_session(session_id: str | None = None, user_id: str | None = None) -> Memory

Return a Memory instance bound to a session.

If session_id is None, generates a new one. Session is not created in Redis until first write.

Parameters:

Name	Type	Description	Default
session_id	str | None	Session ID to use (or None to generate)	None
user_id	str | None	Optional user ID for memory isolation	None

Returns:

Type	Description
Memory	New Memory instance bound to the session

Source code in redis_agent_kit/memory.py

def with_session(
    self,
    session_id: str | None = None,
    user_id: str | None = None,
) -> Memory:
    """
    Return a Memory instance bound to a session.

    If session_id is None, generates a new one.
    Session is not created in Redis until first write.

    Args:
        session_id: Session ID to use (or None to generate)
        user_id: Optional user ID for memory isolation

    Returns:
        New Memory instance bound to the session
    """
    m = Memory(
        redis_client=self._redis,
        namespace=self._namespace,
        enabled=self._enabled,
    )
    m._session_id = session_id or str(ULID())
    m._user_id = user_id
    return m

add_message async

add_message(role: str, content: str) -> str | None

Add a message to working memory.

Creates session on first call if it doesn't exist.

Parameters:

Name	Type	Description	Default
role	str	Message role (user, assistant, system)	required
content	str	Message content	required

Returns:

Type	Description
str | None	Session ID, or None if memory disabled

Source code in redis_agent_kit/memory.py

async def add_message(self, role: str, content: str) -> str | None:
    """
    Add a message to working memory.

    Creates session on first call if it doesn't exist.

    Args:
        role: Message role (user, assistant, system)
        content: Message content

    Returns:
        Session ID, or None if memory disabled
    """
    if not self._enabled:
        return None

    _require_ams()
    from agent_memory_server.models import MemoryMessage

    session = await self._ensure_session()
    if session is None:
        return None

    session.messages.append(MemoryMessage(role=role, content=content))
    await self._save_session(session)
    return self._session_id

get_messages async

get_messages(limit: int | None = None) -> MessageList

Get messages from current session.

Parameters:

Name	Type	Description	Default
limit	int | None	Max messages to return (None for all)	None

Returns:

Type	Description
MessageList	MessageList with 'role' and 'content' dicts. Supports:
MessageList	.dict() - list of dicts
MessageList	.json() - JSON string
MessageList	.markdown() - formatted markdown string

Source code in redis_agent_kit/memory.py

async def get_messages(self, limit: int | None = None) -> MessageList:
    """
    Get messages from current session.

    Args:
        limit: Max messages to return (None for all)

    Returns:
        MessageList with 'role' and 'content' dicts. Supports:
        - .dict() - list of dicts
        - .json() - JSON string
        - .markdown() - formatted markdown string
    """
    if not self._enabled:
        return MessageList()

    session = await self._get_session()
    if session is None:
        return MessageList()

    messages = session.messages
    if limit:
        messages = messages[-limit:]

    return MessageList({"role": m.role, "content": m.content} for m in messages)

get_history async

get_history(limit: int | None = 10, format: str = '{role}: {content}', separator: str = '\n') -> str

Get conversation history formatted as text.

Convenience method that formats messages for LLM context.

Parameters:

Name	Type	Description	Default
limit	int | None	Max messages to include (None for all)	10
format	str	Format string for each message (supports {role}, {content})	'{role}: {content}'
separator	str	String to join messages	'\n'

Returns:

Type	Description
str	Formatted conversation history string

Example

Default format

history = await ctx.memory.get_history(limit=10)

"user: Hello\nassistant: Hi there!"

Custom format

history = await ctx.memory.get_history( format="[{role}] {content}", separator="\n---\n" )

Source code in redis_agent_kit/memory.py

async def get_history(
    self,
    limit: int | None = 10,
    format: str = "{role}: {content}",
    separator: str = "\n",
) -> str:
    """
    Get conversation history formatted as text.

    Convenience method that formats messages for LLM context.

    Args:
        limit: Max messages to include (None for all)
        format: Format string for each message (supports {role}, {content})
        separator: String to join messages

    Returns:
        Formatted conversation history string

    Example:
        # Default format
        history = await ctx.memory.get_history(limit=10)
        # "user: Hello\\nassistant: Hi there!"

        # Custom format
        history = await ctx.memory.get_history(
            format="[{role}] {content}",
            separator="\\n---\\n"
        )
    """
    messages = await self.get_messages(limit=limit)
    return separator.join(format.format(role=m["role"], content=m["content"]) for m in messages)

get_context async

get_context() -> str | None

Get context string from working memory.

Returns summarized context if available, otherwise None.

Source code in redis_agent_kit/memory.py

async def get_context(self) -> str | None:
    """
    Get context string from working memory.

    Returns summarized context if available, otherwise None.
    """
    if not self._enabled:
        return None

    session = await self._get_session()
    if session is None:
        return None

    return str(session.context) if session.context is not None else None

set_data async

set_data(key: str, value: Any) -> None

Set arbitrary data in session.

Parameters:

Name	Type	Description	Default
key	str	Data key	required
value	Any	Data value (must be JSON serializable)	required

Source code in redis_agent_kit/memory.py

async def set_data(self, key: str, value: Any) -> None:
    """
    Set arbitrary data in session.

    Args:
        key: Data key
        value: Data value (must be JSON serializable)
    """
    if not self._enabled:
        return

    _require_ams()
    session = await self._ensure_session()
    if session is None:
        return

    if session.data is None:
        session.data = {}
    session.data[key] = value
    await self._save_session(session)

get_data async

get_data(key: str, default: Any = None) -> Any

Get arbitrary data from session.

Parameters:

Name	Type	Description	Default
key	str	Data key	required
default	Any	Default value if key not found	None

Returns:

Type	Description
Any	Value or default

Source code in redis_agent_kit/memory.py

async def get_data(self, key: str, default: Any = None) -> Any:
    """
    Get arbitrary data from session.

    Args:
        key: Data key
        default: Default value if key not found

    Returns:
        Value or default
    """
    if not self._enabled:
        return default

    session = await self._get_session()
    if session is None or session.data is None:
        return default

    return session.data.get(key, default)

search async

search(text: str, limit: int = 10, memory_type: str | None = None, topics: list[str] | None = None) -> MemorySearchResults

Search long-term memories by semantic similarity.

Parameters:

Name	Type	Description	Default
text	str	Search query	required
limit	int	Max results	10
memory_type	str | None	Filter by type (semantic, episodic, etc.)	None
topics	list[str] | None	Filter by topics	None

Returns:

Type	Description
MemorySearchResults	MemorySearchResults with 'id', 'text', 'memory_type', 'score', etc. Supports:
MemorySearchResults	.dict() - list of dicts
MemorySearchResults	.json() - JSON string
MemorySearchResults	.markdown() - formatted markdown string

Source code in redis_agent_kit/memory.py

async def search(
    self,
    text: str,
    limit: int = 10,
    memory_type: str | None = None,
    topics: list[str] | None = None,
) -> MemorySearchResults:
    """
    Search long-term memories by semantic similarity.

    Args:
        text: Search query
        limit: Max results
        memory_type: Filter by type (semantic, episodic, etc.)
        topics: Filter by topics

    Returns:
        MemorySearchResults with 'id', 'text', 'memory_type', 'score', etc. Supports:
        - .dict() - list of dicts
        - .json() - JSON string
        - .markdown() - formatted markdown string
    """
    if not self._enabled:
        return MemorySearchResults()

    _require_ams()
    self._configure_ams()

    from agent_memory_server import long_term_memory

    result = await long_term_memory.search_long_term_memories(
        text=text,
        user_id=self._user_id,
        namespace=self._namespace,
        limit=limit,
        memory_type=memory_type,
        topics=topics,
        redis_client=self._redis,
    )

    if result is None:
        return MemorySearchResults()

    return MemorySearchResults(
        {
            "id": m.id,
            "text": m.text,
            "memory_type": m.memory_type,
            "topics": m.topics,
            "entities": m.entities,
            "score": getattr(m, "dist", None),
        }
        for m in result.memories
    )

search_text async

search_text(text: str, limit: int = 10, separator: str = '\n', memory_type: str | None = None, topics: list[str] | None = None) -> str

Search long-term memories and return text content only.

Convenience method that returns just the text from search results, formatted for direct use in LLM context.

Parameters:

Name	Type	Description	Default
text	str	Search query	required
limit	int	Max results	10
separator	str	String to join memory texts	'\n'
memory_type	str | None	Filter by type (semantic, episodic, etc.)	None
topics	list[str] | None	Filter by topics	None

Returns:

Type	Description
str	Concatenated memory texts, or empty string if no results

Example

Get relevant context for a query

memory_context = await ctx.memory.search_text(ctx.message, limit=3) prompt = f"Context:\n{memory_context}\n\nQuestion: {ctx.message}"

Source code in redis_agent_kit/memory.py

async def search_text(
    self,
    text: str,
    limit: int = 10,
    separator: str = "\n",
    memory_type: str | None = None,
    topics: list[str] | None = None,
) -> str:
    """
    Search long-term memories and return text content only.

    Convenience method that returns just the text from search results,
    formatted for direct use in LLM context.

    Args:
        text: Search query
        limit: Max results
        separator: String to join memory texts
        memory_type: Filter by type (semantic, episodic, etc.)
        topics: Filter by topics

    Returns:
        Concatenated memory texts, or empty string if no results

    Example:
        # Get relevant context for a query
        memory_context = await ctx.memory.search_text(ctx.message, limit=3)
        prompt = f"Context:\\n{memory_context}\\n\\nQuestion: {ctx.message}"
    """
    results = await self.search(
        text=text,
        limit=limit,
        memory_type=memory_type,
        topics=topics,
    )
    return separator.join(r.get("text", "") for r in results if r.get("text"))

create_memory async

create_memory(text: str, memory_type: str = 'semantic', topics: list[str] | None = None, entities: list[str] | None = None) -> str | None

Create a long-term memory explicitly.

For passive memory creation, just use working memory - the system auto-promotes based on configuration.

Parameters:

Name	Type	Description	Default
text	str	Memory content	required
memory_type	str	Type (semantic, episodic, etc.)	'semantic'
topics	list[str] | None	Associated topics	None
entities	list[str] | None	Associated entities	None

Returns:

Type	Description
str | None	Memory ID, or None if disabled

Source code in redis_agent_kit/memory.py

async def create_memory(
    self,
    text: str,
    memory_type: str = "semantic",
    topics: list[str] | None = None,
    entities: list[str] | None = None,
) -> str | None:
    """
    Create a long-term memory explicitly.

    For passive memory creation, just use working memory -
    the system auto-promotes based on configuration.

    Args:
        text: Memory content
        memory_type: Type (semantic, episodic, etc.)
        topics: Associated topics
        entities: Associated entities

    Returns:
        Memory ID, or None if disabled
    """
    if not self._enabled:
        return None

    _require_ams()
    self._configure_ams()

    from agent_memory_server import long_term_memory
    from agent_memory_server.models import MemoryRecord

    memory = MemoryRecord(
        text=text,
        user_id=self._user_id,
        namespace=self._namespace,
        memory_type=memory_type,
        topics=topics or [],
        entities=entities or [],
    )

    result = await long_term_memory.create_long_term_memory(
        memory=memory,
        redis_client=self._redis,
    )
    return result.id if result else None

delete_memories async

delete_memories(memory_ids: list[str]) -> int

Delete long-term memories by ID.

Parameters:

Name	Type	Description	Default
memory_ids	list[str]	List of memory IDs to delete	required

Returns:

Type	Description
int	Number of memories deleted

Source code in redis_agent_kit/memory.py

async def delete_memories(self, memory_ids: list[str]) -> int:
    """
    Delete long-term memories by ID.

    Args:
        memory_ids: List of memory IDs to delete

    Returns:
        Number of memories deleted
    """
    if not self._enabled or not memory_ids:
        return 0

    _require_ams()
    self._configure_ams()

    from agent_memory_server import long_term_memory

    result: int = await long_term_memory.delete_long_term_memories(
        memory_ids=memory_ids,
        user_id=self._user_id,
        namespace=self._namespace,
        redis_client=self._redis,
    )
    return result

delete_session async

delete_session() -> bool

Delete current working memory session.

Returns:

Type	Description
bool	True if deleted, False if not found or disabled

Source code in redis_agent_kit/memory.py

async def delete_session(self) -> bool:
    """
    Delete current working memory session.

    Returns:
        True if deleted, False if not found or disabled
    """
    if not self._enabled or not self._session_id:
        return False

    _require_ams()
    self._configure_ams()

    from agent_memory_server import working_memory
    from agent_memory_server.config import settings as ams_settings

    result: bool = await working_memory.delete_working_memory(
        session_id=self._session_id,
        user_id=self._user_id,
        namespace=self._namespace,
        redis_client=self._redis,
        config=ams_settings,
    )
    self._session_exists = False
    return result

MessageList

MessageList

Bases: list[dict[str, str]]

A list of messages with formatting methods.

Supports .dict(), .json(), and .markdown() for different output formats.

dict

dict() -> list[dict[str, str]]

Return messages as a list of dicts.

Source code in redis_agent_kit/memory.py

def dict(self) -> list[dict[str, str]]:
    """Return messages as a list of dicts."""
    return list(self)

json

json(**kwargs: Any) -> str

Return messages as a JSON string.

Source code in redis_agent_kit/memory.py

def json(self, **kwargs: Any) -> str:
    """Return messages as a JSON string."""
    return json.dumps(self.dict(), **kwargs)

markdown

markdown(format: str = '**{role}**: {content}', separator: str = '\n\n') -> str

Return messages formatted as markdown.

Parameters:

Name	Type	Description	Default
format	str	Format string for each message (supports {role}, {content})	'**{role}**: {content}'
separator	str	String to join messages	'\n\n'

Returns:

Type	Description
str	Formatted markdown string

Source code in redis_agent_kit/memory.py

def markdown(
    self,
    format: str = "**{role}**: {content}",
    separator: str = "\n\n",
) -> str:
    """
    Return messages formatted as markdown.

    Args:
        format: Format string for each message (supports {role}, {content})
        separator: String to join messages

    Returns:
        Formatted markdown string
    """
    if not self:
        return ""
    return separator.join(format.format(**m) for m in self)

MemorySearchResults

MemorySearchResults

Bases: list[dict[str, Any]]

A list of memory search results with formatting methods.

Supports .dict(), .json(), and .markdown() for different output formats.

dict

dict() -> list[dict[str, Any]]

Return results as a list of dicts.

Source code in redis_agent_kit/memory.py

def dict(self) -> list[dict[str, Any]]:
    """Return results as a list of dicts."""
    return list(self)

json

json(**kwargs: Any) -> str

Return results as a JSON string.

Source code in redis_agent_kit/memory.py

def json(self, **kwargs: Any) -> str:
    """Return results as a JSON string."""
    return json.dumps(self.dict(), **kwargs)

markdown

markdown(format: str = '- {text}', separator: str = '\n', include_metadata: bool = False) -> str

Return results formatted as markdown.

Parameters:

Name	Type	Description	Default
format	str	Format string for each result (supports {text}, {memory_type}, etc.)	'- {text}'
separator	str	String to join results	'\n'
include_metadata	bool	Include memory_type and topics in output	False

Returns:

Type	Description
str	Formatted markdown string

Source code in redis_agent_kit/memory.py

def markdown(
    self,
    format: str = "- {text}",
    separator: str = "\n",
    include_metadata: bool = False,
) -> str:
    """
    Return results formatted as markdown.

    Args:
        format: Format string for each result (supports {text}, {memory_type}, etc.)
        separator: String to join results
        include_metadata: Include memory_type and topics in output

    Returns:
        Formatted markdown string
    """
    if not self:
        return ""

    if include_metadata:
        lines = []
        for m in self:
            line = format.format(**m)
            meta = []
            if m.get("memory_type"):
                meta.append(f"type: {m['memory_type']}")
            if m.get("topics"):
                meta.append(f"topics: {', '.join(m['topics'])}")
            if meta:
                line += f" ({'; '.join(meta)})"
            lines.append(line)
        return separator.join(lines)

    return separator.join(format.format(**m) for m in self)

NoOpMemory

NoOpMemory

NoOpMemory()

Bases: Memory

Memory implementation that does nothing (for disabled memory).

Source code in redis_agent_kit/memory.py

def __init__(self) -> None:
    super().__init__(redis_client=None, namespace=None, enabled=False)

session_id property

session_id: str | None

Current session ID (may not exist in Redis yet).

user_id property

user_id: str | None

Current user ID.

enabled property

enabled: bool

Whether memory features are enabled.

with_session

with_session(session_id: str | None = None, user_id: str | None = None) -> Memory

Return a Memory instance bound to a session.

If session_id is None, generates a new one. Session is not created in Redis until first write.

Parameters:

Name	Type	Description	Default
session_id	str | None	Session ID to use (or None to generate)	None
user_id	str | None	Optional user ID for memory isolation	None

Returns:

Type	Description
Memory	New Memory instance bound to the session

Source code in redis_agent_kit/memory.py

def with_session(
    self,
    session_id: str | None = None,
    user_id: str | None = None,
) -> Memory:
    """
    Return a Memory instance bound to a session.

    If session_id is None, generates a new one.
    Session is not created in Redis until first write.

    Args:
        session_id: Session ID to use (or None to generate)
        user_id: Optional user ID for memory isolation

    Returns:
        New Memory instance bound to the session
    """
    m = Memory(
        redis_client=self._redis,
        namespace=self._namespace,
        enabled=self._enabled,
    )
    m._session_id = session_id or str(ULID())
    m._user_id = user_id
    return m

get_history async

get_history(limit: int | None = 10, format: str = '{role}: {content}', separator: str = '\n') -> str

Get conversation history formatted as text.

Convenience method that formats messages for LLM context.

Parameters:

Name	Type	Description	Default
limit	int | None	Max messages to include (None for all)	10
format	str	Format string for each message (supports {role}, {content})	'{role}: {content}'
separator	str	String to join messages	'\n'

Returns:

Type	Description
str	Formatted conversation history string

Example

Default format

history = await ctx.memory.get_history(limit=10)

"user: Hello\nassistant: Hi there!"

Custom format

history = await ctx.memory.get_history( format="[{role}] {content}", separator="\n---\n" )

Source code in redis_agent_kit/memory.py

async def get_history(
    self,
    limit: int | None = 10,
    format: str = "{role}: {content}",
    separator: str = "\n",
) -> str:
    """
    Get conversation history formatted as text.

    Convenience method that formats messages for LLM context.

    Args:
        limit: Max messages to include (None for all)
        format: Format string for each message (supports {role}, {content})
        separator: String to join messages

    Returns:
        Formatted conversation history string

    Example:
        # Default format
        history = await ctx.memory.get_history(limit=10)
        # "user: Hello\\nassistant: Hi there!"

        # Custom format
        history = await ctx.memory.get_history(
            format="[{role}] {content}",
            separator="\\n---\\n"
        )
    """
    messages = await self.get_messages(limit=limit)
    return separator.join(format.format(role=m["role"], content=m["content"]) for m in messages)

search_text async

search_text(text: str, limit: int = 10, separator: str = '\n', memory_type: str | None = None, topics: list[str] | None = None) -> str

Search long-term memories and return text content only.

Convenience method that returns just the text from search results, formatted for direct use in LLM context.

Parameters:

Name	Type	Description	Default
text	str	Search query	required
limit	int	Max results	10
separator	str	String to join memory texts	'\n'
memory_type	str | None	Filter by type (semantic, episodic, etc.)	None
topics	list[str] | None	Filter by topics	None

Returns:

Type	Description
str	Concatenated memory texts, or empty string if no results

Example

Get relevant context for a query

memory_context = await ctx.memory.search_text(ctx.message, limit=3) prompt = f"Context:\n{memory_context}\n\nQuestion: {ctx.message}"

Source code in redis_agent_kit/memory.py

async def search_text(
    self,
    text: str,
    limit: int = 10,
    separator: str = "\n",
    memory_type: str | None = None,
    topics: list[str] | None = None,
) -> str:
    """
    Search long-term memories and return text content only.

    Convenience method that returns just the text from search results,
    formatted for direct use in LLM context.

    Args:
        text: Search query
        limit: Max results
        separator: String to join memory texts
        memory_type: Filter by type (semantic, episodic, etc.)
        topics: Filter by topics

    Returns:
        Concatenated memory texts, or empty string if no results

    Example:
        # Get relevant context for a query
        memory_context = await ctx.memory.search_text(ctx.message, limit=3)
        prompt = f"Context:\\n{memory_context}\\n\\nQuestion: {ctx.message}"
    """
    results = await self.search(
        text=text,
        limit=limit,
        memory_type=memory_type,
        topics=topics,
    )
    return separator.join(r.get("text", "") for r in results if r.get("text"))