Multimodalen Assistenten in der Cloud mit Gemini (Python) erstellen und bereitstellen

1. Einführung

In diesem Codelab erstellen Sie eine Anwendung in Form einer Chat-Weboberfläche, über die Sie mit ihr kommunizieren, Dokumente oder Bilder hochladen und darüber diskutieren können. Die Anwendung selbst ist in zwei Dienste unterteilt: Frontend und Backend. So können Sie schnell einen Prototyp erstellen und ausprobieren, wie er sich anfühlt. Außerdem können Sie sich ein Bild davon machen, wie der API-Vertrag für die Integration beider aussieht.

Im Codelab gehen Sie so vor:

  1. Google Cloud-Projekt vorbereiten und alle erforderlichen APIs aktivieren
  2. Front-End-Dienst – Chatoberfläche mit der Gradio-Bibliothek erstellen
  3. Erstellen Sie den Backend-Dienst – HTTP-Server mit FastAPI, der die eingehenden Daten in den Gemini SDK-Standard umformatiert und die Kommunikation mit der Gemini API ermöglicht.
  4. Umgebungsvariablen verwalten und erforderliche Dateien für die Bereitstellung der Anwendung in Cloud Run einrichten
  5. Anwendung in Cloud Run bereitstellen

5bcfa1cce6618305.png

Architekturübersicht

b102df2c3f1adabf.jpeg

Voraussetzungen

Aufgaben in diesem Lab

  • Mit dem Gemini SDK Text und andere Datentypen (multimodal) senden und Textantworten generieren
  • Chatverlauf im Gemini SDK strukturieren, um den Konversationskontext beizubehalten
  • Frontend-Web-Prototyping mit Gradio
  • Entwicklung von Back-End-Diensten mit FastAPI und Pydantic
  • Umgebungsvariablen in einer YAML-Datei mit Pydantic-Einstellungen verwalten
  • Anwendung mit Dockerfile in Cloud Run bereitstellen und Umgebungsvariablen mit YAML-Datei angeben

Voraussetzungen

  • Chrome-Webbrowser
  • Ein Gmail-Konto
  • Ein Cloud-Projekt mit aktivierter Abrechnung

Dieses Codelab richtet sich an Entwickler aller Stufen (einschließlich Anfänger) und verwendet in der Beispielanwendung Python. Python-Kenntnisse sind jedoch nicht erforderlich, um die vorgestellten Konzepte zu verstehen.

2. Hinweis

Cloud-Projekt im Cloud Shell-Editor einrichten

In diesem Codelab wird davon ausgegangen, dass Sie bereits ein Google Cloud-Projekt mit aktivierter Abrechnung haben. Wenn Sie noch keine haben, können Sie der Anleitung unten folgen, um loszulegen.

  1. 2. Wählen Sie in der Google Cloud Console auf der Seite der Projektauswahl ein Google Cloud-Projekt aus oder erstellen Sie eines.
  2. Die Abrechnung für das Cloud-Projekt muss aktiviert sein. So prüfen Sie, ob die Abrechnung für ein Projekt aktiviert ist.
  3. Sie verwenden Cloud Shell, eine Befehlszeilenumgebung, die in Google Cloud ausgeführt wird und bq bereits vorinstalliert hat. Klicken Sie oben in der Google Cloud Console auf „Cloud Shell aktivieren“.

1829c3759227c19b.png

  1. Nachdem Sie eine Verbindung zu Cloud Shell hergestellt haben, prüfen Sie mit dem folgenden Befehl, ob Sie bereits authentifiziert sind und das Projekt auf Ihre Projekt-ID festgelegt ist:
gcloud auth list
  1. Führen Sie in Cloud Shell den folgenden Befehl aus, um zu prüfen, ob der gcloud-Befehl Ihr Projekt kennt.
gcloud config list project
  1. Wenn Ihr Projekt nicht festgelegt ist, verwenden Sie den folgenden Befehl, um es festzulegen:
gcloud config set project <YOUR_PROJECT_ID>

Alternativ können Sie die PROJECT_ID-ID auch in der Console aufrufen.

4032c45803813f30.jpeg

Klicken Sie darauf. Rechts sehen Sie dann Ihr gesamtes Projekt und die Projekt-ID.

8dc17eb4271de6b5.jpeg

  1. Aktivieren Sie die erforderlichen APIs mit dem folgenden Befehl. Dies kann einige Minuten dauern. Bitte haben Sie etwas Geduld.
