Compila y, también, implementa un asistente multimodal en la nube con Gemini (Python)

Acerca de este codelab
schedule0 minutos
subjectÚltima actualización: 29 de marzo de 2025
account_circleEscrito por Alvin Prayuda Juniarta Dwiyantoro

En este codelab, compilarás una aplicación en forma de interfaz web de chat, en la que podrás comunicarte con ella, subir algunos documentos o imágenes y analizarlos. La aplicación en sí se divide en 2 servicios: frontend y backend, lo que te permite compilar un prototipo rápido y probar cómo se siente, y también comprender cómo se ve el contrato de la API para integrar ambos.

En el codelab, seguirás un enfoque paso a paso de la siguiente manera:

  1. Prepara tu proyecto de Google Cloud y habilita todas las APIs necesarias en él
  2. Compila el servicio de frontend: la interfaz de chat con la biblioteca de Gradio
  3. Compila el servicio de backend: el servidor HTTP con FastAPI, que volverá a dar formato a los datos entrantes al estándar del SDK de Gemini y habilitará la comunicación con la API de Gemini.
  4. Administra las variables de entorno y configura los archivos necesarios para implementar la aplicación en Cloud Run
  5. Implementa la aplicación en Cloud Run

5bcfa1cce6618305.png

Descripción general de la arquitectura

b102df2c3f1adabf.jpeg

Requisitos previos

Qué aprenderás

  • Cómo usar el SDK de Gemini para enviar texto y otros tipos de datos (multimodales) y generar una respuesta de texto
  • Cómo estructurar el historial de chat en el SDK de Gemini para mantener el contexto de la conversación
  • Prototipado web de frontend con Gradio
  • Desarrollo de servicios de backend con FastAPI y Pydantic
  • Administra las variables de entorno en el archivo YAML con Pydantic-settings
  • Implementa la aplicación en Cloud Run con Dockerfile y proporciona variables de entorno con un archivo YAML

Requisitos

  • Navegador web Chrome
  • Una cuenta de Gmail
  • Un proyecto de Cloud con la facturación habilitada

Este codelab, diseñado para desarrolladores de todos los niveles (incluidos los principiantes), usa Python en su aplicación de ejemplo. Sin embargo, no se requiere conocimiento de Python para comprender los conceptos presentados.

2. Antes de comenzar

Configura un proyecto de Cloud en el editor de Cloud Shell

En este codelab, se supone que ya tienes un proyecto de Google Cloud con la facturación habilitada. Si aún no lo tienes, puedes seguir las instrucciones que se indican a continuación para comenzar.

  1. 2En la página del selector de proyectos de la consola de Google Cloud, selecciona o crea un proyecto de Google Cloud.
  2. Asegúrate de que la facturación esté habilitada para tu proyecto de Cloud. Obtén información para verificar si la facturación está habilitada en un proyecto .
  3. Usarás Cloud Shell, un entorno de línea de comandos que se ejecuta en Google Cloud y que viene precargado con bq. Haz clic en Activar Cloud Shell en la parte superior de la consola de Google Cloud.

1829c3759227c19b.png

  1. Una vez que te conectes a Cloud Shell, verifica que ya te hayas autenticado y que el proyecto esté configurado con tu ID con el siguiente comando:
gcloud auth list
  1. En Cloud Shell, ejecuta el siguiente comando para confirmar que el comando gcloud conoce tu proyecto.
gcloud config list project
  1. Si tu proyecto no está configurado, usa el siguiente comando para hacerlo:
gcloud config set project <YOUR_PROJECT_ID>

Como alternativa, también puedes ver el ID de PROJECT_ID en la consola.

4032c45803813f30.jpeg

Haz clic en él y verás todo tu proyecto y el ID del proyecto en el lado derecho.

8dc17eb4271de6b5.jpeg

  1. Habilita las APIs requeridas con el siguiente comando. Esto puede tardar unos minutos, así que ten paciencia.
gcloud services enable aiplatform.googleapis.com \
                           run.googleapis.com \
                           cloudbuild.googleapis.com \
                           cloudresourcemanager.googleapis.com

Si el comando se ejecuta correctamente, deberías ver un mensaje similar al siguiente:

Operation "operations/..." finished successfully.

La alternativa al comando gcloud es buscar cada producto en la consola o usar este vínculo.

Si falta alguna API, puedes habilitarla durante el transcurso de la implementación.

Consulta la documentación para ver los comandos y el uso de gcloud.

