Haystack 集成指南：使用 GoogleAIGeminiGenerator 与 GoogleAIGeminiChatGenerator 构建 Gemini 多模态生成与对话应用

Haystack 集成指南：使用 GoogleAIGeminiGenerator 与 GoogleAIGeminiChatGenerator 构建 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

本文以 version-2.18 的 Google AI 集成 API 参考文档 为核心，系统讲解 Haystack 生态中通过 Google AI Studio 调用 Gemini 系列模型的两个生成组件：GoogleAIGeminiGenerator（一次性多模态文本生成）与GoogleAIGeminiChatGenerator（多轮对话补全）。你将掌握 API Key 配置、全部初始化与运行参数、文本/图片/多轮对话/函数调用四种实战写法，以及将组件接入 RAG Pipeline 与异步运行的方法。

适用版本说明：本文组件形态与参数签名以仓库内version-2.18参考文档为准。该版本对应 Haystack 2.x 时代，组件由google-ai-haystack集成包提供。

一、集成概览：两个组件，两条生成路径

Google AI 集成通过 Google AI Studio 提供对 Gemini 系列多模态模型（如gemini-2.0-flash、gemini-1.5-pro）的访问，共暴露两个组件，分别位于haystack_integrations.components.generators.google_ai包中：

两者都要求 Google AI Studio 的 API Key 进行认证，默认从GOOGLE_API_KEY环境变量读取；也都支持通过generation_config、safety_settings精细控制生成行为，并通过streaming_callback实现流式输出。

在 Pipeline 中的典型位置：GoogleAIGeminiGenerator通常放在PromptBuilder之后，GoogleAIGeminiChatGenerator通常放在ChatPromptBuilder之后。

二、环境准备：安装与 API Key

先安装集成包：

pip install google-ai-haystack

Google AI Studio 的 API Key 有两种注入方式，官方推荐使用环境变量（避免密钥硬编码进代码）：

import os from haystack.utils import Secret from haystack_integrations.components.generators.google_ai import GoogleAIGeminiGenerator # 方式一：环境变量（推荐，组件默认行为） os.environ["GOOGLE_API_KEY"] = "<MY_API_KEY>" gemini = GoogleAIGeminiGenerator(model="gemini-2.0-flash") # 方式二：显式传入 Secret（从环境变量读取） gemini = GoogleAIGeminiGenerator( model="gemini-2.0-flash", api_key=Secret.from_env_var("GOOGLE_API_KEY"), ) # 方式三：显式传入密钥字符串 gemini = GoogleAIGeminiGenerator( model="gemini-2.0-flash", api_key=Secret.from_token("<MY_API_KEY>"), )

从构造函数签名可见，api_key的类型是Secret，默认值为Secret.from_env_var("GOOGLE_API_KEY")，这正是"不传api_key也能从环境变量读取"的底层实现（见参考文档GoogleAIGeminiGenerator.__init__）。密钥优先从环境变量获取，其次可以显式覆盖。

三、GoogleAIGeminiGenerator：一次性多模态文本生成

3.1 构造签名与参数说明

def __init__(*, api_key: Secret = Secret.from_env_var("GOOGLE_API_KEY"), model: str = "gemini-2.0-flash", generation_config: Optional[Union[GenerationConfig, dict[str, Any]]] = None, safety_settings: Optional[dict[HarmCategory, HarmBlockThreshold]] = None, streaming_callback: Optional[Callable[[StreamingChunk], None]] = None)

其中StreamingChunk是 Haystack 核心库中定义的流式数据封装类，见 haystack/dataclasses/streaming_chunk.py：它承载content（文本片段）、meta（元数据）、component_info（产生该分块的组件信息）等字段，并定义了SyncStreamingCallbackT/AsyncStreamingCallbackT两种回调类型别名；在异步上下文中，同步回调会被接受但会告警"将内联运行在事件循环上、可能阻塞事件循环"。这套统一抽象让 Google AI 组件与 Haystack 其他生成器在流式行为上保持一致。

3.2 基础文本生成

from haystack.utils import Secret from haystack_integrations.components.generators.google_ai import GoogleAIGeminiGenerator gemini = GoogleAIGeminiGenerator(model="gemini-2.0-flash", api_key=Secret.from_token("<MY_API_KEY>")) res = gemini.run(parts=["What is the most interesting thing you know?"]) for answer in res["replies"]: print(answer)

3.3 多模态输入：图文混合提示

Gemini 的核心能力是多模态。run方法的parts是一个变长参数（Variadic[Union[str, ByteStream, Part]]），可以同时接收字符串、ByteStream与Part对象——这意味着图片、音频、视频都可以与文本一起进入模型。二进制内容统一用 Haystack 的ByteStream数据类承载（定义见 haystack/dataclasses/byte_stream.py，支持data字节数据与mime_type媒体类型字段）：

