Создание и развертывание мультимодального помощника в облаке с помощью Gemini (Python)

1. Введение

В этом практическом занятии вы создадите приложение в виде веб-интерфейса чата, с помощью которого сможете общаться, загружать документы или изображения и обсуждать их. Само приложение разделено на два сервиса: фронтенд и бэкенд; это позволит вам быстро создать прототип, оценить его функциональность и понять, как выглядит API-контракт для интеграции обоих сервисов.

В ходе выполнения практического задания вы будете использовать следующий пошаговый подход:

1. Подготовьте свой проект в Google Cloud и включите в него все необходимые API.
2. Создайте фронтенд-сервис — интерфейс чата — с использованием библиотеки Gradio.
3. Создайте серверную часть — HTTP-сервер с использованием FastAPI , который будет преобразовывать входящие данные в формат, соответствующий стандарту Gemini SDK, и обеспечивать связь с API Gemini.
4. Управление переменными среды и настройка необходимых файлов для развертывания приложения в облаке.
5. Разверните приложение в облаке.

Обзор архитектуры

Предварительные требования

• Уверенное владение API Gemini и SDK Google Gen AI.
• Понимание базовой архитектуры полного стека с использованием HTTP-сервисов.

Что вы узнаете

• Как использовать SDK Gemini для отправки текста и других типов данных (мультимодальный формат) и генерации текстового ответа.
• Как структурировать историю чата в SDK Gemini для сохранения контекста разговора
• Фронтенд-прототипирование веб-сайтов с помощью Gradio
• Разработка бэкэнд-сервисов с использованием FastAPI и Pydantic.
• Управление переменными окружения в YAML-файле с помощью Pydantic-settings
• Разверните приложение в Cloud Run с помощью Dockerfile и укажите переменные среды в YAML-файле.

Что вам понадобится

• Веб-браузер Chrome
• Аккаунт Gmail
• Облачный проект с включенной функцией выставления счетов.

Этот практический урок, разработанный для разработчиков всех уровней (включая начинающих), использует Python в качестве примера приложения. Однако знание Python не требуется для понимания представленных концепций.

2. Прежде чем начать

Настройка облачного проекта в редакторе Cloud Shell

В этом практическом задании предполагается, что у вас уже есть проект Google Cloud с включенной оплатой. Если у вас его еще нет, вы можете следовать инструкциям ниже, чтобы начать работу.

1. 2. В консоли Google Cloud на странице выбора проекта выберите или создайте проект Google Cloud.
2. Убедитесь, что для вашего облачного проекта включена функция выставления счетов. Узнайте, как проверить, включена ли функция выставления счетов для проекта .
3. Вы будете использовать Cloud Shell — среду командной строки, работающую в Google Cloud и поставляемую с предустановленным bq. Нажмите «Активировать Cloud Shell» в верхней части консоли Google Cloud.

4. После подключения к Cloud Shell необходимо проверить, прошли ли вы аутентификацию и установлен ли идентификатор вашего проекта, используя следующую команду:

gcloud auth list

5. Выполните следующую команду в Cloud Shell, чтобы убедиться, что команда gcloud знает о вашем проекте.

gcloud config list project

6. Если ваш проект не задан, используйте следующую команду для его установки:

gcloud config set project <YOUR_PROJECT_ID>

В качестве альтернативы, вы также можете увидеть идентификатор PROJECT_ID в консоли.

Нажмите на него, и справа отобразятся все ваши проекты и их идентификаторы.

7. Включите необходимые API с помощью команды, указанной ниже. Это может занять несколько минут, поэтому, пожалуйста, наберитесь терпения.

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

После успешного выполнения команды вы должны увидеть сообщение, похожее на показанное ниже:

Operation "operations/..." finished successfully.

Альтернативой команде gcloud является поиск каждого продукта в консоли или использование этой ссылки .

Если какой-либо API отсутствует, вы всегда можете включить его в процессе реализации.

Для получения информации о командах gcloud и их использовании обратитесь к документации .

Рабочий каталог приложения установки