gcloud services enable aiplatform.googleapis.com \
                           run.googleapis.com \
                           cloudbuild.googleapis.com \
                           cloudresourcemanager.googleapis.com

Wenn der Befehl erfolgreich ausgeführt wurde, sollte eine Meldung ähnlich der folgenden angezeigt werden:

Operation "operations/..." finished successfully.

Alternativ können Sie in der Console nach den einzelnen Produkten suchen oder diesen Link verwenden.

Wenn eine API fehlt, können Sie sie jederzeit während der Implementierung aktivieren.

Weitere Informationen zu gcloud-Befehlen und deren Verwendung finden Sie in der Dokumentation.

Arbeitsverzeichnis der Anwendung einrichten

  1. Klicken Sie auf die Schaltfläche „Editor öffnen“. Dadurch wird der Cloud Shell-Editor geöffnet, in dem Sie Ihren Code schreiben können. b16d56e4979ec951.png
  2. Achten Sie darauf, dass das Cloud Code-Projekt in der unteren linken Ecke (Statusleiste) des Cloud Shell-Editors festgelegt ist, wie im Bild unten hervorgehoben. Es muss sich dabei um das aktive Google Cloud-Projekt handeln, in dem Sie die Abrechnung aktiviert haben. Klicken Sie auf Autorisieren, wenn Sie dazu aufgefordert werden. Nach der Initialisierung des Cloud Shell-Editors kann es einige Zeit dauern, bis die Schaltfläche Cloud Code – Anmelden angezeigt wird. Bitte haben Sie etwas Geduld. Wenn Sie den vorherigen Befehl bereits ausgeführt haben, verweist die Schaltfläche möglicherweise direkt auf Ihr aktiviertes Projekt und nicht auf die Anmeldeschaltfläche.

f5003b9c38b43262.png

  1. Klicken Sie in der Statusleiste auf das aktive Projekt und warten Sie, bis das Cloud Code-Pop-up geöffnet wird. Wählen Sie im Pop-up-Fenster die Option „Neue Anwendung“ aus.

70f80078e01a02d8.png

  1. Wählen Sie in der Liste der Anwendungen Gemini Generative AI und dann Gemini API Python aus.

362ff332256d6933.jpeg

85957565316308d9.jpeg

  1. Speichern Sie die neue Anwendung unter einem beliebigen Namen. In diesem Beispiel verwenden wir gemini-multimodal-chat-assistant. Klicken Sie dann auf OK.

8409d8db18690fdf.png

Sie sollten sich jetzt im Arbeitsverzeichnis der neuen Anwendung befinden und die folgenden Dateien sehen:

1ef5bb44f1d2c2a4.png

Als Nächstes bereiten wir unsere Python-Umgebung vor.

Umgebung einrichten

Virtuelle Python-Umgebung vorbereiten

Im nächsten Schritt wird die Entwicklungsumgebung vorbereitet. In diesem Codelab verwenden wir Python 3.12 und den uv Python-Projektmanager, um das Erstellen und Verwalten der Python-Version und der virtuellen Umgebung zu vereinfachen.

  1. Wenn Sie das Terminal noch nicht geöffnet haben, klicken Sie auf Terminal -> Neues Terminal oder verwenden Sie die Tastenkombination Strg + Umschalt + C.

f8457daf0bed059e.jpeg

  1. Laden Sie uv herunter und installieren Sie Python 3.12 mit dem folgenden Befehl:
curl -LsSf https://astral.sh/uv/0.6.6/install.sh | sh && \
source $HOME/.local/bin/env && \
uv python install 3.12
  1. Initialisieren wir nun das Python-Projekt mit uv.
uv init
  1. Im Verzeichnis werden main.py, .python-version und pyproject.toml erstellt. Diese Dateien sind für die Verwaltung des Projekts im Verzeichnis erforderlich. Python-Abhängigkeiten und ‑Konfigurationen können in pyproject.toml und .python-version angegeben werden. Die für dieses Projekt verwendete Python-Version ist standardisiert. Weitere Informationen finden Sie in der Dokumentation.
main.py
.python-version
pyproject.toml
  1. Zum Testen überschreiben Sie main.py mit dem folgenden Code:
def main():
   print("Hello from gemini-multimodal-chat-assistant!")

if __name__ == "__main__":
   main()
  1. Führen Sie dann den folgenden Befehl aus:
uv run main.py

Die Ausgabe sollte in etwa so aussehen:

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

