Tworzenie i wdrażanie multimodalnego asystenta w chmurze za pomocą Gemini (Python)

1. Wprowadzenie

W tym laboratorium kodowania utworzysz aplikację w postaci internetowego interfejsu czatu, w którym możesz się komunikować, przesyłać dokumenty lub obrazy i omawiać je. Aplikacja jest podzielona na 2 usługi: frontend i backend. Dzięki temu możesz szybko utworzyć prototyp i sprawdzić, jak działa, a także dowiedzieć się, jak wygląda umowa API, aby zintegrować obie usługi.

W ramach ćwiczeń z programowania będziesz wykonywać kolejne czynności:

1. Przygotowywanie projektu Google Cloud i włączanie w nim wszystkich wymaganych interfejsów API
2. Utwórz usługę frontendu – interfejs czatu za pomocą biblioteki Gradio.
3. Utwórz usługę backendu – serwer HTTP korzystający z FastAPI, który będzie przekształcać dane przychodzące do standardu pakietu Gemini SDK i umożliwiać komunikację z interfejsem Gemini API.
4. zarządzać zmiennymi środowiskowymi i konfigurować wymagane pliki potrzebne do wdrożenia aplikacji w Cloud Run,
5. Wdrażanie aplikacji w Cloud Run

Omówienie architektury

Wymagania wstępne

• Umiejętność korzystania z Gemini API i pakietu Google Gen AI SDK.
• Znajomość podstawowej architektury pełnego stosu z użyciem usługi HTTP

Czego się nauczysz

• Jak używać pakietu Gemini SDK do przesyłania tekstu i innych typów danych (multimodalnych) oraz generowania odpowiedzi tekstowych
• Jak przekształcić historię czatu w pakiet SDK Gemini, aby zachować kontekst rozmowy
• Prototypowanie frontendu za pomocą Gradio
• Tworzenie usługi backendu za pomocą FastAPI i Pydantic
• Zarządzanie zmiennymi środowiskowymi w pliku YAML za pomocą Pydantic-settings
• Wdrażanie aplikacji w Cloud Run przy użyciu pliku Dockerfile i podawanie zmiennych środowiskowych za pomocą pliku YAML

Czego potrzebujesz

• przeglądarki Chrome,
• konto Gmail,
• Projekt w chmurze z włączonymi płatnościami

Ten przewodnik, przeznaczony dla deweloperów na wszystkich poziomach zaawansowania (w tym dla początkujących), wykorzystuje w przykładowej aplikacji język Python. Znajomość Pythona nie jest jednak wymagana do zrozumienia przedstawionych koncepcji.

2. Zanim zaczniesz

Konfigurowanie projektu w Cloud Shell Editor

W tym samouczku zakładamy, że masz już projekt Google Cloud z włączonymi płatnościami. Jeśli jeszcze go nie masz, możesz zacząć, wykonując czynności opisane poniżej.

1. 2 W konsoli Google Cloud na stronie selektora projektu wybierz lub utwórz projekt Google Cloud.
2. Sprawdź, czy w projekcie Cloud włączone są płatności. Dowiedz się, jak sprawdzić, czy w projekcie są włączone płatności .
3. Będziesz używać Cloud Shell, czyli środowiska wiersza poleceń działającego w Google Cloud, które jest wstępnie załadowane narzędziem bq. U góry konsoli Google Cloud kliknij Aktywuj Cloud Shell.

4. Po połączeniu z Cloud Shell sprawdź, czy uwierzytelnianie zostało już przeprowadzone, a projekt jest już ustawiony na Twój identyfikator projektu, używając tego polecenia:

gcloud auth list

5. Aby potwierdzić, że polecenie gcloud zna Twój projekt, uruchom w Cloud Shell to polecenie:

gcloud config list project

6. Jeśli projekt nie jest ustawiony, użyj tego polecenia, aby go ustawić:

gcloud config set project <YOUR_PROJECT_ID>

