企业官网建设流程全解析

Haystack Google 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 2.23 版本对应的 Google AI 集成参考文档（google_ai.md）为核心，系统讲解google-ai-haystack包中GoogleAIGeminiGenerator与GoogleAIGeminiChatGenerator两个组件的完整用法。你将掌握：如何通过 Google AI Studio 的 API Key 接入 Gemini 系列模型、如何用一行代码完成文本生成与多模态（图文混合）推理、如何基于ChatMessage数据类维护多轮对话、如何让 Gemini 自主发起函数调用（Function Calling）并接入 Agent 与 Pipeline，以及该集成的弃用现状与官方推荐的迁移路径。

一、集成概览：两个组件，两种交互范式

从集成参考文档可以看出，Google AI 集成由两个模块组成，分别对应两种不同的模型交互方式：

两者的共同点在于：都通过 Google AI Studio（Gemini Developer API）鉴权，默认从GOOGLE_API_KEY环境变量读取密钥，默认模型为gemini-2.0-flash，都支持流式输出与to_dict/from_dict序列化。

需要特别说明的是，这两个组件并不在本仓库（haystack 核心库）内，而是由独立的google-ai-haystack集成包提供，导入路径为haystack_integrations.components.generators.google_ai。本仓库文档（googleaigeminigenerator.mdx、googleaigeminichatgenerator.mdx）确认了该集成支持gemini-2.5-pro-exp-03-25、gemini-2.0-flash、gemini-1.5-pro、gemini-1.5-flash等模型，完整模型清单以 Google Gemini API 官方模型文档为准。

二、安装与鉴权准备

2.1 安装集成包

集成文档给出了明确的安装命令，与所有 Haystack 集成包一致，通过 pip 安装：

pip install google-ai-haystack

安装完成后即可使用haystack_integrations.components.generators.google_ai下的两个组件。

2.2 获取 API Key 的两种方式

鉴权使用 Google AI Studio 的 API Key，参考文档给出了两种配置途径：

方式一：环境变量（推荐）。两个组件的api_key参数默认值均为Secret.from_env_var("GOOGLE_API_KEY")，即只要在环境中设置了GOOGLE_API_KEY，初始化时无需传参：

import os os.environ["GOOGLE_API_KEY"] = "<MY_API_KEY>" from haystack_integrations.components.generators.google_ai import GoogleAIGeminiChatGenerator gemini_chat = GoogleAIGeminiChatGenerator()

方式二：初始化时显式传入。通过haystack.utils的Secret封装：

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>"))

关于Secret的底层机制，可以查看 haystack/utils/auth.py 的源码：Secret是一个抽象基类，Secret.from_token(...)创建基于 Token 的密钥，Secret.from_env_var(...)创建基于环境变量的密钥（支持传入多个候选环境变量并按序解析，可通过strict参数控制未设置时是否抛异常），且Secret本身不可被直接序列化，这是为了安全地在序列化时避免密钥泄露。

API Key 的申请入口为 Google AI Studio（域名 aistudio.google.com）；生成式组件文档中同样标注了该获取渠道。

三、GoogleAIGeminiGenerator：单次生成与多模态输入

GoogleAIGeminiGenerator的定位是"通过 Google AI Studio 使用多模态 Gemini 模型生成文本"，它接收的输入parts是异构的——可以是字符串、ByteStream或Part对象的任意组合，这正是 Gemini 原生多模态能力的体现。

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)

各参数含义与要点：

• api_key：Google AI Studio API Key，默认从GOOGLE_API_KEY环境变量读取，也支持Secret.from_token显式注入。
• model：模型名称，默认gemini-2.0-flash。可用模型以 Google Gemini API 官方模型文档为准，集成文档确认当前支持gemini-2.5-pro-exp-03-25、gemini-2.0-flash、gemini-1.5-pro、gemini-1.5-flash。
• generation_config：生成配置，可传入GenerationConfig对象或参数字典。它决定了采样温度、max_output_tokens、top_p、top_k、stop_sequences、response_mime_type等生成行为参数。传字典时会被转换为GenerationConfig使用，这是最灵活的传参方式。
• safety_settings：安全设置，以HarmCategory为键、HarmBlockThreshold为值的字典，用于控制模型对有害内容的拦截级别。
• streaming_callback：流式回调函数，每收到一个流式 token 时被调用一次，回调参数为StreamingChunk对象。

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)

run方法的完整签名（来自参考文档）：

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

要点解读：

• parts是变长参数（Variadic），意味着你可以直接传多个位置参数，也可以传一个列表并解包（如run(parts=[...])）。
• 输出字典固定包含replies键，值为字符串列表，每项是一条生成的备选回复。
• streaming_callback也可以在run时传入，覆盖初始化时设置的回调。

3.3 多模态示例：图文混合推理

参考文档提供了一个完整的多模态示例——下载四张机器人图片后与文字问题一起交给模型：

