Haystack 2.x Data Classes API 详解贯穿 Pipeline 的数据载体设计【免费下载链接】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 的 dataclasses 是整套编排框架的数据骨架Document、ByteStream、ChatMessage、StreamingChunk等类在组件之间、Pipeline 输入输出之间承载文本、二进制、对话与流式数据。本文以官方参考文档 Data Classes2.20 版 API 参考为主线逐一拆解每个模块的字段、构造方法与序列化约定并结合当前仓库源码补充 ID 生成、内容分片content part、OpenAI 格式互转等实现细节帮助你在构建 RAG、Agent、多模态 Pipeline 时正确使用这些数据结构。模块总览与懒加载结构Data Classes 统一位于 haystack/dataclasses 包下。从 haystack/dataclasses/init.py 的_import_structure定义看2.20 参考文档覆盖以下模块answerExtractedAnswer、GeneratedAnswerbyte_streamByteStreamchat_messageChatMessage、ChatRole、TextContent、ToolCall、ToolCallResult、ReasoningContentdocumentDocumentimage_contentImageContentsparse_embeddingSparseEmbeddingstreaming_chunkStreamingChunk、ToolCallDelta、ComponentInfo、select_streaming_callback值得注意的是该文件使用LazyImporter实现子模块懒加载haystack/dataclasses/init.py 中sys.modules[__name__] LazyImporter(...)只有真正访问某个名称时才导入对应子模块从而控制包加载成本。文档开头描述的定位是 Core classes that carry data through the system——这些类不执行业务逻辑而是作为组件之间的标准交换格式。ByteStream二进制数据载体ByteStreamhaystack/dataclasses/byte_stream.py是表示二进制对象的基础数据类三个字段为data: bytes—— 存储的二进制数据meta: dict[str, Any]—— 附加元数据默认空 dict且被标记为不参与哈希mime_type: str | None—— 二进制数据的 MIME 类型默认None。构造方法参考文档列出的三个类方法在源码中一一对应dataclass(reprFalse) class ByteStream: data: bytes meta: dict[str, Any] field(default_factorydict, hashFalse) mime_type: str | None field(defaultNone)ByteStream.from_file_path(filepath, mime_typeNone, metaNone, guess_mime_typeFalse)byte_stream.py从文件读取内容构造。当mime_type为空且guess_mime_typeTrue时会通过haystack.utils.misc._guess_mime_type推断 MIME 类型。ByteStream.from_string(text, encodingutf-8, mime_typeNone, metaNone)byte_stream.py按指定编码将字符串编码为字节。ByteStream.to_string(encodingutf-8)反向解码为字符串不包含 metadata若字节无法按指定编码解码会抛出UnicodeDecodeError。写入文件与序列化to_file(destination_path)把data以二进制模式写入目标路径注意metadata 会丢失byte_stream.py。to_dict()返回{data, meta, mime_type}三个键的字典。实现上有个关键细节——JSON 不支持 bytes因此data被转换为整数列表byte_stream.pydef to_dict(self) - dict[str, Any]: # Note: The data is converted to a list of integers for serialization since JSON does not support bytes directly. return {data: list(self.data), meta: self.meta, mime_type: self.mime_type}from_dict(data)以bytes(data[data])还原二进制meta与mime_type缺失时取默认值byte_stream.py。__repr__把data截断到 100 字节再展示避免调试时打印超长内容。另外从源码结构看还存在一个内部方法_to_trace_dictbyte_stream.py把二进制替换为Binary data (N bytes)占位符用于 Tracing 场景避免把大负载发送到 tracing 后端。DocumentRAG 的核心数据单元Documenthaystack/dataclasses/document.py是包含可被查询的数据的基础类可承载文本片段、图片/音频等二进制数据可按score排序并可保存到字典与 JSON。字段定义document.py字段类型说明idstr唯一标识。未显式设置时基于 Document 字段值自动生成contentstr \| None文档文本如果包含文本blobByteStream \| None关联的二进制数据metadict[str, Any]自定义元数据必须可 JSON 序列化scorefloat \| None文档得分通常由 retriever 赋值的排序分数embeddinglist[float] \| None稠密向量表示sparse_embeddingSparseEmbedding \| None稀疏向量表示ID 生成机制__post_init__document.py完成三件事校验content必须是字符串或None把 1.x 遗留的 NumPy 数组 embedding 转换为 float 列表若id为空则调用_create_id()生成。ID 生成逻辑在_create_id()中document.py把content、blob.data、mime_type、排序后的metajson.dumps(..., sort_keysTrue)空 meta 固定为{}以保持旧 ID 稳定、embedding、sparse_embedding拼接成一个字符串再取 SHA-256 十六进制摘要作为 ID。这意味着同一内容含元数据与向量会稳定地映射到同一 ID方便去重与幂等写入。序列化与 1.x 向后兼容to_dict(flatten: bool True)把blob与sparse_embedding递归转成可 JSON 序列化的类型。flattenTrue是默认值用于兼容 Haystack 1.x 的扁平化格式——meta里的键会被摊平到顶层字典但与 Document 字段名冲突的 meta 键会被保留在嵌套的meta字典中document.py。from_dict(data)反向操作把不属于字段名的顶层键归回meta并把blob、sparse_embedding还原为原类型document.py。向后兼容由元类_RemoveLegacyFields处理在Document.__init__调用前content_type、id_hash_keys、dataframe等 1.x 遗留字段会被静默丢弃document.py。content_type属性是纯兼容层content非空时返回text否则抛出ValueError(Content is not set.)document.py。__eq__的比较基于to_dict(flattenFalse)是否一致且要求类型相同——即字典表示完全相同才相等。AnswerExtractedAnswer 与 GeneratedAnsweranswer模块haystack/dataclasses/answer.py提供两种答案结构均实现to_dict/from_dict序列化约定ExtractedAnswer用于抽取式 Reader 的输出字段包括query、score、data答案文本可为 None、document来源 Document可为 None、context上下文文本可为 None、document_offset/context_offsetSpan(start, end)偏移区间、meta。Span是ExtractedAnswer的内嵌 dataclassanswer.py。其from_dict还带一个向后兼容分支旧格式把字段包裹在init_parameters信封中遇到该键会自动解包answer.py。GeneratedAnswer用于生成式 Generator 的输出字段为data答案文本、query、documents引用文档列表、meta。to_dict有一个实用细节如果meta[all_messages]中存的是ChatMessage对象列表会自动逐项调用to_dict()转成字典再序列化answer.pyfrom_dict则对all_messages做反向还原并先拷贝meta以免污染调用方传入的字典。参考文档还声明了AnswerProtocoldata/query/meta 序列化方法使下游组件可以面向协议而非具体类型编程。ChatMessage对话消息与内容分片模型这是 dataclasses 中最核心的模块haystack/dataclasses/chat_message.py。官方文档明确要求请使用from_assistant、from_user、from_system、from_tool类方法创建 ChatMessage而不是直接调用构造器——从源码看ChatMessage.__new__被重写为直接抛异常提示内部_role/_content为私有字段__getattribute__则对已移除的content属性给出更可见的报错提示。ChatRole 与内容分片ChatRolechat_message.py是str枚举四个取值及语义角色值语义USERuser用户消息只包含文本SYSTEMsystem系统消息只包含文本ASSISTANTassistant助手消息可包含文本与 Tool call也可存储元数据TOOLtool工具消息包含工具调用结果ChatRole.from_str(string)把字符串转为枚举未知角色抛出带提示信息的ValueError。消息内容由content part列表组成。2.20 参考文档中的内容类型包括TextContent(text)文本内容ToolCall(tool_name, arguments, idNone, extraNone)模型准备的工具调用id供 OpenAI 风格的 API 关联extra存放 provider 特定信息值须可 JSON 序列化ToolCallResult(result, origin, error)工具调用结果origin指生产该结果的ToolCallReasoningContent(reasoning_text, extra)模型的推理内容通常随 assistant 消息出现ImageContent图片内容见下文。从当前仓库源码结构看该模型还在持续演进ToolCallResult.result的类型别名ToolCallResultContentT已扩展为str | Sequence[TextContent | ImageContent | FileContent]chat_message.py即工具结果可以携带多模态分片内容类型联合中也加入了FileContentchat_message.py。序列化侧_serialize_content_part/_deserialize_content_part按包装键区分分片类型text、tool_call、tool_call_result、image、reasoning、filechat_message.py。其中TextContent走扁平格式{text: ...}其余类型包一层对应键。反序列化的错误信息被刻意写得很长源码注释说明这是为 LLM 在 Agent 运行中构造非法消息时提供指引chat_message.py——这一点体现了数据类错误信息对 Agent 场景的针对性设计。ChatMessage 属性与构造方法参考文档列出的只读属性在源码中全部实现为基于_content的过滤视图chat_message.pyrole/meta/name直接返回内部字段texts/text所有文本分片 / 第一个文本tool_calls/tool_call所有工具调用 / 第一个工具调用tool_call_results/tool_call_result所有工具结果 / 第一个工具结果images/image所有图片 / 第一个图片reasonings/reasoning所有推理内容 / 第一个推理内容。四个类方法的关键签名ChatMessage.from_user(textNone, metaNone, nameNone, *, content_partsNone) ChatMessage.from_system(text, metaNone, nameNone) ChatMessage.from_assistant(textNone, metaNone, nameNone, tool_callsNone, *, reasoningNone) ChatMessage.from_tool(tool_result, origin, errorFalse, metaNone)from_user必须提供text或content_parts二者之一同时提供或都不提供均抛ValueErrorcontent_parts中的裸字符串会被自动包装为TextContentchat_message.py。from_assistant的reasoning参数既接受字符串也接受ReasoningContent对象字符串会被自动包装chat_message.py。from_tool内部构造ToolCallResult(resulttool_result, originorigin, errorerror)作为唯一内容分片chat_message.py。name参数在文档中注明仅 OpenAI 支持。is_from(role)支持传ChatRole或字符串字符串会先经ChatRole.from_str转换。to_dict / from_dict 的多版本兼容from_dictchat_message.py依次兼容三种历史格式当前格式content为分片字典列表2.9.0 之前content为纯字符串自动包装为单个TextContent2.9.0 至 2.12.0 之间_content键承载分片列表。缺少role或content/_content时分别抛出信息详细的ValueError/TypeError。OpenAI 字典格式互转to_openai_dict_format(require_tool_call_idsTrue)与from_openai_dict_format(message)负责与 OpenAI Chat Completions API 的字典格式互转chat_message.py。实现层面的关键约束require_tool_call_idsTrue默认强制每个ToolCall带非空id对接浅层 OpenAI 兼容 API 时可设为False放行无id的工具调用ChatMessage的_meta会被丢弃因为 OpenAI API 不支持含ToolCallResult的消息只能包含该结果一个内容分片非 assistant 角色必须至少有一个内容分片assistant 可为空此时发送空content用户消息中单一纯文本输出content为字符串多分片时输出{type: text/image_url/file, ...}列表图片以data:mime;base64,dataURI 内联from_openai_dict_format对 assistant 的零参工具调用做了容错arguments为空串、null或缺失时一律按{}处理chat_message.py对system与developer角色都映射为系统消息tool_call_id缺失时也会接受文档注明若后续要用于 OpenAI必须自行补齐否则会校验失败。ImageContent多模态消息中的图片ImageContenthaystack/dataclasses/image_content.py承载聊天消息中的图片字段与参数base64_image: str—— base64 编码的图片字符串mime_type: str | None—— 建议显式提供多数 LLM provider 需要未提供时会从 base64 解码结果猜测可能较慢且不完全可靠detail: auto | high | low | None—— 图片细节级别仅 OpenAI 支持meta: dict[str, Any]—— 可选元数据validation: bool True—— 默认开启初始化校验检查 base64 是否合法、未提供 mime_type 时用filetype猜测、校验 MIME 是否为合法图片类型设为False可跳过校验以提速。校验逻辑在__post_init__中image_content.py非法 base64 抛ValueError(The base64 string is not valid)非图片 MIME 抛ValueError猜不到 MIME 时仅告警。合法图片 MIME 集合由FORMAT_TO_MIME映射覆盖 PNG、JPEG、GIF、WebP、BMP、TIFF、ICO 等image_content.py派生。方法与类方法show()借助 PIL 显示图片Jupyter 内走IPython.display终端内走image.show()PIL 为懒加载依赖未安装时提示pip install pillowimage_content.pyfrom_file_path(file_path, *, sizeNone, detailNone, metaNone)从图片文件构造内部复用ImageFileToImageContent组件image_content.py。size(width, height)会在保持宽高比的前提下缩放图片降低文件体积、内存占用与处理时间文档明确PDF 不受支持PDF 转图片应使用PDFToImageContent组件from_url(url, *, retry_attempts2, timeout10, sizeNone, detailNone, metaNone)下载 URL 并转为 base64。内部先用LinkContentFetcher抓取image_content.pyMIME 不在图片集合内时抛ValueError指向 PDF 时给出使用PDFToImageContent的明确指引to_dict/from_dict标准序列化另有内部_to_trace_dict把 base64 替换为Base64 string (N characters)占位符image_content.py避免 tracing 负载过大。SparseEmbedding稀疏向量表示SparseEmbeddinghaystack/dataclasses/sparse_embedding.py以压缩稀疏格式表达向量只有两个字段indices: list[int]—— 非零元素的索引values: list[float]—— 非零元素的值。__post_init__校验两者长度一致不一致抛ValueError(Length of indices and values must be the same.)。to_dict/from_dict直接基于asdict/ 构造器实现。它是Document.sparse_embedding的承载类型服务于稀疏检索场景。StreamingChunk流式输出的标准单元streaming_chunk模块haystack/dataclasses/streaming_chunk.py定义流式生成场景的数据结构。StreamingChunk 字段字段类型 / 默认值说明contentstr该 chunk 的内容字符串必填metadict默认{}与消息块相关的元数据component_infoComponentInfo \| None生成该 chunk 的组件信息indexint \| None该内容块所属的内容块索引tool_callslist[ToolCallDelta] \| None关联的工具调用增量tool_call_resultToolCallResult \| None工具调用结果startbool默认False该 chunk 是否标记内容块开始finish_reasonFinishReason \| None生成结束原因reasoningReasoningContent \| None关联的推理内容finish_reason的标准取值遵循 OpenAI 约定stop、length、tool_calls、content_filter外加 Haystack 特有的tool_call_resultsstreaming_chunk.py 中的FinishReason类型别名。__post_init__有两条互斥校验streaming_chunk.pycontent、tool_calls、tool_call_result、reasoning中最多只能有一项被设置若设置了工具调用、工具结果或推理内容则index必填。to_dict/from_dict完成嵌套类型的递归序列化from_dict缺失content时抛ValueError。ToolCallDelta、ComponentInfo 与回调选择ToolCallDelta(index, tool_nameNone, argumentsNone, idNone, extraNone)流式工具调用的增量表示arguments可以是完整 JSON 也可以是增量片段streaming_chunk.py。ComponentInfo(type, nameNone)封装组件的类型完整模块路径.类名与在 Pipeline 中的名字ComponentInfo.from_component(component)从组件实例提取读取__component_name__属性streaming_chunk.py。回调类型别名为SyncStreamingCallbackT Callable[[StreamingChunk], None]与AsyncStreamingCallbackT Callable[[StreamingChunk], Awaitable[None]]streaming_chunk.py。select_streaming_callback(init_callback, runtime_callback, requires_async)runtime 回调优先于 init 回调。从源码看还有两条规则streaming_chunk.pyasync 上下文中传入同步回调只告警会在事件循环上内联执行、可能阻塞循环同步上下文中传入协程回调则直接抛ValueError因为无从 await。贯穿全局的序列化约定综合以上各模块可以提炼出 Data Classes API 的统一模式to_dict/from_dict成对出现是所有 dataclass 的标配Document、ByteStream、ChatMessage、ToolCall、ToolCallResult、StreamingChunk、SparseEmbedding等保证数据可以跨 Pipeline 持久化、跨语言传输二进制与大数据做降载处理ByteStream.data序列化为整数列表tracing 场景下用_to_trace_dict把二进制/base64 替换为占位符分别见 byte_stream.py 与 image_content.py__repr__截断展示ByteStream与ImageContent都把二进制/base64 截断到 100 字节调试友好版本兼容是显式设计目标Document的 1.x 遗留字段剥离与 numpy embedding 转换、ChatMessage.from_dict的三种历史格式分支、Answer的init_parameters解包都在源码中留有注释说明可变性护栏这些类普遍带_warn_on_inplace_mutation装饰器来自 haystack/utils/dataclasses.py对原地修改给出警告降低 Pipeline 运行中的隐蔽状态污染。适用前提与参考索引本文 API 描述以 2.20 版参考文档为准源码剖析以当前仓库 haystack/dataclasses 的实际实现为准两者存在少量差异如当前源码中FileContent、多模态ToolCallResult.result等演进使用时以所安装版本的实际签名为准。相关源码入口haystack/dataclasses/document.py、haystack/dataclasses/byte_stream.py、haystack/dataclasses/chat_message.py、haystack/dataclasses/image_content.py、haystack/dataclasses/sparse_embedding.py、haystack/dataclasses/streaming_chunk.py、haystack/dataclasses/answer.py。文档中的构造示例均为标准 Python 调用可直接在from haystack.dataclasses import Document, ByteStream, ChatMessage之后复现验证。【免费下载链接】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），仅供参考
