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

IBM watsonx.ai

Haystack 的 IBM watsonx.ai 集成

模块 haystack_integrations.components.generators.watsonx.generator

WatsonxGenerator

使用 IBM 的 watsonx.ai 基础模型进行文本补全。

此组件扩展了 WatsonxChatGenerator，提供了标准的 Generator 接口，该接口使用 prompt 字符串而不是 ChatMessage 对象。

该生成器与 IBM 的基础模型一起使用，包括

  • granite-13b-chat-v2
  • llama-2-70b-chat
  • llama-3-70b-instruct
  • 其他 watsonx.ai 聊天模型

您可以通过 watsonx.ai API 的generation_kwargs 参数来定制生成行为。这些参数直接传递给 watsonx.ai 推理端点。

有关 watsonx.ai API 参数的更多详细信息，请参阅 IBM watsonx.ai 文档。

使用示例

from haystack_integrations.components.generators.watsonx.generator import WatsonxGenerator
from haystack.utils import Secret

generator = WatsonxGenerator(
    api_key=Secret.from_env_var("WATSONX_API_KEY"),
    model="ibm/granite-13b-chat-v2",
    project_id=Secret.from_env_var("WATSONX_PROJECT_ID"),
)

response = generator.run(
    prompt="Explain quantum computing in simple terms",
    system_prompt="You are a helpful physics teacher.",
)
print(response)

输出

{
    "replies": ["Quantum computing uses quantum-mechanical phenomena like...."],
    "meta": [
        {
            "model": "ibm/granite-13b-chat-v2",
            "project_id": "your-project-id",
            "usage": {
                "prompt_tokens": 12,
                "completion_tokens": 45,
                "total_tokens": 57,
            },
        }
    ],
}

WatsonxGenerator.__init__

def __init__(*,
             api_key: Secret = Secret.from_env_var("WATSONX_API_KEY"),
             model: str = "ibm/granite-3-2b-instruct",
             project_id: Secret = Secret.from_env_var("WATSONX_PROJECT_ID"),
             api_base_url: str = "https://us-south.ml.cloud.ibm.com",
             system_prompt: str | None = None,
             generation_kwargs: dict[str, Any] | None = None,
             timeout: float | None = None,
             max_retries: int | None = None,
             verify: bool | str | None = None,
             streaming_callback: StreamingCallbackT | None = None) -> None

创建 WatsonxGenerator 实例。

在初始化组件之前，您可以设置环境变量

  • WATSONX_TIMEOUT 来覆盖默认超时时间
  • WATSONX_MAX_RETRIES 来覆盖默认重试次数

参数:

  • api_key：用于 watsonx.ai 访问的 IBM Cloud API 密钥。可以通过WATSONX_API_KEY 环境变量设置，或直接传递。
  • model：用于补全的模型 ID。默认为 "ibm/granite-13b-chat-v2"。可用模型可在您的 IBM Cloud 账户中找到。
  • project_id：IBM Cloud 项目 ID
  • api_base_url：API 端点的自定义基 URL。默认为 "https://us-south.ml.cloud.ibm.com"。
  • system_prompt：用于文本生成的系统提示。
  • generation_kwargs：用于控制文本生成的附加参数。这些参数直接传递给 watsonx.ai 推理端点。支持的参数包括
  • temperature：控制随机性（越低越确定）
  • max_new_tokens：生成的最大 token 数
  • min_new_tokens：生成的最小 token 数
  • top_p：核心采样概率阈值
  • top_k：要考虑的最高概率 token 数
  • repetition_penalty：重复 token 的惩罚
  • length_penalty：基于输出长度的惩罚
  • stop_sequences：生成应停止的序列列表
  • random_seed：用于可复现结果的种子
  • timeout：API 请求的超时时间（秒）。默认为环境变量WATSONX_TIMEOUT 或 30 秒。
  • max_retries：失败请求的最大重试次数。默认为环境变量WATSONX_MAX_RETRIES 或 5。
  • verify：SSL 验证设置。可以是
  • True：验证 SSL 证书（默认）
  • False：跳过验证（不安全）
  • 用于自定义证书的 CA 捆绑包的路径
  • streaming_callback：用于流式响应的回调函数。

