from haystack_integrations.components.generators.google_vertex import VertexAIGeminiGenerator


gemini = VertexAIGeminiGenerator()
result = gemini.run(parts = ["What is the most interesting thing you know?"])
for answer in result["replies"]:
    print(answer)

>>> 1. **The Origin of Life:** How and where did life begin? The answers to this ...
>>> 2. **The Unseen Universe:** The vast majority of the universe is ...
>>> 3. **Quantum Entanglement:** This eerie phenomenon in quantum mechanics allows ...
>>> 4. **Time Dilation:** Einstein's theory of relativity revealed that time can ...
>>> 5. **The Fermi Paradox:** Despite the vastness of the universe and the ...
>>> 6. **Biological Evolution:** The idea that life evolves over time through natural ...
>>> 7. **Neuroplasticity:** The brain's ability to adapt and change throughout life, ...
>>> 8. **The Goldilocks Zone:** The concept of the habitable zone, or the Goldilocks zone, ...
>>> 9. **String Theory:** This theoretical framework in physics aims to unify all ...
>>> 10. **Consciousness:** The nature of human consciousness and how it arises ...

def __init__(*,
             model: str = "gemini-2.0-flash",
             project_id: Optional[str] = None,
             location: Optional[str] = None,
             generation_config: Optional[Union[GenerationConfig,
                                               Dict[str, Any]]] = None,
             safety_settings: Optional[Dict[HarmCategory,
                                            HarmBlockThreshold]] = None,
             system_instruction: Optional[Union[str, ByteStream, Part]] = None,
             streaming_callback: Optional[Callable[[StreamingChunk],
                                                   None]] = None)

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

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

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

import requests

from haystack.dataclasses.byte_stream import ByteStream
from haystack_integrations.components.generators.google_vertex import VertexAIImageCaptioner

captioner = VertexAIImageCaptioner()

image = ByteStream(
    data=requests.get(
        "https://raw.githubusercontent.com/deepset-ai/haystack-core-integrations/main/integrations/google_vertex/example_assets/robot1.jpg"
    ).content
)
result = captioner.run(image=image)

for caption in result["captions"]:
    print(caption)

>>> two gold robots are standing next to each other in the desert

def __init__(*,
             model: str = "imagetext",
             project_id: Optional[str] = None,
             location: Optional[str] = None,
             **kwargs)

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

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

@component.output_types(captions=List[str])
def run(image: ByteStream)

    from haystack_integrations.components.generators.google_vertex import VertexAICodeGenerator

    generator = VertexAICodeGenerator()

    result = generator.run(prefix="def to_json(data):")

    for answer in result["replies"]:
        print(answer)

    >>> ```python
    >>> import json
    >>>
    >>> def to_json(data):
    >>>   """Converts a Python object to a JSON string.
    >>>
    >>>   Args:
    >>>     data: The Python object to convert.
    >>>
    >>>   Returns:
    >>>     A JSON string representing the Python object.
    >>>   """
    >>>
    >>>   return json.dumps(data)
    >>> ```

def __init__(*,
             model: str = "code-bison",
             project_id: Optional[str] = None,
             location: Optional[str] = None,
             **kwargs)

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

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

@component.output_types(replies=List[str])
def run(prefix: str, suffix: Optional[str] = None)

from pathlib import Path

from haystack_integrations.components.generators.google_vertex import VertexAIImageGenerator

generator = VertexAIImageGenerator()
result = generator.run(prompt="Generate an image of a cute cat")
result["images"][0].to_file(Path("my_image.png"))

def __init__(*,
             model: str = "imagegeneration",
             project_id: Optional[str] = None,
             location: Optional[str] = None,
             **kwargs)

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

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

@component.output_types(images=List[ByteStream])
def run(prompt: str, negative_prompt: Optional[str] = None)

from haystack.dataclasses.byte_stream import ByteStream
from haystack_integrations.components.generators.google_vertex import VertexAIImageQA

qa = VertexAIImageQA()

image = ByteStream.from_file_path("dog.jpg")

res = qa.run(image=image, question="What color is this dog")

print(res["replies"][0])

>>> white

def __init__(*,
             model: str = "imagetext",
             project_id: Optional[str] = None,
             location: Optional[str] = None,
             **kwargs)

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

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

@component.output_types(replies=List[str])
def run(image: ByteStream, question: str)

    from haystack_integrations.components.generators.google_vertex import VertexAITextGenerator

    generator = VertexAITextGenerator()
    res = generator.run("Tell me a good interview question for a software engineer.")

    print(res["replies"][0])

    >>> **Question:**
    >>> You are given a list of integers and a target sum.
    >>> Find all unique combinations of numbers in the list that add up to the target sum.
    >>>
    >>> **Example:**
    >>>
    >>> ```
    >>> Input: [1, 2, 3, 4, 5], target = 7
    >>> Output: [[1, 2, 4], [3, 4]]
    >>> ```
    >>>
    >>> **Follow-up:** What if the list contains duplicate numbers?

