企业官网建设流程全解析

Haystack 集成 IBM watsonx.ai：文本/文档嵌入与 Chat、Text 生成的完整实战指南

【免费下载链接】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 官方集成参考文档（docs-website/reference/integrations-api/watsonx.md）系统讲解如何在 Haystack 中使用 IBM watsonx.ai：通过WatsonxTextEmbedder/WatsonxDocumentEmbedder将查询与文档向量化，通过WatsonxChatGenerator（含多模态与工具调用）和WatsonxGenerator完成生成任务。读完本文，你将掌握 watsonx 集成四个核心组件的全部参数、认证方式、序列化方法，并能搭建一条端到端的 RAG 流水线。

一、集成概览：安装与认证

IBM watsonx.ai 集成是 Haystack 的第三方组件包，以watsonx-haystack为包名独立分发（见 组件文档 中的 Package name 字段）。安装命令：

pip install watsonx-haystack

所有组件都依赖两组 IBM Cloud 凭证，官方推荐通过环境变量注入：

• WATSONX_API_KEY：IBM Cloud API 密钥
• WATSONX_PROJECT_ID：Watson Studio 项目 ID

在组件初始化时，也可以改用 Haystack 的SecretAPI 直接传值。Secret定义于核心库 haystack/utils/auth.py，提供两种构造方式：

from haystack.utils import Secret # 方式一：从环境变量读取（推荐，支持传入多个候选变量，按顺序取第一个已设置的） api_key = Secret.from_env_var("WATSONX_API_KEY") project_id = Secret.from_env_var("WATSONX_PROJECT_ID") # 方式二：直接传入明文 token（不可序列化，适合本地快速调试） api_key = Secret.from_token("<your-api-key>") project_id = Secret.from_token("<your-project-id>")

从源码实现看，Secret.from_env_var接受单个变量名或有序列表，解析时返回第一个已设置的环境变量的值；若全部未设置且strict=True（默认），会抛出异常，从而在流水线运行前就暴露缺失的凭证配置。

二、WatsonxTextEmbedder：查询向量化

WatsonxTextEmbedder用于把单条字符串（典型场景是用户查询）编码为向量，供 embedding Retriever 与文档向量做相似度检索。它位于查询 / RAG 流水线中 Retriever 之前（参考 WatsonxTextEmbedder 组件文档）。

2.1 基础用法

