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

О практической работе
schedule0 минут
subjectПоследнее обновление: 29 марта 2025 г.
account_circleАвторы: Alvin Prayuda Juniarta Dwiyantoro

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

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

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

5bcfa1cce6618305.png

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

b102df2c3f1adabf.jpeg

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

  • Комфортная работа с Gemini API и Google Gen AI SDK.
  • Понимание базовой полностековой архитектуры с использованием службы HTTP.

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

  • Как использовать Gemini SDK для отправки текста и других типов данных (мультимодальных) и создания текстового ответа
  • Как структурировать историю чата в Gemini SDK для сохранения контекста разговора
  • Веб-прототипирование внешнего интерфейса с помощью Gradio
  • Разработка серверных сервисов с помощью FastAPI и Pydantic
  • Управляйте переменными среды в файле YAML с помощью настроек Pydantic.
  • Разверните приложение в 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.

1829c3759227c19b.png

  1. После подключения к Cloud Shell вы проверяете, что вы уже прошли аутентификацию и что для проекта установлен идентификатор вашего проекта, используя следующую команду: 
gcloud auth list
  1. Выполните следующую команду в Cloud Shell, чтобы убедиться, что команда gcloud знает о вашем проекте. 
gcloud config list project
  1. Если ваш проект не установлен, используйте следующую команду, чтобы установить его: 
gcloud config set project <YOUR_PROJECT_ID>

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

4032c45803813f30.jpeg

Нажмите на нее, и вы увидите весь свой проект и идентификатор проекта справа.

8dc17eb4271de6b5.jpeg

  1. Включите необходимые 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, здесь мы можем написать наш код. b16d56e4979ec951.png
  2. Убедитесь, что проект Cloud Code установлен в левом нижнем углу (строке состояния) редактора Cloud Shell, как показано на изображении ниже, и настроен на активный проект Google Cloud, для которого у вас включена оплата. Авторизуйтесь, если будет предложено. После инициализации редактора Cloud Shell может пройти некоторое время, прежде чем появится кнопка Cloud Code — Войти , наберитесь терпения. Если вы уже выполнили предыдущую команду, кнопка также может указывать непосредственно на активированный проект вместо кнопки входа в систему.

f5003b9c38b43262.png

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

70f80078e01a02d8.png

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

362ff332256d6933.jpeg

85957565316308d9.jpeg

  1. Сохраните новое приложение с именем, которое вам нравится, в этом примере мы будем использовать Gemini-multimodal-chat-assistant , затем нажмите «ОК».

8409d8db18690fdf.png

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

1ef5bb44f1d2c2a4.png

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

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

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

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

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

f8457daf0bed059e.jpeg

  1. Загрузите 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
  1. Теперь давайте инициализируем проект Python, используя uv 
uv init
  1. Вы увидите созданные в каталоге main.py, .python-version и pyproject.toml . Эти файлы нужны для поддержания проекта в каталоге. Зависимости и конфигурации Python можно указать в файле pyproject.toml , а .python-version стандартизирует версию Python, используемую для этого проекта. Чтобы узнать больше об этом, вы можете просмотреть эту документацию. 
main.py
.python-version
pyproject.toml
  1. Чтобы проверить это, перезапишите main.py в следующий код 
def main():
   print("Hello from gemini-multimodal-chat-assistant!")

if __name__ == "__main__":
   main()
  1. Затем выполните следующую команду 
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

Вы увидите, что раздел «зависимости» 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 .

  1. Затем создайте файл 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.

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

5bcfa1cce6618305.png

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

Мы создадим фронтенд-сервис с помощью 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 + щелчок по локальной URL-ссылке. Кроме того, вы также можете получить доступ к внешнему приложению, нажав кнопку «Просмотр в Интернете» в правом верхнем углу облачного редактора и выбрав «Просмотр на порту 8080».

49cbdfdf77964065.jpeg

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

bd0464140308cfbe.png

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

Объяснение кода

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

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 -тип данных. Однако Gemini API не может поддерживать все существующие типы MIME, поэтому важно знать, какие типы MIME поддерживаются Gemini, и это можно прочитать в этой документации . Вы можете найти информацию о каждой возможности Gemini API (например, Vision ).

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

4. Создайте серверную службу с использованием FastAPI.

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

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

3e52a362475553dc.jpeg

  1. После этого убедитесь, что вы находитесь в рабочем каталоге gemini-multimodal-chat-assistant, а затем выполните следующую команду: 
uv run backend.py
  1. В случае успеха он отобразит такой вывод 
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 мы определяем маршрут с помощью декоратора приложения . Мы также используем 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 см. в этой документации.

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

Как упоминалось ранее в разделе внешнего интерфейса, одним из способов отправки нетекстовых или мультимодальных данных является отправка данных в виде строки 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. Вы можете начать экспериментировать, загружая файлы и задавая вопросы! Будьте осторожны: некоторые типы файлов еще не поддерживаются и могут вызвать ошибку.

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

ee9c849a276d378.png

6. Развертывание в Cloud Run

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

b102df2c3f1adabf.jpeg

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

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