Haystack 集成 Google GenAI：从 Gemini 文本/多模态嵌入到带思考能力的 ChatGenerator 与 Token 计数完整实战指南

Haystack 集成 Google GenAI：从 Gemini 文本/多模态嵌入到带思考能力的 ChatGenerator 与 Token 计数完整实战指南

【免费下载链接】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 GenAI 集成参考文档 为核心骨架，系统讲解google-genai-haystack集成在 Haystack 中的全部组件：三类 Embedder（文本文档、多模态文档、查询文本）、支持思考能力与工具调用的GoogleGenAIChatGenerator，以及基于countTokens接口的GoogleGenAITokenCounter。读完本文，你将掌握从三种认证方式的选择、嵌入参数调优、多模态文件嵌入的路径安全配置，到组装完整 RAG/Agent 管线的全部实战细节。

一、集成概览：一套包，五类能力

google-genai-haystack是基于 Google 官方 Gen AI SDK（python-genai）构建的 Haystack 集成包，兼容 Gemini Developer API 与 Vertex AI 两套后端。参考文档将其能力划分为五个模块，全部以haystack_integrations命名空间暴露：

安装方式统一为：

pip install google-genai-haystack

从组件关系上看，三类 Embedder 负责把"文档"与"查询"映射到同一向量空间（这是向量检索成立的前提），ChatGenerator 负责生成环节，TokenCounter 则服务于 Agent 上下文管理与成本预估。配套的分组件使用指南可在 embedders 文档目录 与 generators 文档目录 中找到。

二、认证方式：三套配置覆盖两种后端

参考文档中所有组件都共享同一套认证设计：api_key参数默认从GOOGLE_API_KEY或GEMINI_API_KEY环境变量读取（strict=False，两个变量都未设置时才报错），并通过api参数在"gemini"（Gemini Developer API）与"vertex"（Vertex AI）之间切换。三种写法如下（以 DocumentEmbedder 为例，其余组件完全同构）：

1. Gemini Developer API（API Key 认证）

from haystack_integrations.components.embedders.google_genai import GoogleGenAIDocumentEmbedder # export the environment variable (GOOGLE_API_KEY or GEMINI_API_KEY) document_embedder = GoogleGenAIDocumentEmbedder(model="gemini-embedding-001")

2. Vertex AI（Application Default Credentials）

from haystack_integrations.components.embedders.google_genai import GoogleGenAIDocumentEmbedder # Using Application Default Credentials (requires gcloud auth setup) document_embedder = GoogleGenAIDocumentEmbedder( api="vertex", vertex_ai_project="my-project", vertex_ai_location="us-central1", model="gemini-embedding-001" )

3. Vertex AI（API Key 认证）

from haystack_integrations.components.embedders.google_genai import GoogleGenAIDocumentEmbedder # export the environment variable (GOOGLE_API_KEY or GEMINI_API_KEY) document_embedder = GoogleGenAIDocumentEmbedder( api="vertex", model="gemini-embedding-001" )

认证机制与 Secret 管理

这套设计背后的核心抽象是 Haystack 的Secret类（实现见 haystack/utils/auth.py）：

• Secret.from_env_var(["GOOGLE_API_KEY", "GEMINI_API_KEY"], strict=False)：按顺序解析第一个已设置的环境变量；strict=False表示变量缺失时不抛异常，留给运行期报错。
• Secret.from_token("<your-api-key>")：直接把密钥写进初始化参数（文档明确标注 token 型 Secret 不可序列化，源码说明），适合交互式脚本；生产环境仍推荐环境变量方式。
• Vertex AI 的 ADC（Application Default Credentials）模式不需要 API Key，但必须显式传入vertex_ai_project与vertex_ai_location（如"us-central1"、"europe-west1"），并提前完成 gcloud 认证配置。

三、GoogleGenAIDocumentEmbedder：文本文档批量向量化

该组件"使用 Google AI 模型计算文档嵌入"，输出会被写入每个Document的embedding字段——这正是 Document 数据类 中embedding: list[float]字段的用途。它通常位于索引管线的DocumentWriter之前。

完整初始化签名

__init__( *, api_key: Secret = Secret.from_env_var( ["GOOGLE_API_KEY", "GEMINI_API_KEY"], strict=False ), api: Literal["gemini", "vertex"] = "gemini", vertex_ai_project: str | None = None, vertex_ai_location: str | None = None, model: str = "gemini-embedding-001", prefix: str = "", suffix: str = "", batch_size: int = 32, progress_bar: bool = True, meta_fields_to_embed: list[str] | None = None, embedding_separator: str = "\n", config: dict[str, Any] | None = None, timeout: float | None = None, max_retries: int | None = None ) -> None