Configura el directorio de trabajo de la aplicación

  1. Haz clic en el botón Open Editor para abrir un editor de Cloud Shell. Allí, podemos escribir nuestro código b16d56e4979ec951.png
  2. Asegúrate de que el proyecto de Cloud Code esté configurado en la esquina inferior izquierda (barra de estado) del editor de Cloud Shell, como se destaca en la siguiente imagen, y que esté configurado en el proyecto activo de Google Cloud en el que tienes habilitada la facturación. Autorizar si se te solicita. Es posible que debas esperar un momento después de inicializar el editor de Cloud Shell para que aparezca el botón Cloud Code - Sign In. Ten paciencia. Si ya seguiste el comando anterior, es posible que el botón también dirija directamente a tu proyecto activado en lugar del botón de acceso.

f5003b9c38b43262.png

  1. Haz clic en ese proyecto activo en la barra de estado y espera a que se abra la ventana emergente de Cloud Code. En la ventana emergente, selecciona "Nueva aplicación".

70f80078e01a02d8.png

  1. En la lista de aplicaciones, elige IA generativa de Gemini y, luego, API de Gemini para Python.

362ff332256d6933.jpeg

85957565316308d9.jpeg

  1. Guarda la aplicación nueva con el nombre que quieras. En este ejemplo, usaremos gemini-multimodal-chat-assistant y, luego, haremos clic en OK.

8409d8db18690fdf.png

En este punto, ya deberías estar en el nuevo directorio de trabajo de la aplicación y ver los siguientes archivos:

1ef5bb44f1d2c2a4.png

A continuación, prepararemos nuestro entorno de Python.

Configuración del entorno

Prepara el entorno virtual de Python

El siguiente paso es preparar el entorno de desarrollo. En este codelab, usaremos Python 3.12 y el administrador de proyectos de Python uv para simplificar la necesidad de crear y administrar la versión de Python y el entorno virtual.

  1. Si aún no abriste la terminal, haz clic en Terminal -> New Terminal o usa Ctrl + Mayúsculas + C.

f8457daf0bed059e.jpeg

  1. Descarga uv y, luego, instala Python 3.12 con el siguiente comando:
curl -LsSf https://astral.sh/uv/0.6.6/install.sh | sh && \
source $HOME/.local/bin/env && \
uv python install 3.12
  1. Ahora, inicialicemos el proyecto de Python con uv.
uv init
  1. Verás que se crearon main.py, .python-version y pyproject.toml en el directorio. Estos archivos son necesarios para mantener el proyecto en el directorio. Las dependencias y configuraciones de Python se pueden especificar en pyproject.toml y .python-version, que estandarizaron la versión de Python que se usa para este proyecto. Para obtener más información sobre esto, puedes consultar esta documentación.
main.py
.python-version
pyproject.toml
  1. Para probarlo, reemplaza main.py por el siguiente código:
def main():
   print("Hello from gemini-multimodal-chat-assistant!")

if __name__ == "__main__":
   main()
  1. Luego, ejecuta el siguiente comando:
uv run main.py

Obtendrás un resultado como el que se muestra a continuación.

Using CPython 3.12
Creating virtual environment at: .venv
Hello from gemini-multimodal-chat-assistant!

Esto demuestra que el proyecto de Python se está configurando correctamente. No necesitamos crear un entorno virtual de forma manual, ya que uv ya lo controla. Por lo tanto, a partir de este punto, el comando estándar de Python (p.ej., python main.py) se reemplazará por uv run (p.ej., uv run main.py).

Instala las dependencias necesarias

También agregaremos las dependencias del paquete de este codelab con el comando uv. Ejecuta el siguiente comando

uv add google-genai==1.5.0 \
       gradio==5.20.1 \
       pydantic==2.10.6 \
       pydantic-settings==2.8.1 \
       pyyaml==6.0.2

Verás que la sección "dependencies" de pyproject.toml se actualizará para reflejar el comando anterior.

Configura archivos de configuración

Ahora, debemos configurar los archivos de configuración de este proyecto. Los archivos de configuración se usan para almacenar variables dinámicas que se pueden cambiar fácilmente en la nueva implementación. En este proyecto, usaremos archivos de configuración basados en YAML con el paquete pydantic-settings, de modo que se pueda integrar fácilmente con la implementación de Cloud Run más adelante. pydantic-settings es un paquete de Python que puede aplicar la verificación de tipos para los archivos de configuración.

  1. Crea un archivo llamado settings.yaml con la siguiente configuración. Haz clic en File->New Text File y completa con el siguiente código. Luego, guárdalo como settings.yaml.