import requests from haystack.utils import Secret from haystack.dataclasses.byte_stream import ByteStream from haystack_integrations.components.generators.google_ai import GoogleAIGeminiGenerator URLS = [ "https://raw.githubusercontent.com/silvanocerza/robots/main/robot1.jpg", "https://raw.githubusercontent.com/silvanocerza/robots/main/robot2.jpg", "https://raw.githubusercontent.com/silvanocerza/robots/main/robot3.jpg", "https://raw.githubusercontent.com/silvanocerza/robots/main/robot4.jpg", ] images = [ ByteStream(data=requests.get(url).content, mime_type="image/jpeg") for url in URLS ] gemini = GoogleAIGeminiGenerator(model="gemini-2.0-flash", api_key=Secret.from_token("<MY_API_KEY>")) result = gemini.run(parts=["What can you tell me about this robots?", *images]) for answer in result["replies"]: print(answer)

要点：

• 用列表解包（*images）把多张图片与提示文本拼接进parts；
• 为每个ByteStream显式指定mime_type="image/jpeg"，帮助模型正确解析内容类型；
• 也可以使用ByteStream.from_file_path(...)从本地文件直接构造二进制流（支持guess_mime_type=True自动推断 MIME 类型）。

3.4 run 方法与输出

@component.output_types(replies=list[str]) def run(parts: Variadic[Union[str, ByteStream, Part]], streaming_callback: Optional[Callable[[StreamingChunk], None]] = None)

• 输入：parts为字符串、ByteStream或Part对象的异构列表；streaming_callback可在运行时覆盖构造时设置的回调。
• 输出：返回字典，仅含一个键replies——模型生成的回复字符串列表（list[str]）。

四、GoogleAIGeminiChatGenerator：多轮对话与函数调用

GoogleAIGeminiChatGenerator用于聊天补全，它基于 Haystack 的ChatMessage数据类与模型交互（ChatMessage及其角色枚举ChatRole的定义见 haystack/dataclasses/chat_message.py，包含user、system、assistant、tool四种角色）。

4.1 构造签名与参数说明

def __init__(*, api_key: Secret = Secret.from_env_var("GOOGLE_API_KEY"), model: str = "gemini-2.0-flash", generation_config: Optional[Union[GenerationConfig, dict[str, Any]]] = None, safety_settings: Optional[dict[HarmCategory, HarmBlockThreshold]] = None, tools: Optional[list[Tool]] = None, tool_config: Optional[content_types.ToolConfigDict] = None, streaming_callback: Optional[StreamingCallbackT] = None)

与GoogleAIGeminiGenerator相比，新增两个与工具调用相关的参数：

其余api_key、model、generation_config、safety_settings、streaming_callback的含义与生成器组件一致。

4.2 多轮对话基础用法

from haystack.utils import Secret from haystack.dataclasses.chat_message import ChatMessage from haystack_integrations.components.generators.google_ai import GoogleAIGeminiChatGenerator gemini_chat = GoogleAIGeminiChatGenerator(model="gemini-2.0-flash", api_key=Secret.from_token("<MY_API_KEY>")) messages = [ChatMessage.from_user("What is the most interesting thing you know?")] res = gemini_chat.run(messages=messages) for reply in res["replies"]: print(reply.text) # 把模型回复追加回消息历史，继续追问，形成多轮上下文 messages += res["replies"] + [ChatMessage.from_user("Tell me more about it")] res = gemini_chat.run(messages=messages) for reply in res["replies"]: print(reply.text)

多轮对话的关键在于：每一轮都要把历史消息完整传回（用户消息 + 模型回复 + 新问题），模型才能基于完整上下文继续作答。

4.3 函数调用（Function Calling）

Gemini 可以"准备"工具调用，再由你的代码真正执行工具。完整流程分三步：

第一步：定义函数并转换为Tool。使用Annotated类型注解为参数提供描述，再通过create_tool_from_function转换（该函数定义在 haystack/tools/from_function.py，会基于函数签名、类型注解与 docstring 自动生成工具参数 JSON Schema；Tool数据类的字段定义见 haystack/tools/tool.py）：

from 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_ai import GoogleAIGeminiChatGenerator # example function to get the current weather 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 f"The 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 = GoogleAIGeminiChatGenerator( model="gemini-2.0-flash-exp", api_key=Secret.from_token("<MY_API_KEY>"), tools=[tool], ) user_message = [ChatMessage.from_user("What is the temperature in celsius in Berlin?")] replies = gemini_chat.run(messages=user_message)["replies"] print(replies[0].tool_calls)

此时模型返回的replies[0].tool_calls是ToolCall对象列表（包含tool_name与arguments，定义见 haystack/dataclasses/chat_message.py），而不是最终答案。

第三步：实际调用工具并让模型总结结果。

# actually invoke the tool tool_messages = tool_invoker.run(messages=replies)["tool_messages"] messages = user_message + replies + tool_messages # transform the tool call result into a human readable message final_replies = gemini_chat.run(messages=messages)["replies"] print(final_replies[0].text)

