使用 Gemini (Python) 在 Cloud 上构建和部署多模态助理

关于此 Codelab

0 分钟

上次更新时间：2025年3月29日

Alvin Prayuda Juniarta Dwiyantoro 编写

此页面由 Cloud Translation API 翻译。

1. 简介

在此 Codelab 中，您将构建一个聊天 Web 界面形式的应用，您可以在其中与应用进行通信、上传一些文档或图片并进行讨论。应用本身分为 2 项服务：前端和后端；这样，您就可以快速构建原型并试用其效果，还可以了解 API 协定如何集成这两项服务。

在本 Codelab 中，您将采用分步方法，具体步骤如下：

准备您的 Google Cloud 项目并在其中启用所有所需的 API
使用 Gradio 库构建前端服务 - 聊天界面
使用 FastAPI 构建后端服务 - HTTP 服务器，该服务器将传入数据重新格式化为 Gemini SDK 标准，并支持与 Gemini API 通信
管理环境变量并设置将应用部署到 Cloud Run 所需的文件
将应用部署到 Cloud Run

架构概览

前提条件

熟悉 Gemini API 和 Google Gen AI SDK
了解使用 HTTP 服务的基本全栈架构

学习内容

如何使用 Gemini SDK 提交文本和其他数据类型（多模态）并生成文本回答
如何将对话记录构建为 Gemini SDK 以维护对话上下文
使用 Gradio 制作前端 Web 原型
使用 FastAPI 和 Pydantic 开发后端服务
使用 Pydantic-settings 在 YAML 文件中管理环境变量
使用 Dockerfile 将应用部署到 Cloud Run，并使用 YAML 文件提供环境变量

所需条件

Chrome 网络浏览器
Gmail 账号
启用了结算功能的 Cloud 项目

此 Codelab 面向各种级别（包括新手）的开发者，其示例应用中使用了 Python。不过，您无需了解 Python 即可理解所介绍的概念。

2. 准备工作

在 Cloud Shell Editor 中设置 Cloud 项目

本 Codelab 假定您已有一个启用了结算功能的 Google Cloud 项目。如果您还没有，可以按照以下说明开始使用。

2 在 Google Cloud 控制台的项目选择器页面上，选择或创建一个 Google Cloud 项目。
确保您的 Cloud 项目已启用结算功能。了解如何检查项目是否已启用结算功能。
您将使用 Cloud Shell，这是一个在 Google Cloud 中运行的命令行环境，它预加载了 bq。点击 Google Cloud 控制台顶部的“激活 Cloud Shell”。

连接到 Cloud Shell 后，您可以使用以下命令检查自己是否已通过身份验证，以及项目是否已设置为您的项目 ID：

gcloud auth list

在 Cloud Shell 中运行以下命令，以确认 gcloud 命令了解您的项目。

gcloud config list project

如果项目未设置，请使用以下命令进行设置：

gcloud config set project <YOUR_PROJECT_ID>

或者，您也可以在控制台中查看 PROJECT_ID ID

点击它，您会在右侧看到您的所有项目和项目 ID

通过以下命令启用所需的 API。此过程可能需要几分钟的时间，请耐心等待。

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

成功执行该命令后，您应该会看到如下所示的消息：

Operation "operations/..." finished successfully.

您可以通过控制台搜索各个产品或使用此链接，以替代 gcloud 命令。

如果缺少任何 API，您随时可以在实现过程中启用它。

如需了解 gcloud 命令和用法，请参阅文档。

设置应用工作目录

点击“打开编辑器”按钮，这会打开 Cloud Shell 编辑器，我们可以在此处编写代码
确保在 Cloud Shell 编辑器的左下角（状态栏）中设置 Cloud Code 项目（如下图所示），并将其设置为已启用结算功能的有效 Google Cloud 项目。出现提示时，授权。初始化 Cloud Shell Editor 后，系统可能需要一段时间才能显示 Cloud Code - Sign In 按钮，请耐心等待。如果您已按照上一条命令操作，该按钮可能也会直接指向已启用的项目，而不是登录按钮

