文档API 参考📓 教程🧑‍🍳 食谱🤝 集成💜 Discord🎨 Studio
API 参考

Data Classes (数据类)

在系统中传递数据的核心类。

模块 answer

ExtractedAnswer

ExtractedAnswer.to_dict

def to_dict() -> dict[str, Any]

将对象序列化为字典。

返回值:

对象的序列化字典表示。

ExtractedAnswer.from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "ExtractedAnswer"

从字典反序列化对象。

参数:

  • data: 对象的字典表示。

返回值:

反序列化的对象。

GeneratedAnswer

GeneratedAnswer.to_dict

def to_dict() -> dict[str, Any]

将对象序列化为字典。

返回值:

对象的序列化字典表示。

GeneratedAnswer.from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "GeneratedAnswer"

从字典反序列化对象。

参数:

  • data: 对象的字典表示。

返回值:

反序列化的对象。

模块 byte_stream

ByteStream

表示 Haystack API 中二进制对象的基类。

参数:

  • data: 存储在 ByteStream 中的二进制数据。
  • meta: 要与 ByteStream 一起存储的附加元数据。
  • mime_type: 二进制数据的 MIME 类型。

ByteStream.to_file

def to_file(destination_path: Path) -> None

将 ByteStream 写入文件。注意：元数据将丢失。

参数:

  • destination_path: 要将 ByteStream 写入的路径。

ByteStream.from_file_path

@classmethod
def from_file_path(cls,
                   filepath: Path,
                   mime_type: Optional[str] = None,
                   meta: Optional[dict[str, Any]] = None,
                   guess_mime_type: bool = False) -> "ByteStream"

从文件中读取的内容创建 ByteStream。

参数:

  • filepath: 文件的有效路径。
  • mime_type: 文件的 MIME 类型。
  • meta: 要与 ByteStream 一起存储的附加元数据。
  • guess_mime_type: 是否根据文件猜测 MIME 类型。

ByteStream.from_string

@classmethod
def from_string(cls,
                text: str,
                encoding: str = "utf-8",
                mime_type: Optional[str] = None,
                meta: Optional[dict[str, Any]] = None) -> "ByteStream"

创建编码字符串的 ByteStream。

参数:

  • text: 要编码的字符串
  • encoding: 用于将字符串转换为字节的编码
  • mime_type: 文件的 MIME 类型。
  • meta: 要与 ByteStream 一起存储的附加元数据。

ByteStream.to_string

def to_string(encoding: str = "utf-8") -> str

将 ByteStream 转换为字符串，不包含元数据。

参数:

  • encoding: 用于将字节转换为字符串的编码。默认为“utf-8”。

引发:

  • None: UnicodeDecodeError：如果 ByteStream 数据无法使用指定的编码进行解码。

返回值:

ByteStream 的字符串表示。

ByteStream.__repr__

def __repr__() -> str

返回 ByteStream 的字符串表示，将数据截断为 100 字节。

ByteStream.to_dict

def to_dict() -> dict[str, Any]

将 ByteStream 转换为字典表示。

返回值:

包含“data”、“meta”和“mime_type”键的字典。

ByteStream.from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "ByteStream"

从字典表示创建 ByteStream。

参数:

  • data: 包含“data”、“meta”和“mime_type”键的字典。

返回值:

一个 ByteStream 实例。

模块 chat_message

ChatRole

表示聊天中角色的枚举。

USER

用户角色。用户的消息只包含文本。

SYSTEM

系统角色。系统的消息只包含文本。

ASSISTANT

助手角色。助手消息可以包含文本和工具调用。它还可以存储元数据。

TOOL

工具角色。来自工具的消息包含工具调用的结果。

ChatRole.from_str

@staticmethod
def from_str(string: str) -> "ChatRole"

将字符串转换为 ChatRole 枚举。

ToolCall

表示模型准备的工具调用，通常包含在助手消息中。

参数:

  • id: 工具调用的 ID。
  • tool_name: 要调用的工具的名称。
  • arguments: 调用工具的参数。

id

noqa: A003

ToolCall.to_dict

def to_dict() -> dict[str, Any]

将 ToolCall 转换为字典。

返回值:

包含“tool_name”、“arguments”和“id”键的字典。

ToolCall.from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "ToolCall"

从字典创建新的 ToolCall 对象。

参数:

  • data: 用于构建 ToolCall 对象的字典。

返回值:

创建的对象。

ToolCallResult