from haystack_integrations.components.embedders.watsonx.text_embedder import WatsonxTextEmbedder from haystack.utils import Secret text_to_embed = "I love pizza!" text_embedder = WatsonxTextEmbedder( model="ibm/slate-30m-english-rtrvr-v2", 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-v2', # 'truncated_input_tokens': 3}}

run(text: str)接收单个字符串，返回字典包含两个键：

• embedding：输入文本的嵌入向量（list[float]）
• meta：模型使用信息，如模型名、被截断的输入 token 数（truncated_input_tokens）

2.2 初始化参数说明

WatsonxTextEmbedder.__init__的完整签名与参数语义：

__init__( *, model: str = "ibm/slate-30m-english-rtrvr-v2", 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 ) -> None

WatsonxTextEmbedder同时提供to_dict()（序列化为字典）与from_dict(data)（从字典反序列化），二者是 Haystack 组件支持 YAML 流水线声明与断点调试的基础能力。

三、WatsonxDocumentEmbedder：文档批量向量化

WatsonxDocumentEmbedder计算整批文档的嵌入，并把向量写回每个Document，用于索引流水线在写入 DocumentStore 之前完成向量化（参考 WatsonxDocumentEmbedder 组件文档）。

3.1 基础用法

from haystack import Document from haystack_integrations.components.embedders.watsonx.document_embedder import WatsonxDocumentEmbedder from haystack.utils import Secret documents = [ Document(content="I love pizza!"), Document(content="Pasta is great too"), ] document_embedder = WatsonxDocumentEmbedder( model="ibm/slate-30m-english-rtrvr-v2", 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, ...]

run(documents: list[Document])返回：

• documents：已附加embedding的文档列表
• meta：模型使用信息

3.2 初始化参数说明

__init__( *, model: str = "ibm/slate-30m-english-rtrvr-v2", 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" ) -> None

相比WatsonxTextEmbedder，文档嵌入器额外暴露了四个面向批量的参数：

truncate_input_tokens、prefix、suffix、timeout、max_retries的语义与WatsonxTextEmbedder相同。prefix/suffix/embedding_separator与元数据嵌入的拼接逻辑，可从 Haystack 同类组件（如 haystack/components/embedders/openai_document_embedder.py）的_prepare_texts_to_embed实现中推断：最终送入模型的文本形如prefix + 分隔符.join(元数据值 + [正文]) + suffix。

3.3 元数据嵌入（Embedding Metadata）

文档通常携带元数据，若其中包含语义上有区分度的字段（标题、章节、标签等），把它们一并编码可以显著改善检索质量。使用meta_fields_to_embed即可开启：

from haystack import Document from haystack_integrations.components.embedders.watsonx.document_embedder import WatsonxDocumentEmbedder from haystack.utils import Secret doc = Document(content="some text", meta={"title": "relevant title", "page number": 18}) embedder = WatsonxDocumentEmbedder( api_key=Secret.from_env_var("WATSONX_API_KEY"), project_id=Secret.from_env_var("WATSONX_PROJECT_ID"), meta_fields_to_embed=["title"], ) docs_w_embeddings = embedder.run(documents=[doc])["documents"]

指定多个字段时，它们会按embedding_separator（默认换行符）与正文拼接后统一编码。

3.4 在索引流水线中使用

from haystack import Document, Pipeline from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack.components.writers import DocumentWriter from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever from haystack_integrations.components.embedders.watsonx.document_embedder import WatsonxDocumentEmbedder from haystack_integrations.components.embedders.watsonx.text_embedder import WatsonxTextEmbedder document_store = InMemoryDocumentStore(embedding_similarity_function="cosine") documents = [ Document(content="My name is Wolfgang and I live in Berlin"), Document(content="I saw a black horse running"), Document(content="Germany has many big cities"), ] indexing_pipeline = Pipeline() indexing_pipeline.add_component("embedder", WatsonxDocumentEmbedder()) indexing_pipeline.add_component("writer", DocumentWriter(document_store=document_store)) indexing_pipeline.connect("embedder", "writer") indexing_pipeline.run({"embedder": {"documents": documents}}) query_pipeline = Pipeline() query_pipeline.add_component("text_embedder", WatsonxTextEmbedder()) query_pipeline.add_component("retriever", InMemoryEmbeddingRetriever(document_store=document_store)) query_pipeline.connect("text_embedder.embedding", "retriever.query_embedding") result = query_pipeline.run({"text_embedder": {"text": "Who lives in Berlin?"}}) print(result["retriever"]["documents"][0]) # Document(id=..., content: 'My name is Wolfgang and I live in Berlin', score: ...)

该示例完整展示了 watsonx 嵌入器与 Haystack 核心组件（InMemoryDocumentStore、DocumentWriter、InMemoryEmbeddingRetriever）的组合方式：索引阶段文档经WatsonxDocumentEmbedder编码后写入存储，查询阶段查询文本经WatsonxTextEmbedder编码后交由 Retriever 计算余弦相似度。

四、WatsonxChatGenerator：多模态 Chat 补全

WatsonxChatGenerator基于 watsonx.ai 基础模型提供 Chat 补全能力，输入输出均采用 Haystack 的ChatMessage格式（定义于 haystack/dataclasses/chat_message.py，提供from_user、from_system、from_assistant、from_tool等构造器），并支持同时包含文本与图片的多模态输入。它通常放在ChatPromptBuilder之后（参考 WatsonxChatGenerator 组件文档）。

4.1 基础用法

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-4-h-small", project_id=Secret.from_env_var("WATSONX_PROJECT_ID"), ) response = client.run(messages) print(response)

4.2 多模态用法

ChatMessage支持通过content_parts携带ImageContent（图片可由文件路径或 base64 构造）：

from haystack.dataclasses import ChatMessage, ImageContent # 从文件路径或 base64 创建图片内容 image_content = ImageContent.from_file_path("path/to/your/image.jpg") # 构造同时包含文本与图片的多模态消息 messages = [ChatMessage.from_user(content_parts=["What's in this image?", image_content])] # 使用多模态模型 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)

4.3 SUPPORTED_MODELS 与初始化参数

组件内置一个非穷举的受支持模型列表（SUPPORTED_MODELS），涵盖 IBM Granite、Meta Llama、Mistral 与 OpenAI 开源模型等多个系列：

SUPPORTED_MODELS: list[str] = [ "ibm/granite-3-1-8b-base", "ibm/granite-3-8b-instruct", "ibm/granite-4-h-small", "ibm/granite-8b-code-instruct", "ibm/granite-guardian-3-8b", "meta-llama/llama-3-1-70b-gptq", "meta-llama/llama-3-1-8b", "meta-llama/llama-3-2-11b-vision-instruct", "meta-llama/llama-3-2-90b-vision-instruct", "meta-llama/llama-3-3-70b-instruct", "meta-llama/llama-3-405b-instruct", "meta-llama/llama-4-maverick-17b-128e-instruct-fp8", "meta-llama/llama-guard-3-11b-vision", "mistral-large-2512", "mistralai/mistral-medium-2505", "mistralai/mistral-small-3-1-24b-instruct-2503", "openai/gpt-oss-120b", ]

初始化签名：

__init__( *, api_key: Secret = Secret.from_env_var("WATSONX_API_KEY"), model: str = "ibm/granite-4-h-small", 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, tools: ToolsType | None = None ) -> None

除与嵌入器一致的api_key/model/project_id/api_base_url/timeout/max_retries外，还需注意：

初始化前还可以设置两个环境变量覆盖默认网络行为：

• WATSONX_TIMEOUT：覆盖默认超时（未设置时默认为 30 秒）
• WATSONX_MAX_RETRIES：覆盖默认重试次数（未设置时默认为 5 次）

generation_kwargs支持的参数直接映射 watsonx.ai 推理端点，常用项包括：

• temperature：控制随机性（越低越确定）
• max_new_tokens/min_new_tokens：生成 token 数上下限
• top_p：核采样概率阈值
• top_k：候选 token 数
• repetition_penalty：重复 token 惩罚
• length_penalty：输出长度惩罚
• stop_sequences：停止生成序列列表
• random_seed：随机种子，用于结果复现

4.4 run 与 run_async

run( *, messages: list[ChatMessage] | str, generation_kwargs: dict[str, Any] | None = None, streaming_callback: StreamingCallbackT | None = None, tools: ToolsType | None = None ) -> dict[str, list[ChatMessage]] run_async( *, messages: list[ChatMessage] | str, generation_kwargs: dict[str, Any] | None = None, streaming_callback: StreamingCallbackT | None = None, tools: ToolsType | None = None ) -> dict[str, list[ChatMessage]]

• messages：ChatMessage列表；若传入普通字符串，会被自动包装为一条user角色的ChatMessage。
• generation_kwargs：运行时传入可覆盖初始化时设置的同名参数。
• streaming_callback：提供时覆盖初始化设置的回调。
• tools：提供时覆盖初始化设置的tools。
• 返回值：{'replies': [ChatMessage, ...]}，即模型生成的回复消息列表。

run_async与run参数完全一致，适合在异步流水线中调用，便于在高并发场景下复用事件循环。

4.5 在流水线中使用 ChatPromptBuilder

from haystack import Pipeline from haystack.components.builders import ChatPromptBuilder from haystack.dataclasses import ChatMessage from haystack_integrations.components.generators.watsonx.chat.chat_generator import WatsonxChatGenerator from haystack.utils import Secret pipe = Pipeline() pipe.add_component("prompt_builder", ChatPromptBuilder()) pipe.add_component( "llm", WatsonxChatGenerator( api_key=Secret.from_env_var("WATSONX_API_KEY"), project_id=Secret.from_env_var("WATSONX_PROJECT_ID"), model="ibm/granite-4-h-small", ), ) pipe.connect("prompt_builder", "llm") country = "Germany" system_message = ChatMessage.from_system( "You are an assistant giving out valuable information to language learners.", ) messages = [ system_message, ChatMessage.from_user("What's the official language of {{ country }}?"), ] res = pipe.run( data={ "prompt_builder": { "template_variables": {"country": country}, "template": messages, }, }, ) print(res)

这里ChatMessage.from_system(...)与ChatMessage.from_user(...)均来自核心库 haystack/dataclasses/chat_message.py，分别构造系统角色与用户角色的消息，模板变量由ChatPromptBuilder在运行时填充。

五、WatsonxGenerator：基于字符串的 Text 补全（已弃用）

WatsonxGenerator继承自WatsonxChatGenerator（Bases: WatsonxChatGenerator），提供面向纯字符串 prompt 的标准 Generator 接口，适合简单文本生成任务。注意：组件文档中已标注弃用声明，建议迁移到同样接受纯字符串输入的WatsonxChatGenerator（参考 WatsonxGenerator 组件文档）。

5.1 基础用法

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-4-h-small", 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-4-h-small", "project_id": "your-project-id", "usage": { "prompt_tokens": 12, "completion_tokens": 45, "total_tokens": 57, }, } ], }