点击状态栏中的相应活跃项目，然后等待 Cloud Code 弹出式窗口打开。在弹出式窗口中，选择“新建应用”。

从应用列表中选择 Gemini 生成式 AI，然后选择 Gemini API Python

使用您喜欢的名称保存新应用，在此示例中，我们将使用 gemini-multimodal-chat-assistant，然后点击确定

此时，您应该已经进入新的应用工作目录，并看到以下文件

接下来，我们将准备 Python 环境

环境设置

准备 Python 虚拟环境

下一步是准备开发环境。在本 Codelab 中，我们将使用 Python 3.12，并使用 uv Python 项目管理器来简化创建和管理 Python 版本和虚拟环境的流程

如果您尚未打开终端，请依次点击 Terminal（终端）-> New Terminal（新建终端），或使用 Ctrl + Shift + C 组合键打开终端

使用以下命令下载 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

现在，我们使用 uv 初始化 Python 项目

uv init

您会看到目录中创建了 main.py、.python-version 和 pyproject.toml。这些文件是维护目录中项目所需的。您可以在 pyproject.toml 中指定 Python 依赖项和配置，并使用 .python-version 标准化此项目使用的 Python 版本。如需详细了解，请参阅此文档

main.py
.python-version
pyproject.toml

如需进行测试，请将 main.py 替换为以下代码

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

if __name__ == "__main__":
   main()

然后，运行以下命令

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）。

安装所需的依赖项

我们还将使用 uv 命令添加此 Codelab 软件包依赖项。运行以下命令

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 的“dependencies”部分将更新为反映之前的命令

设置配置文件

现在，我们需要为此项目设置配置文件。配置文件用于存储动态变量，这些变量可以在重新部署时轻松更改。在此项目中，我们将使用基于 YAML 的配置文件与 pydantic-settings 软件包搭配使用，以便日后轻松与 Cloud Run 部署集成。pydantic-settings 是一个 Python 软件包，可强制对配置文件进行类型检查。

创建名为 settings.yaml 的文件，并添加以下配置。依次点击 File->New Text File，然后使用以下代码进行填充。然后将其另存为 settings.yaml

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

请根据您在创建 Google Cloud 项目时选择的值更新 VERTEXAI_PROJECT_ID 的值。在此 Codelab 中，我们将使用为 VERTEXAI_LOCATION 和 BACKEND_URL 预配置的值。

然后，创建 Python 文件 settings.py，此模块将作为配置文件中配置值的程序化条目。依次点击 File->New Text File，然后使用以下代码进行填充。然后将其另存为 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 构建前端服务

我们将构建一个如下所示的聊天 Web 界面

它包含一个输入字段，供用户发送文本和上传文件。此外，用户还可以在“其他输入”字段中覆盖将发送到 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

您将在 Cloud 控制台中看到类似于以下内容的输出

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

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

之后，您可以Ctrl 键 + 点击本地网址链接，查看 Web 界面。或者，您也可以点击 Cloud 编辑器右上角的网页预览按钮，然后选择在端口 8080 上预览，以访问前端应用

您会看到 Web 界面，但由于后端服务尚未设置，因此在尝试提交聊天时会收到预期错误

现在，让服务运行，暂时不要终止它。与此同时，我们可以在这里讨论重要的代码组件

代码说明

用于将数据从 Web 界面发送到后端的代码位于此部分

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 类型，因此请务必参阅此文档，了解 Gemini 支持的 MIME 类型。您可以在每个 Gemini API 功能（例如 Vision）中找到相关信息

此外，在聊天界面中，还需要发送聊天记录作为额外的上下文，以便 Gemini 记住对话内容。因此，在此 Web 界面中，我们还会发送 Gradio 在每个网页会话中管理的聊天记录，并将其与用户输入的消息一起发送。此外，我们还允许用户修改系统指令并发送该指令

