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_KEYIBM Cloud API 密钥WATSONX_PROJECT_IDWatson 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接受单个变量名或有序列表解析时返回第一个已设置的环境变量的值若全部未设置且strictTrue默认会抛出异常从而在流水线运行前就暴露缺失的凭证配置。二、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( modelibm/slate-30m-english-rtrvr-v2, api_keySecret.from_env_var(WATSONX_API_KEY), api_base_urlhttps://us-south.ml.cloud.ibm.com, project_idSecret.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_tokens2.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参数类型默认值说明modelstribm/slate-30m-english-rtrvr-v2用于计算嵌入的 watsonx 模型名api_keySecretSecret.from_env_var(WATSONX_API_KEY)IBM watsonx API 密钥可用环境变量设置api_base_urlstrhttps://us-south.ml.cloud.ibm.comwatsonx.ai 服务地址可替换为其他区域或自建端点project_idSecretSecret.from_env_var(WATSONX_PROJECT_ID)Watson Studio 项目 IDtruncate_input_tokensint \| NoneNone输入文本最多使用的 token 数为None时使用完整输入不超过模型上限prefixstr追加到每个待嵌入文本开头的字符串suffixstr追加到每个待嵌入文本末尾的字符串timeoutfloat \| NoneNoneAPI 请求超时时间秒max_retriesint \| NoneNoneAPI 请求最大重试次数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(contentI love pizza!), Document(contentPasta is great too), ] document_embedder WatsonxDocumentEmbedder( modelibm/slate-30m-english-rtrvr-v2, api_keySecret.from_env_var(WATSONX_API_KEY), api_base_urlhttps://us-south.ml.cloud.ibm.com, project_idSecret.from_env_var(WATSONX_PROJECT_ID), ) result document_embedder.run(documentsdocuments) 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文档嵌入器额外暴露了四个面向批量的参数参数类型默认值说明batch_sizeint1000单次 API 调用嵌入的文档数量concurrency_limitint5并行请求数meta_fields_to_embedlist[str] \| NoneNone需要连同文档正文一起嵌入的元数据字段名列表embedding_separatorstr\n拼接元数据字段与文档正文时使用的分隔符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(contentsome text, meta{title: relevant title, page number: 18}) embedder WatsonxDocumentEmbedder( api_keySecret.from_env_var(WATSONX_API_KEY), project_idSecret.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_functioncosine) documents [ Document(contentMy name is Wolfgang and I live in Berlin), Document(contentI saw a black horse running), Document(contentGermany has many big cities), ] indexing_pipeline Pipeline() indexing_pipeline.add_component(embedder, WatsonxDocumentEmbedder()) indexing_pipeline.add_component(writer, DocumentWriter(document_storedocument_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_storedocument_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_keySecret.from_env_var(WATSONX_API_KEY), modelibm/granite-4-h-small, project_idSecret.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[Whats in this image?, image_content])] # 使用多模态模型 client WatsonxChatGenerator( api_keySecret.from_env_var(WATSONX_API_KEY), modelmeta-llama/llama-3-2-11b-vision-instruct, project_idSecret.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外还需注意参数类型默认值说明generation_kwargsdict[str, Any] \| NoneNone透传给 watsonx.ai 推理端点的生成参数见下文verifybool \| str \| NoneNoneSSL 校验设置True校验默认、False跳过不安全、或传入 CA bundle 路径使用自定义证书streaming_callbackStreamingCallbackT \| NoneNone流式响应回调每个新 token 到达时被调用toolsToolsType \| NoneNoneTool/Toolset对象列表或单个Toolset供模型准备函数调用初始化前还可以设置两个环境变量覆盖默认网络行为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_asyncrun( *, 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]]messagesChatMessage列表若传入普通字符串会被自动包装为一条user角色的ChatMessage。generation_kwargs运行时传入可覆盖初始化时设置的同名参数。streaming_callback提供时覆盖初始化设置的回调。tools提供时覆盖初始化设置的tools。返回值{replies: [ChatMessage, ...]}即模型生成的回复消息列表。run_async与run参数完全一致适合在异步流水线中调用便于在高并发场景下复用事件循环。4.5 在流水线中使用 ChatPromptBuilderfrom 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_keySecret.from_env_var(WATSONX_API_KEY), project_idSecret.from_env_var(WATSONX_PROJECT_ID), modelibm/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(Whats 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继承自WatsonxChatGeneratorBases: WatsonxChatGenerator提供面向纯字符串 prompt 的标准 Generator 接口适合简单文本生成任务。注意组件文档中已标注弃用声明建议迁移到同样接受纯字符串输入的WatsonxChatGenerator参考 WatsonxGenerator 组件文档。5.1 基础用法from haystack_integrations.components.generators.watsonx.generator import WatsonxGenerator from haystack.utils import Secret generator WatsonxGenerator( api_keySecret.from_env_var(WATSONX_API_KEY), modelibm/granite-4-h-small, project_idSecret.from_env_var(WATSONX_PROJECT_ID), ) response generator.run( promptExplain quantum computing in simple terms, system_promptYou 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_keySecret.from_env_var(WATSONX_API_KEY), project_idSecret.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_dictwatsonx 集成的四个组件都实现了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.aiHaystack 集成提供了四条清晰的组件路径索引向量化WatsonxDocumentEmbedder批量编码文档配合meta_fields_to_embed提升检索质量通过batch_size与concurrency_limit控制吞吐查询向量化WatsonxTextEmbedder编码单条查询接 embedding Retriever 完成相似度召回对话生成WatsonxChatGenerator基于ChatMessage完成同步/异步 Chat 补全支持多模态图片输入、流式回调与tools函数调用文本生成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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
