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, mit der Sie kommunizieren, Dokumente oder Bilder hochladen und diese diskutieren können. Die Anwendung selbst ist in zwei Dienste unterteilt: Frontend und Backend. So können Sie schnell einen Prototyp erstellen und ausprobieren und auch nachvollziehen, wie der API-Vertrag für die Integration beider Dienste aussieht.

In diesem Codelab gehen Sie schrittweise so vor:

  1. Google Cloud-Projekt vorbereiten und alle erforderlichen APIs aktivieren
  2. Front-End-Dienst – Chatoberfläche mit der Gradio-Bibliothek erstellen
  3. Backend-Dienst – HTTP-Server – mit FastAPI erstellen, der die eingehenden Daten in das Gemini SDK-Standardformat umwandelt und die Kommunikation mit der Gemini API ermöglicht
  4. Umgebungsvariablen verwalten und erforderliche Dateien einrichten, die zum Bereitstellen der Anwendung in Cloud Run benötigt werden
  5. Anwendung in Cloud Run bereitstellen

5bcfa1cce6618305.png

Architekturübersicht

b102df2c3f1adabf.jpeg

Voraussetzungen

Lerninhalte

  • So verwenden Sie das Gemini SDK, um Text und andere Datentypen (multimodal) zu übermitteln und eine Textantwort zu generieren
  • Chatverlauf im Gemini SDK strukturieren, um den Unterhaltungskontext beizubehalten
  • Frontend-Webprototyping mit Gradio
  • Entwicklung von Backend-Diensten mit FastAPI und Pydantic
  • Umgebungsvariablen in der YAML-Datei mit Pydantic-settings verwalten
  • Anwendung mit Dockerfile in Cloud Run bereitstellen und Umgebungsvariablen mit YAML-Datei bereitstellen

Voraussetzungen

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

In diesem Codelab, das sich an Entwickler*innen aller Erfahrungsstufen (auch Anfänger*innen) richtet, wird Python in der Beispielanwendung verwendet. 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 kein Konto haben, können Sie der Anleitung unten folgen, um eines zu erstellen.

  1. 2. Wählen Sie in der Google Cloud Console auf der Seite zur 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 in der bq vorinstalliert ist. Klicken Sie oben in der Google Cloud Console auf „Cloud Shell aktivieren“.

1829c3759227c19b.png

  1. Sobald die Verbindung mit der Cloud Shell hergestellt ist, prüfen Sie mit dem folgenden Befehl, ob Sie bereits authentifiziert sind und für das Projekt schon Ihre Projekt-ID eingestellt ist:
gcloud auth list
  1. Führen Sie den folgenden Befehl in Cloud Shell aus, um zu bestätigen, dass 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 sehen.

4032c45803813f30.jpeg

Klicken Sie darauf. Rechts sehen Sie dann alle Ihre Projekte und die Projekt-ID.

8dc17eb4271de6b5.jpeg

  1. Aktivieren Sie die erforderlichen APIs mit dem unten gezeigten Befehl. Das kann einige Minuten dauern.
gcloud services enable aiplatform.googleapis.com \
                           run.googleapis.com \
                           cloudbuild.googleapis.com \
                           cloudresourcemanager.googleapis.com

Bei erfolgreicher Ausführung des Befehls sollte eine Meldung wie die unten gezeigte angezeigt werden:

Operation "operations/..." finished successfully.

Alternativ zum gcloud-Befehl können Sie in der Konsole nach den einzelnen Produkten suchen oder diesen Link verwenden.

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

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“, um den Cloud Shell-Editor zu öffnen. Hier können Sie Ihren Code schreiben b16d56e4979ec951.png.
  2. Achten Sie darauf, dass das Cloud Code-Projekt in der Statusleiste unten links im Cloud Shell-Editor festgelegt ist, wie im Bild unten dargestellt, und auf das aktive Google Cloud-Projekt festgelegt ist, in dem die Abrechnung aktiviert ist. Autorisieren, wenn Sie dazu aufgefordert werden. Nach der Initialisierung des Cloud Shell-Editors kann es eine Weile dauern, bis die Schaltfläche Cloud Code – Anmelden angezeigt wird. Wenn Sie den vorherigen Befehl bereits ausgeführt haben, wird die Schaltfläche möglicherweise direkt auf Ihr aktiviertes Projekt verwiesen, anstatt auf die Schaltfläche „Anmelden“.

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 „Neue Anwendung“ aus.