4. 使用 FastAPI 构建后端服务

接下来，我们需要构建后端，以处理之前讨论过的载荷、上一条用户消息、聊天记录和系统说明。我们将使用 FastAPI 创建 HTTP 后端服务。

创建新文件，依次点击 File->New Text File，然后复制并粘贴以下代码，并将其另存为 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。之后，我们可以尝试运行后端服务。请记得，在上一步中，我们正确运行了前端服务，现在我们需要打开新的终端并尝试运行此后端服务

创建一个新终端。前往底部区域的终端，找到“+”按钮以创建新的终端。或者，您也可以按 Ctrl + Shift + C 打开新的终端

之后，确保您位于工作目录 gemini-multimodal-chat-assistant 中，然后运行以下命令

uv run backend.py

如果成功，则会显示如下所示的输出

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 协定。我们使用 POST 方法指定用于生成响应的路由为 /chat。以下代码中声明的这些功能

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 客户端时将其作为 history 参数值插入。您可以检查以下代码

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] 数据类型。每个内容都必须至少包含一个角色和部分值。角色是指消息的来源，可能是用户或模型。其中，parts 是指问题本身，可以是纯文本，也可以是不同模态的组合。如需详细了解如何构建 Content 参数，请参阅此文档

处理非文本（多模态）数据

如前面“前端”部分所述，发送非文本或多模态数据的方法之一是将数据发送为 base64 字符串。我们还需要为数据指定 MIME 类型，以便正确解读数据，例如，如果我们发送带有 .jpg 后缀的图片数据，则提供 image/jpeg MIME 类型。

此部分代码会将 base64 数据转换为 Gemini SDK 中的 Part.from_bytes 格式

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. 集成测试

现在，您应该可以在不同的 Cloud 控制台标签页中运行多个服务：

在端口 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 上的 Web 应用与 Google 助理无缝聊天，并发送文档。您可以开始上传文件并提出问题，开始进行实验！请注意，系统尚不支持某些文件类型，并且会引发错误。

您还可以通过文本框下方的其他输入字段修改系统说明

6. 部署到 Cloud Run

当然，我们希望向其他人展示这款出色的应用。为此，我们可以打包此应用，并将其部署到 Cloud Run 作为可供他人访问的公共服务。为此，我们来回顾一下架构

在此 Codelab 中，我们将前端服务和后端服务都放入一个容器中。我们需要借助监督功能来管理这两项服务。

创建新文件，依次点击 File->New Text File，然后复制并粘贴以下代码，并将其另存为 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，点击 File->New Text File（文件 -> 新建文本文件），然后复制并粘贴以下代码，并将其另存为 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 命令设置项目 ID：

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 容器并将其推送到 Artifact Registry。系统还会提示您将在该区域创建 Artifact Registry 代码库，请在此回答“Y”。此外，当系统询问您是否要允许未经身份验证的调用时，请说“y”。请注意，由于这是演示版应用，因此我们允许未经身份验证的访问。建议为企业应用和生产应用使用适当的身份验证。

部署完成后，您应该会收到类似于以下内容的链接：

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

接下来，您可以继续在无痕式窗口或移动设备上使用您的应用。应该已经上线了。

7. 挑战

现在是时候大显身手，磨练您的探索技能了。您是否有能力更改代码，以便 Google 助理支持读取音频文件或视频文件？

8. 清理

为避免系统因本 Codelab 中使用的资源向您的 Google Cloud 账号收取费用，请按照以下步骤操作：

在 Google Cloud 控制台中，前往管理资源页面。
在项目列表中，选择要删除的项目，然后点击删除。
在对话框中输入项目 ID，然后点击关停以删除项目。
或者，您也可以前往控制台中的 Cloud Run，选择刚刚部署的服务并进行删除。