表示工具调用的结果。

参数:

  • result: 工具调用的结果。
  • origin: 产生此结果的工具调用。
  • error: 工具调用是否导致了错误。

ToolCallResult.to_dict

def to_dict() -> dict[str, Any]

将 ToolCallResult 转换为字典。

返回值:

包含“result”、“origin”和“error”键的字典。

ToolCallResult.from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "ToolCallResult"

从字典创建 ToolCallResult。

参数:

  • data: 用于构建 ToolCallResult 对象的字典。

返回值:

创建的对象。

TextContent

聊天消息的文本内容。

参数:

  • text: 消息的文本内容。

TextContent.to_dict

def to_dict() -> dict[str, Any]

将 TextContent 转换为字典。

TextContent.from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "TextContent"

从字典创建 TextContent。

ReasoningContent

表示模型准备的可选推理内容，通常包含在助手消息中。

参数:

  • reasoning_text: 模型生成的推理文本。
  • extra: 关于推理内容的附加信息的字典。用于存储提供商特定的信息。为避免序列化问题，值应为 JSON 可序列化。

ReasoningContent.to_dict

def to_dict() -> dict[str, Any]

将 ReasoningContent 转换为字典。

返回值:

包含“reasoning_text”和“extra”键的字典。

ReasoningContent.from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "ReasoningContent"

创建新的 ReasoningContent 对象。

参数:

  • data: 用于构建 ReasoningContent 对象的字典。

返回值:

创建的对象。

ChatMessage

表示 LLM 聊天对话中的消息。

使用from_assistant, from_user, from_system 和from_tool 类方法用于创建 ChatMessage。

ChatMessage.__new__

def __new__(cls, *args, **kwargs)

此方法被重写，以使对ChatMessage 数据类的更改更加明显。

ChatMessage.__getattribute__

def __getattribute__(name)

此方法被重写，以使content 属性的删除更加明显。

ChatMessage.role

@property
def role() -> ChatRole

返回发送消息的实体的角色。

ChatMessage.meta

@property
def meta() -> dict[str, Any]

返回与消息关联的元数据。

ChatMessage.name

@property
def name() -> Optional[str]

返回与消息关联的名称。

ChatMessage.texts

@property
def texts() -> list[str]

返回消息中包含的所有文本的列表。

ChatMessage.text

@property
def text() -> Optional[str]

返回消息中包含的第一个文本。

ChatMessage.tool_calls

@property
def tool_calls() -> list[ToolCall]

返回消息中包含的所有工具调用的列表。

ChatMessage.tool_call

@property
def tool_call() -> Optional[ToolCall]

返回消息中包含的第一个工具调用。

ChatMessage.tool_call_results

@property
def tool_call_results() -> list[ToolCallResult]

返回消息中包含的所有工具调用结果的列表。

ChatMessage.tool_call_result

@property
def tool_call_result() -> Optional[ToolCallResult]

返回消息中包含的第一个工具调用结果。

ChatMessage.images

@property
def images() -> list[ImageContent]

返回消息中包含的所有图像的列表。

ChatMessage.image

@property
def image() -> Optional[ImageContent]

返回消息中包含的第一个图像。

ChatMessage.reasonings

@property
def reasonings() -> list[ReasoningContent]

返回消息中包含的所有推理内容的列表。

ChatMessage.reasoning

@property
def reasoning() -> Optional[ReasoningContent]

返回消息中包含的第一个推理内容。

ChatMessage.is_from

def is_from(role: Union[ChatRole, str]) -> bool

检查消息是否来自特定角色。

参数:

  • role: 要检查的角色。

返回值:

如果消息来自指定角色，则为 True，否则为 False。

ChatMessage.from_user

@classmethod
def from_user(
    cls,
    text: Optional[str] = None,
    meta: Optional[dict[str, Any]] = None,
    name: Optional[str] = None,
    *,
    content_parts: Optional[Sequence[Union[TextContent, str,
                                           ImageContent]]] = None
) -> "ChatMessage"

创建一条来自用户的消息。

参数:

  • text: 消息的文本内容。指定此项或 content_parts。
  • meta: 与消息关联的附加元数据。
  • name: 可选的参与者名称。此字段仅由 OpenAI 支持。
  • content_parts: 要包含在消息中的内容部分列表。指定此项或 text。

返回值:

一个新的 ChatMessage 实例。

ChatMessage.from_system