Możesz też zobaczyć PROJECT_ID id w konsoli.

Kliknij go, a po prawej stronie zobaczysz wszystkie swoje projekty i identyfikator projektu.

7. Włącz wymagane interfejsy API za pomocą polecenia pokazanego poniżej. Może to potrwać kilka minut, więc zachowaj cierpliwość.

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

Po pomyślnym wykonaniu polecenia powinien wyświetlić się komunikat podobny do tego poniżej:

Operation "operations/..." finished successfully.

Alternatywą dla polecenia gcloud jest wyszukanie poszczególnych usług w konsoli lub skorzystanie z tego linku.

Jeśli pominiesz jakiś interfejs API, możesz go włączyć w trakcie wdrażania.

Informacje o poleceniach gcloud i ich użyciu znajdziesz w dokumentacji.

Konfigurowanie katalogu roboczego aplikacji

1. Kliknij przycisk Otwórz edytor. Spowoduje to otwarcie edytora Cloud Shell, w którym możesz pisać kod.
2. Sprawdź, czy projekt Cloud Code jest ustawiony w lewym dolnym rogu (na pasku stanu) edytora Cloud Shell, jak pokazano na ilustracji poniżej, i czy jest ustawiony jako aktywny projekt Google Cloud, w którym masz włączone płatności. W razie potrzeby kliknij Autoryzuj. Po zainicjowaniu edytora Cloud Shell może minąć trochę czasu, zanim pojawi się przycisk Cloud Code – zaloguj się. Prosimy o cierpliwość. Jeśli wykonasz poprzednie polecenie, przycisk może też prowadzić bezpośrednio do aktywowanego projektu zamiast do przycisku logowania.

3. Kliknij aktywny projekt na pasku stanu i poczekaj na otwarcie wyskakującego okienka Cloud Code. W wyskakującym okienku wybierz „Nowa aplikacja”.

4. Z listy aplikacji wybierz Generatywna AI Gemini, a potem Gemini API Python.

5. Zapisz nową aplikację pod wybraną nazwą. W tym przykładzie użyjemy nazwy gemini-multimodal-chat-assistant, a następnie kliknij OK.

W tym momencie powinna być już otwarta nowa aplikacja w katalogu roboczym, w którym znajdują się te pliki:

Następnie przygotujemy środowisko Pythona.

Konfiguracja środowiska

Przygotowywanie wirtualnego środowiska Pythona

Następnym krokiem jest przygotowanie środowiska programistycznego. W tym samouczku użyjemy Pythona 3.12 i menedżera projektów Pythona uv, aby uprościć tworzenie i zarządzanie wersją Pythona oraz środowiskiem wirtualnym.

1. Jeśli terminal nie jest jeszcze otwarty, otwórz go , klikając Terminal –> Nowy terminal lub używając skrótu Ctrl + Shift + C.

2. Pobierz uv i zainstaluj Pythona 3.12 za pomocą tego polecenia:

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

3. Teraz zainicjujmy projekt w Pythonie za pomocą uv

uv init

4. W katalogu zobaczysz utworzone pliki main.py, .python-version i pyproject.toml. Te pliki są potrzebne do utrzymania projektu w katalogu. Zależności i konfiguracje Pythona można określić w plikach pyproject.toml i .python-version, które standardowo określają wersję Pythona używaną w tym projekcie. Więcej informacji znajdziesz w dokumentacji.

main.py
.python-version
pyproject.toml

5. Aby go przetestować, zastąp plik main.py tym kodem:

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

if __name__ == "__main__":
   main()

6. Następnie uruchom to polecenie:

uv run main.py

Otrzymasz dane wyjściowe podobne do tych poniżej.

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

Oznacza to, że projekt w Pythonie jest prawidłowo konfigurowany. Nie musieliśmy ręcznie tworzyć środowiska wirtualnego, ponieważ uv już to robi. Od tego momentu standardowe polecenie Pythona (np. python main.py) zostanie zastąpione poleceniem uv run (np. uv run main.py).