Das Python-Projekt wird also richtig eingerichtet. Wir mussten keine virtuelle Umgebung manuell erstellen, da dies bereits von uv erledigt wird. Ab diesem Zeitpunkt wird der Standard-Python-Befehl (z.B. python main.py) durch uv run ersetzt (z.B. uv run main.py).

Erforderliche Abhängigkeiten installieren

Wir fügen diesem Codelab-Paket Abhängigkeiten mit dem Befehl uv hinzu. Führen Sie den folgenden Befehl aus:

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

Sie sehen, dass der Abschnitt „dependencies“ in pyproject.toml anhand des vorherigen Befehls aktualisiert wird.

Konfigurationsdateien einrichten

Jetzt müssen wir Konfigurationsdateien für dieses Projekt einrichten. In Konfigurationsdateien werden dynamische Variablen gespeichert, die bei der Neubereitstellung leicht geändert werden können. In diesem Projekt verwenden wir YAML-basierte Konfigurationsdateien mit dem Paket pydantic-settings, damit es später problemlos in die Cloud Run-Bereitstellung eingebunden werden kann. pydantic-settings ist ein Python-Paket, mit dem die Typprüfung für die Konfigurationsdateien erzwungen werden kann.

  1. Erstellen Sie eine Datei namens settings.yaml mit der folgenden Konfiguration. Klicken Sie auf Datei -> Neue Textdatei und fügen Sie den folgenden Code ein. Speichern Sie sie dann als settings.yaml.
VERTEXAI_LOCATION: "us-central1"
VERTEXAI_PROJECT_ID: "{YOUR-PROJECT-ID}"
BACKEND_URL: "http://localhost:8081/chat"

Bitte aktualisieren Sie die Werte für VERTEXAI_PROJECT_ID gemäß den Angaben, die Sie beim Erstellen des Google Cloud-Projekts ausgewählt haben. Für dieses Codelab verwenden wir die vorkonfigurierten Werte für VERTEXAI_LOCATION und BACKEND_URL .

  1. Erstellen Sie dann die Python-Datei settings.py. Dieses Modul dient als programmatischer Eintrag für die Konfigurationswerte in unseren Konfigurationsdateien. Klicken Sie auf Datei -> Neue Textdatei und fügen Sie den folgenden Code ein. Speichern Sie den Code dann als settings.py. Im Code ist zu sehen, dass wir explizit festgelegt haben, dass die Datei settings.yaml gelesen werden soll.
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()

Mit diesen Konfigurationen können wir unsere Laufzeit flexibel aktualisieren. Bei der Erstbereitstellung verwenden wir die Konfiguration settings.yaml, um die erste Standardkonfiguration zu erhalten. Danach können wir die Umgebungsvariablen flexibel über die Konsole aktualisieren und neu bereitstellen, da sie gegenüber der Standard-YAML-Konfiguration eine höhere Priorität haben.

Jetzt können wir mit dem nächsten Schritt fortfahren: dem Erstellen der Dienste.

3. Frontend-Dienst mit Gradio erstellen

Wir erstellen eine Chat-Weboberfläche, die so aussieht:

5bcfa1cce6618305.png

Es enthält ein Eingabefeld, über das Nutzer Text senden und Dateien hochladen können. Außerdem können Nutzer die Systemanweisung, die an die Gemini API gesendet wird, im Feld „Zusätzliche Eingaben“ überschreiben.

Wir erstellen den Front-End-Dienst mit Gradio. Benennen Sie main.py in frontend.py um und überschreiben Sie den Code mit dem folgenden Code:

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

Anschließend können wir versuchen, den Frontend-Dienst mit dem folgenden Befehl auszuführen. Vergessen Sie nicht, die Datei main.py in frontend.py umzubenennen.

uv run frontend.py

In der Cloud Console wird eine ähnliche Ausgabe angezeigt.

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

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

Anschließend können Sie die Weboberfläche prüfen, indem Sie Strg + klicken, um den Link zur lokalen URL aufzurufen. Alternativ können Sie auch über die Schaltfläche Webvorschau oben rechts im Cloud-Editor auf die Frontend-Anwendung zugreifen und Vorschau auf Port 8080 auswählen.

49cbdfdf77964065.jpeg

Die Weboberfläche wird angezeigt, aber beim Senden eines Chats wird der erwartete Fehler angezeigt, da der Backend-Dienst noch nicht eingerichtet ist.

bd0464140308cfbe.png

Lassen Sie den Dienst jetzt laufen und beenden Sie ihn noch nicht. In der Zwischenzeit können wir die wichtigsten Codekomponenten besprechen.