@classmethod
def from_system(cls,
                text: str,
                meta: Optional[dict[str, Any]] = None,
                name: Optional[str] = None) -> "ChatMessage"

创建一条来自系统的消息。

参数:

  • text: 消息的文本内容。
  • meta: 与消息关联的附加元数据。
  • name: 可选的参与者名称。此字段仅由 OpenAI 支持。

返回值:

一个新的 ChatMessage 实例。

ChatMessage.from_assistant

@classmethod
def from_assistant(
        cls,
        text: Optional[str] = None,
        meta: Optional[dict[str, Any]] = None,
        name: Optional[str] = None,
        tool_calls: Optional[list[ToolCall]] = None,
        *,
        reasoning: Optional[Union[str,
                                  ReasoningContent]] = None) -> "ChatMessage"

创建一条来自助手的消息。

参数:

  • text: 消息的文本内容。
  • meta: 与消息关联的附加元数据。
  • name: 可选的参与者名称。此字段仅由 OpenAI 支持。
  • tool_calls: 要包含在消息中的工具调用。
  • reasoning: 要包含在消息中的推理内容。

返回值:

一个新的 ChatMessage 实例。

ChatMessage.from_tool

@classmethod
def from_tool(cls,
              tool_result: str,
              origin: ToolCall,
              error: bool = False,
              meta: Optional[dict[str, Any]] = None) -> "ChatMessage"

创建一条来自工具的消息。

参数:

  • tool_result: 工具调用的结果。
  • origin: 产生此结果的工具调用。
  • error: 工具调用是否导致了错误。
  • meta: 与消息关联的附加元数据。

返回值:

一个新的 ChatMessage 实例。

ChatMessage.to_dict

def to_dict() -> dict[str, Any]

将 ChatMessage 转换为字典。

返回值:

对象的序列化版本。

ChatMessage.from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "ChatMessage"

从字典创建新的 ChatMessage 对象。

参数:

  • data: 用于构建 ChatMessage 对象的字典。

返回值:

创建的对象。

ChatMessage.to_openai_dict_format

def to_openai_dict_format(
        require_tool_call_ids: bool = True) -> dict[str, Any]

将 ChatMessage 转换为 OpenAI Chat API 期望的字典格式。

参数:

  • require_tool_call_ids: 如果为 True（默认），则强制每个 Tool Call 包含一个非空的id 属性。设置为 False 以允许没有id 的 Tool Call，这可能适用于浅层 OpenAI 兼容 API。

引发:

  • ValueError: 如果消息格式无效，或者如果require_tool_call_ids 为 True 且任何 Tool Call 缺少id 属性。

返回值:

OpenAI Chat API 期望的 ChatMessage 格式。

ChatMessage.from_openai_dict_format

@classmethod
def from_openai_dict_format(cls, message: dict[str, Any]) -> "ChatMessage"

从 OpenAI Chat API 期望的格式的字典创建 ChatMessage。

注意：虽然 OpenAI 的 API 要求tool_call_id 在工具调用和工具消息中都存在，但此方法接受没有它的消息，以支持浅层 OpenAI 兼容 API。如果您打算将生成的 ChatMessage 与 OpenAI 一起使用，则必须包含tool_call_id，否则会遇到验证错误。

参数:

  • message: 用于构建 ChatMessage 对象的 OpenAI 字典。

引发:

  • ValueError: 如果消息字典缺少必需的字段。

返回值:

创建的 ChatMessage 对象。

模块 document

_BackwardCompatible

处理 Document 向后兼容性的元类。

_BackwardCompatible.__call__

def __call__(cls, *args, **kwargs)

在 Document.init 之前调用，处理旧版字段。

Embedding 在 1.x 中存储为 NumPy 数组，因此我们将其转换为浮点数列表。其他旧版字段已被删除。

Document

包含要查询的数据的基类。

可以包含文本片段和图像或音频的文件路径。文档可以按分数排序，并保存到/从字典和 JSON。

参数:

  • id: 文档的唯一标识符。未设置时，将根据 Document 字段的值生成。
  • content: 文档的文本，如果文档包含文本。
  • blob: 与文档关联的二进制数据，如果文档有任何关联的二进制数据。
  • meta: 文档的其他自定义元数据。必须是 JSON 可序列化的。
  • score: 文档的分数。用于排名，通常由检索器分配。
  • embedding: 文档的密集向量表示。
  • sparse_embedding: 文档的稀疏向量表示。

Document.__eq__

def __eq__(other)

