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 集成由两个模块组成分别对应两种不同的模型交互方式组件所属模块交互范式输出类型GoogleAIGeminiGeneratorhaystack_integrations.components.generators.google_ai.gemini单次生成支持多模态 partsreplies: list[str]GoogleAIGeminiChatGeneratorhaystack_integrations.components.generators.google_ai.chat.gemini多轮对话 函数调用replies: list[ChatMessage]两者的共同点在于都通过 Google AI StudioGemini 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(modelgemini-2.0-flash, api_keySecret.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_keyGoogle 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(modelgemini-2.0-flash, api_keySecret.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(replieslist[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(datarequests.get(url).content, mime_typeimage/jpeg) for url in URLS ] gemini GoogleAIGeminiGenerator(modelgemini-2.0-flash, api_keySecret.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_typeimage/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(modelgemini-2.0-flash, api_keySecret.from_token(MY_API_KEY)) messages [ChatMessage.from_user(What is the most interesting thing you know?)] res gemini_chat.run(messagesmessages) for reply in res[replies]: print(reply.text) messages res[replies] [ChatMessage.from_user(Tell me more about it)] res gemini_chat.run(messagesmessages) for reply in res[replies]: print(reply.text)注意与生成式组件两个关键差异入参不同run接收的是messages: list[ChatMessage]而不是parts。出参不同run返回的replies是ChatMessage列表访问文本要用reply.text而非直接打印。run与run_async的完整签名参考文档component.output_types(replieslist[ChatMessage]) def run(messages: list[ChatMessage], streaming_callback: Optional[StreamingCallbackT] None, *, tools: Optional[list[Tool]] None) component.output_types(replieslist[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让模型决定何时调用、携带什么参数再由代码真正执行工具最后把工具结果回传给模型生成最终答案。第一步定义函数并转为 Toolfrom 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 fThe 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( modelgemini-2.0-flash-exp, api_keySecret.from_token(MY_API_KEY), 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)此时模型不会直接给出最终答案而是在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(messagesreplies)[tool_messages] messages user_message replies tool_messages # transform the tool call result into a human readable message final_replies gemini_chat.run(messagesmessages)[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_resultresult, origintool_call)构造工具消息效果等价。五、流式输出让 token 实时到达两个组件都支持流式输出。集成参考文档指出将回调函数传给streaming_callback初始化参数组件会在每个新 token 到达时调用它。from haystack.dataclasses.streaming_chunk import StreamingChunk def streaming_callback(chunk: StreamingChunk): print(chunk.content, end, flushTrue) gemini GoogleAIGeminiGenerator( modelgemini-2.0-flash, api_keySecret.from_env_var(GOOGLE_API_KEY), streaming_callbackstreaming_callback, )回调参数StreamingChunk定义于 haystack/dataclasses/streaming_chunk.py它是一个数据类核心字段包括content当前 chunk 的文本内容meta与 chunk 相关的元数据字典component_info产生该 chunk 的组件名称与类型index内容块索引配合流式工具调用时使用tool_calls/tool_call_result流式场景下的工具调用增量信息start是否为新内容块的起始 chunkfinish_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: Whats the official language of {{ country }}? pipe Pipeline() pipe.add_component(retriever, InMemoryBM25Retriever(document_storedocstore)) pipe.add_component(prompt_builder, PromptBuilder(templatetemplate)) pipe.add_component(gemini, GoogleAIGeminiGenerator(modelgemini-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_generatorGoogleAIGeminiChatGenerator(modelgemini-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它正是以ChatMessageToolCall的数据模型为底座与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 月后失去支持。官方推荐迁移到新的GoogleGenAIChatGeneratorgoogle-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后者可切换apivertex并配置项目与区域。存量项目若已在用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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考