Haystack ChatMessageStore 实验组件详解用 InMemoryChatMessageStore 管理多会话对话历史【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack本文聚焦 Haystack 实验性组件包haystack-experimental中的InMemoryChatMessageStore一种以chat_history_id为命名空间、在内存中按会话隔离存储ChatMessage的对话历史仓库。通过阅读本文你将掌握该组件的构造参数、完整 CRUD 方法签名与异常语义并能直接复用文档中的可运行示例为自己的 Agent 或对话式 RAG 应用搭建会话级消息存取能力。ChatMessageStore 是什么对话记忆的存取层在多轮对话与 Agent 场景中模型上下文需要不断累积用户提问、助手回答乃至工具调用记录。Haystack 的ChatMessage数据类定义于 haystack/dataclasses/chat_message.py承载单条消息而 ChatMessageStore 则负责把这些消息按会话批量写入、检索、计数与删除构成对话系统的记忆存取层。本文主讲的InMemoryChatMessageStore属于haystack-experimental 实验性包中的组件其完整引用路径为haystack_experimental.chat_message_stores.in_memory.InMemoryChatMessageStore。之所以在 haystack 主仓库之外是因为它仍处于实验演进阶段仓库中的发布说明 add-haystack-experimental-dependency-96ff02e71bc2af13.yaml 也印证了主框架对实验包的依赖关系。本文对应 API 参考原文位于 docs-website/reference_versioned_docs/version-2.18/experiments-api/experimental_chatmessage_store_api.md该 API 在 2.18 至 2.31 的版本化参考文档中持续存在接口保持稳定。从文档描述看该存储的核心设计是一张以会话 ID 为键、以消息列表为值的内存映射每个chat_history_id对应一份独立的ChatMessage列表互不干扰。核心机制chat_history_id 会话命名空间chat_history_id是本组件最重要的概念。文档明确指出它作为每次对话conversation或聊天会话chat session的唯一标识符充当隔离不同会话消息的命名空间每一个chat_history_id取值都对应内存中一份独立的ChatMessage列表。因此在使用约定上有一条黄金法则写入、读取或删除消息时务必传入同一个唯一chat_history_id例如 session ID 或 conversation ID这样才能保证不同会话的消息互不重叠。一旦多个会话共用一个 ID消息就会混在一起反之同一会话使用不同 ID 则会导致历史失忆。快速上手完整可运行示例文档给出了一个可以直接运行的示例演示了最核心的写—读闭环from haystack.dataclasses import ChatMessage from haystack_experimental.chat_message_stores.in_memory import InMemoryChatMessageStore message_store InMemoryChatMessageStore() messages [ ChatMessage.from_assistant(Hello, how can I help you?), ChatMessage.from_user(Hi, I have a question about Python. What is a Protocol?), ] message_store.write_messages(chat_history_iduser_456_session_123, messagesmessages) retrieved_messages message_store.retrieve_messages(chat_history_iduser_456_session_123) print(retrieved_messages)其中ChatMessage.from_assistant(...)与ChatMessage.from_user(...)是 ChatMessage 提供的类方法构造器源码中还包括from_systemchat_message.py#L470、from_assistantchat_message.py#L482、from_toolchat_message.py#L518等分别用于构造系统、助手、用户、工具角色的消息正好覆盖 Agent 循环中的各类消息来源。值得注意的细节示例中write_messages与retrieve_messages使用了同一个chat_history_iduser_456_session_123这正是上一节命名空间原则的直接体现——读取时若传入其他 ID将拿不到刚写入的消息。构造参数详解InMemoryChatMessageStore的构造函数签名如下def __init__(skip_system_messages: bool True, last_k: int | None 10) - None参数类型默认值说明skip_system_messagesboolTrue是否跳过存储系统消息system message默认跳过last_kint \| None10检索时默认返回最近的消息条数不指定时为 10 条两点理解skip_system_messagesTrue意味着写入时系统角色消息默认被过滤避免每次检索都带回重复的系统提示节省上下文空间last_k10是检索侧的兜底窗口——调用retrieve_messages时不传last_k就会回退到构造时的这个值。API 方法逐个拆解组件提供 7 个公开方法覆盖存储的完整生命周期写入、检索、计数、删除与序列化。write_messages写入消息def write_messages(chat_history_id: str, messages: list[ChatMessage]) - intchat_history_id消息所属会话的命名空间键messages要写入的ChatMessage列表RaisesValueError当messages不是ChatMessage列表时抛出Returns成功写入的消息条数int。写入的返回值让你可以校验实际写入了多少条例如与len(messages)对比以排查被skip_system_messages过滤掉的系统消息。retrieve_messages检索消息def retrieve_messages(chat_history_id: str, last_k: int | None None) - list[ChatMessage]chat_history_id从哪个会话取消息last_k返回最近多少条消息若传None则回退使用构造函数传入的last_k默认 10——这就是构造参数与调用参数联动的设计RaisesValueError当last_k不为None且小于 0 时抛出ReturnsChatMessage列表。该方法的last_k覆盖语义非常实用既可以在构造时设定全局默认窗口又能在某些场景如首轮只需最近 3 条按需覆盖。count_messages统计消息数量def count_messages(chat_history_id: str) - int返回指定会话下已存储的消息条数适合在做上下文预算、token 控制前快速评估会话长度。delete_messages 与 delete_all_messages删除消息def delete_messages(chat_history_id: str) - None def delete_all_messages() - Nonedelete_messages(chat_history_id)删除指定会话的全部消息保留其他会话delete_all_messages()清空所有会话的消息——用于整体重置如测试环境清理需谨慎调用。to_dict 与 from_dict序列化往返def to_dict() - dict[str, Any] classmethod def from_dict(cls, data: dict[str, Any]) - InMemoryChatMessageStoreto_dict将组件序列化为字典返回序列化后的数据from_dict从字典反序列化并返回组件实例。这对方法保证组件可以被 Haystack 的管线序列化体系YAML/JSON 保存与加载收纳是将其接入Pipeline或PipelineTool等场景的基础能力。典型应用Agent 会话记忆的存取闭环综合以上 API一个典型的会话记忆使用模式是写入多轮追加→ 检索取最近窗口→ 交给生成组件 → 再写入新消息 → 会话结束时按 ID 清理。伪代码化的完整流程如下from haystack.dataclasses import ChatMessage from haystack_experimental.chat_message_stores.in_memory import InMemoryChatMessageStore store InMemoryChatMessageStore(skip_system_messagesTrue, last_k10) session_id user_456_session_123 # 1. 会话开始时写入系统指令skip_system_messages 默认会过滤需按需调整 store.write_messages(chat_history_idsession_id, messages[ ChatMessage.from_system(You are a helpful Python assistant.), ]) # 2. 每轮对话追加消息 store.write_messages(chat_history_idsession_id, messages[ ChatMessage.from_user(What is a Protocol?), ]) # 3. 检索最近 10 条作为生成上下文 history store.retrieve_messages(chat_history_idsession_id) # 4. 追加助手回复 store.write_messages(chat_history_idsession_id, messages[ ChatMessage.from_assistant(A Protocol is a structural typing interface in Python...), ]) # 5. 可选检查会话长度 / 会话结束后清理 print(store.count_messages(chat_history_idsession_id)) # store.delete_messages(chat_history_idsession_id) # 结束该会话 # store.delete_all_messages() # 清空所有会话在实际 Agent 循环中retrieve_messages返回的历史列表通常直接拼入 ChatPromptBuilder 或 ChatGenerator 的messages输入实现多轮上下文延续。使用注意事项与边界内存态存储该类名为InMemory数据仅存活于进程内存中进程退出即丢失不适用于需要持久化的生产场景对持久化需求应关注文档存储或其他记忆方案。实验性 API组件位于haystack_experimental包接口可能随版本演进调整升级时需留意对应版本的 API 参考文档。系统消息过滤skip_system_messagesTrue是默认行为若你的会话依赖系统消息参与每次生成需要显式评估其影响。ID 必须唯一且一致写、读、删必须使用同一个chat_history_id这是本组件正确工作的前提。异常语义write_messages对非ChatMessage列表抛ValueErrorretrieve_messages对负数last_k抛ValueError调用方应据此做输入校验。延伸阅读本文 API 参考原文experimental_chatmessage_store_api.md相同 API 亦存在于 2.192.31 各版本参考目录消息数据类源码haystack/dataclasses/chat_message.py主仓库与实验包的依赖关系说明add-haystack-experimental-dependency-96ff02e71bc2af13.yaml同目录下的其他实验组件参考experimental_agents_api.md、experimental_writers_api.md【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
长河君 · 2026/9/12 12:09:52 · 阅读 1.2k Haystack ChatMessageStore 实验组件详解：用 InMemoryChatMessageStore 管理多会话对话历史
Haystack ChatMessageStore 实验组件详解用 InMemoryChatMessageStore 管理多会话对话历史【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack本文聚焦 Haystack 实验性组件包haystack-experimental中的InMemoryChatMessageStore一种以chat_history_id为命名空间、在内存中按会话隔离存储ChatMessage的对话历史仓库。通过阅读本文你将掌握该组件的构造参数、完整 CRUD 方法签名与异常语义并能直接复用文档中的可运行示例为自己的 Agent 或对话式 RAG 应用搭建会话级消息存取能力。ChatMessageStore 是什么对话记忆的存取层在多轮对话与 Agent 场景中模型上下文需要不断累积用户提问、助手回答乃至工具调用记录。Haystack 的ChatMessage数据类定义于 haystack/dataclasses/chat_message.py承载单条消息而 ChatMessageStore 则负责把这些消息按会话批量写入、检索、计数与删除构成对话系统的记忆存取层。本文主讲的InMemoryChatMessageStore属于haystack-experimental 实验性包中的组件其完整引用路径为haystack_experimental.chat_message_stores.in_memory.InMemoryChatMessageStore。之所以在 haystack 主仓库之外是因为它仍处于实验演进阶段仓库中的发布说明 add-haystack-experimental-dependency-96ff02e71bc2af13.yaml 也印证了主框架对实验包的依赖关系。本文对应 API 参考原文位于 docs-website/reference_versioned_docs/version-2.18/experiments-api/experimental_chatmessage_store_api.md该 API 在 2.18 至 2.31 的版本化参考文档中持续存在接口保持稳定。从文档描述看该存储的核心设计是一张以会话 ID 为键、以消息列表为值的内存映射每个chat_history_id对应一份独立的ChatMessage列表互不干扰。核心机制chat_history_id 会话命名空间chat_history_id是本组件最重要的概念。文档明确指出它作为每次对话conversation或聊天会话chat session的唯一标识符充当隔离不同会话消息的命名空间每一个chat_history_id取值都对应内存中一份独立的ChatMessage列表。因此在使用约定上有一条黄金法则写入、读取或删除消息时务必传入同一个唯一chat_history_id例如 session ID 或 conversation ID这样才能保证不同会话的消息互不重叠。一旦多个会话共用一个 ID消息就会混在一起反之同一会话使用不同 ID 则会导致历史失忆。快速上手完整可运行示例文档给出了一个可以直接运行的示例演示了最核心的写—读闭环from haystack.dataclasses import ChatMessage from haystack_experimental.chat_message_stores.in_memory import InMemoryChatMessageStore message_store InMemoryChatMessageStore() messages [ ChatMessage.from_assistant(Hello, how can I help you?), ChatMessage.from_user(Hi, I have a question about Python. What is a Protocol?), ] message_store.write_messages(chat_history_iduser_456_session_123, messagesmessages) retrieved_messages message_store.retrieve_messages(chat_history_iduser_456_session_123) print(retrieved_messages)其中ChatMessage.from_assistant(...)与ChatMessage.from_user(...)是 ChatMessage 提供的类方法构造器源码中还包括from_systemchat_message.py#L470、from_assistantchat_message.py#L482、from_toolchat_message.py#L518等分别用于构造系统、助手、用户、工具角色的消息正好覆盖 Agent 循环中的各类消息来源。值得注意的细节示例中write_messages与retrieve_messages使用了同一个chat_history_iduser_456_session_123这正是上一节命名空间原则的直接体现——读取时若传入其他 ID将拿不到刚写入的消息。构造参数详解InMemoryChatMessageStore的构造函数签名如下def __init__(skip_system_messages: bool True, last_k: int | None 10) - None参数类型默认值说明skip_system_messagesboolTrue是否跳过存储系统消息system message默认跳过last_kint \| None10检索时默认返回最近的消息条数不指定时为 10 条两点理解skip_system_messagesTrue意味着写入时系统角色消息默认被过滤避免每次检索都带回重复的系统提示节省上下文空间last_k10是检索侧的兜底窗口——调用retrieve_messages时不传last_k就会回退到构造时的这个值。API 方法逐个拆解组件提供 7 个公开方法覆盖存储的完整生命周期写入、检索、计数、删除与序列化。write_messages写入消息def write_messages(chat_history_id: str, messages: list[ChatMessage]) - intchat_history_id消息所属会话的命名空间键messages要写入的ChatMessage列表RaisesValueError当messages不是ChatMessage列表时抛出Returns成功写入的消息条数int。写入的返回值让你可以校验实际写入了多少条例如与len(messages)对比以排查被skip_system_messages过滤掉的系统消息。retrieve_messages检索消息def retrieve_messages(chat_history_id: str, last_k: int | None None) - list[ChatMessage]chat_history_id从哪个会话取消息last_k返回最近多少条消息若传None则回退使用构造函数传入的last_k默认 10——这就是构造参数与调用参数联动的设计RaisesValueError当last_k不为None且小于 0 时抛出ReturnsChatMessage列表。该方法的last_k覆盖语义非常实用既可以在构造时设定全局默认窗口又能在某些场景如首轮只需最近 3 条按需覆盖。count_messages统计消息数量def count_messages(chat_history_id: str) - int返回指定会话下已存储的消息条数适合在做上下文预算、token 控制前快速评估会话长度。delete_messages 与 delete_all_messages删除消息def delete_messages(chat_history_id: str) - None def delete_all_messages() - Nonedelete_messages(chat_history_id)删除指定会话的全部消息保留其他会话delete_all_messages()清空所有会话的消息——用于整体重置如测试环境清理需谨慎调用。to_dict 与 from_dict序列化往返def to_dict() - dict[str, Any] classmethod def from_dict(cls, data: dict[str, Any]) - InMemoryChatMessageStoreto_dict将组件序列化为字典返回序列化后的数据from_dict从字典反序列化并返回组件实例。这对方法保证组件可以被 Haystack 的管线序列化体系YAML/JSON 保存与加载收纳是将其接入Pipeline或PipelineTool等场景的基础能力。典型应用Agent 会话记忆的存取闭环综合以上 API一个典型的会话记忆使用模式是写入多轮追加→ 检索取最近窗口→ 交给生成组件 → 再写入新消息 → 会话结束时按 ID 清理。伪代码化的完整流程如下from haystack.dataclasses import ChatMessage from haystack_experimental.chat_message_stores.in_memory import InMemoryChatMessageStore store InMemoryChatMessageStore(skip_system_messagesTrue, last_k10) session_id user_456_session_123 # 1. 会话开始时写入系统指令skip_system_messages 默认会过滤需按需调整 store.write_messages(chat_history_idsession_id, messages[ ChatMessage.from_system(You are a helpful Python assistant.), ]) # 2. 每轮对话追加消息 store.write_messages(chat_history_idsession_id, messages[ ChatMessage.from_user(What is a Protocol?), ]) # 3. 检索最近 10 条作为生成上下文 history store.retrieve_messages(chat_history_idsession_id) # 4. 追加助手回复 store.write_messages(chat_history_idsession_id, messages[ ChatMessage.from_assistant(A Protocol is a structural typing interface in Python...), ]) # 5. 可选检查会话长度 / 会话结束后清理 print(store.count_messages(chat_history_idsession_id)) # store.delete_messages(chat_history_idsession_id) # 结束该会话 # store.delete_all_messages() # 清空所有会话在实际 Agent 循环中retrieve_messages返回的历史列表通常直接拼入 ChatPromptBuilder 或 ChatGenerator 的messages输入实现多轮上下文延续。使用注意事项与边界内存态存储该类名为InMemory数据仅存活于进程内存中进程退出即丢失不适用于需要持久化的生产场景对持久化需求应关注文档存储或其他记忆方案。实验性 API组件位于haystack_experimental包接口可能随版本演进调整升级时需留意对应版本的 API 参考文档。系统消息过滤skip_system_messagesTrue是默认行为若你的会话依赖系统消息参与每次生成需要显式评估其影响。ID 必须唯一且一致写、读、删必须使用同一个chat_history_id这是本组件正确工作的前提。异常语义write_messages对非ChatMessage列表抛ValueErrorretrieve_messages对负数last_k抛ValueError调用方应据此做输入校验。延伸阅读本文 API 参考原文experimental_chatmessage_store_api.md相同 API 亦存在于 2.192.31 各版本参考目录消息数据类源码haystack/dataclasses/chat_message.py主仓库与实验包的依赖关系说明add-haystack-experimental-dependency-96ff02e71bc2af13.yaml同目录下的其他实验组件参考experimental_agents_api.md、experimental_writers_api.md【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考