比较 Document 是否相等。

如果两个 Document 的字典表示相同，则认为它们相等。

Document.__post_init__

def __post_init__()

根据 init 参数生成 ID。

Document.to_dict

def to_dict(flatten: bool = True) -> dict[str, Any]

将 Document 转换为字典。

blob 字段被转换为 JSON 可序列化类型。

参数:

  • flatten: 是否展平meta 字段。默认为True 以向后兼容 Haystack 1.x。

Document.from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "Document"

从字典创建新的 Document 对象。

该blob 字段被转换为其原始类型。

Document.content_type

@property
def content_type()

返回文档内容的类型。

这对于保持与 1.x 的向后兼容性至关重要。

模块 image_content

ImageContent

聊天消息的图像内容。

参数:

  • base64_image: 表示图像的 base64 字符串。
  • mime_type: 图像的 MIME 类型（例如，“image/png”、“image/jpeg”）。建议提供此值，因为大多数 LLM 提供商都需要它。如果未提供，则从 base64 字符串中猜测 MIME 类型，这可能会很慢且不总是可靠。
  • detail: 可选的图像详细程度（仅由 OpenAI 支持）。“auto”、“high”或“low”之一。
  • meta: 图像的可选元数据。
  • validation: 如果为 True（默认），则执行验证过程
  • 检查 base64 字符串是否有效；
  • 如果未提供 MIME 类型，则猜测 MIME 类型；
  • 检查 MIME 类型是否为有效的图像 MIME 类型。设置为 False 以跳过验证并加快初始化速度。

ImageContent.__repr__

def __repr__() -> str

返回 ImageContent 的字符串表示，将 base64_image 截断为 100 字节。

ImageContent.show

def show() -> None

显示图像。

ImageContent.to_dict

def to_dict() -> dict[str, Any]

将 ImageContent 转换为字典。

ImageContent.from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "ImageContent"

从字典创建 ImageContent。

ImageContent.from_file_path

@classmethod
def from_file_path(cls,
                   file_path: Union[str, Path],
                   *,
                   size: Optional[tuple[int, int]] = None,
                   detail: Optional[Literal["auto", "high", "low"]] = None,
                   meta: Optional[dict[str, Any]] = None) -> "ImageContent"

从文件路径创建 ImageContent 对象。

它公开了与ImageFileToImageContent 组件类似的功能。对于 PDF 到 ImageContent 的转换，请使用PDFToImageContent 组件。

参数:

  • file_path: 图像文件的路径。不支持 PDF 文件。对于 PDF 到 ImageContent 的转换，请使用PDFToImageContent 组件。
  • size: 如果提供，则将图像调整为适合指定的尺寸（宽度，高度），同时保持纵横比。这可以减小文件大小、内存使用量和处理时间，这在处理具有分辨率限制的模型或向远程服务传输图像时非常有用。
  • detail: 可选的图像详细程度（仅由 OpenAI 支持）。“auto”、“high”或“low”之一。
  • meta: 图像的附加元数据。

返回值:

一个 ImageContent 对象。

ImageContent.from_url

@classmethod
def from_url(cls,
             url: str,
             *,
             retry_attempts: int = 2,
             timeout: int = 10,
             size: Optional[tuple[int, int]] = None,
             detail: Optional[Literal["auto", "high", "low"]] = None,
             meta: Optional[dict[str, Any]] = None) -> "ImageContent"

从 URL 创建 ImageContent 对象。图像将被下载并转换为 base64 字符串。

对于 PDF 到 ImageContent 的转换，请使用PDFToImageContent 组件。

参数:

  • url: 图像的 URL。不支持 PDF 文件。对于 PDF 到 ImageContent 的转换，请使用PDFToImageContent 组件。
  • retry_attempts: 重试获取 URL 内容的次数。
  • timeout: 请求的超时时间（秒）。
  • size: 如果提供，则将图像调整为适合指定的尺寸（宽度，高度），同时保持纵横比。这可以减小文件大小、内存使用量和处理时间，这在处理具有分辨率限制的模型或向远程服务传输图像时非常有用。
  • detail: 可选的图像详细程度（仅由 OpenAI 支持）。“auto”、“high”或“low”之一。
  • meta: 图像的附加元数据。

引发:

  • ValueError: 如果 URL 不指向图像或指向 PDF 文件。

返回值:

一个 ImageContent 对象。

模块 sparse_embedding

SparseEmbedding

表示稀疏嵌入的类。