VERTEXAI_LOCATION: "us-central1"
VERTEXAI_PROJECT_ID: "{YOUR-PROJECT-ID}"
BACKEND_URL: "http://localhost:8081/chat"

Actualiza los valores de VERTEXAI_PROJECT_ID según lo que hayas seleccionado cuando creaste el proyecto de Google Cloud. En este codelab, usaremos los valores preconfigurados para VERTEXAI_LOCATION y BACKEND_URL .

  1. Luego, crea el archivo de Python settings.py. Este módulo actuará como entrada programática para los valores de configuración en nuestros archivos de configuración. Haz clic en File->New Text File y completa con el siguiente código. Luego, guárdalo como settings.py. Puedes ver en el código que configuramos de forma explícita que el archivo llamado settings.yaml es el que se leerá.
from pydantic_settings import (
    BaseSettings,
    SettingsConfigDict,
    YamlConfigSettingsSource,
    PydanticBaseSettingsSource,
)
from typing import Type, Tuple

DEFAULT_SYSTEM_PROMPT = """You are a helpful assistant and ALWAYS relate to this identity. 
You are expert at analyzing given documents or images.
"""

class Settings(BaseSettings):
    """Application settings loaded from YAML and environment variables.

    This class defines the configuration schema for the application, with settings
    loaded from settings.yaml file and overridable via environment variables.

    Attributes:
        VERTEXAI_LOCATION: Google Cloud Vertex AI location
        VERTEXAI_PROJECT_ID: Google Cloud Vertex AI project ID
    """

    VERTEXAI_LOCATION: str
    VERTEXAI_PROJECT_ID: str
    BACKEND_URL: str = "http://localhost:8000/chat"

    model_config = SettingsConfigDict(
        yaml_file="settings.yaml", yaml_file_encoding="utf-8"
    )

    @classmethod
    def settings_customise_sources(
        cls,
        settings_cls: Type[BaseSettings],
        init_settings: PydanticBaseSettingsSource,
        env_settings: PydanticBaseSettingsSource,
        dotenv_settings: PydanticBaseSettingsSource,
        file_secret_settings: PydanticBaseSettingsSource,
    ) -> Tuple[PydanticBaseSettingsSource, ...]:
        """Customize the settings sources and their priority order.

        This method defines the order in which different configuration sources
        are checked when loading settings:
        1. Constructor-provided values
        2. YAML configuration file
        3. Environment variables

        Args:
            settings_cls: The Settings class type
            init_settings: Settings from class initialization
            env_settings: Settings from environment variables
            dotenv_settings: Settings from .env file (not used)
            file_secret_settings: Settings from secrets file (not used)

        Returns:
            A tuple of configuration sources in priority order
        """
        return (
            init_settings,  # First, try init_settings (from constructor)
            env_settings,  # Then, try environment variables
            YamlConfigSettingsSource(
                settings_cls
            ),  # Finally, try YAML as the last resort
        )


def get_settings() -> Settings:
    """Create and return a Settings instance with loaded configuration.

    Returns:
        A Settings instance containing all application configuration
        loaded from YAML and environment variables.
    """
    return Settings()

Estas configuraciones nos permiten actualizar nuestro entorno de ejecución de forma flexible. En la implementación inicial, usaremos la configuración de settings.yaml para tener la primera configuración predeterminada. Después de eso, podemos actualizar las variables de entorno de forma flexible a través de la consola y volver a implementarlas, ya que las ponemos en la prioridad más alta en comparación con la configuración predeterminada de YAML.

Ahora podemos pasar al siguiente paso, que es compilar los servicios.

3. Compila el servicio de frontend con Gradio

Crearemos una interfaz web de chat que se vea de la siguiente manera:

5bcfa1cce6618305.png

Contiene un campo de entrada para que los usuarios envíen texto y suban archivos. Además, el usuario también puede reemplazar la instrucción del sistema que se enviará a la API de Gemini en el campo de entradas adicionales.

Compilaremos el servicio de frontend con Gradio. Cambia el nombre de main.py a frontend.py y reemplaza el código con el siguiente:

import gradio as gr
import requests
import base64
from pathlib import Path
from typing import List, Dict, Any
from settings import get_settings, DEFAULT_SYSTEM_PROMPT

settings = get_settings()