Erläuterung zum Code

Der Code zum Senden von Daten von der Weboberfläche an das Backend befindet sich in diesem Teil.

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

Wenn wir multimodale Daten an Gemini senden und für Dienste zugänglich machen möchten, können wir die Daten in den im Code deklarierten Datentyp base64 konvertieren. Außerdem müssen wir den MIME der Daten angeben. Die Gemini API kann jedoch nicht alle vorhandenen MIME-Typen unterstützen. Daher ist es wichtig zu wissen, welche MIME-Typen von Gemini unterstützt werden. Weitere Informationen finden Sie in dieser Dokumentation. Sie finden die Informationen in den einzelnen Funktionen der Gemini API (z. B. Vision).

Außerdem ist es in einer Chatoberfläche wichtig, den Chatverlauf als zusätzlichen Kontext zu senden, damit Gemini sich an die Unterhaltung erinnern kann. Daher senden wir über diese Weboberfläche auch den Chatverlauf, der von Gradio pro Websitzung verwaltet wird, zusammen mit der vom Nutzer eingegebenen Nachricht. Außerdem können Nutzer die Systemanleitung ändern und ebenfalls senden.

4. Backend-Dienst mit FastAPI erstellen

Als Nächstes müssen wir das Backend erstellen, das die zuvor besprochene Nutzlast, die letzte Nutzernachricht, den Chatverlauf und die Systemanweisung verarbeiten kann. Wir verwenden FastAPI, um den HTTP-Backenddienst zu erstellen.

Erstellen Sie eine neue Datei, klicken Sie auf Datei -> Neue Textdatei,kopieren Sie den folgenden Code und fügen Sie ihn ein. Speichern Sie die Datei dann als 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)

Speichern Sie das Script als backend.py. Danach können wir versuchen, den Back-End-Dienst auszuführen. Denken Sie daran, dass wir im vorherigen Schritt den Frontend-Dienst ausgeführt haben. Jetzt müssen wir ein neues Terminal öffnen und versuchen, diesen Backend-Dienst auszuführen.

  1. Erstellen Sie ein neues Terminal. Rufen Sie unten Ihr Terminal auf und klicken Sie auf das Pluszeichen, um ein neues Terminal zu erstellen. Alternativ können Sie Strg + Umschalttaste + C drücken, um ein neues Terminal zu öffnen.

3e52a362475553dc.jpeg

  1. Prüfen Sie danach, ob Sie sich im Arbeitsverzeichnis gemini-multimodal-chat-assistant befinden, und führen Sie den folgenden Befehl aus:
uv run backend.py
  1. Bei Erfolg wird eine Ausgabe wie diese angezeigt:
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)

Erläuterung zum Code

HTTP-Route zum Empfangen von Chatanfragen definieren

In FastAPI definieren wir die Route mit dem Decorator app. Außerdem verwenden wir Pydantic, um den API-Vertrag zu definieren. Wir geben an, dass die Route zum Generieren der Antwort /chat mit der Methode POST ist. Diese Funktionen werden im folgenden Code deklariert

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

Format für den Gemini SDK-Chatverlauf vorbereiten

Eines der wichtigsten Dinge, die wir verstehen müssen, ist, wie wir den Chatverlauf so umstrukturieren können, dass er später als Argumentwert history eingefügt werden kann, wenn wir einen Gemini-Client initialisieren. Sie können den Code unten überprüfen.

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

Damit der Chatverlauf in das Gemini SDK aufgenommen werden kann, müssen die Daten im Datentyp List[Content] formatiert werden. Jeder Inhalt muss mindestens einen Wert für role und parts haben. role bezieht sich auf die Quelle der Nachricht, also user oder model. Dabei bezieht sich parts auf den Prompt selbst, der nur aus Text oder einer Kombination verschiedener Modalitäten bestehen kann. Weitere Informationen zum Strukturieren von Inhaltsargumenten findest du in dieser Dokumentation.

Nicht textbasierte ( multimodale) Daten verarbeiten

Wie bereits im Abschnitt zum Frontend erwähnt, können Sie nicht textbasierte oder multimodale Daten als base64 senden. Außerdem müssen wir den MIME-Typ für die Daten angeben, damit sie richtig interpretiert werden können. Geben wir beispielsweise den MIME-Typ image/jpeg an, wenn wir Bilddaten mit dem Suffix .jpg senden.