这里ToolInvoker负责真正执行工具并生成tool角色的消息，把"用户消息 + 模型工具调用 + 工具执行结果"拼接后再次送入生成器，模型即可基于真实结果输出最终答案。

4.4 run / run_async：同步与异步执行

@component.output_types(replies=list[ChatMessage]) def run(messages: list[ChatMessage], streaming_callback: Optional[StreamingCallbackT] = None, *, tools: Optional[list[Tool]] = None)

@component.output_types(replies=list[ChatMessage]) async def run_async(messages: list[ChatMessage], streaming_callback: Optional[StreamingCallbackT] = None, *, tools: Optional[list[Tool]] = None)

• messages：ChatMessage实例列表，代表输入对话历史；
• streaming_callback：运行时覆盖流式回调；
• tools：运行时覆盖初始化时设置的工具列表——注意它是关键字参数（*之后），且优先级高于构造时的tools；
• 返回字典含replies键：模型生成的ChatMessage列表；
• run_async是run的异步版本，供Pipeline.run_async或高并发场景使用。

五、接入 Pipeline：RAG 与聊天流水线

5.1 RAG 检索增强生成流水线

将GoogleAIGeminiGenerator接到检索器与提示模板之后，即构成一条完整 RAG 链路（用例出自 googleaigeminigenerator.mdx）：

import os from haystack.components.retrievers.in_memory import InMemoryBM25Retriever from haystack.components.builders import PromptBuilder from haystack import Pipeline from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack_integrations.components.generators.google_ai import ( GoogleAIGeminiGenerator, ) os.environ["GOOGLE_API_KEY"] = "<MY_API_KEY>" docstore = InMemoryDocumentStore() template = """ Given the following information, answer the question. Context: {% for document in documents %} {{ document.content }} {% endfor %} Question: What's the official language of {{ country }}? """ pipe = Pipeline() pipe.add_component("retriever", InMemoryBM25Retriever(document_store=docstore)) pipe.add_component("prompt_builder", PromptBuilder(template=template)) pipe.add_component("gemini", GoogleAIGeminiGenerator(model="gemini-pro")) pipe.connect("retriever", "prompt_builder.documents") pipe.connect("prompt_builder", "gemini") pipe.run({"prompt_builder": {"country": "France"}})

5.2 聊天流水线

用ChatPromptBuilder组织模板消息，再连接GoogleAIGeminiChatGenerator（用例出自 googleaigeminichatgenerator.mdx）：

import os from haystack.components.builders import ChatPromptBuilder from haystack.dataclasses import ChatMessage from haystack import Pipeline from haystack_integrations.components.generators.google_ai import ( GoogleAIGeminiChatGenerator, ) ## no parameter init, we don't use any runtime template variables prompt_builder = ChatPromptBuilder() os.environ["GOOGLE_API_KEY"] = "<MY_API_KEY>" gemini_chat = GoogleAIGeminiChatGenerator() 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, }, }, ) print(res)

六、序列化：to_dict / from_dict

两个组件都实现了标准的 Haystack 组件序列化协议，用于将组件持久化（例如导出为 YAML/JSON 或存入版本库后还原）：

与所有 Haystack 组件一致，这两个方法配合 Haystack 的Pipeline序列化机制（Pipeline.dumps/Pipeline.loads）即可实现完整流水线的导出与导入。需要注意的是：序列化时Secret类型的 API Key 会以安全的引用形式存储（指向环境变量名），而不是把明文密钥写入序列化产物。

七、流式输出实践

两个组件都支持流式输出：把回调函数传给streaming_callback后，模型生成的每个 token 会以StreamingChunk的形式被实时推送给回调（而不是等完整回复生成后才返回）。典型写法：

def on_token(chunk: StreamingChunk) -> None: print(chunk.content, end="", flush=True) gemini = GoogleAIGeminiGenerator( model="gemini-2.0-flash", api_key=Secret.from_env_var("GOOGLE_API_KEY"), streaming_callback=on_token, ) res = gemini.run(parts=["Tell me a short story"])

streaming_callback既可以像上面一样在构造时设置，也可以在run/run_async调用时以参数形式传入（运行时设置会覆盖构造时的设置）。

八、版本提醒与迁移建议

仓库内 googleaigeminigenerator.mdx 与 googleaigeminichatgenerator.mdx 均带有明确的弃用警告：

该集成使用已被弃用的google-generativeaiSDK，该 SDK 将在 2025 年 8 月之后失去支持。官方建议切换到新的 GoogleGenAIChatGenerator 集成。

因此在规划新项目时，建议优先评估google-genai-haystack提供的新集成；而本文所述组件在已运行的旧项目中仍可继续使用，迁移时注意参数名与输出结构的变化即可。更完整的 API 参考可继续查阅仓库内的 Google AI 集成参考文档 及各版本对应的 GoogleAIGeminiGenerator、GoogleAIGeminiChatGenerator 指南。

九、快速决策速查表

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