ractogateway.redis.chat_memory

Sliding-window conversation memory stored in Redis.

Each conversation is kept as a Redis List of JSON-encoded {"role", "content"} message dicts. The list is capped at max_turns * 2 entries (one user + one assistant message per turn) using LTRIM after every append().

Why Redis Lists?

Redis Lists support O(1) push and O(n) range reads — perfect for maintaining a bounded conversation history that must be accessible to multiple server replicas. Unlike in-memory approaches, the history survives rolling deployments and can be shared between a web server and a background worker.

Compatibility with ChatConfig.history

get_history() returns list[dict[str, str]] with "role" and "content" keys. This is the exact format used by all three provider adapters under the hood. You can pass the result directly to ChatConfig(history=memory.get_history(conv_id)) after wrapping each dict in your Message model.

Example:

from ractogateway.redis import RedisChatMemory, ChatMemoryConfig

memory = RedisChatMemory(
    url="redis://localhost:6379/0",
    config=ChatMemoryConfig(max_turns=20, ttl_seconds=1800),
)

# Store messages as the conversation progresses:
memory.append("conv_abc", "user", "What is the capital of France?")
memory.append("conv_abc", "assistant", "Paris.")

# Retrieve history to pass back into the kit:
history = memory.get_history("conv_abc")
# → [{"role": "user", "content": "What is the capital of France?"},
#    {"role": "assistant", "content": "Paris."}]

class ractogateway.redis.chat_memory.RedisChatMemory(*, url='redis://localhost:6379/0', client=None, config=None)

Bases: object

Shared, bounded conversation history backed by Redis.

Parameters:

• url (str) – Redis connection URL. Ignored when client is provided.

• client (Any | None) – Pre-built redis.Redis client.

• config (ChatMemoryConfig | None) – ChatMemoryConfig controlling turn limit, TTL, and key namespace. Defaults are applied when None.

append(conversation_id, role, content)

Append a message to the conversation history.

After appending, the list is trimmed to the last config.max_turns * 2 messages (oldest dropped first). If a TTL is configured, it is refreshed on every append so the window slides with activity.

Parameters:

• conversation_id (str) – Opaque identifier for the conversation (e.g. session UUID).

• role (str) – The message author: "user", "assistant", or "system".

• content (str) – Text content of the message.

Return type:

None

get_history(conversation_id)

Return all stored messages as a list of {"role", "content"} dicts.

The list is ordered oldest-first, matching the ChatConfig.history convention.

Returns an empty list when the conversation does not exist or has expired.

Return type:

list[dict[str, str]]

clear(conversation_id)

Delete the conversation history from Redis.

Return type:

None

count(conversation_id)

Return the number of messages stored for this conversation.

Returns 0 when the conversation does not exist.

Return type:

int