Créer et déployer un assistant multimodal dans le cloud avec Gemini (Python)

1. Introduction

Dans cet atelier de programmation, vous allez créer une application sous la forme d'une interface Web de chat, dans laquelle vous pourrez communiquer, importer des documents ou des images, et en discuter. L'application elle-même est divisée en deux services : l'interface utilisateur et le backend. Cela vous permet de créer un prototype rapide et de tester son fonctionnement, mais aussi de comprendre à quoi ressemble le contrat d'API pour intégrer les deux.

Au cours de cet atelier de programmation, vous allez suivre une approche par étapes :

Préparez votre projet Google Cloud et activez toutes les API requises.
Créer le service de frontend : interface de chat à l'aide de la bibliothèque Gradio
Créez le service de backend (serveur HTTP) à l'aide de FastAPI, qui reformatera les données entrantes au format standard du SDK Gemini et permettra la communication avec l'API Gemini.
Gérer les variables d'environnement et configurer les fichiers requis pour déployer l'application sur Cloud Run
Déployer l'application sur Cloud Run

Présentation de l'architecture

Prérequis

Vous êtes à l'aise avec l'API Gemini et le SDK Google Gen AI.
Comprendre l'architecture full stack de base à l'aide du service HTTP

Points abordés

Comment utiliser le SDK Gemini pour envoyer du texte et d'autres types de données (multimodales) et générer une réponse textuelle
Structurer l'historique des discussions dans le SDK Gemini pour conserver le contexte des conversations
Prototyper l'interface Web avec Gradio
Développement de services de backend avec FastAPI et Pydantic
Gérer les variables d'environnement dans un fichier YAML avec Pydantic-settings
Déployer une application sur Cloud Run à l'aide d'un fichier Dockerfile et fournir des variables d'environnement avec un fichier YAML

Prérequis

Navigateur Web Chrome
Un compte Gmail
Un projet Cloud pour lequel la facturation est activée

Cet atelier de programmation, conçu pour les développeurs de tous niveaux (y compris les débutants), utilise Python dans son exemple d'application. Toutefois, aucune connaissance de Python n'est requise pour comprendre les concepts présentés.

2. Avant de commencer

Configurer un projet Cloud dans l'éditeur Cloud Shell

Cet atelier de programmation suppose que vous disposez déjà d'un projet Google Cloud pour lequel la facturation est activée. Si vous ne l'avez pas encore, vous pouvez suivre les instructions ci-dessous pour commencer.

2Dans la console Google Cloud, sur la page du sélecteur de projet, sélectionnez ou créez un projet Google Cloud.
Assurez-vous que la facturation est activée pour votre projet Cloud. Découvrez comment vérifier si la facturation est activée sur un projet .
Vous allez utiliser Cloud Shell, un environnement de ligne de commande exécuté dans Google Cloud et fourni avec bq. Cliquez sur "Activer Cloud Shell" en haut de la console Google Cloud.

Une fois connecté à Cloud Shell, vérifiez que vous êtes déjà authentifié et que le projet est défini sur votre ID de projet à l'aide de la commande suivante :

gcloud auth list

Exécutez la commande suivante dans Cloud Shell pour vérifier que la commande gcloud connaît votre projet.

gcloud config list project

Si votre projet n'est pas défini, utilisez la commande suivante pour le définir :

gcloud config set project <YOUR_PROJECT_ID>

Vous pouvez également voir l'ID PROJECT_ID dans la console.

Cliquez dessus pour afficher tous vos projets et l'ID du projet sur la droite.

Activez les API requises à l'aide de la commande ci-dessous. Cette opération peut prendre quelques minutes. Veuillez patienter.

gcloud services enable aiplatform.googleapis.com \
                           run.googleapis.com \
                           cloudbuild.googleapis.com \
                           cloudresourcemanager.googleapis.com

Si la commande s'exécute correctement, un message semblable à celui ci-dessous s'affiche :

Operation "operations/..." finished successfully.

Vous pouvez également accéder à la console en recherchant chaque produit ou en utilisant ce lien.

Si vous oubliez une API, vous pourrez toujours l'activer au cours de l'implémentation.

Consultez la documentation pour connaître les commandes gcloud ainsi que leur utilisation.

Configurer le répertoire de travail de l'application

