文档API 参考📓 教程🧑‍🍳 食谱🤝 集成💜 Discord🎨 Studio
API 参考

Google AI

Google AI integration for Haystack

Module haystack_integrations.components.generators.google_ai.gemini

GoogleAIGeminiGenerator

Generates text using multimodal Gemini models through Google AI Studio.

使用示例

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)

多模态示例

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)

GoogleAIGeminiGenerator.__init__

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)

Initializes aGoogleAIGeminiGenerator instance.

To get an API key, visit: https://makersuite.google.com

参数:

  • api_key: Google AI Studio API key.
  • model: Name of the model to use. For available models, see https://ai.google.dev/gemini-api/docs/models/gemini
  • generation_config: The generation configuration to use. This can either be aGenerationConfig object or a dictionary of parameters. For available parameters, see the GenerationConfig API reference.
  • safety_settings: The safety settings to use. A dictionary withHarmCategory as keys andHarmBlockThreshold as values. For more information, see the API reference
  • streaming_callback: 当从流中接收到新 token 时调用的回调函数。回调函数接受 StreamingChunk 作为参数。

GoogleAIGeminiGenerator.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

GoogleAIGeminiGenerator.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "GoogleAIGeminiGenerator"

从字典反序列化组件。

参数:

  • data: 要反序列化的字典。

返回值:

反序列化后的组件。

GoogleAIGeminiGenerator.run

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

Generates text based on the given input parts.

参数:

  • parts: A heterogeneous list of strings,ByteStream orPart objects.
  • streaming_callback: 当从流中接收到新 token 时调用的回调函数。

返回值:

A dictionary containing the following key

  • replies: A list of strings containing the generated responses.

Module haystack_integrations.components.generators.google_ai.chat.gemini

GoogleAIGeminiChatGenerator

Completes chats using Gemini models through Google AI Studio.

It uses the ChatMessage dataclass to interact with the model.

使用示例

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

With function calling

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

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)

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

GoogleAIGeminiChatGenerator.__init__

def __init__(*,
             api_key: Secret = Secret.from_env_var("GOOGLE_API_KEY"),
             model: str = "gemini-1.5-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)

Initializes aGoogleAIGeminiChatGenerator instance.

To get an API key, visit: https://aistudio.google.com/

参数:

  • api_key: Google AI Studio API key. To get a key, see Google AI Studio.
  • model: Name of the model to use. For available models, see https://ai.google.dev/gemini-api/docs/models/gemini.
  • generation_config: The generation configuration to use. This can either be aGenerationConfig object or a dictionary of parameters. For available parameters, see the API reference.
  • safety_settings: The safety settings to use. A dictionary withHarmCategory as keys andHarmBlockThreshold as values. For more information, see the API reference
  • tools：模型可以准备调用的工具列表。
  • tool_config: The tool config to use. See the documentation for ToolConfig.
  • streaming_callback: 当从流中接收到新 token 时调用的回调函数。回调函数接受 StreamingChunk 作为参数。

GoogleAIGeminiChatGenerator.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

GoogleAIGeminiChatGenerator.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "GoogleAIGeminiChatGenerator"

从字典反序列化组件。

参数:

  • data: 要反序列化的字典。

返回值:

反序列化后的组件。

GoogleAIGeminiChatGenerator.run

@component.output_types(replies=List[ChatMessage])
def run(messages: List[ChatMessage],
        streaming_callback: Optional[StreamingCallbackT] = None,
        *,
        tools: Optional[List[Tool]] = None)

Generates text based on the provided messages.

参数:

  • messages：一组ChatMessage instances, representing the input messages.
  • streaming_callback: 当从流中接收到新 token 时调用的回调函数。
  • tools：模型可以准备调用的工具列表。如果设置，它将覆盖tools 参数。

返回值:

A dictionary containing the following key

  • replies: A list containing the generated responses asChatMessage instances.

GoogleAIGeminiChatGenerator.run_async

@component.output_types(replies=List[ChatMessage])
async def run_async(messages: List[ChatMessage],
                    streaming_callback: Optional[StreamingCallbackT] = None,
                    *,
                    tools: Optional[List[Tool]] = None)

Async version of the run method. Generates text based on the provided messages.

参数:

  • messages：一组ChatMessage instances, representing the input messages.
  • streaming_callback: 当从流中接收到新 token 时调用的回调函数。
  • tools：模型可以准备调用的工具列表。如果设置，它将覆盖tools 参数。

返回值:

A dictionary containing the following key

  • replies: A list containing the generated responses asChatMessage instances.