1. Нажмите кнопку «Открыть редактор», это откроет редактор Cloud Shell, где мы можем писать свой код.
2. Убедитесь, что проект Cloud Code указан в левом нижнем углу (строке состояния) редактора Cloud Shell, как показано на изображении ниже, и что он соответствует активному проекту Google Cloud, в котором включена оплата. Авторизуйтесь, если потребуется. После инициализации редактора Cloud Shell может потребоваться некоторое время, чтобы появилась кнопка «Cloud Code — Войти» , пожалуйста, подождите. Если вы уже выполнили предыдущую команду, кнопка может также указывать непосредственно на ваш активированный проект, а не на кнопку входа.

3. Щелкните по активному проекту в строке состояния и дождитесь открытия всплывающего окна Cloud Code. Во всплывающем окне выберите «Новое приложение».

4. Из списка приложений выберите Gemini Generative AI , затем выберите Gemini API Python.

5. Сохраните новое приложение под любым именем, в этом примере мы будем использовать gemini-multimodal-chat-assistant , затем нажмите OK.

На этом этапе вы уже должны находиться в рабочей директории нового приложения и видеть следующие файлы.

Далее мы подготовим нашу среду Python.

Настройка среды

Подготовка виртуальной среды Python

Следующий шаг — подготовка среды разработки. В этом практическом задании мы будем использовать Python 3.12 и менеджер проектов uv python , чтобы упростить создание и управление версиями Python и виртуальными средами.

1. Если вы еще не открыли терминал, откройте его, щелкнув «Терминал» -> «Новый терминал» или используя сочетание клавиш Ctrl + Shift + C.

2. Загрузите uv и установите Python 3.12 с помощью следующей команды.

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

3. Теперь инициализируем проект Python с помощью uv

uv init

4. В этой директории будут созданы файлы main.py, .python-version и pyproject.toml . Эти файлы необходимы для поддержки проекта в данной директории. Зависимости и конфигурации Python можно указать в файлах pyproject.toml и .python-version , которые стандартизируют версию Python, используемую в этом проекте. Более подробную информацию можно найти в этой документации.

main.py
.python-version
pyproject.toml

5. Для проверки замените код в файле main.py следующим содержимым.

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

if __name__ == "__main__":
   main()

6. Затем выполните следующую команду.

uv run main.py

В результате вы получите результат, показанный ниже.

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

Это показывает, что проект Python настраивается правильно. Нам не потребовалось вручную создавать виртуальное окружение, поскольку uv уже это делает. Поэтому с этого момента стандартная команда Python (например , python main.py ) будет заменена на команду uv run (например , uv run main.py ).

Установите необходимые зависимости.

Мы также добавим зависимости этого пакета codelab, используя команду uv . Выполните следующую команду.

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

Вы увидите, что раздел "dependencies" в файле pyproject.toml будет обновлен в соответствии с предыдущей командой.

Настройка конфигурационных файлов

Теперь нам нужно настроить конфигурационные файлы для этого проекта. Конфигурационные файлы используются для хранения динамических переменных, которые можно легко изменить при повторном развертывании. В этом проекте мы будем использовать конфигурационные файлы на основе YAML с пакетом pydantic-settings , что позволит легко интегрировать их с развертыванием Cloud Run в дальнейшем. pydantic-settings — это пакет Python, который позволяет принудительно проверять типы в конфигурационных файлах.

1. Создайте файл с именем settings.yaml со следующей конфигурацией. Нажмите «Файл» -> «Новый текстовый файл» и заполните его следующим кодом. Затем сохраните его как settings.yaml.

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

Пожалуйста, обновите значения параметра VERTEXAI_PROJECT_ID в соответствии с теми, которые вы выбрали при создании проекта Google Cloud. Для этого практического задания мы будем использовать предварительно настроенные значения параметров VERTEXAI_LOCATION и BACKEND_URL .

2. Затем создайте файл Python settings.py ; этот модуль будет выступать в качестве программной точки входа для значений конфигурации в наших конфигурационных файлах. Нажмите «Файл» -> «Новый текстовый файл» и заполните его следующим кодом. Затем сохраните его как settings.py. В коде видно, что мы явно указали, что будет считываться файл с именем settings.yaml.

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

Эти конфигурации позволяют нам гибко обновлять среду выполнения. При первоначальном развертывании мы будем использовать конфигурацию settings.yaml , чтобы иметь первую конфигурацию по умолчанию. После этого мы сможем гибко обновлять переменные среды через консоль и повторно развертывать приложение, поскольку мы присваиваем переменным среды более высокий приоритет по сравнению с конфигурацией YAML по умолчанию.