参数详解

使用示例

from haystack import Document from haystack_integrations.components.embedders.google_genai import GoogleGenAIDocumentEmbedder doc = Document(content="I love pizza!") document_embedder = GoogleGenAIDocumentEmbedder() result = document_embedder.run([doc]) print(result['documents'][0].embedding) # [0.017020374536514282, -0.023255806416273117, ...]

run(documents)返回字典含两个键：documents（已携带嵌入向量的文档列表）与meta（模型用量信息，如 token 数）。

进阶：把语义化元数据一并嵌入

文档自带元数据如果语义鲜明，可以一并编码以提升检索质量，参考 GoogleGenAIDocumentEmbedder 组件指南：

from haystack import Document from haystack.utils import Secret from haystack_integrations.components.embedders.google_genai import GoogleGenAIDocumentEmbedder doc = Document(content="some text", meta={"title": "relevant title", "page number": 18}) embedder = GoogleGenAIDocumentEmbedder( api_key=Secret.from_token("<your-api-key>"), meta_fields_to_embed=["title"], ) docs_w_embeddings = embedder.run(documents=[doc])["documents"]

拼接规则为：content + embedding_separator + meta 字段值，默认分隔符是换行符"\n"，可通过embedding_separator调整。

四、GoogleGenAITextEmbedder：查询字符串向量化

与 DocumentEmbedder 对应，GoogleGenAITextEmbedder"使用 Google AI 模型嵌入字符串"，典型场景是查询侧：把用户问题编码后交给嵌入检索器（Embedding Retriever）做相似度检索。

初始化签名与差异点

__init__( *, api_key: Secret = Secret.from_env_var( ["GOOGLE_API_KEY", "GEMINI_API_KEY"], strict=False ), api: Literal["gemini", "vertex"] = "gemini", vertex_ai_project: str | None = None, vertex_ai_location: str | None = None, model: str = "gemini-embedding-001", prefix: str = "", suffix: str = "", config: dict[str, Any] | None = None, timeout: float | None = None, max_retries: int | None = None ) -> None

它没有batch_size/progress_bar/meta_fields_to_embed等批处理参数（因为每次只嵌一个字符串），其余参数语义与 DocumentEmbedder 完全一致。

使用示例

from haystack_integrations.components.embedders.google_genai import GoogleGenAITextEmbedder text_to_embed = "I love pizza!" text_embedder = GoogleGenAITextEmbedder() print(text_embedder.run(text_to_embed)) # {'embedding': [0.017020374536514282, -0.023255806416273117, ...], # 'meta': {'model': 'gemini-embedding-001-v2', # 'usage': {'prompt_tokens': 4, 'total_tokens': 4}}}

run(text)返回embedding（输入文本的向量）与meta（模型名与 token 用量）。注意meta中的模型名会带上服务端实际使用的版本后缀（如gemini-embedding-001-v2），可作为确认所用模型版本的依据。

五、GoogleGenAIMultimodalDocumentEmbedder：图片/PDF/视频/音频统一嵌入

该组件是文档中唯一面向非文本内容的嵌入器：支持图片、PDF、视频与音频文件，并把它们映射到同一个向量空间，从而可以与文本嵌入直接做相似度比较。文本文档请用GoogleGenAIDocumentEmbedder，查询字符串请用GoogleGenAITextEmbedder。

初始化签名

__init__( *, api_key: Secret = Secret.from_env_var( ["GOOGLE_API_KEY", "GEMINI_API_KEY"], strict=False ), api: Literal["gemini", "vertex"] = "gemini", vertex_ai_project: str | None = None, vertex_ai_location: str | None = None, file_path_meta_field: str = "file_path", root_path: str | None = None, image_size: tuple[int, int] | None = None, model: str = "gemini-embedding-2", batch_size: int = 6, progress_bar: bool = True, config: dict[str, Any] | None = None, timeout: float | None = None, max_retries: int | None = None ) -> None

参数详解

其余api_key/api/vertex_ai_project/vertex_ai_location/progress_bar/timeout/max_retries语义与前两类组件相同。

使用示例

from haystack import Document from haystack_integrations.components.embedders.google_genai import GoogleGenAIMultimodalDocumentEmbedder doc = Document(content=None, meta={"file_path": "path/to/image.jpg"}) document_embedder = GoogleGenAIMultimodalDocumentEmbedder() result = document_embedder.run([doc]) print(result['documents'][0].embedding) # [0.017020374536514282, -0.023255806416273117, ...]

