RedisChatMessageHistory#

class langchain_redis.chat_message_history.RedisChatMessageHistory(
session_id: str,
redis_url: str = 'redis://localhost:6379',
key_prefix: str = 'chat:',
ttl: int | None = None,
index_name: str = 'idx:chat_history',
redis_client: Redis | None = None,
**kwargs: Any,
)[source]#

Redis-based implementation of chat message history using RedisVL.

This class provides a way to store and retrieve chat message history using Redis with RedisVL for efficient indexing, querying, and document management.

redis_client#

The Redis client instance.

Type:

Redis

session_id#

A unique identifier for the chat session.

Type:

str

key_prefix#

Prefix for Redis keys to namespace the messages.

Type:

str

ttl#

Time-to-live for message entries in seconds.

Type:

Optional[int]

index_name#

Name of the Redis search index for message retrieval.

Type:

str

Parameters:

  • session_id (str) – A unique identifier for the chat session.

  • redis_url (str, optional) – URL of the Redis instance. Defaults to “redis://localhost:6379”.

  • key_prefix (str, optional) – Prefix for Redis keys. Defaults to “chat:”.

  • ttl (Optional[int], optional) – Time-to-live for entries in seconds. Defaults to None (no expiration).

  • index_name (str, optional) – Name of the Redis search index. Defaults to “idx:chat_history”.

  • redis_client (Optional[Redis], optional) – Existing Redis client instance. If provided, redis_url is ignored.

  • **kwargs – Additional keyword arguments to pass to the Redis client.

Raises:

  • ValueError – If session_id is empty or None.

  • ResponseError – If Redis connection fails or RedisVL operations fail.

Example

from langchain_redis import RedisChatMessageHistory
from langchain_core.messages import HumanMessage, AIMessage

history = RedisChatMessageHistory(
    session_id="user123",
    redis_url="redis://localhost:6379",
    ttl=3600  # Expire chat history after 1 hour
)

# Add messages to the history
history.add_message(HumanMessage(content="Hello, AI!"))
history.add_message(
  AIMessage(content="Hello, human! How can I assist you today?")
)

# Retrieve all messages
messages = history.messages
for message in messages:
    print(f"{message.type}: {message.content}")

# Clear the history for the session
history.clear()

Note

  • This class uses RedisVL for managing Redis JSON storage and search indexes, providing efficient querying and retrieval.

  • A Redis search index is created to enable fast lookups and search capabilities over the chat history.

  • If TTL is set, message entries will automatically expire after the specified duration.

  • The session_id is used to group messages belonging to the same conversation or user session.

  • RedisVL automatically handles tokenization and escaping for search queries.

Attributes

id

Return the session ID.

messages

Retrieve all messages for the current session, sorted by timestamp.

Methods

__init__(session_id[, redis_url, ...])

aadd_messages(messages)

Async add a list of messages.

aclear()

Async remove all messages from the store.

add_ai_message(message)

Convenience method for adding an AI message string to the store.

add_message(message)

Add a message to the chat history using RedisVL.

add_messages(messages)

Add a list of messages.

add_user_message(message)

Convenience method for adding a human message string to the store.

aget_messages()

Async version of getting messages.

clear()

Clear all messages from the chat history for the current session.

delete()

Delete all sessions and the chat history index from Redis.

search_messages(query[, limit])

Search for messages in the chat history that match the given query.
__init__(
session_id: str,
redis_url: str = 'redis://localhost:6379',
key_prefix: str = 'chat:',
ttl: int | None = None,
index_name: str = 'idx:chat_history',
redis_client: Redis | None = None,
**kwargs: Any,
) → None[source]#
Parameters:

  • session_id (str)

  • redis_url (str)

  • key_prefix (str)

  • ttl (int | None)

  • index_name (str)

  • redis_client (Redis | None)

  • kwargs (Any)

Return type:

None

async aadd_messages(
messages: Sequence[BaseMessage],
) → None#

Async add a list of messages.

Parameters:

messages (Sequence[BaseMessage]) – A sequence of BaseMessage objects to store.

Return type:

None

async aclear() → None#

Async remove all messages from the store.

Return type:

None

add_ai_message(
message: AIMessage | str,
) → None#

Convenience method for adding an AI message string to the store.

Please note that this is a convenience method. Code should favor the bulk add_messages interface instead to save on round-trips to the underlying persistence layer.

This method may be deprecated in a future release.

Parameters:

message (AIMessage | str) – The AI message to add.

Return type:

None

add_message(
message: BaseMessage,
) → None[source]#

Add a message to the chat history using RedisVL.

This method adds a new message to the Redis store for the current session using RedisVL’s document loading capabilities.

Parameters:

message (BaseMessage) – The message to add to the history. This should be an instance of a class derived from BaseMessage, such as HumanMessage, AIMessage, or SystemMessage.

Raises:

  • ResponseError – If Redis connection fails or RedisVL operations fail.

  • ValueError – If message is None or invalid.

Return type:

None

Example