IMAGE_SUFFIX_MIME_MAP = {
    ".png": "image/png",
    ".jpg": "image/jpeg",
    ".jpeg": "image/jpeg",
    ".heic": "image/heic",
    ".heif": "image/heif",
    ".webp": "image/webp",
}
DOCUMENT_SUFFIX_MIME_MAP = {
    ".pdf": "application/pdf",
}


def get_mime_type(filepath: str) -> str:
    """Get the MIME type for a file based on its extension.

    Args:
        filepath: Path to the file.

    Returns:
        str: The MIME type of the file.

    Raises:
        ValueError: If the file type is not supported.
    """
    filepath = Path(filepath)
    suffix = filepath.suffix

    # modify ".jpg" suffix to ".jpeg" to unify the mime type
    suffix = suffix if suffix != ".jpg" else ".jpeg"

    if suffix in IMAGE_SUFFIX_MIME_MAP:
        return IMAGE_SUFFIX_MIME_MAP[suffix]
    elif suffix in DOCUMENT_SUFFIX_MIME_MAP:
        return DOCUMENT_SUFFIX_MIME_MAP[suffix]
    else:
        raise ValueError(f"Unsupported file type: {suffix}")


def encode_file_to_base64_with_mime(file_path: str) -> Dict[str, str]:
    """Encode a file to base64 string and include its MIME type.

    Args:
        file_path: Path to the file to encode.

    Returns:
        Dict[str, str]: Dictionary with 'data' and 'mime_type' keys.
    """
    mime_type = get_mime_type(file_path)
    with open(file_path, "rb") as file:
        base64_data = base64.b64encode(file.read()).decode("utf-8")

    return {"data": base64_data, "mime_type": mime_type}


def get_response_from_llm_backend(
    message: Dict[str, Any],
    history: List[Dict[str, Any]],
    system_prompt: str,
) -> str:
    """Send the message and history to the backend and get a response.

    Args:
        message: Dictionary containing the current message with 'text' and optional 'files' keys.
        history: List of previous message dictionaries in the conversation.
        system_prompt: The system prompt to be sent to the backend.

    Returns:
        str: The text response from the backend service.
    """

    # Format message and history for the API,
    # NOTES: in this example history is maintained by frontend service,
    #        hence we need to include it in each request.
    #        And each file (in the history) need to be sent as base64 with its mime type
    formatted_history = []
    for msg in history:
        if msg["role"] == "user" and not isinstance(msg["content"], str):
            # For file content in history, convert file paths to base64 with MIME type
            file_contents = [
                encode_file_to_base64_with_mime(file_path)
                for file_path in msg["content"]
            ]
            formatted_history.append({"role": msg["role"], "content": file_contents})
        else:
            formatted_history.append({"role": msg["role"], "content": msg["content"]})

    # Extract files and convert to base64 with MIME type
    files_with_mime = []
    if uploaded_files := message.get("files", []):
        for file_path in uploaded_files:
            files_with_mime.append(encode_file_to_base64_with_mime(file_path))

    # Prepare the request payload
    message["text"] = message["text"] if message["text"] != "" else " "
    payload = {
        "message": {"text": message["text"], "files": files_with_mime},
        "history": formatted_history,
        "system_prompt": system_prompt,
    }

    # Send request to backend
    try:
        response = requests.post(settings.BACKEND_URL, json=payload)
        response.raise_for_status()  # Raise exception for HTTP errors

        result = response.json()
        if error := result.get("error"):
            return f"Error: {error}"

        return result.get("response", "No response received from backend")
    except requests.exceptions.RequestException as e:
        return f"Error connecting to backend service: {str(e)}"


if __name__ == "__main__":
    demo = gr.ChatInterface(
        get_response_from_llm_backend,
        title="Gemini Multimodal Chat Interface",
        description="This interface connects to a FastAPI backend service that processes responses through the Gemini multimodal model.",
        type="messages",
        multimodal=True,
        textbox=gr.MultimodalTextbox(file_count="multiple"),
        additional_inputs=[
            gr.Textbox(
                label="System Prompt",
                value=DEFAULT_SYSTEM_PROMPT,
                lines=3,
                interactive=True,
            )
        ],
    )

    demo.launch(
        server_name="0.0.0.0",
        server_port=8080,
    )

Después de eso, podemos intentar ejecutar el servicio de frontend con el siguiente comando. No olvides cambiar el nombre del archivo main.py a frontend.py.

