Haystack 数据类Data Classes完全指南驱动 LLM 应用的数据骨架与序列化协议【免费下载链接】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 2.22 版本的官方 API 参考docs-website/reference_versioned_docs/version-2.22/haystack-api/data_classes_api.md系统讲解驱动整个框架运转的核心数据类Document、ByteStream、ChatMessage及其内容部件、ExtractedAnswer/GeneratedAnswer、SparseEmbedding、ImageContent与StreamingChunk。读完本文你将掌握这些数据对象的字段语义、构造方式、与 OpenAI 等外部 API 的互转协议以及贯穿所有数据类的to_dict/from_dict序列化机制从而能在 RAG 管道、Agent 工作流与流式输出场景中正确地组织、传递和持久化数据。说明所有代码与行为描述均以当前仓库 haystack/dataclasses 目录下的源码实现为准版本 2.22示例可直接复制运行。数据类在 Haystack 中的地位Haystack 是一个开源 AI 编排框架用于构建上下文工程化、生产就绪的 LLM 应用其核心能力体现在模块化 Pipeline 与 Agent 工作流的设计上。无论是检索、路由、记忆还是生成管道中流动的每一条数据都由 haystack/dataclasses 中定义的数据类承载因此这套数据类被称为“驱动系统运转的数据骨架”。从源码的模块组织见 haystack/dataclasses/__init__.py可以看出数据类被划分为几大职责域文档与二进制内容Document、ByteStream、SparseEmbedding对应 RAG 检索链路中的存储与召回对象对话消息ChatMessage、ChatRole、TextContent、ToolCall、ToolCallResult、ReasoningContent、ImageContent对应 LLM 对话与 Agent 工具调用链路问答结果Answer、ExtractedAnswer、GeneratedAnswer对应抽取式阅读器与生成式模型两种答案形态流式输出StreamingChunk、ToolCallDelta、ComponentInfo以及StreamingCallbackT/AsyncStreamingCallbackT回调类型对应生成模型的增量输出。这些数据类普遍遵循两个设计约定由装饰器_warn_on_inplace_mutation与各类的to_dict/from_dict方法体现见 haystack/utils/dataclasses.py统一的序列化协议几乎每个数据类都实现to_dict()与from_dict()保证对象可无损地存入 JSON、落盘、跨进程传输并在反序列化后还原为完整类型可变性防护装饰器_warn_on_inplace_mutation会在就地修改字段时发出警告避免因 dataclass 哈希值失效而引发隐蔽 bug。下面按模块逐一深入。Document可被查询的基础数据单元Document是 Haystack 中最核心的数据类定义在 haystack/dataclasses/document.py代表“一段可被查询的数据”。它既可以承载纯文本content也可以携带二进制数据blob类型为ByteStream还可以附加向量、稀疏向量、评分与任意元数据。字段语义与类型约束字段类型说明idstr文档唯一标识不显式设置时由__post_init__依据各字段值自动生成SHA-256 哈希contentstr \| None文档的文本内容若传入非字符串构造函数抛出ValueErrorblobByteStream \| None与文档关联的二进制数据如图片、音频metadict[str, Any]自定义元数据必须是 JSON 可序列化的scorefloat \| None文档得分通常由 retriever 在召回时赋值用于排序embeddinglist[float] \| None文档的稠密向量表示sparse_embeddingSparseEmbedding \| None文档的稀疏向量表示源码层面的实现细节值得注意见 document.pyID 自动生成__post_init__中执行self.id self.id or self._create_id()。_create_id将content、blob.data、blob.mime_type、排序后的metaJSON、embedding、sparse_embedding拼接后做 SHA-256 哈希因此内容完全相同的文档会得到相同的 ID这为去重提供了天然依据。同时 meta 按键排序保证 meta 字典的键顺序不影响 ID。1.x 兼容性若embedding传入 NumPy 数组Haystack 1.x 时代的存储格式构造函数会自动调用ndarray.tolist()转为list[float]元类_RemoveLegacyFields则会静默移除content_type、id_hash_keys、dataframe这些 1.x 遗留字段见 document.py。相等性__eq__通过比较双方to_dict(flattenFalse)的字典表示来判断两个Document是否相等见 document.py。序列化to_dict 的扁平化与冲突处理Document.to_dict(flatten: bool True)的行为与 Haystack 1.x 的向后兼容性紧密相关见 document.py当flattenTrue默认时meta字典会被展开平铺到顶层但若某个 meta 键与文档字段名如id、content、score冲突该键会被保留在嵌套的meta字典中避免覆盖文档字段。blob与sparse_embedding会分别调用各自数据类的to_dict()转换成 JSON 可序列化的形式。当flattenFalse时返回未扁平化的原始字典结构Document.__eq__正是基于此模式比较。Document.from_dict见 document.py执行逆过程把扁平化的 meta 键重新收拢回meta字典并把blob、sparse_embedding还原为ByteStream、SparseEmbedding对象。因此Document可以在 JSON/YAML 管道配置与运行时对象之间无缝往返。用法示例from haystack.dataclasses import Document, ByteStream doc Document( contentHaystack is an open-source AI orchestration framework., meta{source: docs, language: en}, score0.95, ) print(doc.id) # 自动生成: 64 位十六进制 SHA-256 哈希 print(doc.content_type) # text1.x 兼容属性 # 二进制文档 blob_doc Document( blobByteStream.from_file_path(images/logo.png, guess_mime_typeTrue), meta{kind: image}, ) # 序列化往返 d doc.to_dict() # meta 被扁平化平铺到顶层 restored Document.from_dict(d) assert doc restoredByteStream管道中的二进制对象ByteStream是 Haystack 表示二进制数据的统一载体定义在 haystack/dataclasses/byte_stream.py字段为data: bytes、meta: dict、mime_type: str | None。文档Document.blob、文件抓取器、各类转换器如 PDF/图片转换都通过它传递二进制内容。构造与转换方法方法签名行为to_file(destination_path: Path) - None将二进制数据写入文件meta 信息会丢失见 byte_stream.pyfrom_file_path(filepath, mime_typeNone, metaNone, guess_mime_typeFalse)从文件读取内容构造ByteStream当mime_type未提供且guess_mime_typeTrue时通过文件内容自动猜测 MIME 类型底层调用_guess_mime_type见 haystack/utils/misc.pyfrom_string(text, encodingutf-8, mime_typeNone, metaNone)用指定编码把字符串编码为字节to_string(encodingutf-8) - str解码回字符串解码失败抛出UnicodeDecodeError序列化细节to_dict()返回{data: list(self.data), meta: ..., mime_type: ...}——注意bytes 被转换为整数列表因为 JSON 无法直接序列化 bytes见 byte_stream.pyfrom_dict则用bytes(data[data])还原。此外ByteStream还提供_to_trace_dict()见 byte_stream.py用于 tracing 场景将二进制内容替换为占位字符串Binary data (N bytes)避免向 tracing 后端发送大体积负载。__repr__会把data截断到前 100 字节再加...防止在日志中打印巨大二进制内容。用法示例from pathlib import Path from haystack.dataclasses import ByteStream # 文件 - ByteStream自动猜测 MIME 类型 stream ByteStream.from_file_path(data/sample.pdf, guess_mime_typeTrue) print(stream.mime_type) # application/pdf # 字符串 - ByteStream - 字符串 stream2 ByteStream.from_string(hello haystack, mime_typetext/plain) assert stream2.to_string() hello haystack # 序列化往返bytes - list[int] d stream.to_dict() restored ByteStream.from_dict(d) # 写回磁盘注意 meta 丢失 stream.to_file(Path(data/copy.pdf))ChatMessage 与聊天内容部件ChatMessage表示 LLM 聊天对话中的一条消息是 Agent 与 Chat Generator 类组件之间传递消息的标准载体定义在 haystack/dataclasses/chat_message.py。构造约定官方文档明确指出应使用from_assistant、from_user、from_system、from_tool四个类方法来创建消息而不是直接调用构造函数构造函数接受带下划线的内部字段_role、_content、_name、_meta不鼓励直接使用。ChatRole消息角色枚举ChatRole继承自str, Enum提供四个角色见 chat_message.py枚举值字符串值语义USERuser用户消息只包含文本SYSTEMsystem系统消息只包含文本ASSISTANTassistant助手消息可包含文本与 Tool 调用也可携带元数据TOOLtool工具消息包含一次工具调用的结果ChatRole.from_str(string)静态方法可将字符串转为枚举未知字符串抛出ValueError并列出所有支持的角色。消息内容部件ChatMessage的_content是一个内容部件序列Sequence[TextContent | ToolCall | ToolCallResult | ImageContent | ReasoningContent | FileContent]每种内容类型对应一个内容部件数据类TextContent(text: str)文本内容ToolCall(tool_name, arguments, idNone, extraNone)模型准备好的工具调用arguments为调用参数 dictextra用于存放提供商特有信息必须 JSON 可序列化ToolCallResult(result, origin, error)工具调用的结果。result可以是字符串也可以是TextContent | ImageContent | FileContent序列origin指向产生该结果的ToolCallerror标记调用是否出错见 chat_message.pyReasoningContent(reasoning_text, extraNone)模型输出的推理过程文本用于思维链类模型extra存放提供商特有信息见 chat_message.pyImageContent图片内容见下节FileContent文件内容如 PDF通过 base64 数据承载。序列化时每种部件会包上自己的序列化键text、tool_call、tool_call_result、image、reasoning、file见 chat_message.py 中的_CONTENT_PART_CLASSES_TO_SERIALIZATION_KEYS与_serialize_content_part/_deserialize_content_part。常用属性与方法ChatMessage对内容部件提供了丰富的只读访问属性见 chat_message.py属性/方法返回说明roleChatRole消息角色metadict与消息关联的元数据namestr \| None参与者名称仅 OpenAI 支持texts/textlist[str]/str \| None消息中全部 / 第一条文本tool_calls/tool_calllist[ToolCall]/ToolCall \| None全部 / 第一个 Tool 调用tool_call_results/tool_call_resultlist[ToolCallResult]/ToolCallResult \| None全部 / 第一个工具结果images/imagelist[ImageContent]/ImageContent \| None全部 / 第一张图片files/filelist[FileContent]/FileContent \| None全部 / 第一个文件reasonings/reasoninglist[ReasoningContent]/ReasoningContent \| None全部 / 第一条推理内容is_from(role)bool判断消息是否来自指定角色接受ChatRole或字符串__len__返回内容部件数量。__new__与__getattribute__被重新实现用于在改动 dataclass 字段时如移除旧的content属性向使用者发出更明显的提示。四个构造类方法# 用户消息text 与 content_parts 二选一 user_msg ChatMessage.from_user(text你好帮我查一下文档) # 多模态用户消息文本 图片 multi_msg ChatMessage.from_user(content_parts[看这张图, ImageContent.from_file_path(chart.png)]) # 系统消息 system_msg ChatMessage.from_system(text你是一个严谨的助手。) # 助手消息可携带 tool_calls 与 reasoning assistant_msg ChatMessage.from_assistant( text我来查询一下。, tool_calls[ToolCall(idcall_1, tool_namesearch, arguments{query: haystack})], reasoning用户想知道 haystack 的用法, ) # 工具结果消息origin 必须指向对应 ToolCall tool_msg ChatMessage.from_tool( tool_result找到 3 条相关文档, originassistant_msg.tool_call, # 引用助手消息中的 ToolCall )各方法的参数约束源码层面验证见 chat_message.pyfrom_usertext与content_parts必须且只能提供一个否则抛ValueErrorcontent_parts只接受str、TextContent、ImageContent、FileContentfrom_systemtext为必填from_assistantreasoning接受str或ReasoningContenttool_calls为ToolCall列表from_tooltool_result为字符串或内容部件序列origin必填error默认False。与 OpenAI Chat API 的互转ChatMessage提供两套与 OpenAI Chat Completions API 的双向转换见 chat_message.pyto_openai_dict_format(require_tool_call_ids: bool True)将消息转为 OpenAI 期望的字典格式_meta会被丢弃OpenAI API 不支持用户多模态消息中的图片转为image_url类型的 content 部件MIME 缺失时默认image/jpeg工具结果消息要求ToolCall带非空id除非require_tool_call_idsFalse用于兼容浅层 OpenAI 兼容 API助手消息的tool_calls转为{type: function, function: {...}}结构参数用json.dumps(..., ensure_asciiFalse)序列化校验规则非助手消息不能为空含ToolCallResult的消息不能再携带其他内容OpenAI 兼容限制推理内容reasoning会被忽略Chat Completions API 不支持。from_openai_dict_format(message)将 OpenAI 格式的字典还原为ChatMessagetool_call_id缺失时仍可构造兼容浅层兼容 API但若后续要发给 OpenAI 必须补上处理 OpenAI 兼容服务器可能发送的空参数空串/null/缺失arguments——一律按{}处理支持developer角色映射为 system 消息校验不合法格式并抛出ValueError。ChatMessage 的序列化格式to_dict()输出{role: ..., meta: ..., name: ..., content: [部件字典, ...]}。from_dict兼容三种历史格式见 chat_message.py当前格式content为部件字典列表2.9.0 之前的格式content为纯字符串2.9.0 ~ 2.12.0 之间的格式使用_content键。这保证了老版本序列化数据的平滑迁移。同时_to_trace_dict()会把图片/文件的 base64 数据替换为占位符避免 tracing 负载过大。问答结果ExtractedAnswer 与 GeneratedAnswerhaystack/dataclasses/answer.py定义了答案的两种形态见 answer.py。Answer 协议Answer是一个用runtime_checkable标注的Protocol定义了数据类需要实现的协议接口data: Any、query: str、meta: dict以及to_dict/from_dict方法。它只约束形状不参与实例化。ExtractedAnswer抽取式答案由抽取式 Readerextractive reader产出包含查询、得分、答案文本以及可选的文档/上下文定位信息字段类型说明querystr原始查询scorefloat答案得分datastr \| None答案文本documentDocument \| None答案来源文档contextstr \| None答案所在上下文片段document_offsetSpan \| None答案在文档中的字符区间context_offsetSpan \| None答案在上下文中的字符区间metadict元数据内部的Span数据类只有start、end两个整数字段。to_dict将document以to_dict(flattenFalse)序列化、Span用asdict序列化from_dict兼容旧版init_parameters包装格式见 answer.py。GeneratedAnswer生成式答案由 Generator 产出包含答案文本、查询、引用文档与元数据字段类型说明datastr生成的答案文本querystr原始查询documentslist[Document]生成答案时引用的文档metadict元数据序列化时若meta中的all_messages是ChatMessage列表会先转为字典列表documents以flattenFalse形式序列化。from_dict则把字典形式的all_messages还原为ChatMessage对象见 answer.py。ImageContent聊天中的图片内容ImageContent表示聊天消息中的图片内容见 haystack/dataclasses/image_content.py字段包括字段类型说明base64_imagestr图片的 base64 字符串mime_typestr \| None图片 MIME 类型如image/png、image/jpeg建议显式提供大多数 LLM 提供方要求缺失时会从 base64 数据猜测可能较慢且不一定可靠detailLiteral[auto, high, low] \| None图片细节级别仅 OpenAI 支持metadict元数据validationbool默认True开启时校验 base64 合法性、猜测缺失的 MIME 类型、并检查 MIME 是否为合法图片类型设为False可跳过校验加速初始化源码中内置了一张FORMAT_TO_MIME格式映射表支持 PNG、JPEG、GIF、WEBP、TIFF 等常见格式与IMAGE_MIME_TYPES集合用于校验见 image_content.py。__post_init__中base64 无效抛ValueErrorMIME 无法猜测时记录警告MIME 非法图片类型抛ValueError见 image_content.py。三个构造路径类方法签名要点行为from_file_path(file_path, *, sizeNone, detailNone, metaNone)从本地图片文件构造不支持 PDF底层委托给ImageFileToImageContent转换组件见 haystack/components/converters/image.pyfrom_url(url, *, retry_attempts2, timeout10, sizeNone, detailNone, metaNone)下载 URL 图片并转为 base64内部使用LinkContentFetcher见 haystack/components/fetchers/link_content.pyURL 指向非图片或 PDF 时抛ValueErrorshow() - None显示图片需要 Pillowpip install pillow在 Jupyter 中用IPython.display展示size参数(width, height)元组用于按比例缩放图片减小文件体积、内存占用与处理耗时适合有分辨率约束的模型或需要传输到远程服务的场景。PDF 转ImageContent请使用PDFToImageContent组件。用法示例from haystack.dataclasses import ImageContent # 本地文件自动猜测 MIME img ImageContent.from_file_path(images/diagram.png, size(512, 512), detailauto) # 从 URL 下载失败重试 2 次超时 10 秒 img2 ImageContent.from_url(https://example.com/photo.jpg, retry_attempts3, timeout15) # 直接构造显式 MIME 类型可避免猜测 img3 ImageContent(base64_imageiVBORw0KGgo..., mime_typeimage/png) # 放入用户消息 msg ChatMessage.from_user(content_parts[请分析这张图, img])SparseEmbedding稀疏向量表示SparseEmbedding用两个等长列表表示稀疏向量见 haystack/dataclasses/sparse_embedding.pyindices: list[int]非零元素的下标列表values: list[float]非零元素的值列表。__post_init__校验两者长度必须一致否则抛ValueError。to_dict/from_dict与其余数据类一致。它通常用于 BM25、SPLADE 等稀疏检索场景与Document.sparse_embedding字段配合使用可同时承载稠密向量embedding与稀疏向量以支持混合检索。from haystack.dataclasses import SparseEmbedding, Document sparse SparseEmbedding(indices[3, 17, 42], values[0.8, 0.5, 0.9]) doc Document(contentsparse retrieval, sparse_embeddingsparse) # 序列化往返 d doc.to_dict() restored Document.from_dict(d) assert restored.sparse_embedding sparseStreamingChunk 与流式输出协议StreamingChunk封装一段流式生成的内容及其元数据是流式输出的基本单位见 haystack/dataclasses/streaming_chunk.py。字段语义字段类型说明contentstr消息块的文本内容metadict与消息块相关的元数据component_infoComponentInfo \| None生成该块的组件信息类型与名称indexint \| None该块属于哪个内容块content block的可选索引tool_callslist[ToolCallDelta] \| None与消息块关联的工具调用增量tool_call_resultToolCallResult \| None工具调用结果startbool是否为某个内容块的起始块finish_reasonFinishReason \| None生成结束原因reasoningReasoningContent \| None与消息块关联的推理内容校验规则与 finish_reason__post_init__实施两条约束见 streaming_chunk.pycontent、tool_calls、tool_call_result、reasoning中最多只能设置一个否则抛ValueError一个块只能表达一种载荷若设置了tool_calls/tool_call_result/reasoning则index必须同时设置用于定位块所属的内容块。FinishReason是Literal类型别名见 streaming_chunk.py取值遵循 OpenAI 约定stop、length、tool_calls、content_filter外加 Haystack 特有值tool_call_results。ToolCallDelta流式工具调用增量ToolCallDelta(index, tool_nameNone, argumentsNone, idNone, extraNone)表示流式返回的工具调用增量见 streaming_chunk.py。arguments可以是完整 JSON 参数也可以是参数增量delta由各 LLM 提供方的流式协议决定。ComponentInfo流式块的来源信息ComponentInfo(type: str, name: str | None)记录产生流式块的组件见 streaming_chunk.py。from_component(component)类方法从Component实例自动提取type为f{模块}.{类名}的完整限定名name取组件的__component_name__属性即加入 Pipeline 时赋予的名称。回调选择函数select_streaming_callback(init_callback, runtime_callback, requires_async)用于在初始化回调与运行时回调之间做出选择运行时回调优先于初始化回调requires_async指定所选回调是否必须兼容异步见 streaming_chunk.py该文件同时定义了SyncStreamingCallbackT与AsyncStreamingCallbackT回调类型别名。用法示例from haystack.dataclasses import StreamingChunk, ComponentInfo def on_chunk(chunk: StreamingChunk) - None: if chunk.start: print(--- 内容块开始 ---) if chunk.content: print(chunk.content, end) if chunk.finish_reason: print(f\n[finish_reason{chunk.finish_reason}]) chunk StreamingChunk( contentHello, component_infoComponentInfo.from_component(chat_generator), startTrue, ) on_chunk(chunk) # 将 chunk 传给 Generator 的 streaming_callback 即可接入流式输出序列化机制总结纵观所有数据类序列化协议高度统一这是 Haystack 管道配置YAML/JSON、断点快照、缓存与追踪tracing得以运转的基础数据类to_dict输出特殊处理Document扁平化字典blob→ByteStream.to_dictsparse_embedding→SparseEmbedding.to_dictmeta 平铺但冲突键保留嵌套ByteStreamdata整数列表metamime_typebytes 转整数列表以兼容 JSONChatMessagerole/meta/name/content部件字典列表兼容三种历史格式部件按序列化键包装ExtractedAnswer/GeneratedAnswer各字段 嵌套Document兼容旧init_parameters包装ImageContentasdict全字段base64 校验与 MIME 猜测在__post_init__SparseEmbeddingindicesvalues长度一致性校验StreamingChunk全字段含嵌套部件to_dict载荷互斥校验、index必填校验这些数据类还被断点/管道快照机制haystack/dataclasses/breakpoints.py与管道序列化机制haystack/core/pipeline广泛复用管道在运行中断点时通过to_dict固化PipelineState恢复时通过from_dict还原数据类的序列化协议正是这一能力的前提。关于这些数据类在 Agent、检索、生成组件中的完整消费方式可继续阅读 docs-website/docs 目录下的概念与管道组件文档。结语Haystack 的数据类看似朴素实则是整个框架的“数据契约”Document让检索结果可统一表达ChatMessage让对话与工具调用可统一建模StreamingChunk让流式输出可结构化消费而贯穿其中的to_dict/from_dict协议保证了这些对象可以穿越管道边界、文件系统与网络。理解了这套数据骨架你就掌握了 Haystack 管道中数据流动的底层语言无论是编写自定义组件、调试管道还是构建复杂的 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），仅供参考