Cliquez sur le bouton "Ouvrir l'éditeur". Un éditeur Cloud Shell s'ouvre. Vous pouvez y écrire votre code .
Assurez-vous que le projet Cloud Code est défini en bas à gauche (barre d'état) de l'éditeur Cloud Shell, comme indiqué dans l'image ci-dessous, et qu'il est défini sur le projet Google Cloud actif pour lequel la facturation est activée. Cliquez sur Autoriser si vous y êtes invité. L'initialisation de l'éditeur Cloud Shell peut prendre un certain temps. Soyez patient jusqu'à ce que le bouton Cloud Code – Se connecter s'affiche. Si vous avez déjà suivi la commande précédente, le bouton peut également pointer directement vers votre projet activé au lieu du bouton de connexion.

Cliquez sur le projet actif dans la barre d'état et attendez que le pop-up Cloud Code s'ouvre. Dans le pop-up qui s'affiche, sélectionnez "Nouvelle application".

Dans la liste des applications, sélectionnez IA générative Gemini, puis Gemini API Python.

Enregistrez la nouvelle application avec le nom de votre choix. Dans cet exemple, nous utiliserons gemini-multimodal-chat-assistant , puis cliquez sur OK.

À ce stade, vous devriez déjà vous trouver dans le répertoire de travail de la nouvelle application et voir les fichiers suivants :

Ensuite, nous allons préparer notre environnement Python.

Configurer l'environnement

Préparer l'environnement virtuel Python

L'étape suivante consiste à préparer l'environnement de développement. Dans cet atelier de programmation, nous utiliserons Python 3.12 et le gestionnaire de projets Python uv pour simplifier la création et la gestion de la version Python et de l'environnement virtuel.

Si vous n'avez pas encore ouvert le terminal, ouvrez-le en cliquant sur Terminal > Nouveau terminal ou en utilisant le raccourci Ctrl+Maj+C.

Téléchargez uv et installez Python 3.12 avec la commande suivante :

curl -LsSf https://astral.sh/uv/0.6.6/install.sh | sh && \
source $HOME/.local/bin/env && \
uv python install 3.12

Initialisons maintenant le projet Python à l'aide de uv.

uv init

Les fichiers main.py, .python-version et pyproject.toml sont créés dans le répertoire. Ces fichiers sont nécessaires pour gérer le projet dans le répertoire. Les dépendances et configurations Python peuvent être spécifiées dans les fichiers pyproject.toml et .python-version, qui standardisent la version de Python utilisée pour ce projet. Pour en savoir plus, consultez ces documents.

main.py
.python-version
pyproject.toml

Pour le tester, remplacez le contenu de main.py par le code suivant.

def main():
   print("Hello from gemini-multimodal-chat-assistant!")

if __name__ == "__main__":
   main()

Exécutez ensuite la commande suivante :

uv run main.py

Vous obtiendrez un résultat semblable à celui ci-dessous.

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

Cela montre que le projet Python est correctement configuré. Nous n'avons pas eu besoin de créer manuellement un environnement virtuel, car uv s'en charge déjà. À partir de maintenant, la commande Python standard (par exemple, python main.py) sera remplacée par uv run (par exemple, uv run main.py).

Installer les dépendances requises

Nous ajouterons également les dépendances du package de cet atelier de programmation à l'aide de la commande uv. Exécutez la commande suivante :

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

Vous verrez que la section "dependencies" de pyproject.toml sera mise à jour pour refléter la commande précédente.

Configurer les fichiers de configuration

Nous devons maintenant configurer les fichiers de configuration pour ce projet. Les fichiers de configuration permettent de stocker des variables dynamiques qui peuvent être facilement modifiées lors du redéploiement. Dans ce projet, nous utiliserons des fichiers de configuration basés sur YAML avec le package pydantic-settings, afin qu'il puisse être facilement intégré au déploiement Cloud Run ultérieurement. pydantic-settings est un package Python qui peut appliquer la vérification du type pour les fichiers de configuration.

Créez un fichier nommé settings.yaml avec la configuration suivante. Cliquez sur Fichier > Nouveau fichier texte et ajoutez le code suivant. Enregistrez-le ensuite sous le nom settings.yaml.

VERTEXAI_LOCATION: "us-central1"
VERTEXAI_PROJECT_ID: "{YOUR-PROJECT-ID}"
BACKEND_URL: "http://localhost:8081/chat"

Veuillez mettre à jour les valeurs de VERTEXAI_PROJECT_ID en fonction de ce que vous avez sélectionné lors de la création du projet Google Cloud. Pour cet atelier de programmation, nous allons utiliser les valeurs préconfigurées pour VERTEXAI_LOCATION et BACKEND_URL .

Créez ensuite le fichier Python settings.py. Ce module servira de point d'entrée programmatique pour les valeurs de configuration dans nos fichiers de configuration. Cliquez sur Fichier > Nouveau fichier texte et ajoutez le code suivant. Enregistrez-le ensuite sous le nom settings.py. Vous pouvez voir dans le code que nous avons explicitement défini le fichier nommé settings.yaml comme fichier à lire.

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

Ces configurations nous permettent de mettre à jour notre environnement d'exécution de manière flexible. Lors du déploiement initial, nous nous appuierons sur la configuration settings.yaml pour obtenir la première configuration par défaut. Ensuite, nous pouvons mettre à jour les variables d'environnement de manière flexible via la console et les redéployer, car nous avons défini les variables d'environnement comme étant plus prioritaires que la configuration YAML par défaut.

Nous pouvons maintenant passer à l'étape suivante, qui consiste à créer les services.

3. Créer un service de frontend à l'aide de Gradio

Nous allons créer une interface Web de chat qui ressemble à ceci :

Il contient un champ de saisie permettant aux utilisateurs d'envoyer du texte et d'importer des fichiers. De plus, l'utilisateur peut également remplacer les instructions système qui seront envoyées à l'API Gemini dans le champ des entrées supplémentaires.

Nous allons créer le service de frontend à l'aide de Gradio. Renommez main.py en frontend.py et remplacez le code par le code suivant.

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

Ensuite, nous pouvons essayer d'exécuter le service de frontend avec la commande suivante. N'oubliez pas de renommer le fichier main.py en frontend.py.

uv run frontend.py

Un résultat semblable à celui-ci s'affiche dans la console Cloud.

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

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

Vous pouvez ensuite consulter l'interface Web en cliquant sur la touche Ctrl et en cliquant sur le lien de l'URL locale. Vous pouvez également accéder à l'application frontend en cliquant sur le bouton Aperçu sur le Web en haut à droite de Cloud Editor, puis en sélectionnant Prévisualiser sur le port 8080.

L'interface Web s'affiche, mais vous obtenez une erreur attendue lorsque vous essayez d'envoyer un chat, car le service backend n'est pas encore configuré.

Laissez le service s'exécuter et ne l'arrêtez pas tout de suite. En attendant, nous pouvons discuter des composants de code importants ici.

Explication du code

Le code permettant d'envoyer des données de l'interface Web au backend se trouve dans cette partie.

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
    ...

Lorsque nous souhaitons envoyer des données multimodales à Gemini et les rendre accessibles entre les services, l'un des mécanismes que nous pouvons utiliser consiste à convertir les données en type de données base64, comme indiqué dans le code. Nous devons également déclarer le type MIME des données. Toutefois, l'API Gemini ne peut pas prendre en charge tous les types MIME existants. Il est donc important de connaître les types MIME compatibles avec Gemini. Pour en savoir plus, consultez cette documentation. Vous trouverez des informations sur chacune des fonctionnalités de l'API Gemini (par exemple, Vision).

De plus, dans une interface de chat, il est également important d'envoyer l'historique des discussions comme contexte supplémentaire pour donner à Gemini une "mémoire" de la conversation. Dans cette interface Web, nous envoyons également l'historique des discussions géré par session Web par Gradio, ainsi que le message saisi par l'utilisateur. Nous permettons également à l'utilisateur de modifier l'instruction système et de l'envoyer.

4. Créer un service de backend avec FastAPI

Ensuite, nous devrons créer le backend qui pourra gérer la charge utile évoquée précédemment, à savoir le dernier message de l'utilisateur, l'historique des discussions et l'instruction système. Nous utiliserons FastAPI pour créer le service de backend HTTP.

Créez un fichier en cliquant sur File->New Text File (Fichier > Nouveau fichier texte), puis copiez-collez le code suivant et enregistrez-le sous le nom 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)