Теперь мы можем перейти к следующему шагу — созданию сервисов.

3. Создайте фронтенд-сервис с использованием Gradio.

Мы создадим веб-интерфейс для чата, который будет выглядеть следующим образом.

Он содержит поле ввода, в которое пользователи могут отправлять текст и загружать файлы. Кроме того, пользователь может также переопределить системную инструкцию, которая будет отправлена ​​в API Gemini, в дополнительном поле ввода.

Мы будем создавать фронтенд-сервис с помощью Gradio . Переименуйте файл main.py в frontend.py и замените код, используя следующий код.

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

После этого можно попробовать запустить фронтенд-сервис с помощью следующей команды. Не забудьте переименовать файл main.py в frontend.py.

uv run frontend.py

В консоли вашего облака вы увидите результат, похожий на этот.

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

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

После этого вы можете проверить веб-интерфейс, щелкнув по локальной ссылке с помощью Ctrl+клик . Также вы можете получить доступ к фронтенд-приложению, нажав кнопку « Предварительный просмотр веб-страниц» в правом верхнем углу облачного редактора и выбрав «Предварительный просмотр на порту 8080».

Вы увидите веб-интерфейс, однако при попытке отправить сообщение в чат вы получите ожидаемую ошибку, поскольку серверная часть еще не настроена.

Теперь дайте сервису работать и пока не останавливайте его. А пока мы можем обсудить важные компоненты кода здесь.

Пояснение к коду

Код для отправки данных из веб-интерфейса на бэкэнд находится в этой части.

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

Когда мы хотим отправить мультимодальные данные в Gemini и сделать эти данные доступными между сервисами, одним из механизмов может быть преобразование данных в формат base64 , как указано в коде. Также необходимо указать MIME- тип данных. Однако API Gemini не поддерживает все существующие MIME-типы, поэтому важно знать, какие MIME-типы поддерживает Gemini, информацию о которых можно найти в этой документации . Вы можете найти эту информацию в описании каждой из возможностей API Gemini (например, Vision ).

Кроме того, в интерфейсе чата важно также отправлять историю чата в качестве дополнительного контекста, чтобы Gemini «памятил» разговор. Поэтому в этом веб-интерфейсе мы также отправляем историю чата, которая управляется Gradio для каждой веб-сессии, вместе с сообщениями, введенными пользователем. Дополнительно мы также предоставляем пользователю возможность изменять системные инструкции и отправляем их тоже.

4. Создайте бэкэнд-сервис с использованием FastAPI.

Далее нам потребуется создать бэкэнд, способный обрабатывать ранее обсуждавшиеся данные: полезную нагрузку, последнее сообщение пользователя, историю чата и системные инструкции . Для создания HTTP-сервиса мы будем использовать FastAPI .

Создайте новый файл, нажмите «Файл» -> «Новый текстовый файл», скопируйте и вставьте следующий код, затем сохраните его как 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)

Не забудьте сохранить его как backend.py. После этого можно попробовать запустить бэкэнд-сервис. Помните, что на предыдущем шаге мы запустили фронтенд-сервис, верно? Теперь нам нужно открыть новое окно терминала и попробовать запустить этот бэкэнд-сервис.

1. Создайте новый терминал. Перейдите в меню терминала внизу и найдите кнопку «+», чтобы создать новый терминал. Также вы можете нажать Ctrl + Shift + C, чтобы открыть новый терминал.

2. После этого убедитесь, что вы находитесь в рабочей директории gemini-multimodal-chat-assistant, а затем выполните следующую команду.

uv run backend.py

3. В случае успеха будет показан следующий результат.

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)

Пояснение к коду

Определение HTTP-маршрута для получения запроса чата

В FastAPI мы определяем маршрут с помощью декоратора app . Мы также используем Pydantic для определения контракта API. Мы указываем, что маршрут для генерации ответа находится в маршруте /chat с методом POST . Эти функции объявлены в следующем коде.

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

Один из важных моментов, который необходимо понять, — это как можно реструктурировать историю чата, чтобы её можно было использовать в качестве значения аргумента истории при последующей инициализации клиента Gemini. Вы можете ознакомиться с кодом ниже.

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

