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, avec laquelle vous pourrez communiquer, importer des documents ou des images et en discuter. L'application elle-même est divisée en deux services: le frontend et le backend. Vous pouvez ainsi créer un prototype rapide et tester son fonctionnement, et comprendre comment le contrat d'API permet de les intégrer tous les deux.

Au cours de l'atelier de programmation, vous allez suivre une approche par étapes, comme suit:

  1. Préparer votre projet Google Cloud et y activer toutes les API requises
  2. Créer le service de frontend : interface de chat à l'aide de la bibliothèque Gradio
  3. Créez le service backend (serveur HTTP) à l'aide de FastAPI, qui reformatera les données entrantes selon la norme du SDK Gemini et permettra la communication avec l'API Gemini.
  4. Gérer les variables d'environnement et configurer les fichiers requis pour déployer l'application dans Cloud Run
  5. Déployer l'application dans Cloud Run

5bcfa1cce6618305.png

Présentation de l'architecture

b102df2c3f1adabf.jpeg

Prérequis

Points abordés

  • Utiliser le SDK Gemini pour envoyer du texte et d'autres types de données (multimodaux) et générer une réponse textuelle
  • Structurer l'historique des discussions dans le SDK Gemini pour conserver le contexte de la conversation
  • Prototypage Web côté client avec Gradio
  • Développement de services 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 application exemple. Toutefois, vous n'avez pas besoin de connaître Python pour comprendre les concepts présentés.

2. Avant de commencer

Configurer un projet Cloud dans l'éditeur Cloud Shell

Cet atelier de programmation part du principe 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.

  1. 2 Dans la console Google Cloud, sur la page du sélecteur de projet, sélectionnez ou créez un projet Google Cloud.
  2. 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 .
  3. 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.

1829c3759227c19b.png

  1. Une fois connecté à Cloud Shell, vérifiez que vous êtes déjà authentifié et que le projet est défini avec votre ID de projet à l'aide de la commande suivante:
gcloud auth list
  1. Exécutez la commande suivante dans Cloud Shell pour vérifier que la commande gcloud connaît votre projet.
gcloud config list project
  1. 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 afficher l'ID PROJECT_ID dans la console.

4032c45803813f30.jpeg

Cliquez dessus pour afficher l'ensemble de votre projet et son ID sur la droite.

8dc17eb4271de6b5.jpeg

  1. 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 l'exécution de la commande aboutit, un message semblable à celui ci-dessous s'affiche:

Operation "operations/..." finished successfully.

Vous pouvez également rechercher chaque produit dans la console ou utiliser ce lien.

Si une API est manquante, vous pouvez 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

  1. Cliquez sur le bouton "Ouvrir l'éditeur". Un éditeur Cloud Shell s'ouvre, dans lequel vous pouvez écrire votre code. b16d56e4979ec951.png
  2. 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. Autoriser si vous y êtes invité. Après l'initialisation de l'éditeur Cloud Shell, l'affichage du bouton Cloud Code - Se connecter peut prendre un certain temps. Veuillez patienter. Si vous suivez déjà la commande précédente, le bouton peut également pointer directement vers votre projet activé au lieu du bouton de connexion.

f5003b9c38b43262.png

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

70f80078e01a02d8.png

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

362ff332256d6933.jpeg

85957565316308d9.jpeg

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

8409d8db18690fdf.png

À ce stade, vous devriez déjà être dans le nouveau répertoire de travail de l'application et voir les fichiers suivants :

1ef5bb44f1d2c2a4.png

Ensuite, nous allons préparer notre environnement Python.

Configuration de l'environnement

Préparer l'environnement virtuel Python

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

  1. Si vous n'avez pas encore ouvert le terminal, cliquez sur Terminal -> Nouveau terminal ou appuyez sur Ctrl+Maj+C.

f8457daf0bed059e.jpeg

  1. 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
  1. Initialisons maintenant le projet Python à l'aide de uv.
uv init
  1. 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 les 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 cette documentation.
main.py
.python-version
pyproject.toml
  1. Pour le tester, remplacez le fichier main.py par le code suivant.
def main():
   print("Hello from gemini-multimodal-chat-assistant!")

if __name__ == "__main__":
   main()
  1. Exécutez ensuite la commande suivante :
uv run main.py

Vous obtiendrez un résultat semblable à celui-ci :

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

Cela indique que le projet Python est configuré correctement. Nous n'avons pas eu besoin de créer manuellement un environnement virtuel, car uv s'en charge déjà. À partir de ce moment, 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 de ce package d'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 "dépendances" de pyproject.toml est mise à jour pour refléter la commande précédente.

Configurer les fichiers de configuration