70f80078e01a02d8.png

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

362ff332256d6933.jpeg

85957565316308d9.jpeg

  1. Speichern Sie die neue Anwendung unter dem gewünschten 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 bereiten Sie die Entwicklungsumgebung vor. In diesem Codelab verwenden wir Python 3.12 und den uv-Python-Projektmanager, um das Erstellen und Verwalten von Python-Versionen und virtuellen Umgebungen zu vereinfachen.

  1. Wenn Sie das Terminal noch nicht geöffnet haben, klicken Sie auf Terminal -> Neues Terminal oder verwenden Sie 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 die Dateien main.py, .python-version und pyproject.toml erstellt. Diese Dateien sind erforderlich, um das Projekt im Verzeichnis zu verwalten. Python-Abhängigkeiten und -Konfigurationen können in den Dateien pyproject.toml und .python-version angegeben werden. Die Python-Version, die für dieses Projekt verwendet wird, ist standardisiert. Weitere Informationen finden Sie in dieser Dokumentation.
main.py
.python-version
pyproject.toml
  1. Überschreiben Sie die Datei „main.py“ mit dem folgenden Code, um sie zu testen.
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 zeigt, dass das Python-Projekt richtig eingerichtet wird. Wir mussten keine virtuelle Umgebung manuell erstellen, da uv dies bereits übernimmt. Ab diesem Punkt wird der Standard-Python-Befehl (z.B. python main.py) durch uv run (z.B. uv run main.py) ersetzt.

Erforderliche Abhängigkeiten installieren

Wir fügen die Abhängigkeiten dieses Codelab-Pakets ebenfalls 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

Der Abschnitt „dependencies“ in pyproject.toml wird entsprechend dem vorherigen Befehl aktualisiert.

Konfigurationsdateien einrichten

Als Nächstes müssen wir Konfigurationsdateien für dieses Projekt einrichten. Konfigurationsdateien werden zum Speichern dynamischer Variablen verwendet, die bei der erneuten Bereitstellung einfach geändert werden können. In diesem Projekt verwenden wir YAML-basierte Konfigurationsdateien mit dem pydantic-settings-Paket, damit es später problemlos in die Cloud Run-Bereitstellung integriert werden kann. pydantic-settings ist ein Python-Paket, mit dem die Typüberprüfung für die Konfigurationsdateien erzwungen werden kann.

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

Aktualisieren Sie die Werte für VERTEXAI_PROJECT_ID entsprechend den Einstellungen, die Sie beim Erstellen des Google Cloud-Projekts ausgewählt haben. In diesem 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 Einstiegspunkt für die Konfigurationswerte in unseren Konfigurationsdateien. Klicken Sie auf Datei > Neue Textdatei und fügen Sie den folgenden Code ein. Speichern Sie die Datei dann als settings.py. Im Code sehen Sie, dass wir explizit festgelegt haben, dass die Datei mit dem Namen settings.yaml gelesen wird.
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()

Diese Konfigurationen ermöglichen es uns, unsere Laufzeit flexibel zu aktualisieren. Bei der ersten Bereitstellung wird die Konfiguration settings.yaml verwendet, damit wir die erste Standardkonfiguration haben. Danach können wir die Umgebungsvariablen flexibel über die Konsole aktualisieren und neu bereitstellen, da die Umgebungsvariablen eine höhere Priorität als die standardmäßige YAML-Konfiguration haben.

Nun können wir mit dem nächsten Schritt fortfahren und die Dienste erstellen.