5.2 初始化与运行参数

__init__( *, api_key: Secret = Secret.from_env_var("WATSONX_API_KEY"), model: str = "ibm/granite-4-h-small", 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

与WatsonxChatGenerator相比，多出system_prompt参数，用于在初始化时设定系统提示；同时因接口为纯文本生成，不含tools参数。WATSONX_TIMEOUT（默认 30 秒）与WATSONX_MAX_RETRIES（默认 5 次）环境变量同样生效。

run( *, prompt: str, system_prompt: str | None = None, streaming_callback: StreamingCallbackT | None = None, generation_kwargs: dict[str, Any] | None = None ) -> dict[str, Any] 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：可选的系统提示；不传时使用__init__中设置的值。
• streaming_callback、generation_kwargs：覆盖规则与WatsonxChatGenerator相同。
• 返回值包含replies（生成文本字符串列表）与meta（每次生成对应的元数据，含模型名、结束原因、token 用量统计）。

在流水线中，WatsonxGenerator通常放在PromptBuilder之后：

from haystack import Pipeline from haystack.components.builders import PromptBuilder from haystack_integrations.components.generators.watsonx.generator import WatsonxGenerator from haystack.utils import Secret template = """ You are an assistant giving out valuable information to language learners. Answer this question, be brief. Question: {{ query }}? """ pipe = Pipeline() pipe.add_component("prompt_builder", PromptBuilder(template)) pipe.add_component( "llm", WatsonxGenerator( api_key=Secret.from_env_var("WATSONX_API_KEY"), project_id=Secret.from_env_var("WATSONX_PROJECT_ID"), ), ) pipe.connect("prompt_builder", "llm") query = "What language is spoken in Germany?" res = pipe.run(data={"prompt_builder": {"query": query}}) print(res)

六、序列化：to_dict 与 from_dict

watsonx 集成的四个组件都实现了to_dict() -> dict[str, Any]与from_dict(data: dict[str, Any])方法。这是 Haystack 组件序列化协议的一部分：to_dict把组件（包括Secret的配置信息）转换为可 JSON 化的字典，from_dict从字典重建组件实例。利用该协议，可以将完整流水线导出为 YAML 声明文件，或通过 pipeline 快照与断点调试（若存在对应章节）保存运行中间态。Secret本身的序列化/反序列化逻辑定义于 haystack/utils/auth.py：基于环境变量的密钥会被序列化为{"type": "env_var", "env_vars": [...]}形式，保证导出的流水线在目标环境中可用同一组环境变量解析凭证。

七、实战小结

围绕 IBM watsonx.ai，Haystack 集成提供了四条清晰的组件路径：

1. 索引向量化：WatsonxDocumentEmbedder批量编码文档，配合meta_fields_to_embed提升检索质量，通过batch_size与concurrency_limit控制吞吐；
2. 查询向量化：WatsonxTextEmbedder编码单条查询，接 embedding Retriever 完成相似度召回；
3. 对话生成：WatsonxChatGenerator基于ChatMessage完成同步/异步 Chat 补全，支持多模态图片输入、流式回调与tools函数调用；
4. 文本生成：WatsonxGenerator提供字符串接口的 Text 补全，已被官方标记为弃用，新项目建议迁移到WatsonxChatGenerator。

所有组件共享一致的认证模型（WATSONX_API_KEY/WATSONX_PROJECT_ID环境变量或Secret直传）、一致的网络配置（api_base_url、timeout、max_retries、WATSONX_TIMEOUT/WATSONX_MAX_RETRIES覆盖）以及一致的序列化协议（to_dict/from_dict），可以无缝嵌入 索引与查询流水线 中，与InMemoryDocumentStore、DocumentWriter、InMemoryEmbeddingRetriever、PromptBuilder/ChatPromptBuilder等 Haystack 核心组件自由组合。

【免费下载链接】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