Nous allons maintenant configurer des fichiers de configuration pour ce projet. Les fichiers de configuration permettent de stocker des variables dynamiques qui peuvent facilement être 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 de type pour les fichiers de configuration.

  1. Créez un fichier nommé settings.yaml avec la configuration suivante. Cliquez sur File->New Text File (Fichier > Nouveau fichier texte) et ajoutez le code suivant. Enregistrez-le 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 conformément à ce que vous avez sélectionné lors de la création du projet Google Cloud. Pour cet atelier de programmation, nous utiliserons les valeurs préconfigurées pour VERTEXAI_LOCATION et BACKEND_URL .

  1. Créez ensuite le fichier Python settings.py. Ce module servira d'entrée programmatique pour les valeurs de configuration dans nos fichiers de configuration. Cliquez sur File->New Text File (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 défini explicitement que le fichier nommé settings.yaml est celui qui sera lu.
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 appuyons sur la configuration settings.yaml pour obtenir la première configuration par défaut. Nous pouvons ensuite mettre à jour les variables d'environnement de manière flexible via la console et les redéployer, car nous leur attribuons une priorité plus élevée 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 se présente comme suit :

5bcfa1cce6618305.png

Il contient un champ de saisie permettant aux utilisateurs d'envoyer du texte et d'importer des fichiers. De plus, l'utilisateur peut également écraser l'instruction système qui sera envoyée à l'API Gemini dans le champ d'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,
    )

Nous pouvons ensuite 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 vérifier l'interface Web en appuyant sur Ctrl+clic sur le lien de l'URL locale. Vous pouvez également accéder à l'application frontale en cliquant sur le bouton Web Preview (Aperçu sur le Web) en haut à droite de l'éditeur Cloud, puis en sélectionnant Preview on port 8080 (Prévisualiser sur le port 8080).

49cbdfdf77964065.jpeg

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

bd0464140308cfbe.png

Laissez maintenant 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, nous pouvons utiliser un mécanisme consistant à les convertir 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 n'est pas compatible avec tous les types MIME existants. Il est donc important de savoir quels types MIME sont compatibles avec Gemini. Vous pouvez les consulter dans cette documentation. Vous trouverez ces informations dans 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 en tant que contexte supplémentaire pour fournir à Gemini une "mémoire" de la conversation. Dans cette interface Web, nous envoyons également l'historique de chat géré par Gradio par session Web, ainsi que le message saisi par l'utilisateur. De plus, nous permettons à l'utilisateur de modifier l'instruction système et de l'envoyer également.

4. Créer un service de backend à l'aide de FastAPI

Ensuite, nous devons créer le backend qui peut gérer la charge utile précédemment discutée, 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, cliquez 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. Nous pouvons ensuite essayer d'exécuter le service backend. N'oubliez pas que, à l'étape précédente, nous avons exécuté le service de premier plan. Nous devons maintenant ouvrir un nouveau terminal et essayer d'exécuter ce service de backend.

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

3e52a362475553dc.jpeg

  1. 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
  1. Si l'opération réussit, le résultat est semblable à ceci :
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 le chemin HTTP pour recevoir la requête Chat

Dans FastAPI, nous définissons la route à l'aide du décorateur app. Nous utilisons également Pydantic pour définir le contrat de l'API. Nous spécifions que le chemin d'accès pour générer la réponse se trouve dans le chemin /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 qu'il puisse être inséré en tant que valeur d'argument history lorsque nous allons initialiser un client Gemini plus tard. 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 dans le SDK Gemini, nous devons mettre en forme les données dans le type de données List[Content]. Chaque contenu doit comporter au moins une valeur role et parts. role fait référence à la source du message, qu'il s'agisse d'un utilisateur ou d'un modèle. Les parties font référence à la requête elle-même, qui peut être uniquement du texte ou une combinaison de différentes modalités. Découvrez comment structurer les arguments Content en détail dans cette documentation.

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

Comme indiqué précédemment dans la section "Frontend", l'une des façons d'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 afin 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 un 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:

  • Service de front-end exécuté sur le port 8080
* Running on local URL:  http://0.0.0.0:8080

To create a public link, set `share=True` in `launch()`.
  • Service backend exécuté 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 dans le chat sans problème avec l'assistant depuis l'application Web sur le port 8080. Vous pouvez commencer à tester en important des fichiers et en posant des questions. Notez que certains types de fichiers ne sont pas encore compatibles et qu'ils génèrent une erreur.

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

ee9c849a276d378.png

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 dans Cloud Run en tant que service public accessible par d'autres utilisateurs. Pour ce faire, revoyons l'architecture.

b102df2c3f1adabf.jpeg

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

Créez un fichier, cliquez sur File->New Text File (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

Nous allons ensuite avoir besoin de notre Dockerfile. Cliquez sur File->New Text File (Fichier > Nouveau fichier texte), puis copiez-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 dans 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 transmettra à 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 l'accès non authentifié ici, 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 recevoir un lien semblable à celui-ci:

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

Vous pouvez 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

C'est maintenant votre tour de briller et de perfectionner vos compétences en matière d'exploration. Avez-vous les compétences nécessaires pour modifier le code afin 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 ne soient facturées sur votre compte Google Cloud, procédez comme suit:

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