Haystack × Google AIGemini 生成与多轮对话组件集成实战指南【免费下载链接】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_ai.md 为核心系统讲解如何通过google-ai-haystack集成包在 Haystack 中接入 Google Gemini 模型既可以用GoogleAIGeminiGenerator完成文本与多模态生成也可以用GoogleAIGeminiChatGenerator完成多轮对话与函数调用并将它们嵌入 Pipeline、Agent 与流式输出场景。读完本文你将掌握两个组件的完整参数语义、消息与字节流数据模型、工具调用闭环以及从安装认证到序列化部署的全链路用法。集成概览两个组件如何分工google-ai-haystack是 Haystack 官方维护的 Google AI 集成包向 Haystack 生态提供两个核心生成组件对应参考文档中的两个模块generators.google_ai.gemini与generators.google_ai.chat.gemini组件所属模块定位输入输出GoogleAIGeminiGenerator...generators.google_ai.gemini通过 Google AI Studio 使用多模态 Gemini 模型生成文本parts字符串、ByteStream、Part的异构列表replieslist[str]生成回复GoogleAIGeminiChatGenerator...generators.google_ai.chat.gemini通过 Google AI Studio 完成聊天补全messageslist[ChatMessage]replieslist[ChatMessage]简单地说需要一次性生成如 RAG 问答、内容摘要用GoogleAIGeminiGenerator需要多轮对话状态管理、函数调用Tool Calling用GoogleAIGeminiChatGenerator。前者以Haystack的ByteStream与Part承载文本、图片等异构内容后者以 Haystack 统一消息模型 ChatMessage 承载对话历史。从使用指南googleaigeminigenerator.mdx看组件支持gemini-2.5-pro-exp-03-25、gemini-2.0-flash、gemini-1.5-pro、gemini-1.5-flash等模型参考文档中两个组件的默认模型均为gemini-2.0-flash。可用的模型清单以 Google 官方 Gemini API 文档为准。安装与认证准备安装集成包两个组件都位于google-ai-haystack包内先安装它pip install google-ai-haystack获取并配置 API Key两个组件初始化时的默认api_key均指向环境变量GOOGLE_API_KEYapi_key: Secret Secret.from_env_var(GOOGLE_API_KEY)也就是说你可以直接用环境变量完成认证推荐避免把密钥写进代码或序列化文件import os os.environ[GOOGLE_API_KEY] MY_API_KEY也可以在构造时显式传入from haystack.utils import Secret api_keySecret.from_token(MY_API_KEY)Secret是 Haystack 统一的密钥封装定义于 haystack/utils/auth.pySecret.from_token()创建基于 Token 的密钥不可序列化避免泄露Secret.from_env_var(GOOGLE_API_KEY)创建基于环境变量的密钥支持传入多个候选环境变量并按顺序解析第一个已设置的变量strictTrue时若全部缺失会抛出异常。这保证了同一个组件在本地开发、CI 与生产环境间迁移时无需改动代码密钥只存在于运行时环境。API Key 的申请入口为 Google AI Studio原 MakerSuite控制台申请后妥善保管不要提交到版本库。GoogleAIGeminiGenerator文本与多模态生成基础文本生成GoogleAIGeminiGenerator以parts为输入一次性把提示词交给 Gemini 模型生成回复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()的输出是一个字典键replies对应list[str]——模型可能返回多条候选回复因此用循环遍历打印。参考文档在 googleaigeminigenerator.mdx 中还给出了使用modelgemini-1.5-pro的同类示例其中模型会输出结构化的编号列表如关于费米悖论、意识起源、暗物质等话题的十条回答。多模态输入基于 ByteStreamGoogleAIGeminiGenerator的一大亮点是原生多模态输入parts可以同时混入字符串与图片以及音频、视频等二进制内容。图片以 Haystack 的ByteStream数据类承载定义于 haystack/dataclasses/byte_stream.py它包含data: bytes二进制数据、mime_type媒体类型与meta元数据三个字段。参考文档给出的多模态示例用requests从 URL 下载 4 张机器人图片包装成带image/jpegMIME 类型的ByteStream再与文本提示词一起交给模型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)注意parts是变长参数Variadic*images会把图片列表展开为多个位置参数与文本提示词混排。配套使用指南googleaigeminigenerator.mdx展示了该能力的典型效果模型能逐张识别出图片中的机器人分别对应《星球大战》的 C-3PO 与 R2-D2、《大都会》的 Maria、《地球停转之日》的 Gort 以及《银河系漫游指南》的 Marvin。除str与ByteStream外parts还接受Part对象。ByteStream还提供了from_file_path()与from_string()工厂方法见 byte_stream.py前者可从本地文件直接构造二进制流更适合把本地图片、PDF 等文件送入 Gemini。GoogleAIGeminiGenerator 参数与运行接口初始化参数构造函数签名源自参考文档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 密钥建议用环境变量方式注入modelstrgemini-2.0-flash模型名称可用的模型清单参见 Google Gemini API 模型文档generation_configGenerationConfig或dictNone生成配置可传入对象或参数字典如温度、max_output_tokens、top_p等参数范围见 Google 生成内容 API 参考safety_settingsdict[HarmCategory, HarmBlockThreshold]None安全设置键为危害类别、值为拦截阈值用于控制模型对特定内容类别的过滤强度streaming_callbackCallable[[StreamingChunk], None]None流式回调每当流式通道收到一个新 token 时被调用参数为StreamingChunkgeneration_config的灵活性在于既可以传官方的GenerationConfig对象也可以直接传普通字典底层会完成参数映射方便在 YAML 序列化场景下使用纯数据表达生成参数。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与初始化参数同名若在run时传入会覆盖构造时的回调用于实现按 token 粒度消费流式输出。返回字典仅含一个键replies值为list[str]即模型生成的全部回复文本。序列化to_dict 与 from_dictdef to_dict() - dict[str, Any] classmethod def from_dict(cls, data: dict[str, Any]) - GoogleAIGeminiGeneratorto_dict()将组件序列化为字典from_dict()从字典还原组件实例。这是 Haystack Pipeline 序列化协议的一部分——组件在保存/加载 YAML 管道时通过这两个方法往返配合Secret的环境变量模式密钥本身不会进入序列化数据从而保证管道文件可安全入库与分发。GoogleAIGeminiChatGenerator多轮对话当需要多轮对话时GoogleAIGeminiChatGenerator通过 Haystack 的ChatMessage数据类与模型交互参考文档明确注明这一点。ChatMessage 消息模型ChatMessage定义于 haystack/dataclasses/chat_message.py是 Haystack 统一的消息载体表示一次 LLM 对话中的一条消息通过类方法构造ChatMessage.from_user(text)用户消息ChatMessage.from_assistant(...)助手模型回复ChatMessage.from_system(...)系统提示ChatMessage.from_tool(tool_result, origin)工具调用结果。从源码看chat_message.pyChatMessage通过texts/text访问文本内容通过tool_calls/tool_call访问模型发起的工具调用通过tool_call_results/tool_call_result访问工具执行结果——这些属性正是下面函数调用闭环的数据基础。基础对话循环参考文档给出的多轮对话示例先发一条用户消息拿到回复再把回复与新的用户消息拼接后继续追问实现上下文累积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)注意两点第一replies中的元素是ChatMessage打印内容用reply.text属性返回第一条文本见 chat_message.py第二每次调用都需要把完整历史messages传回模型才能记住前文。配套使用指南googleaigeminichatgenerator.mdx给出了利用GOOGLE_API_KEY环境变量、省略显式api_key的等价写法并演示了推荐一部电影 → 询问主演的两轮对话。函数调用Tool CallingGoogleAIGeminiChatGenerator支持函数调用模型可以准备调用返回ToolCall由你的代码真正执行再把结果回传模型据此生成最终答案。定义工具先用create_tool_from_function把普通 Python 函数包装成 HaystackTool函数定义见 haystack/tools/from_function.py通过Annotated元数据为参数生成 JSON Schema 描述from typing import Annotated from haystack.tools import create_tool_from_function # 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传入构造参数tools让模型可以发起该函数的调用from haystack.dataclasses.chat_message import ChatMessage from haystack.components.tools import ToolInvoker 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) # 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)整个闭环分三步模型准备调用gemini_chat.run(messagesuser_message)返回的回复中携带tool_calls从源码看ChatMessage.tool_calls属性返回消息内所有ToolCall见 chat_message.py执行工具ToolInvoker示例导入自haystack.components.tools实际可用路径以你所安装的 Haystack 版本为准接收模型回复调用真实函数并把结果包装为tool_messages回传结果把用户消息、模型回复、工具结果拼接后再次调用gemini_chat.run()模型基于工具输出生成可读的最终答案。配套使用指南还演示了不依赖ToolInvoker、手动执行调用的等价写法googleaigeminichatgenerator.mdx遍历replies[0].tool_calls用tool.invoke(**tool_call.arguments)执行并用ChatMessage.from_tool(tool_resultresult, origintool_call)构造工具结果消息。示例输出显示模型会正确解析出locationBerlin、unitcelsius两个参数。运行时动态传入工具run()方法的tools参数可以在每次调用时覆盖初始化时的工具列表参考文档明确说明If set, it will override the tools parameter set during component initialization这为实现按轮次动态切换可用工具提供了可能例如根据用户意图在运行时决定开放哪些工具。与 Agent 集成自动化的工具循环手动驱动工具调用闭环在多轮、多工具场景下会比较繁琐更推荐把生成器交给 Haystack 的Agent源码位于 haystack/components/agents/agent.py。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) # The temperature in Berlin is 20 degrees Celsius.在 Pipeline 中使用两个组件都是标准 Haystack 组件带component.output_types装饰器可以无缝接入 Pipeline。GoogleAIGeminiGenerator 构建 RAG 管道参考使用指南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}})GoogleAIGeminiChatGenerator 构建对话管道聊天场景中GoogleAIGeminiChatGenerator通常位于ChatPromptBuilder之后见 googleaigeminichatgenerator.mdxChatPromptBuilder渲染的prompt直接连接到gemini.messagesimport 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)该示例的运行结果是一段关于罗马历史的编年体回答从前 753 年建城到西罗马帝国灭亡验证了模板变量 → 提示词渲染 → Gemini 生成的完整链路。流式输出streaming_callback 与 StreamingChunk两个组件都支持流式输出。把回调函数传给streaming_callback构造时或run时均可每当流式通道收到新 token回调就会被调用参数是StreamingChunk。StreamingChunk是 Haystack 统一的流式块数据类定义于 haystack/dataclasses/streaming_chunk.py字段包括content本次流式块承载的文本片段meta与块相关的元数据字典component_info产生该块的组件信息名称与类型index内容块序号tool_calls与块关联的ToolCallDelta列表流式场景下工具调用参数会增量到达finish_reason生成结束原因stop、length、tool_calls、content_filter、tool_call_results等见 streaming_chunk.py。典型用法是把每次到达的chunk.content实时追加渲染实现边生成边显示的类 ChatGPT 体验。异步支持run_asyncGoogleAIGeminiChatGenerator额外提供异步版本run_asyncasync def签名与run完全一致component.output_types(replieslist[ChatMessage]) async def run_async(messages: list[ChatMessage], streaming_callback: Optional[StreamingCallbackT] None, *, tools: Optional[list[Tool]] None)参数语义与run相同messages为输入对话消息列表streaming_callback处理流式 tokentools覆盖构造时的工具列表。在需要高并发吞吐如异步 Web 服务中并行处理多个会话的场景下应使用run_async配合asyncio运行避免阻塞事件循环。这也与 Haystack 主库整体对异步 Pipeline 的支持保持一致主仓库中haystack/components/agents/agent.py同样提供异步能力。GoogleAIGeminiChatGenerator 参数速查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全部参数之外聊天生成器额外提供两个工具相关参数tools可让模型准备调用的工具列表配合上面函数调用一节使用tool_config工具调用配置ToolConfig字典用于控制工具调用的启用方式与强制程度例如是否强制调用某个函数。to_dict()/from_dict()的序列化语义与生成器组件一致保证组件可被 Haystack Pipeline 的 YAML 序列化体系完整往返。弃用状态与迁移建议需要特别留意的是仓库内配套使用指南googleaigeminigenerator.mdx 与 googleaigeminichatgenerator.mdx均在文档开头明确标注了弃用警告该集成使用已弃用的 google-generativeai SDK该 SDK 在 2025 年 8 月之后将不再获得支持。官方建议迁移到新的GoogleGenAIChatGenerator集成。因此在规划新项目时应优先评估迁移到新的 Google GenAI 集成已有基于google-ai-haystack的项目应制定升级计划。本文档所述的 API 细节适用于仍在使用的存量系统可作为迁移期间的理解依据。参考资源以下仓库内文件可供继续深入集成参考文档docs-website/reference/integrations-api/google_ai.md生成器使用指南googleaigeminigenerator.mdx聊天生成器使用指南googleaigeminichatgenerator.mdx消息数据类haystack/dataclasses/chat_message.py二进制流数据类haystack/dataclasses/byte_stream.py流式块数据类haystack/dataclasses/streaming_chunk.py密钥管理haystack/utils/auth.py函数转工具haystack/tools/from_function.pyAgent 组件haystack/components/agents/agent.py【免费下载链接】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），仅供参考