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

۱. مقدمه

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

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

  1. پروژه Google Cloud خود را آماده کنید و تمام API های مورد نیاز را روی آن فعال کنید
  2. ساخت رابط کاربری چت سرویس frontend با استفاده از کتابخانه Gradio
  3. ساخت سرویس backend - سرور 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-settings
  • با استفاده از Dockerfile، برنامه را روی Cloud Run مستقر کنید و متغیرهای محیطی را با فایل YAML ارائه دهید.

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

  • مرورگر وب کروم
  • یک حساب جیمیل
  • یک پروژه ابری با قابلیت پرداخت صورتحساب

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

۲. قبل از شروع

راه‌اندازی پروژه ابری در ویرایشگر Cloud Shell

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

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

۱۸۲۹c۳۷۵۹۲۲۷c۱۹b.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. روی دکمه‌ی Open Editor کلیک کنید، این کار یک ویرایشگر Cloud Shell باز می‌کند، می‌توانیم کد خود را اینجا بنویسیم b16d56e4979ec951.png
  2. مطمئن شوید که پروژه Cloud Code در گوشه پایین سمت چپ (نوار وضعیت) ویرایشگر Cloud Shell، همانطور که در تصویر زیر مشخص شده است، تنظیم شده باشد و روی پروژه فعال Google Cloud که در آن صورتحساب را فعال کرده‌اید، تنظیم شده باشد. در صورت درخواست، آن را تأیید کنید . ممکن است پس از مقداردهی اولیه ویرایشگر Cloud Shell، مدتی طول بکشد تا دکمه Cloud Code - Sign In ظاهر شود، لطفاً صبور باشید. اگر قبلاً دستور قبلی را دنبال کرده‌اید، ممکن است این دکمه به جای دکمه ورود، مستقیماً به پروژه فعال شده شما اشاره کند.

f5003b9c38b43262.png

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

70f80078e01a02d8.png

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

۳۶۲ff۳۳۲۲۵۶d۶۹۳۳.jpeg

۸۵۹۵۷۵۶۵۳۱۶۳۰۸d۹.jpeg

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

8409d8db18690fdf.png

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

۱ef5bb44f1d2c2a4.png

در مرحله بعد، محیط پایتون خود را آماده خواهیم کرد

تنظیمات محیط

آماده‌سازی محیط مجازی پایتون

مرحله بعدی آماده‌سازی محیط توسعه است. ما در این آزمایشگاه کد از پایتون ۳.۱۲ استفاده خواهیم کرد و از uv python project manager برای ساده‌سازی نیاز به ایجاد و مدیریت نسخه پایتون و محیط مجازی استفاده خواهیم کرد.

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

f8457daf0bed059e.jpeg

  1. uv را دانلود و پایتون ۳.۱۲ را با دستور زیر نصب کنید 
curl -LsSf https://astral.sh/uv/0.6.6/install.sh | sh && \
source $HOME/.local/bin/env && \
uv python install 3.12
  1. حالا بیایید پروژه پایتون را با استفاده از uv مقداردهی اولیه کنیم. 
uv init
  1. خواهید دید که main.py، .python-version و pyproject.toml در دایرکتوری ایجاد شده‌اند. این فایل‌ها برای نگهداری پروژه در دایرکتوری مورد نیاز هستند. وابستگی‌ها و پیکربندی‌های پایتون را می‌توان در pyproject.toml مشخص کرد و .python-version نسخه پایتون مورد استفاده برای این پروژه را استانداردسازی می‌کند. برای مطالعه بیشتر در این مورد می‌توانید به این مستندات مراجعه کنید. 
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!

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

نصب وابستگی‌های مورد نیاز

ما وابستگی‌های بسته codelab را نیز با استفاده از دستور uv اضافه خواهیم کرد. دستور زیر را اجرا کنید 

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

خواهید دید که بخش "dependencies" در pyproject.toml به‌روزرسانی می‌شود تا دستور قبلی را منعکس کند.

فایل‌های پیکربندی راه‌اندازی

حالا باید فایل‌های پیکربندی این پروژه را تنظیم کنیم. فایل‌های پیکربندی برای ذخیره متغیرهای پویا استفاده می‌شوند که به راحتی در هنگام استقرار مجدد قابل تغییر هستند. در این پروژه از فایل‌های پیکربندی مبتنی بر YAML با بسته pydantic-settings استفاده خواهیم کرد، بنابراین می‌توان آن را به راحتی با استقرار Cloud Run در آینده ادغام کرد. pydantic-settings یک بسته پایتون است که می‌تواند بررسی نوع را برای فایل‌های پیکربندی اعمال کند.

  1. یک فایل با نام 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 استفاده خواهیم کرد.

  1. سپس، فایل پایتون 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 در اولویت بالاتری قرار می‌دهیم.

حالا می‌توانیم به مرحله بعدی، یعنی ساخت سرویس‌ها، برویم.

۳. ساخت سرویس Frontend با استفاده از Gradio

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

5bcfa1cce6618305.png

این شامل یک فیلد ورودی برای کاربران جهت ارسال متن و آپلود فایل است. علاوه بر این، کاربر می‌تواند دستورالعمل سیستم را که به 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 محلی، رابط وب را بررسی کنید. همچنین می‌توانید با کلیک روی دکمه پیش‌نمایش وب در سمت راست بالای ویرایشگر ابری و انتخاب پیش‌نمایش روی پورت ۸۰۸۰، به برنامه frontend دسترسی پیدا کنید.