注意Document的content为None，文件路径放在meta["file_path"]（可通过file_path_meta_field改名）。这与 Document 数据类 中"content可为 None、meta承载扩展信息"的设计一致。

异常语义（值得注意的健壮性设计）

参考文档为run与run_async明确列出了三类异常，便于调用方捕获处理：

• TypeError：输入不是Document列表；
• ValueError：文档缺失文件路径元数据字段、文件路径越出root_path、或 MIME 类型不受支持；
• RuntimeError：部分文档转换失败。

其中root_path的路径越界校验是安全相关的关键设计——当文档元数据可能来自不可信来源（如外部导入的文档集）时，设置root_path可有效防止路径遍历攻击。

六、GoogleGenAIChatGenerator：Gemini 对话生成与思考能力

GoogleGenAIChatGenerator通过 Google Gen AI SDK 完成 Gemini 对话补全，是集成中功能最丰富的组件：支持思考（thinking）、工具调用、结构化输出、流式与多模态输入。

支持的模型

SUPPORTED_MODELS: list[str] = [ "gemini-3.7-flash", "gemini-3.6-flash", "gemini-3.5-flash-lite", "gemini-3.1-pro-preview", "gemini-3-flash-preview", "gemini-3.1-flash-lite-preview", "gemini-2.5-pro", "gemini-2.5-flash", "gemini-2.5-flash-lite", ]

该列表是非穷尽的（参考文档原文即注明 "A non-exhaustive list"），更完整的模型 ID 以 Google 官方模型文档为准。

思考能力（Thinking Support）

针对 Gemini 2.5 与 Gemini 3 系列模型，组件开放了思考配置：

• 推理透明化：模型可以展示推理过程；
• 思维签名（Thought Signatures）：多轮对话配合工具时，通过加密的"保存状态"维持跨轮次的思维上下文——因此使用工具时应把之前的助手回复保留在聊天历史中；
• 可配置思考预算：精确控制分配给推理的 token。

通过generation_kwargs={"thinking_budget": value}配置：

Gemini 3 系列及更新模型还支持thinking_level（思考深度），可选minimal（接近"无思考"，仅复杂编码任务会有极少量思考，延迟最低）、low（适合简单指令遵循、对话与高吞吐场景）、medium（多数任务的均衡档）、high（默认动态档，推理最深但首 token 延迟显著增大）。

初始化签名

__init__( *, api_key: Secret = Secret.from_env_var( ["GOOGLE_API_KEY", "GEMINI_API_KEY"], strict=False ), api: Literal["gemini", "vertex"] = "gemini", vertex_ai_project: str | None = None, vertex_ai_location: str | None = None, model: str = "gemini-3.7-flash", generation_kwargs: dict[str, Any] | None = None, safety_settings: list[dict[str, Any]] | None = None, streaming_callback: StreamingCallbackT | None = None, tools: ToolsType | None = None, timeout: float | None = None, max_retries: int | None = None ) -> None

关键参数：generation_kwargs承载 temperature、max_tokens 以及上文的thinking_budget/thinking_level；safety_settings控制内容过滤；streaming_callback在收到新 token 时被调用；tools接受Tool和/或Toolset的列表或单个Toolset（每个工具名须唯一）；timeout/max_retries未设置时采用 Google Gen AI 客户端默认值。

使用示例：开启思考并读取推理内容

from haystack.dataclasses.chat_message import ChatMessage from haystack.tools import Tool, Toolset from haystack_integrations.components.generators.google_genai import GoogleGenAIChatGenerator # Initialize the chat generator with thinking support chat_generator = GoogleGenAIChatGenerator( model="gemini-3.7-flash", generation_kwargs={"thinking_budget": 1024} # Enable thinking with 1024 token budget ) # Generate a response messages = [ChatMessage.from_user("Tell me about the future of AI")] response = chat_generator.run(messages=messages) print(response["replies"][0].text) # Access reasoning content if available message = response["replies"][0] if message.reasonings: for reasoning in message.reasonings: print("Reasoning:", reasoning.reasoning_text)

replies中每条ChatMessage的reasonings字段承载推理内容——该字段在 Haystack 中由 ReasoningContent 内容类型 表示，其中reasoning_text即推理文本。

使用示例：带思考的工具调用