WatsonxGenerator.to_dict

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

将组件序列化为字典。

返回值:

序列化后的组件（字典格式）。

WatsonxGenerator.from_dict

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

从字典反序列化此组件。

参数:

  • data：此组件的字典表示。

返回值:

反序列化的组件实例。

WatsonxGenerator.run

@component.output_types(replies=list[str], meta=list[dict[str, Any]])
def run(*,
        prompt: str,
        system_prompt: str | None = None,
        streaming_callback: StreamingCallbackT | None = None,
        generation_kwargs: dict[str, Any] | None = None) -> dict[str, Any]

同步生成文本补全。

参数:

  • prompt：用于文本生成的输入 prompt 字符串。
  • system_prompt：可选的系统提示，用于提供上下文或生成指令。如果未提供，将使用__init__ 方法中设置的系统提示。
  • streaming_callback：当从流接收到新 token 时调用的回调函数。如果提供，此函数将覆盖__init__ 方法中设置的streaming_callback。
  • generation_kwargs：用于文本生成的附加关键字参数。这些参数可能会覆盖在streaming_callback。支持的参数包括 temperature、max_new_tokens、top_p 等。

返回值:

包含以下键的字典

  • replies：作为字符串生成的文本补全列表。
  • meta：包含每次生成信息的元数据字典列表，包括模型名称、结束原因和 token 使用统计信息。

WatsonxGenerator.run_async

@component.output_types(replies=list[str], meta=list[dict[str, Any]])
async def run_async(
        *,
        prompt: str,
        system_prompt: str | None = None,
        streaming_callback: StreamingCallbackT | None = None,
        generation_kwargs: dict[str, Any] | None = None) -> dict[str, Any]

异步生成文本补全。

参数:

  • prompt：用于文本生成的输入 prompt 字符串。
  • system_prompt：可选的系统提示，用于提供上下文或生成指令。
  • streaming_callback：当从流接收到新 token 时调用的回调函数。如果提供，此函数将覆盖__init__ 方法中设置的streaming_callback。
  • generation_kwargs：用于文本生成的附加关键字参数。这些参数可能会覆盖在streaming_callback。支持的参数包括 temperature、max_new_tokens、top_p 等。

返回值:

包含以下键的字典

  • replies：作为字符串生成的文本补全列表。
  • meta：包含每次生成信息的元数据字典列表，包括模型名称、结束原因和 token 使用统计信息。

模块 haystack_integrations.components.generators.watsonx.chat.chat_generator

WatsonxChatGenerator

使用 IBM 的 watsonx.ai 基础模型进行聊天补全。

此组件与 IBM 的 watsonx.ai 平台交互，使用各种基础模型生成聊天响应。它支持 ChatMessage 格式的输入和输出，包括带有文本和图像的多模态输入。

该生成器与 IBM 的基础模型一起使用，包括

  • granite-13b-chat-v2
  • llama-2-70b-chat
  • llama-3-70b-instruct
  • llama-3-2-11b-vision-instruct (多模态)
  • llama-3-2-90b-vision-instruct (多模态)
  • pixtral-12b (多模态)
  • 其他 watsonx.ai 聊天模型

您可以通过 watsonx.ai API 的generation_kwargs 参数来定制生成行为。这些参数直接传递给 watsonx.ai 推理端点。

有关 watsonx.ai API 参数的更多详细信息，请参阅 IBM watsonx.ai 文档。

使用示例

from haystack_integrations.components.generators.watsonx.chat.chat_generator import WatsonxChatGenerator
from haystack.dataclasses import ChatMessage
from haystack.utils import Secret

messages = [ChatMessage.from_user("Explain quantum computing in simple terms")]