N'oubliez pas de l'enregistrer sous le nom backend.py. Ensuite, nous pouvons essayer d'exécuter le service de backend. N'oubliez pas que nous avons exécuté le service frontend à l'étape précédente. Nous devons maintenant ouvrir un nouveau terminal et essayer d'exécuter ce service backend.

Créez un terminal. Accédez à votre terminal dans la zone inférieure et recherchez le bouton "+" pour créer un terminal. Vous pouvez également appuyer sur Ctrl+Maj+C pour ouvrir un nouveau terminal.

Ensuite, assurez-vous d'être dans le répertoire de travail gemini-multimodal-chat-assistant, puis exécutez la commande suivante.

uv run backend.py

Si l'opération réussit, un résultat semblable à celui-ci s'affiche :

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)

Explication du code

Définir la route HTTP pour recevoir les demandes de chat

Dans FastAPI, nous définissons la route à l'aide du décorateur app. Nous utilisons également Pydantic pour définir le contrat d'API. Nous spécifions que la route pour générer une réponse se trouve dans la route /chat avec la méthode POST. Ces fonctionnalités sont déclarées dans le code suivant.

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
    ...

Préparer le format de l'historique des discussions du SDK Gemini

Il est important de comprendre comment restructurer l'historique des discussions afin de pouvoir l'insérer en tant que valeur d'argument history lorsque nous initialiserons un client Gemini ultérieurement. Vous pouvez inspecter le code ci-dessous.

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