Dieser Teil des Codes wandelt die base64-Daten in das Part.from_bytes-Format aus dem Gemini SDK um.

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

Jetzt sollten mehrere Dienste auf verschiedenen Cloud Console-Tabs ausgeführt werden:

  • Frontend-Dienst, der auf Port 8080 ausgeführt wird
* Running on local URL:  http://0.0.0.0:8080

To create a public link, set `share=True` in `launch()`.
  • Backenddienst, der auf Port 8081 ausgeführt wird
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)

Derzeit sollten Sie Ihre Dokumente über die Webanwendung auf Port 8080 nahtlos an den Assistenten im Chat senden können. Sie können mit dem Experimentieren beginnen, indem Sie Dateien hochladen und Fragen stellen. Bestimmte Dateitypen werden noch nicht unterstützt und es wird ein Fehler ausgegeben.

Sie können die Systemanweisungen auch über das Feld Zusätzliche Eingaben unter dem Textfeld bearbeiten.

ee9c849a276d378.png

6. In Cloud Run bereitstellen

Natürlich möchten wir diese tolle App auch anderen zeigen. Dazu können wir diese Anwendung verpacken und als öffentlichen Dienst in Cloud Run bereitstellen, auf den andere zugreifen können. Sehen wir uns dazu noch einmal die Architektur an.

b102df2c3f1adabf.jpeg

In diesem Codelab platzieren wir sowohl den Frontend- als auch den Backend-Dienst in einem Container. Für die Verwaltung beider Dienste benötigen wir supervisord.

Erstellen Sie eine neue Datei, klicken Sie auf Datei -> Neue Textdatei,kopieren Sie den folgenden Code und fügen Sie ihn ein. Speichern Sie die Datei dann als 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

Als Nächstes benötigen wir unser Dockerfile. Klicken Sie auf Datei -> Neue Textdatei, kopieren Sie den folgenden Code und speichern Sie ihn als 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"]

Wir haben jetzt alle Dateien, die zum Bereitstellen unserer Anwendungen in Cloud Run erforderlich sind. Jetzt können wir sie bereitstellen. Rufen Sie das Cloud Shell-Terminal auf und prüfen Sie, ob das aktuelle Projekt auf Ihr aktives Projekt konfiguriert ist. Falls nicht, können Sie die Projekt-ID mit dem Befehl „gcloud configure“ festlegen:

gcloud config set project [PROJECT_ID]

Führen Sie dann den folgenden Befehl aus, um ihn in Cloud Run bereitzustellen.

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

Sie werden aufgefordert, einen Namen für Ihren Dienst einzugeben, z. B. gemini-multimodal-chat-assistant. Da sich das Dockerfile in unserem Arbeitsverzeichnis der Anwendung befindet, wird der Docker-Container erstellt und in die Artifact Registry übertragen. Sie werden auch aufgefordert, zu bestätigen, dass das Artifact Registry-Repository in der Region erstellt werden soll. Antworten Sie dazu mit Y. Antworten Sie auch mit y, wenn Sie gefragt werden, ob Sie nicht authentifizierte Aufrufe zulassen möchten. Hinweis: Wir erlauben hier den nicht authentifizierten Zugriff, da es sich um eine Demoanwendung handelt. Wir empfehlen, für Ihre Unternehmens- und Produktionsanwendungen eine geeignete Authentifizierung zu verwenden.

Nach Abschluss der Bereitstellung sollten Sie einen Link ähnlich dem folgenden erhalten:

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

Sie können die Anwendung jetzt über das Inkognitofenster oder Ihr Mobilgerät verwenden. Sie sollte bereits live sein.

7. Herausforderung

Jetzt ist es an der Zeit, Ihre explorativen Fähigkeiten zu verbessern. Haben Sie die nötigen Kenntnisse, um den Code so zu ändern, dass der Assistant Audio- oder Videodateien lesen kann?

8. Bereinigen

So vermeiden Sie, dass Ihrem Google Cloud-Konto die in diesem Codelab verwendeten Ressourcen in Rechnung gestellt werden:

  1. Rufen Sie in der Google Cloud Console die Seite Ressourcen verwalten auf.
  2. Wählen Sie in der Projektliste das Projekt aus, das Sie löschen möchten, und klicken Sie auf Löschen.
  3. Geben Sie im Dialogfeld die Projekt-ID ein und klicken Sie auf Beenden, um das Projekt zu löschen.
  4. Alternativ können Sie in der Console Cloud Run aufrufen, den gerade bereitgestellten Dienst auswählen und löschen.