client = WatsonxChatGenerator(
    api_key=Secret.from_env_var("WATSONX_API_KEY"),
    model="ibm/granite-13b-chat-v2",
    project_id=Secret.from_env_var("WATSONX_PROJECT_ID"),
)
response = client.run(messages)
print(response)

多模态使用示例

from haystack.dataclasses import ChatMessage, ImageContent

# Create an image from file path or base64
image_content = ImageContent.from_file_path("path/to/your/image.jpg")

# Create a multimodal message with both text and image
messages = [ChatMessage.from_user(content_parts=["What's in this image?", image_content])]

# Use a multimodal model
client = WatsonxChatGenerator(
    api_key=Secret.from_env_var("WATSONX_API_KEY"),
    model="meta-llama/llama-3-2-11b-vision-instruct",
    project_id=Secret.from_env_var("WATSONX_PROJECT_ID"),
)
response = client.run(messages)
print(response)

WatsonxChatGenerator.__init__

def __init__(*,
             api_key: Secret = Secret.from_env_var("WATSONX_API_KEY"),
             model: str = "ibm/granite-3-2b-instruct",
             project_id: Secret = Secret.from_env_var("WATSONX_PROJECT_ID"),
             api_base_url: str = "https://us-south.ml.cloud.ibm.com",
             generation_kwargs: dict[str, Any] | None = None,
             timeout: float | None = None,
             max_retries: int | None = None,
             verify: bool | str | None = None,
             streaming_callback: StreamingCallbackT | None = None) -> None

创建 WatsonxChatGenerator 实例。

在初始化组件之前，您可以设置环境变量

  • WATSONX_TIMEOUT 来覆盖默认超时时间
  • WATSONX_MAX_RETRIES 来覆盖默认重试次数

参数:

  • api_key：用于 watsonx.ai 访问的 IBM Cloud API 密钥。可以通过WATSONX_API_KEY 环境变量设置，或直接传递。
  • model：用于补全的模型 ID。默认为 "ibm/granite-13b-chat-v2"。可用模型可在您的 IBM Cloud 账户中找到。
  • project_id：IBM Cloud 项目 ID
  • api_base_url：API 端点的自定义基 URL。默认为 "https://us-south.ml.cloud.ibm.com"。
  • generation_kwargs：用于控制文本生成的附加参数。这些参数直接传递给 watsonx.ai 推理端点。支持的参数包括
  • temperature：控制随机性（越低越确定）
  • max_new_tokens：生成的最大 token 数
  • min_new_tokens：生成的最小 token 数
  • top_p：核心采样概率阈值
  • top_k：要考虑的最高概率 token 数
  • repetition_penalty：重复 token 的惩罚
  • length_penalty：基于输出长度的惩罚
  • stop_sequences：生成应停止的序列列表
  • random_seed：用于可复现结果的种子
  • timeout：API 请求的超时时间（秒）。默认为环境变量WATSONX_TIMEOUT 或 30 秒。
  • max_retries：失败请求的最大重试次数。默认为环境变量WATSONX_MAX_RETRIES 或 5。
  • verify：SSL 验证设置。可以是
  • True：验证 SSL 证书（默认）
  • False：跳过验证（不安全）
  • 用于自定义证书的 CA 捆绑包的路径
  • streaming_callback：用于流式响应的回调函数。

WatsonxChatGenerator.to_dict

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

将组件序列化为字典。

返回值:

序列化后的组件（字典格式）。

WatsonxChatGenerator.from_dict

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

从字典反序列化此组件。

参数:

  • data：此组件的字典表示。

返回值:

反序列化的组件实例。

WatsonxChatGenerator.run

@component.output_types(replies=list[ChatMessage])
def run(
    *,
    messages: list[ChatMessage],
    generation_kwargs: dict[str, Any] | None = None,
    streaming_callback: StreamingCallbackT | None = None
) -> dict[str, list[ChatMessage]]

