ساخت و استقرار دستیار چند وجهی روی ابر با جمینی (پایتون)، ساخت و استقرار دستیار چند وجهی روی ابر با جمینی (پایتون)، ساخت و استقرار دستیار چند وجهی روی ابر با جمینی (پایتون)

درباره این codelab

۰ دقیقه

آخرین به‌روزرسانی: ۹ فروردین ۱۴۰۴

نویسنده: Alvin Prayuda Juniarta Dwiyantoro

این صفحه به‌وسیله ‏Cloud Translation API‏ ترجمه شده است.

1. مقدمه

در این کد لبه شما اپلیکیشنی را در قالب یک رابط وب چت می سازید که در آن می توانید با آن ارتباط برقرار کنید، برخی از اسناد یا تصاویر را آپلود کرده و در مورد آنها بحث کنید. خود برنامه به 2 سرویس تقسیم می شود: frontend و backend. به شما امکان می دهد یک نمونه اولیه سریع بسازید و احساس آن را امتحان کنید و همچنین درک کنید که قرارداد API چگونه به نظر می رسد تا هر دوی آنها را یکپارچه کند.

از طریق کد لبه، شما یک رویکرد گام به گام را به شرح زیر به کار خواهید گرفت:

پروژه Google Cloud خود را آماده کنید و تمام API مورد نیاز را روی آن فعال کنید
ساخت سرویس جلویی - رابط چت با استفاده از کتابخانه Gradio
ساخت سرویس پشتیبان - سرور HTTP با استفاده از FastAPI که داده های دریافتی را به استاندارد Gemini SDK قالب بندی می کند و ارتباط با Gemini API را فعال می کند.
متغیرهای محیطی را مدیریت کنید و فایل‌های مورد نیاز برای استقرار برنامه در Cloud Run را تنظیم کنید
برنامه را در Cloud Run مستقر کنید

نمای کلی معماری

پیش نیازها

کار راحت با Gemini API و Google Gen AI SDK
درک معماری پایه تمام پشته با استفاده از سرویس HTTP

چیزی که یاد خواهید گرفت

نحوه استفاده از Gemini SDK برای ارسال متن و سایر انواع داده (چند وجهی) و تولید پاسخ متنی
نحوه ساختاردهی تاریخچه چت به Gemini SDK برای حفظ زمینه گفتگو
نمونه سازی وب فرانت اند با Gradio
توسعه خدمات Backend با FastAPI و Pydantic
متغیرهای محیطی را در فایل YAML با تنظیمات Pydantic مدیریت کنید
برنامه را با استفاده از Dockerfile در Cloud Run مستقر کنید و متغیرهای محیطی را با فایل YAML ارائه دهید

آنچه شما نیاز دارید

مرورگر وب کروم
یک اکانت جیمیل
یک پروژه Cloud با فعال کردن صورت‌حساب

این کد لبه که برای توسعه دهندگان همه سطوح (از جمله مبتدیان) طراحی شده است، از پایتون در برنامه نمونه خود استفاده می کند. با این حال، دانش پایتون برای درک مفاهیم ارائه شده مورد نیاز نیست.

2. قبل از شروع

راه اندازی Cloud Project در Cloud Shell Editor

این کد لبه فرض می کند که شما قبلاً یک پروژه Google Cloud با فعال بودن صورتحساب دارید. اگر هنوز آن را ندارید، می توانید دستورالعمل های زیر را برای شروع دنبال کنید.

2 در Google Cloud Console ، در صفحه انتخاب پروژه، یک پروژه Google Cloud را انتخاب یا ایجاد کنید.
مطمئن شوید که صورتحساب برای پروژه Cloud شما فعال است. با نحوه بررسی فعال بودن صورت‌حساب در پروژه آشنا شوید.
شما از Cloud Shell استفاده خواهید کرد، یک محیط خط فرمان در حال اجرا در Google Cloud که با bq از قبل بارگذاری شده است. روی Activate Cloud Shell در بالای کنسول Google Cloud کلیک کنید.