参数:

  • indices: 嵌入中非零元素的索引列表。
  • values: 嵌入中非零元素的值列表。

SparseEmbedding.__post_init__

def __post_init__()

检查索引和值列表是否长度相同。

如果不同，则引发 ValueError。

SparseEmbedding.to_dict

def to_dict() -> dict[str, Any]

将 SparseEmbedding 对象转换为字典。

返回值:

序列化的稀疏嵌入。

SparseEmbedding.from_dict

@classmethod
def from_dict(cls, sparse_embedding_dict: dict[str, Any]) -> "SparseEmbedding"

从字典反序列化稀疏嵌入。

参数:

  • sparse_embedding_dict: 要从中反序列化的字典。

返回值:

反序列化的稀疏嵌入。

模块 streaming_chunk

ToolCallDelta

表示模型准备的工具调用，通常包含在助手消息中。

参数:

  • index: 工具调用在工具调用列表中的索引。
  • tool_name: 要调用的工具的名称。
  • arguments: 参数的完整 JSON 格式或参数的增量。
  • id: 工具调用的 ID。

id

noqa: A003

ToolCallDelta.to_dict

def to_dict() -> dict[str, Any]

返回 ToolCallDelta 的字典表示。

返回值:

包含“index”、“tool_name”、“arguments”和“id”键的字典。

ToolCallDelta.from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "ToolCallDelta"

从序列化表示创建 ToolCallDelta。

参数:

  • data: 包含 ToolCallDelta 属性的字典。

返回值:

一个 ToolCallDelta 实例。

ComponentInfo

该ComponentInfo 类封装了组件的信息。

参数:

  • type: 组件的类型。
  • name: 将组件添加到管道时为其指定的名称。

ComponentInfo.from_component

@classmethod
def from_component(cls, component: Component) -> "ComponentInfo"

通过指定组件名称和触发断点的访问计数来创建从Component 实例创建 ComponentInfo 对象。

参数:

  • component: 包含给定组件类型和名称的Component 实例创建 ComponentInfo 对象。

返回值:

该ComponentInfo 对象。

ComponentInfo.to_dict

def to_dict() -> dict[str, Any]

返回 ComponentInfo 的字典表示。

返回值:

包含“type”和“name”键的字典。

ComponentInfo.from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "ComponentInfo"

从序列化表示创建 ComponentInfo。

参数:

  • data: 包含 ComponentInfo 属性的字典。

返回值:

一个 ComponentInfo 实例。

StreamingChunk

该StreamingChunk 类封装了流式内容的片段以及相关的元数据。

此结构促进了系统化地处理和处理流式数据。

参数:

  • content: 消息块的内容（字符串）。
  • meta: 包含与消息块相关的元数据的字典。
  • component_info: 一个ComponentInfo 对象，包含有关生成该块的组件的信息，例如组件名称和类型。
  • index: 一个可选的整数索引，表示此块属于哪个内容块。
  • tool_calls: 一个可选的 ToolCallDelta 对象列表，表示与消息块关联的工具调用。
  • tool_call_result: 一个可选的 ToolCallResult 对象，表示工具调用的结果。
  • start: 一个布尔值，指示此块是否标记为内容块的开始。
  • finish_reason: 一个可选值，指示生成完成的原因。标准值遵循 OpenAI 的约定：“stop”、“length”、“tool_calls”、“content_filter”，以及 Haystack 特有的值“tool_call_results”。
  • reasoning: 一个可选的 ReasoningContent 对象，表示与消息块关联的推理内容。

StreamingChunk.to_dict

def to_dict() -> dict[str, Any]

返回 StreamingChunk 的字典表示。

返回值:

调用对象的序列化字典表示。

StreamingChunk.from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "StreamingChunk"

从序列化表示创建反序列化的 StreamingChunk 实例。

参数:

  • data: 包含 StreamingChunk 属性的字典。

返回值:

一个 StreamingChunk 实例。

select_streaming_callback

def select_streaming_callback(
        init_callback: Optional[StreamingCallbackT],
        runtime_callback: Optional[StreamingCallbackT],
        requires_async: bool) -> Optional[StreamingCallbackT]

根据可选的初始回调和运行时回调选择正确的流式回调。

运行时回调优先于初始回调。

参数:

  • init_callback: 初始回调。
  • runtime_callback: 运行时回调。
  • requires_async: 所选回调是否必须兼容异步。

返回值:

选定的回调。