49cbdfdf77964065.jpeg

رابط وب را مشاهده خواهید کرد، با این حال هنگام تلاش برای ارسال چت به دلیل عدم راه‌اندازی سرویس backend، خطای مورد انتظار را دریافت خواهید کرد.

bd0464140308cfbe.png

حالا، بگذارید سرویس اجرا شود و فعلاً آن را از بین نبرید. در این فاصله می‌توانیم اجزای مهم کد را اینجا مورد بحث قرار دهیم.

توضیح کد

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

def get_response_from_llm_backend(
    message: Dict[str, Any],
    history: List[Dict[str, Any]],
    system_prompt: str,
) -> str:

    ... 
    # Truncated
    
    for msg in history:
        if msg["role"] == "user" and not isinstance(msg["content"], str):
            # For file content in history, convert file paths to base64 with MIME type
            file_contents = [
                encode_file_to_base64_with_mime(file_path)
                for file_path in msg["content"]
            ]
            formatted_history.append({"role": msg["role"], "content": file_contents})
        else:
            formatted_history.append({"role": msg["role"], "content": msg["content"]})

    # Extract files and convert to base64 with MIME type
    files_with_mime = []
    if uploaded_files := message.get("files", []):
        for file_path in uploaded_files:
            files_with_mime.append(encode_file_to_base64_with_mime(file_path))

    # Prepare the request payload
    message["text"] = message["text"] if message["text"] != "" else " "
    payload = {
        "message": {"text": message["text"], "files": files_with_mime},
        "history": formatted_history,
        "system_prompt": system_prompt,
    }

    # Truncated
    ...

وقتی می‌خواهیم داده‌های چندوجهی را به Gemini ارسال کنیم و داده‌ها را بین سرویس‌ها قابل دسترسی کنیم، یکی از مکانیسم‌هایی که می‌توانیم انجام دهیم تبدیل داده‌ها به نوع داده base64 همانطور که در کد اعلام شده است، می‌باشد. همچنین باید نوع MIME داده‌ها را نیز اعلام کنیم. با این حال، API Gemini نمی‌تواند از همه انواع MIME موجود پشتیبانی کند، از این رو مهم است که بدانیم چه نوع MIMEهایی توسط Gemini پشتیبانی می‌شوند که می‌توان آنها را در این مستندات مطالعه کرد. می‌توانید اطلاعات را در هر یک از قابلیت‌های API Gemini (مثلاً Vision ) پیدا کنید.

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

۴. ساخت سرویس Backend با استفاده از FastAPI

در مرحله بعد، باید backend را بسازیم که بتواند payload، آخرین پیام کاربر، تاریخچه چت و دستورالعمل‌های سیستم که قبلاً مورد بحث قرار گرفت را مدیریت کند. ما از FastAPI برای ایجاد سرویس HTTP backend استفاده خواهیم کرد.

یک فایل جدید ایجاد کنید، روی 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 را درست اجرا کردیم، اکنون باید یک ترمینال جدید باز کنیم و سعی کنیم این سرویس backend را اجرا کنیم.

  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] قالب‌بندی کنیم. هر محتوا باید حداقل یک مقدار role و parts داشته باشد. role به منبع پیام اشاره دارد، چه کاربر باشد چه مدل. parts به خود prompt اشاره دارد، که می‌تواند فقط متن یا ترکیبی از روش‌های مختلف باشد. نحوه ساختاردهی آرگومان‌های Content را در جزئیات این مستندات ببینید.

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

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

این بخش از کد، داده‌های base64 را از کیت توسعه نرم‌افزار Gemini به فرمت 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)

۵. آزمون ادغام

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

  • سرویس frontend روی پورت ۸۰۸۰ اجرا می‌شود 
* Running on local URL:  http://0.0.0.0:8080

To create a public link, set `share=True` in `launch()`.
  • سرویس بک‌اند روی پورت ۸۰۸۱ اجرا می‌شود 
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)

در وضعیت فعلی، شما باید بتوانید اسناد خود را در چت به طور یکپارچه با دستیار از طریق برنامه وب روی پورت ۸۰۸۰ ارسال کنید. می‌توانید با آپلود فایل‌ها و پرسیدن سوال، آزمایش را شروع کنید! مراقب باشید که انواع خاصی از فایل‌ها هنوز پشتیبانی نمی‌شوند و باعث ایجاد خطا می‌شوند.

همچنین می‌توانید دستورالعمل‌های سیستم را از فیلد ورودی‌های اضافی زیر کادر متن ویرایش کنید.

ee9c849a276d378.png

۶. استقرار در Cloud Run

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

b102df2c3f1adabf.jpeg

در این آزمایشگاه کد، ما هر دو سرویس frontend و backend را در یک container قرار خواهیم داد. برای مدیریت هر دو سرویس به کمک supervisord نیاز خواهیم داشت.

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

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

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

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

۷. چالش

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

۸. تمیز کردن

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

  1. در کنسول گوگل کلود، به صفحه مدیریت منابع بروید.
  2. در لیست پروژه‌ها، پروژه‌ای را که می‌خواهید حذف کنید انتخاب کنید و سپس روی «حذف» کلیک کنید.
  3. در کادر محاوره‌ای، شناسه پروژه را تایپ کنید و سپس برای حذف پروژه، روی خاموش کردن کلیک کنید.
  4. روش دیگر این است که به Cloud Run در کنسول بروید، سرویسی را که اخیراً مستقر کرده‌اید انتخاب کرده و حذف کنید.