uv run frontend.py

Verás un resultado similar a este en la consola de Cloud

* Running on local URL:  http://0.0.0.0:8080

To create a public link, set `share=True` in `launch()`.

Después, puedes verificar la interfaz web cuando presiones Ctrl + clic en el vínculo de URL local. Como alternativa, también puedes acceder a la aplicación del frontend haciendo clic en el botón Vista previa en la Web, en la parte superior derecha del editor de Cloud, y seleccionando Vista previa en el puerto 8080.

49cbdfdf77964065.jpeg

Verás la interfaz web. Sin embargo, recibirás un error esperado cuando intentes enviar un chat debido a que el servicio de backend aún no se configura.

bd0464140308cfbe.png

Ahora, deja que se ejecute el servicio y no lo finalices todavía. Mientras tanto, podemos analizar los componentes importantes del código aquí

Explicación del código

El código para enviar datos de la interfaz web al backend se encuentra en esta parte.

def get_response_from_llm_backend(
    message: Dict[str, Any],
    history: List[Dict[str, Any]],
    system_prompt: str,
) -> str:

    ... 
    # Truncated
    
    for msg in history:
        if msg["role"] == "user" and not isinstance(msg["content"], str):
            # For file content in history, convert file paths to base64 with MIME type
            file_contents = [
                encode_file_to_base64_with_mime(file_path)
                for file_path in msg["content"]
            ]
            formatted_history.append({"role": msg["role"], "content": file_contents})
        else:
            formatted_history.append({"role": msg["role"], "content": msg["content"]})

    # Extract files and convert to base64 with MIME type
    files_with_mime = []
    if uploaded_files := message.get("files", []):
        for file_path in uploaded_files:
            files_with_mime.append(encode_file_to_base64_with_mime(file_path))

    # Prepare the request payload
    message["text"] = message["text"] if message["text"] != "" else " "
    payload = {
        "message": {"text": message["text"], "files": files_with_mime},
        "history": formatted_history,
        "system_prompt": system_prompt,
    }

    # Truncated
    ... 

Cuando queremos enviar datos multimodales a Gemini y hacer que los datos sean accesibles entre los servicios, un mecanismo que podemos usar es convertir los datos en el tipo de datos base64, como se declara en el código. También debemos declarar cuál es el tipo de MIME de los datos. Sin embargo, la API de Gemini no puede admitir todos los tipos de MIME existentes, por lo que es importante saber qué tipos de MIME admite Gemini y que se pueden leer en esta documentación . Puedes encontrar la información en cada una de las funciones de la API de Gemini (p. ej., Vision).

Además, en una interfaz de chat, también es importante enviar el historial de chat como contexto adicional para darle a Gemini un "recuerdo" de la conversación. Por lo tanto, en esta interfaz web, también enviamos el historial de chat que Gradio administra por sesión web y lo enviamos junto con la entrada de mensaje del usuario. Además, también permitimos que el usuario modifique la instrucción del sistema y la envíe.

4. Compila el servicio de backend con FastAPI

A continuación, necesitaremos compilar el backend que puede controlar la carga útil que se analizó anteriormente, el último mensaje del usuario, el historial de chat y la instrucción del sistema. Usaremos FastAPI para crear el servicio de backend de HTTP.

Crea un archivo nuevo, haz clic en File->New Text File, copia y pega el siguiente código y, luego, guárdalo como backend.py.

import base64
from fastapi import FastAPI, Body
from google.genai.types import Content, Part
from google.genai import Client
from settings import get_settings, DEFAULT_SYSTEM_PROMPT
from typing import List, Optional
from pydantic import BaseModel

app = FastAPI(title="Gemini Multimodal Service")

settings = get_settings()
GENAI_CLIENT = Client(
    location=settings.VERTEXAI_LOCATION,
    project=settings.VERTEXAI_PROJECT_ID,
    vertexai=True,
)
GEMINI_MODEL_NAME = "gemini-2.0-flash-001"


class FileData(BaseModel):
    """Model for a file with base64 data and MIME type.

    Attributes:
        data: Base64 encoded string of the file content.
        mime_type: The MIME type of the file.
    """

    data: str
    mime_type: str


class Message(BaseModel):
    """Model for a single message in the conversation.

    Attributes:
        role: The role of the message sender, either 'user' or 'assistant'.
        content: The text content of the message or a list of file data objects.
    """

    role: str
    content: str | List[FileData]