import requests from haystack.utils import Secret from haystack.dataclasses.byte_stream import ByteStream from haystack_integrations.components.generators.google_ai import GoogleAIGeminiGenerator BASE_URL = ( "https://raw.githubusercontent.com/deepset-ai/haystack-core-integrations" "/main/integrations/google_ai/example_assets" ) URLS = [ f"{BASE_URL}/robot1.jpg", f"{BASE_URL}/robot2.jpg", f"{BASE_URL}/robot3.jpg", f"{BASE_URL}/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)

这里的核心是ByteStream——它是 Haystack 的统一二进制内容载体，定义于 haystack/dataclasses/byte_stream.py。通过ByteStream(data=..., mime_type="image/jpeg")显式声明 MIME 类型，组件即可识别图片内容。parts中的文字与ByteStream图片可以任意顺序混合，模型会同时理解文本指令与图像内容。这也是GoogleAIGeminiGenerator文档中标注的"常用位置在PromptBuilder之后、parts可混合图片/音频/视频/文本"的实战价值所在。

3.4 序列化：to_dict 与 from_dict

参考文档为两个组件都定义了标准的序列化方法：

• to_dict() -> dict[str, Any]：将组件序列化为字典，用于Pipeline的 YAML/JSON 持久化。
• from_dict(cls, data) -> "GoogleAIGeminiGenerator"：类方法，从字典反序列化重建组件实例。

需要注意：由于api_key基于Secret封装，to_dict序列化时不会写入明文密钥（Secret.from_token创建的 TokenSecret 不可序列化），从字典反序列化后仍需通过环境变量或在初始化时重新注入密钥。

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

GoogleAIGeminiChatGenerator的定位是"通过 Google AI Studio 使用 Gemini 模型完成对话补全"，它与模型交互的载体是ChatMessage数据类（定义于 haystack/dataclasses/chat_message.py），该数据类统一了user、system、assistant、tool四种角色，并能携带工具调用（ToolCall）信息——这正是构建 Agent 的基础。

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)

相比生成式组件，聊天组件新增了两个与工具相关的参数：

• tools：工具列表，模型可以据此准备函数调用（Function Calling）。工具通过 Haystack 的Tool类型承载，可以用create_tool_from_function从任意 Python 函数自动生成。
• tool_config：工具调用配置（对应 Gemini 的ToolConfig），用于控制函数调用的行为，如强制调用某个函数（FunctionCallingConfig模式）。

其余参数（api_key、model、generation_config、safety_settings、streaming_callback）与生成式组件语义一致。

4.2 多轮对话示例

参考文档展示了如何手动维护对话历史——每次把模型回复追加回messages列表再发起新一轮对话：

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)

注意与生成式组件两个关键差异：

1. 入参不同：run接收的是messages: list[ChatMessage]，而不是parts。
2. 出参不同：run返回的replies是ChatMessage列表，访问文本要用reply.text而非直接打印。

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)

要点：tools是关键字参数，若在run/run_async时传入，会覆盖初始化时设置的tools——这意味着你可以针对不同轮次的对话动态更换工具集，而不必重建组件。run_async是异步版本，适用于高并发或与异步 Pipeline 集成的场景。

4.3 函数调用完整流程

参考文档给出了一个完整的 Function Calling 示例：定义一个天气查询函数，转成Tool，让模型决定何时调用、携带什么参数，再由代码真正执行工具，最后把工具结果回传给模型生成最终答案。

第一步：定义函数并转为 Tool

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])

create_tool_from_function位于 haystack/tools/from_function.py，它会从函数签名（含Annotated类型注解）自动生成 OpenAI 风格的 JSON Schema 工具描述，Annotated字符串即为参数说明，会原样传递给模型作为参数语义提示。

第二步：模型准备工具调用

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（定义于 haystack/dataclasses/chat_message.py），包含tool_name、arguments（如{'unit': 'celsius', 'location': 'Berlin'}）与可选的id。

第三步：执行工具并把结果回传给模型

# 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（位于 haystack/components/tools/tool_invoker.py）自动遍历replies中的工具调用并执行，把执行结果转换为ChatMessage.from_tool类型的tool_messages。随后把「用户消息 + 模型回复（含 ToolCall）+ 工具结果」拼接成完整对话历史再次调用模型，模型便能基于真实工具结果给出最终回答。

集成文档（googleaigeminichatgenerator.mdx）还展示了不使用ToolInvoker的手动循环写法：遍历replies[0].tool_calls，用tool.invoke(**tool_call.arguments)逐个执行并用ChatMessage.from_tool(tool_result=result, origin=tool_call)构造工具消息，效果等价。

五、流式输出：让 token 实时到达

两个组件都支持流式输出。集成参考文档指出：将回调函数传给streaming_callback初始化参数，组件会在每个新 token 到达时调用它。

from haystack.dataclasses.streaming_chunk import StreamingChunk def streaming_callback(chunk: StreamingChunk): print(chunk.content, end="", flush=True) gemini = GoogleAIGeminiGenerator( model="gemini-2.0-flash", api_key=Secret.from_env_var("GOOGLE_API_KEY"), streaming_callback=streaming_callback, )

回调参数StreamingChunk定义于 haystack/dataclasses/streaming_chunk.py，它是一个数据类，核心字段包括：