پس از اتصال به Cloud Shell، با استفاده از دستور زیر بررسی می‌کنید که قبلاً احراز هویت شده‌اید و پروژه به ID پروژه شما تنظیم شده است:

gcloud auth list

دستور زیر را در Cloud Shell اجرا کنید تا تأیید کنید که دستور gcloud از پروژه شما اطلاع دارد.

gcloud config list project

اگر پروژه شما تنظیم نشده است، از دستور زیر برای تنظیم آن استفاده کنید:

gcloud config set project <YOUR_PROJECT_ID>

همچنین می‌توانید شناسه PROJECT_ID را در کنسول ببینید

روی آن کلیک کنید و تمام پروژه و شناسه پروژه را در سمت راست خواهید دید

API های مورد نیاز را از طریق دستور زیر فعال کنید. این ممکن است چند دقیقه طول بکشد، پس لطفا صبور باشید.

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

در اجرای موفقیت آمیز دستور، باید پیامی مشابه تصویر زیر مشاهده کنید:

Operation "operations/..." finished successfully.

جایگزین دستور gcloud از طریق کنسول با جستجوی هر محصول یا استفاده از این پیوند است.

اگر هر یک از API از دست رفته است، همیشه می توانید آن را در طول پیاده سازی فعال کنید.

برای دستورات و استفاده از gcloud به اسناد مراجعه کنید.

راه اندازی دایرکتوری کاری برنامه

روی دکمه Open Editor کلیک کنید، با این کار یک Cloud Shell Editor باز می شود، ما می توانیم کد خود را اینجا بنویسیم
مطمئن شوید که پروژه Cloud Code در گوشه سمت چپ پایین (نوار وضعیت) ویرایشگر Cloud Shell تنظیم شده است، همانطور که در تصویر زیر مشخص شده است و روی پروژه فعال Google Cloud که در آن صورت‌حساب را فعال کرده‌اید، تنظیم شده است. در صورت درخواست مجوز دهید . ممکن است پس از مقداردهی اولیه ویرایشگر پوسته Cloud کمی طول بکشد تا دکمه Cloud Code - Sign In ظاهر شود، لطفا صبور باشید. اگر از قبل دستور قبلی را دنبال کرده اید، دکمه ممکن است به جای دکمه ورود مستقیماً به پروژه فعال شده شما اشاره کند

روی آن پروژه فعال در نوار وضعیت کلیک کنید و منتظر بمانید تا Cloud Code باز شود. در پنجره پاپ آپ "برنامه جدید" را انتخاب کنید.

از لیست برنامه ها، Gemini Generative AI را انتخاب کنید، سپس Gemini API Python را انتخاب کنید

برنامه جدید را با نامی که دوست دارید ذخیره کنید، در این مثال از gemini-multimodal-chat-assistant استفاده می کنیم، سپس روی OK کلیک کنید.

در این مرحله، شما باید از قبل در دایرکتوری کار برنامه جدید قرار گرفته باشید و فایل های زیر را ببینید

در ادامه محیط پایتون خود را آماده می کنیم

راه اندازی محیط

محیط مجازی پایتون را آماده کنید

مرحله بعدی آماده سازی محیط توسعه است. ما از Python 3.12 در این کد لبه استفاده خواهیم کرد و از مدیر پروژه uv python برای ساده سازی نیاز به ایجاد و مدیریت نسخه پایتون و محیط مجازی استفاده خواهیم کرد.

اگر هنوز ترمینال را باز نکرده‌اید، آن را با کلیک کردن روی Terminal -> New Terminal باز کنید یا از Ctrl + Shift + C استفاده کنید.

uv را دانلود و با دستور زیر پایتون 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 مقداردهی کنیم

uv init