Instalowanie wymaganych zależności

Zależności pakietu ćwiczenia dodamy również za pomocą polecenia uv. Uruchom to polecenie:

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

Zauważysz, że sekcja „dependencies” w pliku pyproject.toml zostanie zaktualizowana zgodnie z poprzednim poleceniem.

Konfigurowanie plików konfiguracji

Teraz musimy skonfigurować pliki konfiguracji dla tego projektu. Pliki konfiguracyjne służą do przechowywania zmiennych dynamicznych, które można łatwo zmienić podczas ponownego wdrażania. W tym projekcie będziemy używać plików konfiguracji opartych na YAML z pakietem pydantic-settings, dzięki czemu będzie można je łatwo zintegrować z wdrożeniem Cloud Run w późniejszym czasie. pydantic-settings to pakiet Pythona, który może wymuszać sprawdzanie typów w plikach konfiguracji.

1. Utwórz plik o nazwie settings.yaml z tą konfiguracją. Kliknij Plik –> Nowy plik tekstowy i wypełnij go tym kodem: Następnie zapisz go jako settings.yaml.

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

Zaktualizuj wartości w polu VERTEXAI_PROJECT_ID zgodnie z tym, co zostało wybrane podczas tworzenia projektu Google Cloud. W tym samouczku użyjemy wstępnie skonfigurowanych wartości VERTEXAI_LOCATION i BACKEND_URL .

2. Następnie utwórz plik Pythona settings.py. Ten moduł będzie służyć jako programowe miejsce wprowadzania wartości konfiguracji w naszych plikach konfiguracyjnych. Kliknij Plik –> Nowy plik tekstowy i wypełnij go tym kodem: Następnie zapisz go jako settings.py. W kodzie widać, że jawnie ustawiliśmy, że plik o nazwie settings.yaml będzie odczytywany.

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

Te konfiguracje umożliwiają nam elastyczne aktualizowanie środowiska wykonawczego. Podczas początkowego wdrażania będziemy korzystać z konfiguracji settings.yaml, aby mieć pierwszą konfigurację domyślną. Następnie możemy elastycznie aktualizować zmienne środowiskowe za pomocą konsoli i wdrażać je ponownie, ponieważ zmienne środowiskowe mają wyższy priorytet niż domyślna konfiguracja YAML.

Możemy teraz przejść do następnego kroku, czyli tworzenia usług.

3. Tworzenie usługi frontendu za pomocą Gradio

Utworzymy internetowy interfejs czatu, który będzie wyglądać tak:

Zawiera pole do wprowadzania danych, w którym użytkownicy mogą wysyłać tekst i przesyłać pliki. Użytkownik może też zastąpić instrukcje systemowe, które zostaną wysłane do interfejsu Gemini API w polu dodatkowych danych wejściowych.

Usługę frontendu utworzymy za pomocą Gradio. Zmień nazwę pliku main.py na frontend.py i zastąp kod poniższym kodem.

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

Następnie możemy spróbować uruchomić usługę frontendu za pomocą tego polecenia. Nie zapomnij zmienić nazwy pliku main.py na frontend.py.

uv run frontend.py

W konsoli Google Cloud zobaczysz dane wyjściowe podobne do tych:

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

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

Następnie możesz sprawdzić interfejs internetowy, klikając z naciśniętym klawiszem Ctrl lokalny link do adresu URL. Możesz też otworzyć aplikację frontendową, klikając przycisk Podgląd w przeglądarce w prawym górnym rogu Cloud Editora i wybierając Podejrzyj na porcie 8080.

Zobaczysz interfejs internetowy, ale podczas próby przesłania czatu pojawi się oczekiwany błąd, ponieważ usługa backendu nie jest jeszcze skonfigurowana.

Teraz pozwól usłudze działać i nie zamykaj jej. W międzyczasie możemy omówić tutaj ważne komponenty kodu.

Wyjaśnienie kodu

