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包中组件模块核心职责运行输入输出键GoogleAIGeminiGeneratorgoogle_ai.gemini使用多模态 Gemini 模型生成文本parts变长参数字符串 /ByteStream/Partreplies: list[str]GoogleAIGeminiChatGeneratorgoogle_ai.chat.gemini使用 Gemini 模型完成聊天补全messages: list[ChatMessage]replies: list[ChatMessage]两者都要求 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-haystackGoogle 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(modelgemini-2.0-flash) # 方式二显式传入 Secret从环境变量读取 gemini GoogleAIGeminiGenerator( modelgemini-2.0-flash, api_keySecret.from_env_var(GOOGLE_API_KEY), ) # 方式三显式传入密钥字符串 gemini GoogleAIGeminiGenerator( modelgemini-2.0-flash, api_keySecret.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)参数类型默认值说明api_keySecretGOOGLE_API_KEY环境变量Google AI Studio API Keymodelstrgemini-2.0-flash使用的模型名如gemini-2.0-flash、gemini-1.5-pro等可用模型以官方模型列表为准generation_configGenerationConfig或dict[str, Any]None生成配置可传对象或参数字典如温度、最大输出 token 等safety_settingsdict[HarmCategory, HarmBlockThreshold]None安全设置键为HarmCategory、值为HarmBlockThreshold的字典streaming_callbackCallable[[StreamingChunk], None]None流式回调每收到一个新 token 时被调用参数为StreamingChunk其中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(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)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(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)要点用列表解包*images把多张图片与提示文本拼接进parts为每个ByteStream显式指定mime_typeimage/jpeg帮助模型正确解析内容类型也可以使用ByteStream.from_file_path(...)从本地文件直接构造二进制流支持guess_mime_typeTrue自动推断 MIME 类型。3.4 run 方法与输出component.output_types(replieslist[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相比新增两个与工具调用相关的参数参数类型说明toolslist[Tool]模型可以为其准备调用function calling的工具列表tool_configToolConfigDict工具调用配置控制模型何时调用工具、调用哪些工具其余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(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)多轮对话的关键在于每一轮都要把历史消息完整传回用户消息 模型回复 新问题模型才能基于完整上下文继续作答。4.3 函数调用Function CallingGemini 可以准备工具调用再由你的代码真正执行工具。完整流程分三步第一步定义函数并转换为Tool。使用Annotated类型注解为参数提供描述再通过create_tool_from_function转换该函数定义在 haystack/tools/from_function.py会基于函数签名、类型注解与 docstring 自动生成工具参数 JSON SchemaTool数据类的字段定义见 haystack/tools/tool.pyfrom 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])第二步把工具传给生成器并让模型准备调用。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对象列表包含tool_name与arguments定义见 haystack/dataclasses/chat_message.py而不是最终答案。第三步实际调用工具并让模型总结结果。# 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负责真正执行工具并生成tool角色的消息把用户消息 模型工具调用 工具执行结果拼接后再次送入生成器模型即可基于真实结果输出最终答案。4.4 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)messagesChatMessage实例列表代表输入对话历史streaming_callback运行时覆盖流式回调tools运行时覆盖初始化时设置的工具列表——注意它是关键字参数*之后且优先级高于构造时的tools返回字典含replies键模型生成的ChatMessage列表run_async是run的异步版本供Pipeline.run_async或高并发场景使用。五、接入 PipelineRAG 与聊天流水线5.1 RAG 检索增强生成流水线将GoogleAIGeminiGenerator接到检索器与提示模板之后即构成一条完整 RAG 链路用例出自 googleaigeminigenerator.mdximport 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}})5.2 聊天流水线用ChatPromptBuilder组织模板消息再连接GoogleAIGeminiChatGenerator用例出自 googleaigeminichatgenerator.mdximport 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 dont 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 或存入版本库后还原方法签名说明to_dictdef to_dict() - dict[str, Any]将组件序列化为字典供 Pipeline 持久化from_dictclassmethod def from_dict(cls, data: dict[str, Any]) - GoogleAIGemini( Chat)Generator从字典反序列化重建组件实例与所有 Haystack 组件一致这两个方法配合 Haystack 的Pipeline序列化机制Pipeline.dumps/Pipeline.loads即可实现完整流水线的导出与导入。需要注意的是序列化时Secret类型的 API Key 会以安全的引用形式存储指向环境变量名而不是把明文密钥写入序列化产物。七、流式输出实践两个组件都支持流式输出把回调函数传给streaming_callback后模型生成的每个 token 会以StreamingChunk的形式被实时推送给回调而不是等完整回复生成后才返回。典型写法def on_token(chunk: StreamingChunk) - None: print(chunk.content, end, flushTrue) gemini GoogleAIGeminiGenerator( modelgemini-2.0-flash, api_keySecret.from_env_var(GOOGLE_API_KEY), streaming_callbackon_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 指南。九、快速决策速查表需求选择关键参数 / 输出一次问答、图文/音视频混输GoogleAIGeminiGeneratorrun(parts[...])→replies: list[str]多轮对话、上下文记忆GoogleAIGeminiChatGeneratorrun(messages[...])→replies: list[ChatMessage]让模型调用你的函数GoogleAIGeminiChatGeneratortools[tool]先取replies[0].tool_calls执行后回填再生成高并发 / 异步流水线GoogleAIGeminiChatGenerator.run_asyncawait run_async(messages...)实时逐 token 输出任一组件设置streaming_callbackRAG 链路GoogleAIGeminiGenerator接在PromptBuilder之后pipe.connect(prompt_builder, 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考