Для отображения истории чата в Gemini SDK необходимо отформатировать данные в типе данных List[Content] . Каждый элемент Content должен содержать как минимум значение role и parts . role обозначает источник сообщения, будь то пользователь или модель. parts обозначает сам запрос, который может представлять собой только текст или комбинацию различных модальностей. Подробное описание структуры аргументов Content см. в этой документации.

Обработка нетекстовых (мультимодальных) данных

Как уже упоминалось в разделе, посвященном фронтенду, один из способов отправки нетекстовых или мультимодальных данных — это отправка данных в виде строки base64 . Также необходимо указать MIME-тип данных, чтобы они могли быть правильно интерпретированы, например, указать MIME-тип image/jpeg, если мы отправляем данные изображения с расширением .jpg .

Этот фрагмент кода преобразует данные base64 в формат Part.from_bytes из 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. Интеграционный тест

Теперь у вас должно быть запущено несколько сервисов в разных вкладках облачной консоли:

• Фронтенд-сервис работает на порту 8080.

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

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

• Серверная служба работает на порту 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)

В текущем состоянии вы должны иметь возможность беспрепятственно отправлять документы в чат с помощью помощника из веб-приложения на порту 8080. Вы можете начать экспериментировать, загружая файлы и задавая вопросы! Имейте в виду, что некоторые типы файлов пока не поддерживаются и могут вызывать ошибку.

Вы также можете редактировать инструкции системы в поле «Дополнительные входы», расположенном под текстовым полем.

6. Развертывание в облаке

Конечно, теперь мы хотим продемонстрировать это замечательное приложение другим. Для этого мы можем упаковать это приложение и развернуть его в Cloud Run в качестве общедоступного сервиса, к которому смогут получить доступ другие пользователи. Для этого давайте вернемся к архитектуре.

В этом практическом задании мы разместим фронтенд и бэкенд сервисы в одном контейнере. Для управления обоими сервисами нам понадобится SupervisorD .

Создайте новый файл, нажмите «Файл» -> «Новый текстовый файл», скопируйте и вставьте следующий код, затем сохраните его как 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

Далее нам понадобится Dockerfile. Нажмите «Файл» -> «Новый текстовый файл», скопируйте и вставьте следующий код, затем сохраните его как 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"]

На данном этапе у нас уже есть все необходимые файлы для развертывания наших приложений в Cloud Run, давайте приступим к развертыванию. Перейдите в терминал Cloud Shell и убедитесь, что текущий проект настроен на ваш активный проект. Если это не так, используйте команду gcloud configure для установки идентификатора проекта:

gcloud config set project [PROJECT_ID]

Затем выполните следующую команду, чтобы развернуть его в Cloud Run.

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

Вам будет предложено ввести имя для вашей службы, например, " gemini-multimodal-chat-assistant ". Поскольку Dockerfile находится в рабочем каталоге нашего приложения, будет создан контейнер Docker и отправлен в реестр артефактов. Также будет предложено создать репозиторий реестра артефактов в указанном регионе; ответьте на этот вопрос " Y" . Также ответьте " Y ", когда вас спросят, хотите ли вы разрешить неаутентифицированные вызовы. Обратите внимание, что мы разрешаем неаутентифицированный доступ, поскольку это демонстрационное приложение. Рекомендуется использовать соответствующую аутентификацию для ваших корпоративных и производственных приложений.

После завершения развертывания вы должны получить ссылку, похожую на приведенную ниже:

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

Смело используйте приложение в режиме инкогнито или на мобильном устройстве. Оно уже должно быть запущено.

7. Вызов

Теперь настало ваше время проявить себя и отточить свои навыки исследования. Сможете ли вы изменить код, чтобы ассистент мог читать аудиофайлы или, возможно, видеофайлы?

8. Уборка

Чтобы избежать списания средств с вашего аккаунта Google Cloud за ресурсы, использованные в этом практическом задании, выполните следующие действия:

1. В консоли Google Cloud перейдите на страницу «Управление ресурсами» .
2. В списке проектов выберите проект, который хотите удалить, и нажмите кнопку «Удалить» .
3. В диалоговом окне введите идентификатор проекта, а затем нажмите «Завершить» , чтобы удалить проект.
4. В качестве альтернативы вы можете перейти в Cloud Run в консоли, выбрать только что развернутую службу и удалить ее.