Pour fournir l'historique des discussions au SDK Gemini, nous devons mettre en forme les données dans le type de données List[Content]. Chaque Content doit avoir au moins une valeur role et parts. role fait référence à la source du message, qu'il s'agisse de user ou de model. parts fait référence à l'invite elle-même, qui peut être uniquement du texte ou une combinaison de différentes modalités. Pour savoir comment structurer les arguments Content, consultez la documentation.

Gérer les données non textuelles ( multimodales)

Comme mentionné précédemment dans la section sur le frontend, l'une des méthodes pour envoyer des données non textuelles ou multimodales consiste à les envoyer sous forme de chaîne base64. Nous devons également spécifier le type MIME des données pour qu'elles puissent être interprétées correctement. Par exemple, nous devons fournir le type MIME image/jpeg si nous envoyons des données d'image avec le suffixe .jpg.

Cette partie du code convertit les données base64 au format Part.from_bytes à partir du SDK 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. Test d'intégration

Vous devriez maintenant avoir plusieurs services exécutés dans différents onglets de la console Cloud :

Le service de frontend s'exécute sur le port 8080.

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

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

Le service de backend s'exécute sur le port 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)

À l'heure actuelle, vous devriez pouvoir envoyer vos documents et discuter avec l'assistant depuis l'application Web sur le port 8080. Vous pouvez commencer à faire des tests en important des fichiers et en posant des questions. Sachez que certains types de fichiers ne sont pas encore acceptés et généreront une erreur.

Vous pouvez également modifier les instructions système dans le champ Entrées supplémentaires situé sous la zone de texte.

6. Déployer sur Cloud Run

Bien sûr, nous voulons présenter cette application incroyable aux autres. Pour ce faire, nous pouvons empaqueter cette application et la déployer sur Cloud Run en tant que service public accessible à d'autres utilisateurs. Pour ce faire, revenons à l'architecture

Dans cet atelier de programmation, nous allons placer les services de frontend et de backend dans un seul conteneur. Nous aurons besoin de l'aide de supervisord pour gérer les deux services.

Créez un fichier en cliquant sur Fichier > Nouveau fichier texte, puis copiez-collez le code suivant et enregistrez-le sous le nom 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

Ensuite, nous aurons besoin de notre fichier Dockerfile. Cliquez sur Fichier > Nouveau fichier texte, puis copiez et collez le code suivant et enregistrez-le sous le nom 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"]

À ce stade, nous disposons déjà de tous les fichiers nécessaires pour déployer nos applications sur Cloud Run. Déployons-les. Accédez au terminal Cloud Shell et assurez-vous que le projet actuel est configuré sur votre projet actif. Si ce n'est pas le cas, utilisez la commande gcloud configure pour définir l'ID du projet :

gcloud config set project [PROJECT_ID]

Exécutez ensuite la commande suivante pour le déployer sur Cloud Run.

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

Vous serez invité à saisir un nom pour votre service, par exemple gemini-multimodal-chat-assistant. Comme nous avons un Dockerfile dans le répertoire de travail de notre application, il créera le conteneur Docker et le transférera vers Artifact Registry. Vous serez également invité à créer le dépôt Artifact Registry dans la région. Répondez Y. Dites également y lorsque vous êtes invité à autoriser les appels non authentifiés. Notez que nous autorisons ici l'accès non authentifié, car il s'agit d'une application de démonstration. Nous vous recommandons d'utiliser une authentification appropriée pour vos applications d'entreprise et de production.

Une fois le déploiement terminé, vous devriez obtenir un lien semblable à celui ci-dessous :

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

N'hésitez pas à utiliser votre application depuis la fenêtre de navigation privée ou votre appareil mobile. Il devrait déjà être en ligne.

7. Défi

Il est maintenant temps de briller et de perfectionner vos compétences en matière d'exploration. Pensez-vous pouvoir modifier le code pour que l'assistant puisse lire des fichiers audio ou vidéo ?

8. Effectuer un nettoyage

Pour éviter que les ressources utilisées dans cet atelier de programmation soient facturées sur votre compte Google Cloud, procédez comme suit :

Dans la console Google Cloud, accédez à la page Gérer les ressources.
Dans la liste des projets, sélectionnez le projet que vous souhaitez supprimer, puis cliquez sur Supprimer.
Dans la boîte de dialogue, saisissez l'ID du projet, puis cliquez sur Arrêter pour supprimer le projet.
Vous pouvez également accéder à Cloud Run dans la console, sélectionner le service que vous venez de déployer, puis le supprimer.