3. Front-End-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 für 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()`.

Danach können Sie die Weboberfläche aufrufen, indem Sie Strg+Klick auf den lokalen URL-Link ausführen. Alternativ können Sie auch auf die Frontend-Anwendung zugreifen, indem Sie oben rechts im Cloud Editor auf die Schaltfläche Webvorschau klicken und Vorschau auf Port 8080 auswählen.

49cbdfdf77964065.jpeg

Die Weboberfläche wird angezeigt, aber beim Versuch, einen Chat zu senden, wird ein erwarteter 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 wichtigen Codekomponenten hier besprechen.

Erläuterung zum Code

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

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 die Daten zwischen Diensten 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-Typ 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 und in dieser Dokumentation nachgelesen werden können. Die Informationen finden Sie in den einzelnen Gemini API-Funktionen (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. Deshalb senden wir in dieser Weboberfläche auch den Chatverlauf, der von Gradio pro Websitzung verwaltet wird, zusammen mit der Nachrichteneingabe des Nutzers. Außerdem ermöglichen wir dem Nutzer, die Systemanweisung zu ändern und ebenfalls zu 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-Backend-Dienst zu erstellen.

Erstellen Sie eine neue Datei, indem Sie auf Datei > Neue Textdatei klicken,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)

Denken Sie daran, die Datei als backend.py zu speichern. Danach können wir versuchen, den Back-End-Dienst auszuführen. Im vorherigen Schritt haben wir den Frontend-Dienst ausgeführt. Jetzt müssen wir ein neues Terminal öffnen und versuchen, diesen Backend-Dienst auszuführen.

  1. Erstellen Sie ein neues Terminal. Gehen Sie unten zu Ihrem Terminal und suchen Sie nach der Schaltfläche „+“, um ein neues Terminal zu erstellen. Alternativ können Sie Strg + Umschalt + C drücken, um ein neues Terminal zu öffnen.

3e52a362475553dc.jpeg

  1. Prüfen Sie, ob Sie sich im Arbeitsverzeichnis gemini-multimodal-chat-assistant befinden, und führen Sie dann 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 für den Empfang von Chatanfragen definieren

In FastAPI definieren wir die Route mit dem Decorator app. Wir verwenden Pydantic auch, um den API-Vertrag zu definieren. Wir geben an, dass sich die Route zum Generieren von Antworten auf der Route /chat mit der Methode POST befindet. 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
    ...

Gemini SDK-Chatverlaufsformat vorbereiten

Eines der wichtigen Dinge, die verstanden werden müssen, ist, wie wir den Chatverlauf so umstrukturieren können, dass er als history-Argumentwert eingefügt werden kann, wenn wir später einen Gemini-Client initialisieren. Sie können den Code unten prü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 wir den Chatverlauf in das Gemini SDK einfügen können, müssen wir die Daten im Datentyp List[Content] formatieren. Jeder Content-Wert muss mindestens einen role- und einen parts-Wert haben. role bezieht sich auf die Quelle der Nachricht, also entweder user oder model. parts bezieht sich auf den Prompt selbst, der nur Text oder eine Kombination aus verschiedenen Modalitäten sein kann. Weitere Informationen

Nicht-Textdaten ( multimodal) verarbeiten

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

In diesem Teil des Codes werden die base64-Daten in das Part.from_bytes-Format aus dem Gemini SDK umgewandelt.

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 in verschiedenen Cloud Console-Tabs ausgeführt werden:

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

To create a public link, set `share=True` in `launch()`.
  • Backend-Dienst wird auf Port 8081 ausgeführt
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 nahtlos über die Webanwendung auf Port 8080 an den Assistenten senden können. Sie können mit dem Experimentieren beginnen, indem Sie Dateien hochladen und Fragen stellen. Beachten Sie, dass bestimmte Dateitypen noch nicht unterstützt werden und einen Fehler auslösen.

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 vorstellen. Dazu können wir diese Anwendung packen 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 werden wir sowohl den Frontend- als auch den Backend-Dienst in einem Container unterbringen. Wir benötigen die Hilfe von supervisord, um beide Dienste zu verwalten.

Erstellen Sie eine neue Datei, indem Sie auf Datei > Neue Textdatei klicken,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 fügen Sie ihn ein. Speichern Sie ihn dann 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. Stellen wir sie also bereit. Rufen Sie das Cloud Shell-Terminal auf und prüfen Sie, ob das aktuelle Projekt für Ihr aktives Projekt konfiguriert ist. Falls nicht, müssen 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 die Anwendung 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 per Push in die Artifact Registry übertragen. Sie werden auch aufgefordert, das Artifact Registry-Repository in der Region zu erstellen. Antworten Sie mit Y. Antworte auch mit y, wenn du gefragt wirst, ob du nicht authentifizierte Aufrufe zulassen möchtest. 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 nun über das Inkognitofenster oder Ihr Mobilgerät verwenden. Sie sollte bereits aktiv sein.

7. Herausforderung

Jetzt ist es an der Zeit, Ihr Wissen zu vertiefen und Ihre Fähigkeiten zu verbessern. Kannst du den Code so ändern, dass der Assistent Audiodateien 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. Wechseln Sie in der Google Cloud Console zur Seite Ressourcen verwalten.
  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 zu Cloud Run wechseln, den gerade bereitgestellten Dienst auswählen und löschen.