main.py، .python-version و pyproject.toml ایجاد شده در فهرست را مشاهده خواهید کرد. این فایل ها برای نگهداری پروژه در دایرکتوری مورد نیاز هستند. وابستگی ها و پیکربندی های پایتون را می توان در pyproject.toml و .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!

این نشان می دهد که پروژه پایتون به درستی راه اندازی شده است. ما نیازی به ایجاد دستی یک محیط مجازی نداشتیم زیرا uv از قبل آن را مدیریت می کند. بنابراین از این نقطه، دستور استاندارد پایتون (به عنوان مثال python main.py ) با uv run (به عنوان مثال uv run main.py ) جایگزین می شود.

Dependencies مورد نیاز را نصب کنید

ما این وابستگی های بسته 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 استفاده خواهیم کرد، بنابراین می توان آن را به راحتی با استقرار Cloud Run بعداً ادغام کرد. pydantic-settings یک بسته پایتون است که می تواند بررسی نوع فایل های پیکربندی را اعمال کند.

یک فایل با نام settings.yaml با تنظیمات زیر ایجاد کنید. روی File->New Text File کلیک کنید و کد زیر را پر کنید. سپس آن را به عنوان 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 استفاده می‌کنیم.

سپس، فایل 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. سرویس Frontend را با استفاده از Gradio بسازید

ما یک رابط وب چت خواهیم ساخت که شبیه این است

این شامل یک فیلد ورودی برای کاربران برای ارسال متن و آپلود فایل است. علاوه بر این، کاربر همچنین می‌تواند دستورالعمل‌های سیستم را که در قسمت ورودی‌های اضافی به Gemini API ارسال می‌شود، بازنویسی کند.

ما سرویس frontend را با استفاده از 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,
    )

پس از آن، می توانیم سرویس frontend را با دستور زیر اجرا کنیم. فراموش نکنید که نام فایل 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 محلی بررسی کنید. همچنین، می‌توانید با کلیک بر روی دکمه پیش‌نمایش وب در سمت راست بالای Cloud Editor، به برنامه frontend دسترسی پیدا کنید و در پورت 8080 Preview را انتخاب کنید.

رابط وب را خواهید دید، با این حال هنگام تلاش برای ارسال چت، به دلیل سرویس پشتیبان که هنوز راه اندازی نشده است، با خطای مورد انتظار مواجه خواهید شد.

حالا، اجازه دهید سرویس اجرا شود و هنوز آن را نکشید. در ضمن ما می توانیم اجزای مهم کد را در اینجا مورد بحث قرار دهیم

توضیح کد

کد ارسال داده از رابط وب به باطن در این قسمت است

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 ) پیدا کنید.

علاوه بر این، در یک رابط چت، ارسال تاریخچه چت به عنوان زمینه اضافی برای دادن "حافظه" مکالمه به Gemini نیز مهم است. بنابراین در این رابط وب، تاریخچه چت که در هر جلسه وب توسط Gradio مدیریت می شود را نیز ارسال می کنیم و آن را همراه با پیام ورودی کاربر ارسال می کنیم. علاوه بر این، کاربر را قادر می سازد تا دستورالعمل سیستم را تغییر دهد و آن را نیز ارسال کند

4. ساخت Backend Service با استفاده از FastAPI

در مرحله بعد، ما نیاز به ساخت backend داریم که بتواند بار مورد بحث قبلی، آخرین پیام کاربر، تاریخچه چت و دستورالعمل سیستم را مدیریت کند. ما از 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 ذخیره کنید. پس از آن می توانیم سرویس Backend را اجرا کنیم. به یاد داشته باشید که در مرحله قبل سرویس frontend را درست اجرا کردیم، اکنون باید ترمینال جدید را باز کنیم و سعی کنیم این سرویس باطن را اجرا کنیم.