• content：当前 chunk 的文本内容；
• meta：与 chunk 相关的元数据字典；
• component_info：产生该 chunk 的组件名称与类型；
• index：内容块索引（配合流式工具调用时使用）；
• tool_calls/tool_call_result：流式场景下的工具调用增量信息；
• start：是否为新内容块的起始 chunk；
• finish_reason：生成结束原因（遵循stop、length、tool_calls、content_filter等约定）。

同一文件还提供了select_streaming_callback工具函数用于在同步/异步回调之间做选择，规则为：运行时回调优先于初始化回调；在异步上下文（run_async）中使用同步回调会告警，而在同步上下文中使用协程回调会直接抛错。这保证了流式回调在同步与异步执行路径中的行为一致性。

六、在 Pipeline 与 Agent 中使用

6.1 生成式组件接入 RAG Pipeline

集成文档（googleaigeminigenerator.mdx）给出了一个典型的 RAG 流水线：InMemoryBM25Retriever检索文档 →PromptBuilder组装带上下文的提示词 →GoogleAIGeminiGenerator生成回答：

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"}})

由于GoogleAIGeminiGenerator的输出replies是字符串列表，而PromptBuilder的输出是prompt字符串，两者可直接连接，构成"检索 → 提示词组装 → 生成"的完整链路。这里的GoogleAIGeminiGenerator在集成文档中被标注为"常用位置在PromptBuilder之后"。

6.2 聊天组件接入对话 Pipeline

聊天组件的典型位置在ChatPromptBuilder之后，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, ) 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, } } )

由于ChatPromptBuilder的输出类型与gemini.messages的输入类型都是ChatMessage列表，两者直接连接即可。

6.3 让 Agent 驱动函数调用循环

集成文档还展示了更省心的方式：把聊天生成器与工具一起交给Agent，由 Agent 自动完成"模型准备调用 → 执行工具 → 结果回传 → 直到得出最终答案"的完整循环，无需手动维护对话历史：

import os from haystack.components.agents import Agent from haystack.dataclasses import ChatMessage from haystack_integrations.components.generators.google_ai import ( GoogleAIGeminiChatGenerator, ) os.environ["GOOGLE_API_KEY"] = "<MY_API_KEY>" agent = Agent( chat_generator=GoogleAIGeminiChatGenerator(model="gemini-2.0-flash"), tools=[tool], ) result = agent.run( messages=[ChatMessage.from_user("What is the temperature in celsius in Berlin?")] ) print(result["last_message"].text)

Agent 机制位于 haystack/components/agents，它正是以ChatMessage+ToolCall的数据模型为底座，与GoogleAIGeminiChatGenerator天然兼容。

七、序列化与安全要点

• to_dict/from_dict：两个组件均实现了标准的序列化协议，可被Pipeline.dumps/Pipeline.loads用于 YAML 持久化；密钥基于Secret不会明文落盘。
• 密钥管理：优先使用GOOGLE_API_KEY环境变量（Secret.from_env_var），避免在代码或配置文件中硬编码；临时脚本可使用Secret.from_token。Secret的解析逻辑见 haystack/utils/auth.py。
• 安全设置：通过safety_settings字典（HarmCategory→HarmBlockThreshold）为生成内容设置安全过滤级别，生产环境建议显式配置。

八、弃用说明与迁移建议

重要提示：集成文档（googleaigeminigenerator.mdx 与 googleaigeminichatgenerator.mdx 均在文首标注了 Deprecation Notice）指出：

该集成使用已弃用的google-generativeaiSDK，该 SDK 将于 2025 年 8 月后失去支持。官方推荐迁移到新的GoogleGenAIChatGenerator（google-genai-haystack包，基于新的 Google Gen AI SDK）。

因此：

• 新项目应直接使用GoogleGenAIChatGenerator（对应文档 googlegenaichatgenerator.mdx），它通过google-genai-haystack包安装，支持gemini-2.5-flash、gemini-2.5-pro等更新模型，同时兼容 Gemini Developer API 与 Vertex AI API（后者可切换api="vertex"并配置项目与区域）。
• 存量项目若已在用GoogleAIGeminiGenerator/GoogleAIGeminiChatGenerator，应制定迁移计划，逐步替换为GoogleGenAIChatGenerator，两者的ChatMessage交互模型与函数调用写法基本一致，迁移成本可控。

结语

GoogleAIGeminiGenerator与GoogleAIGeminiChatGenerator是 Haystack 生态中接入 Gemini 能力的两把钥匙：前者以异构parts输入覆盖文本与多模态推理，适合 RAG 中的单次生成；后者以ChatMessage为对话载体，配合tools参数与ToolInvoker实现完整的函数调用闭环，并可无缝嵌入Pipeline与Agent。掌握这两个组件的参数语义、序列化契约与流式回调机制，你就能把 Gemini 的能力以 Haystack 组件化的方式编排进生产级 LLM 应用；同时务必留意其底层 SDK 已弃用的现状，新项目优先选用官方推荐的GoogleGenAIChatGenerator迁移路径。

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