# Tool usage example with thinking def weather_function(city: str): return f"The weather in {city} is sunny and 25°C" weather_tool = Tool( name="weather", description="Get weather information for a city", parameters={"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}, function=weather_function ) # Can use either List[Tool] or Toolset chat_generator_with_tools = GoogleGenAIChatGenerator( model="gemini-3.7-flash", tools=[weather_tool], # or tools=Toolset([weather_tool]) generation_kwargs={"thinking_budget": -1} # Dynamic thinking allocation ) messages = [ChatMessage.from_user("What's the weather in Paris?")] response = chat_generator_with_tools.run(messages=messages)

组件指南（googlegenaichatgenerator.mdx）进一步说明tools参数支持混用：单个工具列表、单个 Toolset、以及"多个 Toolset + 独立工具"的混合列表，便于把相关工具逻辑分组。工具调用后的结果回合通常由 Haystack 的ToolInvoker执行（见 haystack/components/tools 目录）。

使用示例：结构化输出

from pydantic import BaseModel from haystack.dataclasses.chat_message import ChatMessage from haystack_integrations.components.generators.google_genai import GoogleGenAIChatGenerator class City(BaseModel): name: str country: str population: int chat_generator = GoogleGenAIChatGenerator( model="gemini-3.7-flash", generation_kwargs={"response_format": City} ) messages = [ChatMessage.from_user("Tell me about Paris")] response = chat_generator.run(messages=messages) print(response["replies"][0].text) # JSON output matching the City schema

通过generation_kwargs={"response_format": <Pydantic 模型>}直接约束输出为符合 schema 的 JSON。

使用示例：ChatMessage 内嵌 FileContent 多模态输入

from haystack.dataclasses import ChatMessage, FileContent from haystack_integrations.components.generators.google_genai import GoogleGenAIChatGenerator file_content = FileContent.from_url("https://arxiv.org/pdf/2309.08632") chat_message = ChatMessage.from_user(content_parts=[file_content, "Summarize this paper in 100 words."]) chat_generator = GoogleGenAIChatGenerator() response = chat_generator.run(messages=[chat_message])

FileContent是 haystack/dataclasses/file_content.py 中定义的内容类型，承载 base64 数据与 MIME 类型；from_url通过内置的LinkContentFetcher下载文件并自动推断 MIME 类型（实现见 file_content.py）。图片场景还可使用ImageContent.from_file_path（image_content.py），它支持缩放与格式校验。

run 方法签名与覆盖规则

run( messages: list[ChatMessage] | str, generation_kwargs: dict[str, Any] | None = None, safety_settings: list[dict[str, Any]] | None = None, streaming_callback: StreamingCallbackT | None = None, tools: ToolsType | None = None, ) -> dict[str, Any]

要点：

• messages也接受裸字符串，会自动包装成一条 user 角色的ChatMessage；
• generation_kwargs按key 粒度与初始化时的配置合并：run 传入的 key 优先，仅初始化设置的 key 保留；
• safety_settings、tools传入后会覆盖初始化时设置的对应值；
• 返回值仅含replies（生成的ChatMessage列表）；
• 异常：RuntimeError（生成过程出错）；ValueError（某条ChatMessage不包含TextContent/ToolCall/ToolCallResult中的任何一种，或角色不是 User/System/Assistant）。角色体系定义见 chat_message.py 中的 ChatRole。

run_async为完全对应的异步版本，支持await。

七、GoogleGenAITokenCounter：官方接口的精确 Token 计数

GoogleGenAITokenCounter调用 Google Gen AI SDK 的countTokens端点统计 Gemini 输入 token。与本地估算类计数器不同，它把消息原样发送到官方端点，因此计数包含 Gemini 对消息施加的模型专属格式化（formatting）。输入按GoogleGenAIChatGenerator的发送方式装配：首条 system 消息作为系统指令，其余消息作为请求内容。

初始化签名

__init__( model: str, *, api_key: Secret = Secret.from_env_var( ["GOOGLE_API_KEY", "GEMINI_API_KEY"], strict=False ), api: Literal["gemini", "vertex"] = "gemini", vertex_ai_project: str | None = None, vertex_ai_location: str | None = None, timeout: float | None = None, max_retries: int | None = None ) -> None

model是必填位置参数：token 计数与模型相关，应传入你实际用于生成的同一模型。

使用示例

from haystack.dataclasses import ChatMessage from haystack_integrations.token_counters.google_genai import GoogleGenAITokenCounter counter = GoogleGenAITokenCounter("gemini-3.7-flash") messages = [ChatMessage.from_user("Hello, how are you?")] token_count = counter.count(messages) print(f"Token count: {token_count}")

count(messages, tools=None)返回int（无可测内容时返回0）；传入tools时其 schema 也会计入 token。