یک ترمینال جدید ایجاد کنید. به ترمینال خود در قسمت پایین بروید و دکمه "+" را برای ایجاد یک ترمینال جدید پیدا کنید. یا می توانید 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 decorator تعریف می کنیم. ما همچنین از Pydantic برای تعریف قرارداد API استفاده می کنیم. مشخص می کنیم که مسیر تولید پاسخ در مسیر چت / با روش 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] قالب بندی کنیم. هر محتوا باید حداقل دارای یک نقش و ارزش قطعات باشد. نقش به منبع پیام اشاره دارد، خواه کاربر باشد یا مدل. جایی که قسمت ها به خود فرمان اشاره دارد، جایی که می تواند فقط متن یا ترکیبی از روش های مختلف باشد. نحوه ساختار آرگومان های محتوا را در جزئیات این مستندات ببینید

مدیریت داده های غیر متنی (چند وجهی).

همانطور که قبلاً در قسمت frontend ذکر شد، یکی از راه‌های ارسال داده‌های غیر متنی یا چند وجهی، ارسال داده‌ها به عنوان رشته base64 است. همچنین باید نوع MIME را برای داده ها مشخص کنیم تا بتوان آن را به درستی تفسیر کرد، به عنوان مثال اگر داده های تصویر را با پسوند jpg ارسال کنیم، نوع MIME تصویر/jpeg را ارائه کنیم.

این قسمت از کد، داده های 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. آزمون ادغام

اکنون، باید چندین سرویس را در تب مختلف کنسول ابری اجرا کنید:

سرویس Frontend در پورت 8080 اجرا می شود

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

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

سرویس Backend در پورت 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

اکنون، البته ما می خواهیم این برنامه شگفت انگیز را به دیگران نشان دهیم. برای انجام این کار، می توانیم این برنامه را بسته بندی کنیم و آن را به عنوان یک سرویس عمومی که برای دیگران قابل دسترسی است، در Cloud Run مستقر کنیم. برای انجام این کار، بیایید معماری را دوباره مرور کنیم

در این کد لبه هر دو سرویس frontend و backend را در 1 ظرف قرار می دهیم. برای مدیریت هر دو سرویس به کمک سرپرست نیاز داریم.

فایل جدید ایجاد کنید، روی 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 برای تنظیم شناسه پروژه استفاده کرده اید:

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 فشار می دهد. همچنین از شما می خواهد که مخزن آرتیفکت رجیستری را در منطقه ایجاد کند، به این پاسخ " Y" را بدهید. همچنین وقتی از شما می پرسد که آیا می خواهید فراخوان های احراز هویت نشده را مجاز کنید، بگویید " y " . توجه داشته باشید که ما در اینجا اجازه دسترسی بدون احراز هویت را می دهیم زیرا این یک برنامه آزمایشی است. توصیه این است که از احراز هویت مناسب برای برنامه های تجاری و تولیدی خود استفاده کنید.

پس از تکمیل استقرار، باید پیوندی شبیه به زیر دریافت کنید:

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

ادامه دهید و از برنامه خود از پنجره ناشناس یا دستگاه تلفن همراه خود استفاده کنید. از قبل باید زنده باشد.

7. چالش

اکنون زمان آن است که مهارت های اکتشافی خود را بدرخشید و صیقل دهید. آیا برای تغییر کد نیاز دارید تا دستیار بتواند از خواندن فایل های صوتی یا شاید فایل های ویدیویی پشتیبانی کند؟

8. پاکسازی کنید

برای جلوگیری از تحمیل هزینه به حساب Google Cloud خود برای منابع مورد استفاده در این Codelab، این مراحل را دنبال کنید:

در کنسول Google Cloud، به صفحه مدیریت منابع بروید.
در لیست پروژه، پروژه ای را که می خواهید حذف کنید انتخاب کنید و سپس روی Delete کلیک کنید.
در محاوره، شناسه پروژه را تایپ کنید و سپس روی Shut down کلیک کنید تا پروژه حذف شود.
یا می‌توانید به Cloud Run در کنسول بروید، سرویسی را که به تازگی مستقر کرده‌اید انتخاب کرده و حذف کنید.