Kod do wysyłania danych z interfejsu internetowego do backendu znajduje się w tej części.

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

Jeśli chcemy wysłać dane multimodalne do Gemini i udostępnić je między usługami, możemy przekonwertować je na typ danych base64 zadeklarowany w kodzie. Musimy też zadeklarować typ MIME danych. Interfejs Gemini API nie obsługuje jednak wszystkich typów MIME, dlatego warto wiedzieć, które typy MIME obsługiwane przez Gemini można odczytać w tej dokumentacji . Informacje znajdziesz w sekcji dotyczącej poszczególnych funkcji interfejsu Gemini API (np.Vision).

Dodatkowo w interfejsie czatu ważne jest też przesyłanie historii czatu jako dodatkowego kontekstu, aby Gemini „pamiętał” rozmowę. Dlatego w tym interfejsie internetowym wysyłamy też historię czatu, którą Gradio zarządza w ramach sesji internetowej, wraz z wiadomością wpisaną przez użytkownika. Umożliwiamy też użytkownikowi modyfikowanie instrukcji systemowych i ich wysyłanie.

4. Tworzenie usługi backendu za pomocą FastAPI

Następnie musimy utworzyć backend, który będzie obsługiwać omówiony wcześniej ładunek, ostatnią wiadomość użytkownika, historię czatu i instrukcje systemowe. Do utworzenia usługi backendu HTTP użyjemy FastAPI.

Utwórz nowy plik. Kliknij Plik –> Nowy plik tekstowy, skopiuj i wklej poniższy kod,a następnie zapisz go jako 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)

Nie zapomnij zapisać go jako backend.py. Następnie możemy spróbować uruchomić usługę backendu. Pamiętaj, że w poprzednim kroku uruchomiliśmy usługę frontendową. Teraz musimy otworzyć nowy terminal i spróbować uruchomić tę usługę backendową.

1. Utwórz nowy terminal. Przejdź do terminala w dolnej części i znajdź przycisk „+”, aby utworzyć nowy terminal. Możesz też nacisnąć Ctrl + Shift + C, aby otworzyć nowy terminal.

2. Następnie upewnij się, że jesteś w katalogu roboczym gemini-multimodal-chat-assistant, i uruchom to polecenie:

uv run backend.py

3. Jeśli się uda, wyświetli się taki wynik:

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)

Wyjaśnienie kodu

Określanie trasy HTTP do odbierania żądań czatu

W FastAPI definiujemy trasę za pomocą dekoratora app. Do definiowania kontraktu interfejsu API używamy też biblioteki Pydantic. Określamy, że trasa do wygenerowania odpowiedzi to /chat z metodą POST. Te funkcje są zadeklarowane w tym kodzie:

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

Przygotowywanie historii czatu w formacie pakietu SDK Gemini

Jedną z ważnych kwestii, które należy zrozumieć, jest to, jak przekształcić historię czatu, aby można ją było wstawić jako wartość argumentu history podczas późniejszej inicjalizacji klienta Gemini. Kod możesz sprawdzić poniżej.

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

Aby przekazać historię czatu do pakietu Gemini SDK, musimy sformatować dane w typie danych List[Content]. Każdy element Content musi mieć co najmniej wartość role i parts. role odnosi się do źródła wiadomości, czyli użytkownika lub modelu. Części to sam prompt, który może być tylko tekstem lub kombinacją różnych rodzajów danych. Szczegółowe informacje o tym, jak strukturyzować argumenty Content, znajdziesz w tej dokumentacji.

Obsługa danych innych niż tekstowe ( multimodalnych)

Jak wspomnieliśmy wcześniej w sekcji dotyczącej interfejsu, jednym ze sposobów wysyłania danych innych niż tekstowe lub multimodalne jest wysyłanie ich w postaci ciągu base64. Musimy też określić typ MIME danych, aby można je było prawidłowo zinterpretować. Na przykład jeśli wysyłamy dane obrazu z sufiksem .jpg, podajemy typ MIME image/jpeg.