后端差异：系统指令与工具的支持边界

这是参考文档明确强调的重要坑点：

• Google Gen AI SDK 只有在客户端指向Vertex AI时，才接受在countTokens上传系统指令与工具 schema；
• 在Gemini Developer API上，首条 system 消息会被当作 user 轮次统计（是近似值而非精确值）；传入工具则会直接抛出ValueError，而不是静默返回一个漏掉工具 schema 的计数；
• 仅统计普通消息（无系统指令、无工具）时两个后端都可用。

因此，如果你的 Agent 需要精确统计系统提示词或工具 schema 的 token，请使用api="vertex"。该注意事项在 GoogleGenAITokenCounter 组件文档 中亦有对应描述。

生命周期

warm_up()创建客户端（在应用启动阶段预创建可避免首次调用的延迟）；close()关闭客户端并释放底层 HTTP 资源；to_dict/from_dict支持序列化与反序列化。

八、实战组合：一条完整的 RAG 检索管线

把上述组件组装成可运行的 RAG 场景（示例出自 googlegenaidocumentembedder 组件指南，与 googlegenaitextembedder 组件指南）：

from haystack import Document from haystack import Pipeline from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack.components.writers import DocumentWriter from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever from haystack_integrations.components.embedders.google_genai import ( GoogleGenAITextEmbedder, GoogleGenAIDocumentEmbedder, ) document_store = InMemoryDocumentStore(embedding_similarity_function="cosine") documents = [ Document(content="My name is Wolfgang and I live in Berlin"), Document(content="I saw a black horse running"), Document(content="Germany has many big cities"), ] indexing_pipeline = Pipeline() indexing_pipeline.add_component("embedder", GoogleGenAIDocumentEmbedder()) indexing_pipeline.add_component("writer", DocumentWriter(document_store=document_store)) indexing_pipeline.connect("embedder", "writer") indexing_pipeline.run({"embedder": {"documents": documents}}) query_pipeline = Pipeline() query_pipeline.add_component("text_embedder", GoogleGenAITextEmbedder()) query_pipeline.add_component( "retriever", InMemoryEmbeddingRetriever(document_store=document_store), ) query_pipeline.connect("text_embedder.embedding", "retriever.query_embedding") query = "Who lives in Berlin?" result = query_pipeline.run({"text_embedder": {"text": query}}) print(result["retriever"]["documents"][0]) ## Document(id=..., content: 'My name is Wolfgang and I live in Berlin')

该示例中的关键配合点：

1. 索引侧：GoogleGenAIDocumentEmbedder产出带嵌入的文档 →DocumentWriter写入InMemoryDocumentStore（document_store.py）；
2. 查询侧：GoogleGenAITextEmbedder把查询编码为向量 →InMemoryEmbeddingRetriever（embedding_retriever.py）用余弦相似度召回最相关文档；
3. 两侧必须使用同一模型，才能保证文档向量与查询向量处于同一向量空间。

多模态场景同理：索引时用GoogleGenAIMultimodalDocumentEmbedder嵌入图片/PDF/视频/音频文件，查询时仍用GoogleGenAITextEmbedder编码文本查询，即可实现"文本查图、文本查视频"的跨模态检索。

九、序列化与生命周期管理

参考文档中所有组件都实现了统一的生命周期与序列化接口，便于在 Haystack 的 YAML/JSON 管线声明中使用：

其中to_dict/from_dict是 Haystack 组件可被Pipeline.dump/Pipeline.load持久化、可写入 YAML 管线的基石；而warm_up/close对应组件的资源生命周期管理（如释放 HTTP 连接池），在长运行服务与异步应用中尤其重要。各组件的run_async与run参数、返回值和异常语义保持一致，只是可被await。

结语

至此，google-genai-haystack集成的完整能力已全部覆盖：文本嵌入（GoogleGenAIDocumentEmbedder）、多模态嵌入（GoogleGenAIMultimodalDocumentEmbedder，含root_path路径安全与image_size缩放）、查询编码（GoogleGenAITextEmbedder）、带思考预算/思考深度/工具/结构化输出/多模态输入的GoogleGenAIChatGenerator，以及基于官方countTokens的GoogleGenAITokenCounter（注意其系统指令与工具统计的 Vertex AI 边界）。配合 Haystack 的Pipeline、InMemoryDocumentStore与InMemoryEmbeddingRetriever，即可快速构建生产级的 RAG 与 Agent 应用。进一步细节可查阅仓库中的 Google GenAI 参考文档 及对应的 Embedder 组件指南、ChatGenerator 组件指南。

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