Haystack 实验性检索组件 ChatMessageRetriever从 ChatMessageStore 按会话检索聊天历史【免费下载链接】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/haystackHaystack 是面向生产级 LLM 应用的开源 AI 编排框架而ChatMessageRetriever是其haystack-experimental实验命名空间中用于按会话检索聊天历史的检索组件它从底层的ChatMessageStore聊天消息存储中按chat_history_id拉取一组候选聊天消息并可与本轮新消息合并直接作为ChatGenerator或 Agent 的输入。读完本文你将掌握该组件的构造参数、run()运行语义、与InMemoryChatMessageStore的搭配用法、消息合并规则以及它背后在仓库中的源码级实现依据。组件定位与适用场景从参考文档 experimental_retrievers_api.md 的定义看ChatMessageRetriever属于haystack_experimental.components.retrievers.chat_message_retriever模块其职责一句话概括为Sweep through Document Stores and return a set of candidate documents that are relevant to the query.在存储中扫描并返回与查询相关的一组候选文档/消息。与 Haystack 中面向文档检索的DocumentRetriever不同ChatMessageRetriever 检索的是对话历史消息而不是文档片段。它天然服务于以下场景多轮对话应用把历史会话消息从存储中取出拼入新一轮的提示词上下文Agent 工作流为 Agent 提供会话记忆让它在多轮工具调用之间保持上下文连贯会话隔离借助chat_history_id作为命名空间同一存储可同时承载多个用户/会话互不干扰。该组件当前位于haystack-experimental实验命名空间haystack/core/serialization_security.py中的模块白名单也明确放行了haystack_experimental见 serialization_security.py说明它在反序列化安全边界内被官方认可为允许加载的模块。快速上手最小可用示例参考文档给出的用法示例完整复制即可运行from haystack.dataclasses import ChatMessage from haystack_experimental.components.retrievers import ChatMessageRetriever from haystack_experimental.chat_message_stores.in_memory import 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 InMemoryChatMessageStore() message_store.write_messages(chat_history_iduser_456_session_123, messagesmessages) retriever ChatMessageRetriever(message_store) result retriever.run(chat_history_iduser_456_session_123) print(result[messages])流程拆解用ChatMessage.from_*工厂方法构造消息列表。这些工厂方法定义在 chat_message.py 中from_assistantL482、from_userL428、from_systemL470、from_toolL518分别对应 assistant / user / system / tool 四种角色可携带meta元数据assistant 消息还支持tool_calls与reasoning内容见 chat_message.py。创建InMemoryChatMessageStore用write_messages(chat_history_id, messages)写入会话历史把 store 传入ChatMessageRetriever(message_store)构造检索器调用run(chat_history_id...)检索返回的字典中result[messages]即检索到的消息列表。输出结果为dict[str, list[ChatMessage]]键为messages。构造参数与行为语义构造函数def __init__(chat_message_store: ChatMessageStore, last_k: int | None 10)参数类型默认值说明chat_message_storeChatMessageStore必填ChatMessageStore实例检索器的数据来源last_kint \| None10默认检索最近多少条消息为None表示不限制条数取全部run 方法component.output_types(messageslist[ChatMessage]) def run( chat_history_id: str, *, last_k: int | None None, current_messages: list[ChatMessage] | None None ) - dict[str, list[ChatMessage]]run()支持三个入参chat_history_id必填会话/对话的唯一标识符。每个chat_history_id对应存储在ChatMessageStore中的一份独立聊天历史例如用 session ID 或 conversation ID 来隔离不同会话的消息避免串话。last_k可选本次运行要检索的最近消息条数。该参数优先于构造函数中传入的last_k若未指定则回退使用构造时的last_k。这为构造时定默认、运行时按需覆盖提供了灵活性。current_messages可选需要与检索结果合并的本轮新消息列表。合并规则为该列表中的 system 消息被前置到检索出的历史消息之前而其余消息如 user 消息被追加到历史之后。这样输出可以直接作为ChatGenerator或 Agent 的输入——系统提示词在最前历史在中间本轮新消息在最后形成标准的多轮对话输入形态。若未提供则只返回存储中的消息。异常若last_k不为None且小于 0抛出ValueError。消息存储底座InMemoryChatMessageStoreChatMessageRetriever 本身不存数据真正的读写发生在ChatMessageStore。参考文档配套的 experimental_chatmessage_store_api.md 说明InMemoryChatMessageStore把chat_history_id当作每个会话的命名空间每个 id 对应内存中独立的一组ChatMessage列表写入、读取、删除都围绕该 id 进行从而保证不同会话之间不重叠。其完整方法面如下方法签名说明write_messageswrite_messages(chat_history_id: str, messages: list[ChatMessage]) - int写入消息返回写入条数messages非ChatMessage列表时抛ValueErrorretrieve_messagesretrieve_messages(chat_history_id: str, last_k: int \| None None) - list[ChatMessage]读取消息last_k未指定时使用构造函数中的last_k小于 0 抛ValueErrorcount_messagescount_messages(chat_history_id: str) - int返回指定会话的消息条数delete_messagesdelete_messages(chat_history_id: str) - None删除指定会话的全部消息delete_all_messagesdelete_all_messages() - None清空所有会话的消息to_dict/from_dict—序列化 / 反序列化整个 store其构造函数为def __init__(skip_system_messages: bool True, last_k: int | None 10) - Noneskip_system_messages是否跳过存储 system 消息默认True。开启后写入的 system 提示词不会进入历史存储避免每次对话都把冗长的系统提示重复落库。last_kretrieve_messages未显式传参时的默认最近消息条数默认10。序列化与反序列化ChatMessageRetriever作为 Haystack 组件实现了标准的to_dict/from_dict协议def to_dict() - dict[str, Any] # 将组件序列化为字典 classmethod def from_dict(cls, data: dict[str, Any]) - ChatMessageRetriever # 从字典反序列化这两者与 Haystack 的组件序列化体系serialization.py保持一致意味着 ChatMessageRetriever 可以像其他组件一样被序列化为 YAML/JSON 描述进而参与 Pipeline 的保存与加载、断点快照breakpoint snapshot等流程。InMemoryChatMessageStore同样实现了to_dict/from_dict存储本身可被序列化后整体持久化。与 ChatGenerator / Agent 的衔接模式文档明确指出run()的输出can be directly used as input to a ChatGenerator or an Agent。典型的衔接写法# 假设已有 message_store 与 retriever # 1) 先写入历史如上一步 # 2) 构造本轮新消息 new_user_msg ChatMessage.from_user(Can you give me an example of a Protocol in Python?) # 3) 检索历史并合并本轮消息system 前置、user 追加 result retriever.run( chat_history_iduser_456_session_123, current_messages[new_user_msg], ) # 4) 直接把 messages 喂给 ChatGenerator 或 Agent # chat_generator.run(messagesresult[messages])current_messages的合并设计是有意为之system 消息永远放在历史之前以固定系统角色设定新 user 消息放在历史之后以延续对话得到的就是一条可直接推理的消息序列。对于工具调用类 Agent历史中 assistant 的tool_calls与对应的 tool 结果消息from_tool生成同样会随历史完整取回保证多轮工具调用的因果链不丢失相关消息结构定义见 chat_message.py。边界与注意点实验性 APIhaystack_experimental命名空间下的 API 处于实验阶段接口可能随版本演进调整。当前文档快照对应 Haystack 2.18 时代version-2.18 目录使用时应留意你安装的haystack-experimental包版本与文档的一致性。内存型存储InMemoryChatMessageStore的数据仅存在于进程内存中进程退出即丢失需要持久化会话时应改用具备后端持久化能力的ChatMessageStore实现实验包中另有 mem0 等内存记忆存储可参考 experimental_mem0_memory_store_api.md。参数优先级run(last_k...) 构造函数last_kretrieve_messages(last_k...) 存储构造函数last_k。理解这条覆盖链才能避免传了却没生效的困惑。非法入参last_k 0会抛ValueErrorwrite_messages收到非ChatMessage列表同样抛ValueError。小结ChatMessageRetriever是 Haystack 实验组件中面向会话记忆检索的最小闭环InMemoryChatMessageStore负责按chat_history_id命名空间存取消息ChatMessageRetriever负责按需取出最近last_k条历史并合并本轮current_messages最终产出一份可直接投喂ChatGenerator或 Agent 的消息序列。配合to_dict/from_dict序列化协议它也可以嵌入 Pipeline 快照与序列化流程。理解构造参数与run()的覆盖优先级即可在多轮对话与 Agent 场景中正确接入会话记忆能力。【免费下载链接】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），仅供参考