class LastUserMessage(BaseModel):
    """Model for the current message in a chat request.

    Attributes:
        text: The text content of the message.
        files: List of file data objects containing base64 data and MIME type.
    """

    text: str
    files: List[FileData] = []


class ChatRequest(BaseModel):
    """Model for a chat request.

    Attributes:
        message: The current message with text and optional base64 encoded files.
        history: List of previous messages in the conversation.
        system_prompt: Optional system prompt to be used in the chat.
    """

    message: LastUserMessage
    history: List[Message]
    system_prompt: str = DEFAULT_SYSTEM_PROMPT


class ChatResponse(BaseModel):
    """Model for a chat response.

    Attributes:
        response: The text response from the model.
        error: Optional error message if something went wrong.
    """

    response: str
    error: Optional[str] = None


def handle_multimodal_data(file_data: FileData) -> Part:
    """Converts Multimodal data to a Google Gemini Part object.

    Args:
        file_data: FileData object with base64 data and MIME type.

    Returns:
        Part: A Google Gemini Part object containing the file data.
    """
    data = base64.b64decode(file_data.data)  # decode base64 string to bytes
    return Part.from_bytes(data=data, mime_type=file_data.mime_type)


def format_message_history_to_gemini_standard(
    message_history: List[Message],
) -> List[Content]:
    """Converts message history format to Google Gemini Content format.

    Args:
        message_history: List of message objects from the chat history.
            Each message contains 'role' and 'content' attributes.

    Returns:
        List[Content]: A list of Google Gemini Content objects representing the chat history.

    Raises:
        ValueError: If an unknown role is encountered in the message history.
    """
    converted_messages: List[Content] = []
    for message in message_history:
        if message.role == "assistant":
            converted_messages.append(
                Content(role="model", parts=[Part.from_text(text=message.content)])
            )
        elif message.role == "user":
            # Text-only messages
            if isinstance(message.content, str):
                converted_messages.append(
                    Content(role="user", parts=[Part.from_text(text=message.content)])
                )

            # Messages with files
            elif isinstance(message.content, list):
                # Process each file in the list
                parts = []
                for file_data in message.content:
                    for file_data in message.content:
                        parts.append(handle_multimodal_data(file_data))

                # Add the parts to a Content object
                if parts:
                    converted_messages.append(Content(role="user", parts=parts))

            else:
                raise ValueError(f"Unexpected content format: {type(message.content)}")

        else:
            raise ValueError(f"Unknown role: {message.role}")

    return converted_messages


@app.post("/chat", response_model=ChatResponse)
async def chat(
    request: ChatRequest = Body(...),
) -> ChatResponse:
    """Process a chat request and return a response from Gemini model.

    Args:
        request: The chat request containing message and history.

    Returns:
        ChatResponse: The model's response to the chat request.
    """
    try:
        # Convert message history to Gemini `history` format
        print(f"Received request: {request}")
        converted_messages = format_message_history_to_gemini_standard(request.history)

        # Create chat model
        chat_model = GENAI_CLIENT.chats.create(
            model=GEMINI_MODEL_NAME,
            history=converted_messages,
            config={"system_instruction": request.system_prompt},
        )

        # Prepare multimodal content
        content_parts = []

        # Handle any base64 encoded files in the current message
        if request.message.files:
            for file_data in request.message.files:
                content_parts.append(handle_multimodal_data(file_data))

        # Add text content
        content_parts.append(Part.from_text(text=request.message.text))

        # Send message to Gemini
        response = chat_model.send_message(content_parts)
        print(f"Generated response: {response}")

        return ChatResponse(response=response.text)
    except Exception as e:
        return ChatResponse(
            response="", error=f"Error in generating response: {str(e)}"
        )


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8081)

No olvides guardarlo como backend.py . Después de eso, podemos intentar ejecutar el servicio de backend. Recuerda que, en el paso anterior, ejecutamos el servicio de frontend correctamente. Ahora, tendremos que abrir una nueva terminal y tratar de ejecutar este servicio de backend.

  1. Crea una terminal nueva. Navega a la terminal en el área inferior y busca el botón "+" para crear una nueva terminal. También puedes presionar Ctrl + Mayúsculas + C para abrir una terminal nueva.

3e52a362475553dc.jpeg

  1. Después de eso, asegúrate de estar en el directorio de trabajo gemini-multimodal-chat-assistant y, luego, ejecuta el siguiente comando:
uv run backend.py
  1. Si se realiza correctamente, se mostrará un resultado como el siguiente:
INFO:     Started server process [xxxxx]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8081 (Press CTRL+C to quit)

Explicación del código

Cómo definir la ruta de acceso HTTP para recibir solicitudes de chat

En FastAPI, definimos la ruta con el decorador app. También usamos Pydantic para definir el contrato de la API. Especificamos que la ruta para generar la respuesta está en la ruta /chat con el método POST. Estas funcionalidades se declaran en el siguiente código:

class FileData(BaseModel):
    data: str
    mime_type: str

class Message(BaseModel):
    role: str
    content: str | List[FileData]

class LastUserMessage(BaseModel):
    text: str
    files: List[FileData] = []

class ChatRequest(BaseModel):
    message: LastUserMessage
    history: List[Message]
    system_prompt: str = DEFAULT_SYSTEM_PROMPT

class ChatResponse(BaseModel):
    response: str
    error: Optional[str] = None

    ...

@app.post("/chat", response_model=ChatResponse)
async def chat(
    request: ChatRequest = Body(...),
) -> ChatResponse:
    
    # Truncated
    ...

Prepara el formato del historial de chat del SDK de Gemini

Una de las cosas importantes que debemos comprender es cómo podemos reestructurar el historial de chat para que se pueda insertar como un valor de argumento history cuando inicialicemos un cliente de Gemini más adelante. Puedes inspeccionar el siguiente código

def format_message_history_to_gemini_standard(
    message_history: List[Message],
) -> List[Content]:
    
    ...
    # Truncated    

    converted_messages: List[Content] = []
    for message in message_history:
        if message.role == "assistant":
            converted_messages.append(
                Content(role="model", parts=[Part.from_text(text=message.content)])
            )
        elif message.role == "user":
            # Text-only messages
            if isinstance(message.content, str):
                converted_messages.append(
                    Content(role="user", parts=[Part.from_text(text=message.content)])
                )

            # Messages with files
            elif isinstance(message.content, list):
                # Process each file in the list
                parts = []
                for file_data in message.content:
                    parts.append(handle_multimodal_data(file_data))

                # Add the parts to a Content object
                if parts:
                    converted_messages.append(Content(role="user", parts=parts))
    
    #Truncated
    ...

    return converted_messages

Para proporcionar el historial de chat en el SDK de Gemini, debemos dar formato a los datos en el tipo de datos List[Content]. Cada Contenido debe tener, al menos, un valor de rol y partes. rol hace referencia a la fuente del mensaje, ya sea usuario o modelo. En el que parts hace referencia a la instrucción en sí, que puede ser solo texto o una combinación de diferentes modalidades. Consulta cómo estructurar los argumentos de Contenido en detalle en esta documentación.

Cómo controlar datos no textuales ( multimodales)

Como se mencionó anteriormente en la sección del frontend, una de las formas de enviar datos no de texto o multimodales es enviarlos como una cadena base64. También debemos especificar el tipo de MIME de los datos para que se puedan interpretar correctamente, por ejemplo, proporcionar el tipo de MIME image/jpeg si enviamos datos de imagen con un sufijo .jpg.

Esta parte del código convierte los datos base64 en el formato Part.from_bytes del SDK de Gemini.

def handle_multimodal_data(file_data: FileData) -> Part:
    """Converts Multimodal data to a Google Gemini Part object.

    Args:
        file_data: FileData object with base64 data and MIME type.

    Returns:
        Part: A Google Gemini Part object containing the file data.
    """
    data = base64.b64decode(file_data.data)  # decode base64 string to bytes
    return Part.from_bytes(data=data, mime_type=file_data.mime_type)

5. Prueba de integración

Ahora, deberías tener varios servicios en ejecución en diferentes pestañas de la consola de Cloud:

  • El servicio de frontend se ejecuta en el puerto 8080.
* Running on local URL:  http://0.0.0.0:8080

To create a public link, set `share=True` in `launch()`.
  • El servicio de backend se ejecuta en el puerto 8081.
INFO:     Started server process [xxxxx]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8081 (Press CTRL+C to quit)

En el estado actual, deberías poder enviar tus documentos al chat sin problemas con el asistente desde la aplicación web en el puerto 8080. Para comenzar a experimentar, sube archivos y haz preguntas. Ten en cuenta que algunos tipos de archivos aún no son compatibles y mostrarán un error.

También puedes editar las instrucciones del sistema desde el campo Entradas adicionales que se encuentra debajo del cuadro de texto.