同步生成聊天补全。

参数:

  • messages: 一个 ChatMessage 实例列表，表示输入消息。
  • generation_kwargs：用于文本生成的附加关键字参数。这些参数可能会覆盖在streaming_callback。
  • streaming_callback：当从流接收到新 token 时调用的回调函数。如果提供，此函数将覆盖__init__ 方法中设置的streaming_callback。

返回值:

一个字典，其中包含以下键

  • replies：包含生成响应作为 ChatMessage 实例的列表。

WatsonxChatGenerator.run_async

@component.output_types(replies=list[ChatMessage])
async def run_async(
    *,
    messages: list[ChatMessage],
    generation_kwargs: dict[str, Any] | None = None,
    streaming_callback: StreamingCallbackT | None = None
) -> dict[str, list[ChatMessage]]

异步生成聊天补全。

参数:

  • messages: 一个 ChatMessage 实例列表，表示输入消息。
  • generation_kwargs：用于文本生成的附加关键字参数。这些参数可能会覆盖在streaming_callback。
  • streaming_callback：当从流接收到新 token 时调用的回调函数。如果提供，此函数将覆盖__init__ 方法中设置的streaming_callback。

返回值:

一个字典，其中包含以下键

  • replies：包含生成响应作为 ChatMessage 实例的列表。

模块 haystack_integrations.components.embedders.watsonx.document_embedder

WatsonxDocumentEmbedder

使用 IBM watsonx.ai 模型计算文档嵌入。

使用示例

from haystack import Document
from haystack_integrations.components.embedders.watsonx.document_embedder import WatsonxDocumentEmbedder

documents = [
    Document(content="I love pizza!"),
    Document(content="Pasta is great too"),
]

document_embedder = WatsonxDocumentEmbedder(
    model="ibm/slate-30m-english-rtrvr",
    api_key=Secret.from_env_var("WATSONX_API_KEY"),
    api_base_url="https://us-south.ml.cloud.ibm.com",
    project_id=Secret.from_env_var("WATSONX_PROJECT_ID"),
)

result = document_embedder.run(documents=documents)
print(result["documents"][0].embedding)

# [0.017020374536514282, -0.023255806416273117, ...]

WatsonxDocumentEmbedder.__init__

def __init__(*,
             model: str = "ibm/slate-30m-english-rtrvr",
             api_key: Secret = Secret.from_env_var("WATSONX_API_KEY"),
             api_base_url: str = "https://us-south.ml.cloud.ibm.com",
             project_id: Secret = Secret.from_env_var("WATSONX_PROJECT_ID"),
             truncate_input_tokens: int | None = None,
             prefix: str = "",
             suffix: str = "",
             batch_size: int = 1000,
             concurrency_limit: int = 5,
             timeout: float | None = None,
             max_retries: int | None = None,
             meta_fields_to_embed: list[str] | None = None,
             embedding_separator: str = "\n")

创建 WatsonxDocumentEmbedder 组件。

参数:

  • model：用于计算嵌入的模型名称。默认为 "ibm/slate-30m-english-rtrvr"。
  • api_key：WATSONX API 密钥。可以通过环境变量 WATSONX_API_KEY 设置。
  • api_base_url：watsonx.ai 服务的 WATSONX URL。默认为 "https://us-south.ml.cloud.ibm.com"。
  • project_id：Watson Studio 项目的 ID。可以通过环境变量 WATSONX_PROJECT_ID 设置。
  • truncate_input_tokens：从输入文本使用的最大 token 数。如果设置为None（或未提供），则使用完整的输入文本，最多达到模型的最大 token 限制。
  • prefix: 添加到每个文本开头的字符串。
  • suffix: 添加到每个文本末尾的字符串。
  • batch_size：一次 API 调用中嵌入的文档数量。默认为 1000。
  • concurrency_limit：要发起的并行请求数量。默认为 5。
  • timeout：API 请求的超时时间（秒）。
  • max_retries：API 请求的最大重试次数。

