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

1. 简介

在此 Codelab 中，您将构建一个聊天 Web 界面形式的应用，您可以在其中与应用进行通信、上传一些文档或图片并进行讨论。应用本身分为 2 个服务：前端和后端；让您可以快速构建原型并体验其效果，同时了解 API 合约的外观，以便集成这两个服务。

在此 Codelab 中，您将采用以下分步方法：

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

5bcfa1cce6618305.png

架构概览

b102df2c3f1adabf.jpeg

前提条件

学习内容

  • 如何使用 Gemini SDK 提交文本和其他数据类型（多模态），并生成文本回答
  • 如何将聊天记录构建到 Gemini SDK 中以保持对话上下文
  • 使用 Gradio 进行前端 Web 原型设计
  • 使用 FastAPI 和 Pydantic 进行后端服务开发
  • 使用 Pydantic-settings 管理 YAML 文件中的环境变量
  • 使用 Dockerfile 将应用部署到 Cloud Run，并通过 YAML 文件提供环境变量

所需条件

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

此 Codelab 专为各种水平的开发者（包括新手）设计，并在示例应用中使用 Python。不过，您无需具备 Python 知识即可理解所介绍的概念。

2. 准备工作

在 Cloud Shell 编辑器中设置 Cloud 项目

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

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

1829c3759227c19b.png

  1. 连接到 Cloud Shell 后，您可以使用以下命令检查自己是否已通过身份验证，以及项目是否已设置为您的项目 ID：
gcloud auth list
  1. 在 Cloud Shell 中运行以下命令，以确认 gcloud 命令了解您的项目。
gcloud config list project
  1. 如果项目未设置，请使用以下命令进行设置：
gcloud config set project <YOUR_PROJECT_ID>

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

4032c45803813f30.jpeg

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

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 弹出式窗口打开。在弹出式窗口中，选择“New Application”。

70f80078e01a02d8.png

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

362ff332256d6933.jpeg

85957565316308d9.jpeg

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

8409d8db18690fdf.png

此时，您应该已位于新的应用工作目录中，并看到以下文件

1ef5bb44f1d2c2a4.png

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

环境设置

准备 Python 虚拟环境

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

  1. 如果您尚未打开终端，请依次点击 Terminal -> New Terminal 将其打开，或者使用 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. 现在，我们使用 uv 初始化 Python 项目
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）。

安装必需的依赖项

我们还将使用 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 软件包，可对配置文件强制执行类型检查。

  1. 创建一个名为 settings.yaml 的文件，其中包含以下配置。点击文件 -> 新建文本文件，然后填充以下代码。然后将其另存为 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 的预配置值。

  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 构建前端服务

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

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+点击本地网址链接，查看 Web 界面。或者，您也可以点击 Cloud Editor 右上角的网页预览按钮，然后选择在端口 8080 上预览，以访问前端应用

49cbdfdf77964065.jpeg

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

bd0464140308cfbe.png

现在，让服务运行，暂时不要终止它。在此期间，我们可以讨论一下重要的代码组件

代码说明

用于将数据从 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，以便让 Gemini 记住对话内容，这一点也很重要。因此，在此网页界面中，我们还会发送由 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 中，我们使用 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 客户端时，可以将其作为 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] 数据类型。每条内容都必须至少包含一个角色和部分值。角色是指消息的来源，可以是用户或模型。其中，部分是指提示本身，它可以仅包含文本，也可以是不同模态的组合。如需详细了解如何构建内容实参，请参阅此文档

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

如前端部分所述，发送非文本数据或多模态数据的一种方式是将数据作为 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 应用，在聊天中无缝地向助理发送文档。您可以上传文件并提出问题，开始体验这项功能！请注意，某些文件类型尚不受支持，因此会引发错误。

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

ee9c849a276d378.png

6. 部署到 Cloud Run

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

b102df2c3f1adabf.jpeg

在此 Codelab 中，我们将前端和后端服务都放在 1 个容器中。我们需要借助 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 命令设置项目 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. 挑战

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

8. 清理

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

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