from langchain_redis import RedisChatMessageHistory
from langchain_core.messages import HumanMessage, AIMessage

history = RedisChatMessageHistory(
    session_id="user123",
    redis_url="redis://localhost:6379",
    ttl=3600  # optional: set TTL to 1 hour
)

# Add a human message
history.add_message(HumanMessage(content="Hello, AI!"))

# Add an AI message
history.add_message(
  AIMessage(content="Hello! How can I assist you today?")
)

# Verify messages were added
print(f"Number of messages: {len(history.messages)}")

Note

  • Each message is stored as a separate entry in Redis, associated with the current session_id.

  • Messages are stored using RedisVL’s JSON capabilities for efficient storage and retrieval.

  • If a TTL (Time To Live) was specified when initializing the history, it will be applied to each message.

  • The message’s content, type, and any additional data (like timestamp) are stored.

  • This method is thread-safe and can be used in concurrent environments.

  • The Redis search index is automatically updated to include the new message, enabling future searches.

  • Large message contents may impact performance and storage usage. Consider implementing size limits if dealing with potentially large messages.

add_messages(
messages: Sequence[BaseMessage],
) → None#

Add a list of messages.

Implementations should over-ride this method to handle bulk addition of messages in an efficient manner to avoid unnecessary round-trips to the underlying store.

Parameters:

messages (Sequence[BaseMessage]) – A sequence of BaseMessage objects to store.

Return type:

None

add_user_message(
message: HumanMessage | str,
) → None#

Convenience method for adding a human message string to the store.

Please note that this is a convenience method. Code should favor the bulk add_messages interface instead to save on round-trips to the underlying persistence layer.

This method may be deprecated in a future release.

Parameters:

message (HumanMessage | str) – The human message to add to the store.

Return type:

None

async aget_messages() → list[BaseMessage]#

Async version of getting messages.

Can over-ride this method to provide an efficient async implementation.

In general, fetching messages may involve IO to the underlying persistence layer.

Return type:

list[BaseMessage]

clear() → None[source]#

Clear all messages from the chat history for the current session.

This method removes all messages associated with the current session_id from the Redis store using RedisVL queries.

Raises:

ResponseError – If Redis connection fails or RedisVL operations fail.

Return type:

None

Example

from langchain_redis import RedisChatMessageHistory
from langchain_core.messages import HumanMessage, AIMessage

history = RedisChatMessageHistory(session_id="user123", redis_url="redis://localhost:6379")

# Add some messages
history.add_message(HumanMessage(content="Hello, AI!"))
history.add_message(AIMessage(content="Hello, human!"))

# Clear the history
history.clear()

# Verify that the history is empty
assert len(history.messages) == 0

Note

  • This method only clears messages for the current session_id.

  • It uses RedisVL’s FilterQuery to find all relevant messages and then deletes them individually using the Redis client.

  • The operation removes all messages for the current session only.

  • After clearing, the Redis search index is still maintained, allowing for immediate use of the same session_id for new messages if needed.

  • This operation is irreversible. Make sure you want to remove all messages before calling this method.

delete() → None[source]#

Delete all sessions and the chat history index from Redis.

Raises:

ResponseError – If Redis connection fails or RedisVL operations fail.

Return type:

None

search_messages(
query: str,
limit: int = 10,
) → List[Dict[str, Any]][source]#

Search for messages in the chat history that match the given query.

This method performs a full-text search on the content of messages in the current session using RedisVL’s TextQuery capabilities.

Parameters:

  • query (str) – The search query string to match against message content.

  • limit (int, optional) – The maximum number of results to return. Defaults to 10.

Returns:

A list of dictionaries, each representing a

matching message.

Each dictionary contains the message content and metadata.

Return type:

List[Dict[str, Any]]

Raises:

ResponseError – If Redis connection fails or RedisVL operations fail.

Example

from langchain_redis import RedisChatMessageHistory
from langchain_core.messages import HumanMessage, AIMessage

history = RedisChatMessageHistory(session_id="user123", redis_url="redis://localhost:6379")

# Add some messages
history.add_message(
  HumanMessage(content="Tell me about Machine Learning")
)
history.add_message(
  AIMessage(content="Machine Learning is a subset of AI...")
)
history.add_message(
  HumanMessage(content="What are neural networks?")
)
history.add_message(
  AIMessage(
    content="Neural networks are a key component of deep learning..."
  )
)

# Search for messages containing "learning"
results = history.search_messages("learning", limit=5)

for result in results:
    print(f"Content: {result['content']}")
    print(f"Type: {result['type']}")
    print("---")

Note

  • The search is performed using RedisVL’s TextQuery capabilities, which allows for efficient full-text search.

  • The search is case-insensitive and uses Redis’ default tokenization and stemming.

  • Only messages from the current session (as defined by session_id) are searched.

  • The returned dictionaries include all stored fields, which typically include ‘content’, ‘type’, and any additional metadata stored with the message.

  • This method is useful for quickly finding relevant parts of a conversation without having to iterate through all messages.

Examples using RedisChatMessageHistory