WatsonxDocumentEmbedder.to_dict

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

将组件序列化为字典。

返回值:

序列化后的组件（字典格式）。

WatsonxDocumentEmbedder.from_dict

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

从字典反序列化组件。

参数:

  • data：此组件的字典表示。

返回值:

反序列化的组件实例。

WatsonxDocumentEmbedder.run

@component.output_types(documents=list[Document], meta=dict[str, Any])
def run(documents: list[Document]
        ) -> dict[str, list[Document] | dict[str, Any]]

嵌入一系列文档。

参数:

  • documents：要嵌入的文档列表。

返回值:

一个字典，包含

  • 'documents'：带有嵌入的 Document 列表
  • 'meta'：有关模型使用情况的信息

模块 haystack_integrations.components.embedders.watsonx.text_embedder

WatsonxTextEmbedder

使用 IBM watsonx.ai 基础模型嵌入字符串。

您可以使用它来嵌入用户查询并将查询发送到嵌入检索器。

使用示例

from haystack_integrations.components.embedders.watsonx.text_embedder import WatsonxTextEmbedder

text_to_embed = "I love pizza!"

text_embedder = WatsonxTextEmbedder(
    model="ibm/slate-30m-english-rtrvr",
    api_key=Secret.from_env_var("WATSONX_API_KEY"),
    api_base_url="https://us-south.ml.cloud.ibm.com",
    project_id=Secret.from_env_var("WATSONX_PROJECT_ID"),
)

print(text_embedder.run(text_to_embed))

# {'embedding': [0.017020374536514282, -0.023255806416273117, ...],
#  'meta': {'model': 'ibm/slate-30m-english-rtrvr',
#           'truncated_input_tokens': 3}}

WatsonxTextEmbedder.__init__

def __init__(*,
             model: str = "ibm/slate-30m-english-rtrvr",
             api_key: Secret = Secret.from_env_var("WATSONX_API_KEY"),
             api_base_url: str = "https://us-south.ml.cloud.ibm.com",
             project_id: Secret = Secret.from_env_var("WATSONX_PROJECT_ID"),
             truncate_input_tokens: int | None = None,
             prefix: str = "",
             suffix: str = "",
             timeout: float | None = None,
             max_retries: int | None = None)

创建 WatsonxTextEmbedder 组件。

参数:

  • model：用于计算嵌入的 IBM watsonx 模型名称。默认为 "ibm/slate-30m-english-rtrvr"。
  • api_key：WATSONX API 密钥。可以通过环境变量 WATSONX_API_KEY 设置。
  • api_base_url：watsonx.ai 服务的 WATSONX URL。默认为 "https://us-south.ml.cloud.ibm.com"。
  • project_id：Watson Studio 项目的 ID。可以通过环境变量 WATSONX_PROJECT_ID 设置。
  • truncate_input_tokens：从输入文本使用的最大 token 数。如果设置为None（或未提供），则使用完整的输入文本，最多达到模型的最大 token 限制。
  • prefix：添加到每个待嵌入文本开头的字符串。
  • suffix：添加到每个待嵌入文本末尾的字符串。
  • timeout：API 请求的超时时间（秒）。
  • max_retries：API 请求的最大重试次数。

WatsonxTextEmbedder.to_dict

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

将组件序列化为字典。

返回值:

序列化后的组件（字典格式）。

WatsonxTextEmbedder.from_dict

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

从字典反序列化组件。

参数:

  • data：此组件的字典表示。

返回值:

反序列化的组件实例。

WatsonxTextEmbedder.run

@component.output_types(embedding=list[float], meta=dict[str, Any])
def run(text: str) -> dict[str, list[float] | dict[str, Any]]

嵌入单个字符串。

参数:

  • text：要嵌入的文本。

返回值:

一个字典，包含

  • 'embedding'：输入文本的嵌入
  • 'meta'：有关模型使用情况的信息