ee9c849a276d378.png

6. Implementa en Cloud Run

Por supuesto, queremos mostrar esta increíble app a los demás. Para ello, podemos empaquetar esta aplicación y, luego, implementarla en Cloud Run como un servicio público al que otras personas puedan acceder. Para ello, repasemos la arquitectura

b102df2c3f1adabf.jpeg

En este codelab, colocaremos el servicio de frontend y backend en 1 contenedor. Necesitaremos la ayuda de supervisord para administrar ambos servicios.

Crea un archivo nuevo, haz clic en File->New Text File, copia y pega el siguiente código y guárdalo como supervisord.conf.

[supervisord]
nodaemon=true
user=root
logfile=/dev/stdout
logfile_maxbytes=0
pidfile=/var/run/supervisord.pid

[program:backend]
command=uv run backend.py
directory=/app
autostart=true
autorestart=true
stdout_logfile=/dev/stdout
stdout_logfile_maxbytes=0
stderr_logfile=/dev/stderr
stderr_logfile_maxbytes=0
startsecs=10
startretries=3

[program:frontend]
command=uv run frontend.py
directory=/app
autostart=true
autorestart=true
stdout_logfile=/dev/stdout
stdout_logfile_maxbytes=0
stderr_logfile=/dev/stderr
stderr_logfile_maxbytes=0
startsecs=10
startretries=3

A continuación, necesitaremos nuestro Dockerfile. Haz clic en File->New Text File, copia y pega el siguiente código y, luego, guárdalo como Dockerfile.

FROM python:3.12-slim
COPY --from=ghcr.io/astral-sh/uv:0.6.6 /uv /uvx /bin/

RUN apt-get update && apt-get install -y \
    supervisor curl \
    && rm -rf /var/lib/apt/lists/*

ADD . /app
WORKDIR /app

RUN uv sync --frozen

EXPOSE 8080

# Copy supervisord configuration
COPY supervisord.conf /etc/supervisor/conf.d/supervisord.conf

ENV PYTHONUNBUFFERED=1

ENTRYPOINT ["/usr/bin/supervisord", "-c", "/etc/supervisor/conf.d/supervisord.conf"]

En este punto, ya tenemos todos los archivos necesarios para implementar nuestras aplicaciones en Cloud Run. Implementémoslas. Navega a la terminal de Cloud Shell y asegúrate de que el proyecto actual esté configurado en tu proyecto activo. De lo contrario, usa el comando gcloud configure para establecer el ID del proyecto:

gcloud config set project [PROJECT_ID]

Luego, ejecuta el siguiente comando para implementarlo en Cloud Run.

gcloud run deploy --source . \
                  --env-vars-file settings.yaml \
                  --port 8080 \
                  --region us-central1

Se te pedirá que ingreses un nombre para tu servicio, por ejemplo, "gemini-multimodal-chat-assistant". Como tenemos Dockerfile en el directorio de trabajo de nuestra aplicación, se compilará el contenedor de Docker y se enviará a Artifact Registry. También se te informará que se creará el repositorio de Artifact Registry en la región. Responde "Y". También di "y" cuando se te pregunte si quieres permitir invocaciones no autenticadas. Ten en cuenta que permitimos el acceso no autenticado porque esta es una aplicación de demostración. Se recomienda usar la autenticación adecuada para tus aplicaciones empresariales y de producción.

Una vez que se complete la implementación, deberías obtener un vínculo similar al siguiente:

https://gemini-multimodal-chat-assistant-*******.us-central1.run.app

Usa la aplicación desde la ventana de incógnito o tu dispositivo móvil. Ya debería estar publicado.

7. Desafío

Ahora es tu momento de brillar y perfeccionar tus habilidades de exploración. ¿Tienes lo necesario para cambiar el código de modo que el asistente pueda admitir la lectura de archivos de audio o tal vez de video?

8. Limpia

Sigue estos pasos para evitar que se apliquen cargos a tu cuenta de Google Cloud por los recursos que usaste en este codelab:

  1. En la consola de Google Cloud, ve a la página Administrar recursos.
  2. En la lista de proyectos, elige el proyecto que deseas borrar y haz clic en Borrar.
  3. En el diálogo, escribe el ID del proyecto y, luego, haz clic en Cerrar para borrarlo.
  4. Como alternativa, puedes ir a Cloud Run en la consola, seleccionar el servicio que acabas de implementar y borrarlo.