Redis聊天消息历史#

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的聊天消息历史实现。

该类提供了一种使用Redis存储和检索聊天消息历史的方法。它实现了BaseChatMessageHistory接口，并利用Redis的JSON功能来高效地存储和检索消息。

redis_client#

Redis 客户端实例。

Type:: Redis

session_id#

聊天会话的唯一标识符。

Type:: 字符串

key_prefix#

Redis键的前缀，用于为消息命名空间。

Type:: 字符串

ttl#

消息条目的生存时间（以秒为单位）。

Type:: 可选[int]

index_name#

用于消息检索的Redis搜索索引的名称。

Type:: 字符串

Parameters:

session_id (str) – 聊天会话的唯一标识符。
redis_url (str, optional) – Redis实例的URL。默认为“redis://localhost:6379”。
key_prefix (str, optional) – Redis 键的前缀。默认为“chat:”。
ttl (Optional[int], optional) – 条目的生存时间，单位为秒。默认为 None（无过期时间）。
index_name (str, optional) – Redis搜索索引的名称。默认为“idx:chat_history”。
redis (可选[Redis], 可选) – 现有的 Redis 客户端实例。如果提供了该实例，则忽略 redis_url。
**kwargs (Any) – 传递给Redis客户端的额外关键字参数。
redis_client (Redis | None)
**kwargs

示例

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

history = RedisChatMessageHistory(
    session_id="user123",
    redis_url="redis://localhost:6379",
    ttl=3600  # Messages expire 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
history.clear()

注意

此类使用 Redis JSON 存储消息，从而实现高效的查询和检索。
创建了一个Redis搜索索引，以实现快速查找和未来可能的聊天历史搜索需求。
如果设置了TTL，消息条目将在指定持续时间后自动过期。
session_id 用于将属于同一对话或用户会话的消息分组。

属性

`id`
`messages`

方法

`__init__`(session_id[, redis_url, ...])
`aadd_messages`(messages)	异步添加消息列表。
`aclear`()	异步从存储中移除所有消息
`add_ai_message`(message)	将AI消息字符串添加到存储中的便捷方法。
`add_message`(message)	向聊天记录中添加一条消息。
`add_messages`(messages)	添加消息列表。
`add_user_message`(message)	用于将人类消息字符串添加到存储中的便捷方法。
`aget_messages`()	获取消息的异步版本。
`clear`()	清除当前会话中的所有聊天历史记录。
`search_messages`(query[, limit])	在聊天历史中搜索与给定查询匹配的消息。

__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)[source]#

Parameters:

session_id (str)
redis_url (str)
key_prefix (str)
ttl (int | None)
index_name (str)
redis_client (Redis | None)
kwargs (Any)

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

异步添加消息列表。

Parameters:: messages (Sequence[BaseMessage]) – 要存储的BaseMessage对象序列。
Return type:: 无

async aclear() → None#

异步从存储中移除所有消息

Return type:: 无

add_ai_message(message: AIMessage | str) → None#

向存储中添加AI消息字符串的便捷方法。

请注意，这是一个便捷方法。代码应优先使用批量 add_messages 接口，以减少与底层持久层的往返次数。

此方法可能在未来的版本中被弃用。

Parameters:: message (AIMessage | str) – 要添加的AI消息。
Return type:: 无

add_message(message: BaseMessage) → None[来源]#

向聊天历史记录中添加一条消息。

此方法将新消息添加到当前会话的Redis存储中。

Parameters:: 消息 (BaseMessage) – 要添加到历史记录中的消息。这应该是从BaseMessage派生的类的实例，例如HumanMessage、AIMessage或SystemMessage。
Returns:: 无
Return type:: 无

示例

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)}")

注意

每条消息都作为Redis中的一个单独条目存储，与当前的session_id相关联。
消息使用Redis JSON功能进行存储，以实现高效的存储和检索。
如果在初始化历史记录时指定了TTL（生存时间），它将应用于每条消息。
消息的内容、类型和任何附加数据（如时间戳）都会被存储。
此方法是线程安全的，可以在并发环境中使用。
Redis 搜索索引会自动更新以包含新消息，从而支持未来的搜索。
大消息内容可能会影响性能和存储使用。如果处理可能的大消息，请考虑实施大小限制。

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

添加消息列表。

实现应重写此方法，以高效地处理消息的批量添加，避免对底层存储进行不必要的往返。

Parameters:: messages (Sequence[BaseMessage]) – 要存储的BaseMessage对象序列。
Return type:: 无

add_user_message(message: HumanMessage | str) → None#

用于向存储中添加人类消息字符串的便捷方法。

请注意，这是一个便捷方法。代码应优先使用批量 add_messages 接口，以减少对底层持久层的往返次数。

此方法可能在未来的版本中被弃用。

Parameters:: message (HumanMessage | str) – 要添加到存储中的人类消息。
Return type:: 无

async aget_messages() → list[BaseMessage]#

获取消息的异步版本。

可以重写此方法以提供高效的异步实现。

通常，获取消息可能涉及到底层持久层的IO操作。

Return type:: 列表[BaseMessage]

clear() → None[source]#

清除当前会话中聊天历史的所有消息。

此方法从Redis存储中移除与当前session_id相关的所有消息。

Returns:: 无
Return type:: 无

示例

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

注意

此方法仅清除当前session_id的消息。
它使用Redis搜索查询来查找所有相关消息，然后删除它们。
操作是原子的 - 要么所有消息都被删除，要么一条都不删除。
清除后，Redis 搜索索引仍然保持，允许在需要时立即使用相同的 session_id 来存储新消息。
此操作不可逆。在调用此方法之前，请确保您想要删除所有消息。

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

在聊天历史中搜索与给定查询匹配的消息。

此方法对当前会话中消息的内容执行全文搜索。

Parameters:

query (str) – 用于匹配消息内容的搜索查询字符串。
limit (int, optional) – 返回的最大结果数量。
10. (默认值为)

Returns:

一个字典列表，每个字典代表一个: 匹配的消息。

每个字典包含消息内容和元数据。

Return type:

列表[字典[字符串, 任意类型]]

示例

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("---")

注意

搜索是使用Redis的搜索功能进行的，这使得全文搜索更加高效。
搜索不区分大小写，并使用Redis的默认分词和词干提取。
仅搜索来自当前会话（由session_id定义）的消息。
返回的字典包括所有存储的字段，通常包括‘content’、‘type’以及任何与消息一起存储的附加元数据。
此方法对于快速查找对话的相关部分非常有用，而无需遍历所有消息。

使用 RedisChatMessageHistory 的示例

Redis