def __init__(*,
             model: str = "text-bison",
             project_id: Optional[str] = None,
             location: Optional[str] = None,
             **kwargs)

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

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

@component.output_types(replies=List[str],
                        safety_attributes=Dict[str, float],
                        citations=List[Dict[str, Any]])
def run(prompt: str)

from haystack.dataclasses import ChatMessage
from haystack_integrations.components.generators.google_vertex import VertexAIGeminiChatGenerator

gemini_chat = VertexAIGeminiChatGenerator()

messages = [ChatMessage.from_user("Tell me the name of a movie")]
res = gemini_chat.run(messages)

print(res["replies"][0].text)
>>> The Shawshank Redemption

#### With Tool calling:

```python
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_vertex import VertexAIGeminiChatGenerator

__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 = VertexAIGeminiChatGenerator(
    model="gemini-2.0-flash-exp",
    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)

def __init__(*,
             model: str = "gemini-1.5-flash",
             project_id: Optional[str] = None,
             location: Optional[str] = None,
             generation_config: Optional[Union[GenerationConfig,
                                               Dict[str, Any]]] = None,
             safety_settings: Optional[Dict[HarmCategory,
                                            HarmBlockThreshold]] = None,
             tools: Optional[List[Tool]] = None,
             tool_config: Optional[ToolConfig] = None,
             streaming_callback: Optional[StreamingCallbackT] = None)

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

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

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

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

from haystack import Document
from haystack_integrations.components.embedders.google_vertex import VertexAIDocumentEmbedder

doc = Document(content="I love pizza!")

document_embedder = VertexAIDocumentEmbedder(model="text-embedding-005")

result = document_embedder.run([doc])
print(result['documents'][0].embedding)
# [-0.044606007635593414, 0.02857724390923977, -0.03549133986234665,

def __init__(model: Literal[
    "text-embedding-004",
    "text-embedding-005",
    "textembedding-gecko-multilingual@001",
    "text-multilingual-embedding-002",
    "text-embedding-large-exp-03-07",
],
             task_type: Literal[
                 "RETRIEVAL_DOCUMENT",
                 "RETRIEVAL_QUERY",
                 "SEMANTIC_SIMILARITY",
                 "CLASSIFICATION",
                 "CLUSTERING",
                 "QUESTION_ANSWERING",
                 "FACT_VERIFICATION",
                 "CODE_RETRIEVAL_QUERY",
             ] = "RETRIEVAL_DOCUMENT",
             gcp_region_name: Optional[Secret] = Secret.from_env_var(
                 "GCP_DEFAULT_REGION", strict=False),
             gcp_project_id: Optional[Secret] = Secret.from_env_var(
                 "GCP_PROJECT_ID", strict=False),
             batch_size: int = 32,
             max_tokens_total: int = 20000,
             time_sleep: int = 30,
             retries: int = 3,
             progress_bar: bool = True,
             truncate_dim: Optional[int] = None,
             meta_fields_to_embed: Optional[List[str]] = None,
             embedding_separator: str = "\n") -> None

def get_text_embedding_input(
        batch: List[Document]) -> List[TextEmbeddingInput]

def embed_batch_by_smaller_batches(batch: List[str],
                                   subbatch=1) -> List[List[float]]

def embed_batch(batch: List[str]) -> List[List[float]]

@component.output_types(documents=List[Document])
def run(documents: List[Document])

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

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

from haystack_integrations.components.embedders.google_vertex import VertexAITextEmbedder

text_to_embed = "I love pizza!"

text_embedder = VertexAITextEmbedder(model="text-embedding-005")

print(text_embedder.run(text_to_embed))
# {'embedding': [-0.08127457648515701, 0.03399784862995148, -0.05116401985287666, ...]

def __init__(model: Literal[
    "text-embedding-004",
    "text-embedding-005",
    "textembedding-gecko-multilingual@001",
    "text-multilingual-embedding-002",
    "text-embedding-large-exp-03-07",
],
             task_type: Literal[
                 "RETRIEVAL_DOCUMENT",
                 "RETRIEVAL_QUERY",
                 "SEMANTIC_SIMILARITY",
                 "CLASSIFICATION",
                 "CLUSTERING",
                 "QUESTION_ANSWERING",
                 "FACT_VERIFICATION",
                 "CODE_RETRIEVAL_QUERY",
             ] = "RETRIEVAL_QUERY",
             gcp_region_name: Optional[Secret] = Secret.from_env_var(
                 "GCP_DEFAULT_REGION", strict=False),
             gcp_project_id: Optional[Secret] = Secret.from_env_var(
                 "GCP_PROJECT_ID", strict=False),
             progress_bar: bool = True,
             truncate_dim: Optional[int] = None) -> None

@component.output_types(embedding=List[float])
def run(text: Union[List[Document], List[str], str])

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

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