Haystack 与 Google Vertex AI 集成实战指南从文档嵌入、Gemini 生成到多模态能力全解析【免费下载链接】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 开源项目的 Google Vertex 集成google-vertex-haystack包为核心系统讲解如何在 Haystack 管道中使用 Vertex AI Embeddings API 完成文档/查询向量化以及如何通过 Gemini、PaLM 系生成模型实现文本生成、代码生成、图片生成与图片问答等能力。读完本文你将掌握每个组件的完整参数含义、典型调用方式、认证前提并能将其组合进索引、RAG、Agent 等真实管道中。集成总览一条管道覆盖嵌入与生成Google Vertex AI 集成组件由haystack_integrations.components.embedders.google_vertex与haystack_integrations.components.generators.google_vertex两个命名空间构成前者负责把文本与文档转换成向量Embeddings后者负责调用 Vertex 上的生成式模型。安装方式pip install google-vertex-haystack所有组件均通过 Google Cloud Application Default CredentialsADCs进行认证。前提是当前环境本地 shell 或云上实例能提供一组有权访问 Vertex AI 端点的 GCP 账号凭据具体配置方法可参考 secret-management 概念文档 与 Google 官方 ADC 文档。项目 ID 可通过gcloud projects list查询若未显式传入project_id/gcp_project_id则由认证阶段自动填充。注意当前版本仓库中 vertexaidocumentembedder.mdx 与 vertexaigeminichatgenerator.mdx 等页面标注了弃用提示google-vertex-haystack集成基于已停更的 Google SDK官方推荐迁移到新的google-genai-haystack包如GoogleGenAIDocumentEmbedder、GoogleGenAIChatGenerator。本文仍按 2.19 版本的参考文档完整讲解其 API 行为迁移时可对照使用。一、文本向量化VertexAITextEmbedderVertexAITextEmbedder负责把单条字符串典型场景是查询 query编码为浮点向量用于查询侧嵌入。它与文档侧嵌入组件配合构成检索管道的两端。初始化参数__init__( model: Literal[ text-embedding-004, text-embedding-005, textembedding-gecko-multilingual001, text-multilingual-embedding-002, text-embedding-large-exp-03-07, ], task_type: Literal[ RETRIEVAL_DOCUMENT, RETRIEVAL_QUERY, SEMANTIC_SIMILARITY, CLASSIFICATION, CLUSTERING, QUESTION_ANSWERING, FACT_VERIFICATION, CODE_RETRIEVAL_QUERY, ] RETRIEVAL_QUERY, gcp_region_name: Optional[Secret] Secret.from_env_var(GCP_DEFAULT_REGION, strictFalse), gcp_project_id: Optional[Secret] Secret.from_env_var(GCP_PROJECT_ID, strictFalse), progress_bar: bool True, truncate_dim: Optional[int] None, ) - None参数默认值说明model无必填Vertex AI Embeddings API 中的模型名仅支持上列 5 个值传入其他值会抛出ValueErrortask_typeRETRIEVAL_QUERY生成嵌入所面向的任务类型Vertex 服务端会据此调整嵌入质量查询侧默认取RETRIEVAL_QUERY文档侧默认取RETRIEVAL_DOCUMENTgcp_region_name环境变量GCP_DEFAULT_REGIONAPI 调用使用的默认区域未设置时回退us-central-1gcp_project_id环境变量GCP_PROJECT_IDGCP 项目 ID缺省时由 Google Cloud 认证过程自动设置progress_barTrue处理过程中是否显示进度条truncate_dimNone若指定将嵌入向量截断到该维度Vertex 支持输出维度剪裁以降低存储成本注意gcp_region_name、gcp_project_id的类型是Optional[Secret]与 Haystack 的 Secret 机制 对齐避免密钥硬编码。调用方式from haystack_integrations.components.embedders.google_vertex import VertexAITextEmbedder text_to_embed I love pizza! text_embedder VertexAITextEmbedder(modeltext-embedding-005) print(text_embedder.run(text_to_embed)) # {embedding: [-0.08127457648515701, 0.03399784862995148, -0.05116401985287666, ...]run的签名是run(text: Union[list[Document], list[str], str])——即它不仅能嵌入字符串还能接收Document或字符串列表内部会分批处理以适配 API 的单请求 token 上限。返回值字典仅含一个键embedding输入文本对应的向量。二、文档向量化VertexAIDocumentEmbedderVertexAIDocumentEmbedder为一批Document计算嵌入并把结果写回每个文档的embedding字段是索引管道中DocumentWriter之前的典型组件。初始化参数与VertexAITextEmbedder相比文档嵌入器增加了批量与重试相关的控制项__init__( model: Literal[ text-embedding-004, text-embedding-005, textembedding-gecko-multilingual001, text-multilingual-embedding-002, text-embedding-large-exp-03-07, ], task_type: Literal[...] RETRIEVAL_DOCUMENT, gcp_region_name: Optional[Secret] Secret.from_env_var(GCP_DEFAULT_REGION, strictFalse), gcp_project_id: Optional[Secret] Secret.from_env_var(GCP_PROJECT_ID, strictFalse), batch_size: int 32, max_tokens_total: int 20000, time_sleep: int 30, retries: int 3, progress_bar: bool True, truncate_dim: Optional[int] None, meta_fields_to_embed: Optional[list[str]] None, embedding_separator: str \n, ) - None参数默认值说明batch_size32单批处理的文档数量max_tokens_total20000单次请求允许处理的总 token 上限超限会自动拆分time_sleep30失败重试之间的休眠秒数retries3失败时的最大重试次数meta_fields_to_embedNone需要一并参与嵌入的元数据字段列表embedding_separator\n当拼接content与多个元数据字段时使用的分隔符独立使用from haystack import Document from haystack_integrations.components.embedders.google_vertex import VertexAIDocumentEmbedder doc Document(contentI love pizza!) document_embedder VertexAIDocumentEmbedder(modeltext-embedding-005) result document_embedder.run([doc]) print(result[documents][0].embedding) # [-0.044606007635593414, 0.02857724390923977, -0.03549133986234665, ...run(documents: list[Document])返回{documents: [...]}其中每个文档都被写入了embedding属性。内部处理细节get_text_embedding_input(batch: list[Document]) - list[TextEmbeddingInput]把一批Document转换为 Vertex SDK 的TextEmbeddingInput对象转换时会依据meta_fields_to_embed与embedding_separator拼接内容。embed_batch(batch: list[str]) - list[list[float]]对一批纯文本字符串生成嵌入。embed_batch_by_smaller_batches(batch: list[str], subbatch: int 1) - list[list[float]]把批次再切成更小的子批逐条调用任一项嵌入失败都会抛出带错误详情的异常用于规避单条超限导致整批失败。嵌入 检索的完整管道示例from haystack import Document, Pipeline from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever from haystack_integrations.components.embedders.google_vertex import ( VertexAITextEmbedder, VertexAIDocumentEmbedder, ) 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), ] document_embedder VertexAIDocumentEmbedder(modeltext-embedding-005) documents_with_embeddings document_embedder.run(documents)[documents] document_store.write_documents(documents_with_embeddings) query_pipeline Pipeline() query_pipeline.add_component(text_embedder, VertexAITextEmbedder(modeltext-embedding-005)) query_pipeline.add_component(retriever, InMemoryEmbeddingRetriever(document_storedocument_store)) query_pipeline.connect(text_embedder.embedding, retriever.query_embedding) query Who lives in Berlin? result query_pipeline.run({text_embedder: {text: query}}) print(result[retriever][documents][0]) # Document(id..., content: My name is Wolfgang and I live in Berlin)三、Gemini 聊天生成VertexAIGeminiChatGeneratorVertexAIGeminiChatGenerator通过 Gemini 模型完成多轮聊天补全支持工具调用function calling与流式输出是构建对话式 Agent 的核心生成组件。默认模型为gemini-1.5-flash仓库文档同时确认支持gemini-1.5-pro、gemini-2.0-flashGoogle 官方建议从1.5-pro升级到2.0-flash。初始化参数__init__( *, model: str gemini-1.5-flash, project_id: Optional[str] None, location: Optional[str] None, generation_config: Optional[Union[GenerationConfig, dict[str, Any]]] None, safety_settings: Optional[dict[HarmCategory, HarmBlockThreshold]] None, tools: Optional[list[Tool]] None, tool_config: Optional[ToolConfig] None, streaming_callback: Optional[StreamingCallbackT] None ) - None参数默认值说明modelgemini-1.5-flashGemini 模型名project_id/locationNoneGCP 项目 ID 与区域缺省时分别由认证过程自动设置 / 回退us-central-1generation_configNone生成配置可传GenerationConfig对象或字典支持字段temperature、top_p、top_k、candidate_count、max_output_tokens、stop_sequencessafety_settingsNone内容安全设置键为HarmCategory、值为HarmBlockThresholdtoolsNone模型可准备调用的工具列表tool_configNone工具调用配置如调用模式、允许的函数名streaming_callbackNone流式回调函数接收StreamingChunk参数基本用法from haystack.dataclasses import ChatMessage from haystack_integrations.components.generators.google_vertex import VertexAIGeminiChatGenerator gemini_chat VertexAIGeminiChatGenerator() messages [ChatMessage.from_user(Tell me the name of a movie)] res gemini_chat.run(messages) print(res[replies][0].text) # The Shawshank Redemption messages [res[replies][0], ChatMessage.from_user(Whos the main actor?)] res gemini_chat.run(messages) print(res[replies][0].text) # Tim Robbinsrun(messages, streaming_callbackNone, *, toolsNone)返回{replies: [...]}tools参数若在运行时传入会覆盖初始化时设置的同名参数。run_async提供同名异步版本便于在异步管道中调用。工具调用Function Callingfrom typing import Annotated from haystack.utils import Secret from haystack.dataclasses.chat_message import ChatMessage from haystack.components.tools import ToolInvoker from haystack.tools import create_tool_from_function from haystack_integrations.components.generators.google_vertex import VertexAIGeminiChatGenerator # 定义一个查询天气的函数 def get_current_weather( location: Annotated[str, The city for which to get the weather, e.g. San Francisco] Munich, unit: Annotated[str, The unit for the temperature, e.g. celsius] celsius, ) - str: return fThe weather in {location} is sunny. The temperature is 20 {unit}. tool create_tool_from_function(get_current_weather) tool_invoker ToolInvoker(tools[tool]) gemini_chat VertexAIGeminiChatGenerator(modelgemini-2.0-flash-exp, tools[tool]) user_message [ChatMessage.from_user(What is the temperature in celsius in Berlin?)] replies gemini_chat.run(messagesuser_message)[replies] print(replies[0].tool_calls) # [ToolCall(tool_nameget_current_weather, arguments{unit: celsius, location: Berlin}, idNone)] # 执行工具并回填结果 tool_messages tool_invoker.run(messagesreplies)[tool_messages] messages user_message replies tool_messages final_replies gemini_chat.run(messagesmessages)[replies] print(final_replies[0].text) # The temperature in Berlin is 20 degrees Celsius.如果不想手动维护“模型出工具调用 → 执行 → 回填”的循环可以把生成器与工具直接交给 Haystack 的Agent位于 Agent 组件文档由它自动完成多轮工具调用直至产出最终答案。在管道中使用from haystack.components.builders import ChatPromptBuilder from haystack.dataclasses import ChatMessage from haystack import Pipeline from haystack_integrations.components.generators.google_vertex import VertexAIGeminiChatGenerator prompt_builder ChatPromptBuilder() gemini_chat VertexAIGeminiChatGenerator() pipe Pipeline() pipe.add_component(prompt_builder, prompt_builder) pipe.add_component(gemini, gemini_chat) pipe.connect(prompt_builder.prompt, gemini.messages) location Rome messages [ChatMessage.from_user(Tell me briefly about {{location}} history)] res pipe.run(data{ prompt_builder: { template_variables: {location: location}, template: messages, } })四、多模态生成VertexAIGeminiGeneratorVertexAIGeminiGenerator是面向 Gemini 的多模态文本生成器run的输入parts是变长参数Variadic[Union[str, ByteStream, Part]]可以混合文本、图片、音视频与Part对象一次请求内完成多模态理解。默认模型为gemini-2.0-flash。初始化参数__init__( *, model: str gemini-2.0-flash, project_id: Optional[str] None, location: Optional[str] None, generation_config: Optional[Union[GenerationConfig, dict[str, Any]]] None, safety_settings: Optional[dict[HarmCategory, HarmBlockThreshold]] None, system_instruction: Optional[Union[str, ByteStream, Part]] None, streaming_callback: Optional[Callable[[StreamingChunk], None]] None ) - None与聊天生成器相比多模态生成器额外提供system_instruction系统指令参数字符串、ByteStream或Part。generation_config支持字段同上temperature、top_p、top_k、candidate_count、max_output_tokens、stop_sequences。基础与多模态用法from haystack_integrations.components.generators.google_vertex import VertexAIGeminiGenerator gemini VertexAIGeminiGenerator() result gemini.run(parts[What is the most interesting thing you know?]) for answer in result[replies]: print(answer)多模态提示图片通过ByteStream传入import requests from haystack.dataclasses.byte_stream import ByteStream from haystack_integrations.components.generators.google_vertex import VertexAIGeminiGenerator URLS [ https://raw.githubusercontent.com/silvanocerza/robots/main/robot1.jpg, https://raw.githubusercontent.com/silvanocerza/robots/main/robot2.jpg, ] images [ByteStream(datarequests.get(url).content, mime_typeimage/jpeg) for url in URLS] gemini VertexAIGeminiGenerator() result gemini.run(parts[What can you tell me about these robots?, *images])这里用到的ByteStream是 Haystack 表示二进制对象的统一数据类源码位于 byte_stream.py可通过ByteStream.from_file_path(path)从本地文件创建用to_file(destination_path)落盘还支持from_string/to_string与to_dict/from_dict序列化字段为data、meta、mime_type。RAG 管道示例from haystack.components.retrievers.in_memory import InMemoryBM25Retriever from haystack.components.builders import PromptBuilder from haystack import Pipeline, Document from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack_integrations.components.generators.google_vertex import VertexAIGeminiGenerator docstore InMemoryDocumentStore() docstore.write_documents([ Document(contentRome is the capital of Italy), Document(contentParis is the capital of France), ]) query What is the capital of France? template Given the following information, answer the question. Context: {% for document in documents %} {{ document.content }} {% endfor %} Question: {{ query }}? pipe Pipeline() pipe.add_component(retriever, InMemoryBM25Retriever(document_storedocstore)) pipe.add_component(prompt_builder, PromptBuilder(templatetemplate)) pipe.add_component(gemini, VertexAIGeminiGenerator()) pipe.connect(retriever, prompt_builder.documents) pipe.connect(prompt_builder, gemini) res pipe.run({prompt_builder: {query: query}, retriever: {query: query}})五、其他生成组件一览VertexAITextGeneratorPaLM 文本生成基于text-bison、text-unicorn、text-bison-32k等文本模型run(prompt: str)返回三个键replies生成文本列表、safety_attributes每条回答的安全评分字典、citations引文列表。from haystack_integrations.components.generators.google_vertex import VertexAITextGenerator generator VertexAITextGenerator() res generator.run(Tell me a good interview question for a software engineer.) print(res[replies][0])VertexAICodeGenerator代码生成支持code-bison、code-bison-32k、code-geckorun(prefix: str, suffix: Optional[str] None)接收光标前后的代码片段返回replies列表。**kwargs会透传给底层TextGenerationModel.predict()。from haystack_integrations.components.generators.google_vertex import VertexAICodeGenerator generator VertexAICodeGenerator() result generator.run(prefixdef to_json(data):) for answer in result[replies]: print(answer)VertexAIImageCaptioner图片描述基于 imagetext 模型为图片生成说明文字默认模型imagetext。run(image: ByteStream)返回captions列表。**kwargs透传给ImageTextModel.get_captions()。import requests from haystack.dataclasses.byte_stream import ByteStream from haystack_integrations.components.generators.google_vertex import VertexAIImageCaptioner captioner VertexAIImageCaptioner() image ByteStream(datarequests.get( https://raw.githubusercontent.com/deepset-ai/haystack-core-integrations/main/integrations/google_vertex/example_assets/robot1.jpg ).content) result captioner.run(imageimage) for caption in result[captions]: print(caption) # two gold robots are standing next to each other in the desertVertexAIImageQA图片问答同样基于 imagetext 模型run(image: ByteStream, question: str)返回replies答案列表。**kwargs透传给ImageTextModel.ask_question()。from haystack.dataclasses.byte_stream import ByteStream from haystack_integrations.components.generators.google_vertex import VertexAIImageQA qa VertexAIImageQA() image ByteStream.from_file_path(dog.jpg) res qa.run(imageimage, questionWhat color is this dog) print(res[replies][0]) # whiteVertexAIImageGenerator图片生成基于 imagegeneration 模型run(prompt: str, negative_prompt: Optional[str] None)支持负面提示词返回imagesByteStream列表可配合to_file直接落盘。**kwargs透传给ImageGenerationModel.generate_images()。from pathlib import Path from haystack_integrations.components.generators.google_vertex import VertexAIImageGenerator generator VertexAIImageGenerator() result generator.run(promptGenerate an image of a cute cat) result[images][0].to_file(Path(my_image.png))这些生成组件VertexAITextGenerator、VertexAICodeGenerator、VertexAIImageCaptioner、VertexAIImageQA、VertexAIImageGenerator均继承统一的初始化形态——model、project_id、location加**kwargs透传并实现to_dict/from_dict序列化协议可直接被 Haystack 管道反序列化。六、序列化与管道集成要点所有 Google Vertex 组件都实现了统一的to_dict()/from_dict()协议to_dict() - dict[str, Any]把组件配置序列化为字典便于 YAML/JSON 配置化from_dict(data: dict[str, Any]) - 对应组件从字典还原组件实例。这意味着组件可以无缝嵌入 Haystack 的声明式管道参见 marshal 序列化模块 与管道序列化机制把认证所需的Secret交给 Haystack 的 Secret 管理统一处理密钥以环境变量方式注入GCP_DEFAULT_REGION、GCP_PROJECT_ID避免明文写在配置中。七、选型与迁移建议场景推荐组件配套文档索引向量化VertexAIDocumentEmbedderDocumentWriter、InMemoryDocumentStore查询向量化VertexAITextEmbedderInMemoryEmbeddingRetriever等向量检索器多轮对话 / AgentVertexAIGeminiChatGeneratorToolInvoker、Agent、ChatPromptBuilder多模态理解 / RAG 生成VertexAIGeminiGeneratorPromptBuilder、BM25 检索器代码补全VertexAICodeGenerator—图片描述 / 图片问答VertexAIImageCaptioner/VertexAIImageQAByteStream文生图VertexAIImageGeneratorByteStream.to_file组件之间的参数语义差异集中在嵌入器task_type、batch_size、max_tokens_total、retries、truncate_dim与 Gemini 生成器generation_config、safety_settings、tools、streaming_callback上。嵌入器侧的task_type必须与管道角色匹配查询用RETRIEVAL_QUERY、文档用RETRIEVAL_DOCUMENT否则会影响检索质量。最后再次提醒仓库当前文档vertexaitextembedder.mdx、vertexaigeminigenerator.mdx 等明确标注该集成已归档建议新项目直接采用google-genai-haystack包中的GoogleGenAI*系列组件对于已在运行的旧管道可参考本文 API 逐参数对照迁移。完整的 API 签名与参数说明可查阅 Google Vertex 集成 API 参考本仓库还保留了 2.18 至 3.1 各版本的同名参考文档。【免费下载链接】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），仅供参考