Ta część kodu konwertuje dane base64 na format Part.from_bytes z pakietu Gemini SDK.

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 integracji

Teraz w różnych kartach konsoli chmury powinno działać kilka usług:

• Usługa frontend działa na porcie 8080.

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

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

• Usługa backendu działa na porcie 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)

Obecnie możesz bezproblemowo wysyłać dokumenty na czacie z asystentem z aplikacji internetowej na porcie 8080. Możesz rozpocząć eksperymentowanie, przesyłając pliki i zadając pytania. Pamiętaj, że niektóre typy plików nie są jeszcze obsługiwane i spowodują błąd.

Instrukcje systemowe możesz też edytować w polu Dodatkowe dane wejściowe pod polem tekstowym.

6. Wdrażanie w Cloud Run

Oczywiście chcemy pokazać tę niesamowitą aplikację innym. Aby to zrobić, możemy spakować tę aplikację i wdrożyć ją w Cloud Run jako usługę publiczną, do której inni użytkownicy będą mieli dostęp. Aby to zrobić, wróćmy do architektury

W tym ćwiczeniu umieścimy zarówno usługę frontendu, jak i usługę backendu w 1 kontenerze. Do zarządzania obiema usługami potrzebujemy pomocy supervisord.

Utwórz nowy plik, kliknij Plik –> Nowy plik tekstowy, skopiuj i wklej poniższy kod, a następnie zapisz go jako 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

Następnie potrzebny będzie plik Dockerfile. Kliknij Plik –> Nowy plik tekstowy, skopiuj i wklej poniższy kod, a następnie zapisz go jako 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"]

Mamy już wszystkie pliki potrzebne do wdrożenia aplikacji w Cloud Run. Możemy więc to zrobić. Otwórz terminal Cloud Shell i sprawdź, czy bieżący projekt jest skonfigurowany jako aktywny. Jeśli nie, użyj polecenia gcloud configure, aby ustawić identyfikator projektu:

gcloud config set project [PROJECT_ID]

Następnie uruchom to polecenie, aby wdrożyć go w Cloud Run.

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

Poprosi Cię o wpisanie nazwy usługi, np. „gemini-multimodal-chat-assistant”. Ponieważ w katalogu roboczym aplikacji mamy plik Dockerfile, utworzy on kontener Dockera i przeniesie go do Artifact Registry. Pojawi się też pytanie, czy utworzyć repozytorium Artifact Registry w regionie. Odpowiedz na nie „Y”. Gdy pojawi się pytanie, czy chcesz zezwolić na nieuwierzytelnione wywołania, powiedz „tak”. Pamiętaj, że w tym przypadku zezwalamy na nieuwierzytelniony dostęp, ponieważ jest to aplikacja demonstracyjna. Zalecamy stosowanie odpowiedniego uwierzytelniania w przypadku aplikacji firmowych i produkcyjnych.

Po zakończeniu wdrażania powinien pojawić się link podobny do tego poniżej:

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

Możesz teraz korzystać z aplikacji w oknie incognito lub na urządzeniu mobilnym. Powinien być już widoczny.

7. Wyzwanie

Teraz możesz zabłysnąć i doskonalić swoje umiejętności eksploracyjne. Czy masz umiejętności, które pozwolą Ci zmienić kod, aby asystent mógł odczytywać pliki audio lub wideo?

8. Czyszczenie danych

Aby uniknąć obciążenia konta Google Cloud opłatami za zasoby użyte w tym laboratorium, wykonaj te czynności:

1. W konsoli Google Cloud otwórz stronę Zarządzanie zasobami.
2. Z listy projektów wybierz projekt do usunięcia, a potem kliknij Usuń.
3. W oknie wpisz identyfikator projektu i kliknij Wyłącz, aby usunąć projekt.
4. Możesz też otworzyć Cloud Run w konsoli, wybrać wdrożoną usługę i ją usunąć.