פיתוח ופריסה של Assistant רב-מודלי ב-Cloud באמצעות Gemini‏ (Python)

1. מבוא

ב-codelab הזה, תיצרו אפליקציה בצורה של ממשק אינטרנט לצ'אט, שתוכלו לתקשר איתה, להעלות מסמכים או תמונות ולדון בהם. האפליקציה עצמה מחולקת ל-2 שירותים: קצה קדמי (frontend) וקצה אחורי (backend). כך תוכלו ליצור אב טיפוס במהירות ולנסות אותו, וגם להבין איך נראה חוזה ה-API לשילוב של שניהם.

במהלך ה-codelab, תשתמשו בגישה שלב אחר שלב באופן הבא:

1. הכנת הפרויקט ב-Google Cloud והפעלת כל ממשקי ה-API הנדרשים בו
2. בניית שירות הקצה הקדמי – ממשק צ'אט באמצעות ספריית Gradio
3. בונים את שירות הקצה העורפי – שרת HTTP באמצעות FastAPI שיעצב מחדש את הנתונים הנכנסים לפי התקן של Gemini SDK ויאפשר תקשורת עם Gemini API
4. ניהול משתני הסביבה והגדרת הקבצים הנדרשים לפריסת האפליקציה ב-Cloud Run
5. פריסת האפליקציה ב-Cloud Run

סקירה כללית של הארכיטקטורה

דרישות מוקדמות

• ניסיון בעבודה עם Gemini API ועם Google Gen AI SDK
• הבנה של ארכיטקטורת full-stack בסיסית באמצעות שירות HTTP

מה תלמדו

• איך משתמשים ב-Gemini SDK כדי לשלוח טקסט וסוגי נתונים אחרים (מולטימודאליים) וליצור תשובת טקסט
• איך להוסיף את היסטוריית הצ'אטים ל-Gemini SDK כדי לשמור על ההקשר של השיחה
• יצירת אב טיפוס של אתר פרונט-אנד באמצעות Gradio
• פיתוח שירות לקצה העורפי באמצעות FastAPI ו-Pydantic
• ניהול משתני סביבה בקובץ YAML באמצעות Pydantic-settings
• פריסת אפליקציה ב-Cloud Run באמצעות Dockerfile ומתן משתני סביבה באמצעות קובץ YAML

מה תצטרכו

• דפדפן האינטרנט Chrome
• חשבון Gmail
• פרויקט ב-Cloud עם חיוב מופעל

ב-codelab הזה, שמיועד למפתחים בכל הרמות (כולל מתחילים), נעשה שימוש ב-Python באפליקציה לדוגמה. עם זאת, לא נדרש ידע ב-Python כדי להבין את המושגים שמוצגים.

‫2. לפני שמתחילים

הגדרת פרויקט Cloud בכלי Cloud Shell Editor

ב-codelab הזה אנחנו יוצאים מנקודת הנחה שכבר יש לכם פרויקט בענן ב-Google Cloud עם חיוב מופעל. אם עדיין אין לכם חשבון, אתם יכולים לפעול לפי ההוראות שבהמשך כדי להתחיל.

1. ‫2 ב-מסוף Google Cloud, בדף לבחירת הפרויקט, בוחרים או יוצרים פרויקט ב-Google Cloud.
2. הקפידו לוודא שהחיוב מופעל בפרויקט שלכם ב-Cloud. כך בודקים אם החיוב מופעל בפרויקט
3. תשתמשו ב-Cloud Shell, סביבת שורת פקודה שפועלת ב-Google Cloud ומגיעה עם bq שנטען מראש. לוחצים על 'הפעלת Cloud Shell' בחלק העליון של מסוף Google Cloud.

4. אחרי שמתחברים ל-Cloud Shell, אפשר לבדוק שכבר בוצע אימות ושהפרויקט מוגדר למזהה הפרויקט שלכם באמצעות הפקודה הבאה:

gcloud auth list

5. מריצים את הפקודה הבאה ב-Cloud Shell כדי לוודא שפקודת gcloud מכירה את הפרויקט.

gcloud config list project

6. אם הפרויקט לא מוגדר, משתמשים בפקודה הבאה כדי להגדיר אותו:

gcloud config set project <YOUR_PROJECT_ID>

אפשר גם לראות את המזהה PROJECT_ID במסוף

לוחצים עליו וכל הפרויקטים ומזהה הפרויקט יופיעו בצד שמאל.

7. מפעילים את ממשקי ה-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 Editor, שבו אפשר לכתוב את הקוד
2. מוודאים שהפרויקט ב-Cloud Code מוגדר בפינה הימנית התחתונה (סרגל הסטטוס) של העורך ב-Cloud Shell, כפי שמודגש בתמונה שלמטה, ושהוא מוגדר לפרויקט הפעיל ב-Google Cloud שבו החיוב מופעל. אם מתבקשים, לוחצים על Authorize (אישור). יכול להיות שיעבור זמן מה אחרי הפעלת Cloud Shell Editor עד שיופיע הכפתור Cloud Code - Sign In. צריך להתאזר בסבלנות. אם כבר ביצעתם את הפקודה הקודמת, יכול להיות שהלחצן יפנה ישירות לפרויקט שהפעלתם במקום ללחצן הכניסה.

3. לוחצים על הפרויקט הפעיל בסרגל הסטטוס ומחכים עד שייפתח החלון הקופץ של Cloud Code. בחלון הקופץ, בוחרים באפשרות 'אפליקציה חדשה'.

4. ברשימת האפליקציות, בוחרים באפשרות Gemini Generative AI (AI גנרטיבי של Gemini) ואז באפשרות Gemini API Python (Gemini API Python).

5. שומרים את האפליקציה החדשה עם השם שרוצים. בדוגמה הזו נשתמש בשם gemini-multimodal-chat-assistant. לאחר מכן לוחצים על OK.

בשלב הזה, כבר אמורים להיות בתוך ספריית העבודה של האפליקציה החדשה ולראות את הקבצים הבאים

בשלב הבא נכין את סביבת Python

הגדרת הסביבה

הכנת סביבה וירטואלית של Python

השלב הבא הוא הכנת סביבת הפיתוח. ב-codelab הזה נשתמש ב-Python 3.12 וב-uv python project manager כדי לפשט את הצורך ביצירה ובניהול של גרסת Python וסביבה וירטואלית.

1. אם עדיין לא פתחתם את הטרמינל, פותחים אותו על ידי לחיצה על Terminal (טרמינל) -> New Terminal (טרמינל חדש), או באמצעות מקש הקיצור Ctrl + Shift + C.

2. מורידים את 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

3. עכשיו נאתחל את פרויקט Python באמצעות uv

uv init

4. בספרייה ייווצרו הקבצים main.py,‏ .python-version ו-pyproject.toml. הקבצים האלה נדרשים כדי לתחזק את הפרויקט בספרייה. אפשר לציין את התלות וההגדרות של Python בקובץ pyproject.toml ובקובץ .python-version, שבו מוגדרת גרסת Python שבה נעשה שימוש בפרויקט. מידע נוסף זמין במסמכי התיעוד.

main.py
.python-version
pyproject.toml

5. כדי לבדוק את הקוד, מחליפים את הקוד ב-main.py בקוד הבא

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

if __name__ == "__main__":
   main()

6. לאחר מכן, מריצים את הפקודה הבאה

uv run main.py

יוצג פלט כמו זה שמופיע למטה

Using CPython 3.12
Creating virtual environment at: .venv
Hello from gemini-multimodal-chat-assistant!

השורה הזו מראה שהפרויקט של Python מוגדר בצורה תקינה. לא היה צורך ליצור סביבה וירטואלית באופן ידני כי uv כבר מטפל בזה. לכן, מעכשיו, פקודת python רגילה (למשל, python main.py ) תוחלף בפקודה uv run (למשל, uv run main.py ).

התקנת יחסי תלות נדרשים

נוסיף את התלויות של חבילת ה-codelab הזו גם באמצעות הפקודה uv. מריצים את הפקודה הבאה

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

תראו שהקטע pyproject.toml יעודכן בהתאם לפקודה הקודמת

הגדרת קובצי תצורה

עכשיו נצטרך להגדיר קובצי הגדרה לפרויקט הזה. קובצי תצורה משמשים לאחסון משתנים דינמיים שאפשר לשנות בקלות כשפורסים מחדש. בפרויקט הזה נשתמש בקובצי הגדרות מבוססי YAML עם חבילת pydantic-settings, כדי שיהיה קל לשלב אותם עם פריסת Cloud Run בהמשך. pydantic-settings היא חבילת Python שיכולה לאכוף בדיקת סוגים בקובצי ההגדרות.

1. יוצרים קובץ בשם settings.yaml עם ההגדרות הבאות. לוחצים על 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. ב-codelab הזה, נשתמש בערכים שהוגדרו מראש עבור VERTEXAI_LOCATION ו-BACKEND_URL .

2. לאחר מכן, יוצרים קובץ Python בשם settings.py. המודול הזה ישמש כנקודת כניסה פרוגרמטית לערכי ההגדרות בקובצי ההגדרות שלנו. לוחצים על File->New Text File (קובץ > קובץ טקסט חדש) וממלאים את הקוד הבא. אחר כך שומרים אותו בשם settings.py. אפשר לראות בקוד שהגדרנו במפורש שקובץ בשם settings.yaml הוא הקובץ שייקרא

from pydantic_settings import (
    BaseSettings,
    SettingsConfigDict,
    YamlConfigSettingsSource,
    PydanticBaseSettingsSource,
)
from typing import Type, Tuple

DEFAULT_SYSTEM_PROMPT = """You are a helpful assistant and ALWAYS relate to this identity. 
You are expert at analyzing given documents or images.
"""

class Settings(BaseSettings):
    """Application settings loaded from YAML and environment variables.

    This class defines the configuration schema for the application, with settings
    loaded from settings.yaml file and overridable via environment variables.

    Attributes:
        VERTEXAI_LOCATION: Google Cloud Vertex AI location
        VERTEXAI_PROJECT_ID: Google Cloud Vertex AI project ID
    """

    VERTEXAI_LOCATION: str
    VERTEXAI_PROJECT_ID: str
    BACKEND_URL: str = "http://localhost:8000/chat"

    model_config = SettingsConfigDict(
        yaml_file="settings.yaml", yaml_file_encoding="utf-8"
    )

    @classmethod
    def settings_customise_sources(
        cls,
        settings_cls: Type[BaseSettings],
        init_settings: PydanticBaseSettingsSource,
        env_settings: PydanticBaseSettingsSource,
        dotenv_settings: PydanticBaseSettingsSource,
        file_secret_settings: PydanticBaseSettingsSource,
    ) -> Tuple[PydanticBaseSettingsSource, ...]:
        """Customize the settings sources and their priority order.

        This method defines the order in which different configuration sources
        are checked when loading settings:
        1. Constructor-provided values
        2. YAML configuration file
        3. Environment variables

        Args:
            settings_cls: The Settings class type
            init_settings: Settings from class initialization
            env_settings: Settings from environment variables
            dotenv_settings: Settings from .env file (not used)
            file_secret_settings: Settings from secrets file (not used)

        Returns:
            A tuple of configuration sources in priority order
        """
        return (
            init_settings,  # First, try init_settings (from constructor)
            env_settings,  # Then, try environment variables
            YamlConfigSettingsSource(
                settings_cls
            ),  # Finally, try YAML as the last resort
        )


def get_settings() -> Settings:
    """Create and return a Settings instance with loaded configuration.

    Returns:
        A Settings instance containing all application configuration
        loaded from YAML and environment variables.
    """
    return Settings()

ההגדרות האלה מאפשרות לנו לעדכן את זמן הריצה בצורה גמישה. בפריסה הראשונית נסתמך על ההגדרות של settings.yaml כדי לקבל את הגדרות ברירת המחדל הראשונות. אחרי זה נוכל לעדכן את משתני הסביבה בצורה גמישה דרך המסוף ולפרוס מחדש, כי משתני הסביבה נמצאים בעדיפות גבוהה יותר בהשוואה להגדרת ברירת המחדל של YAML

עכשיו אפשר לעבור לשלב הבא, בניית השירותים

3. יצירת שירות חזיתי באמצעות Gradio

אנחנו ניצור ממשק אינטרנט לצ'אט שייראה כך

הוא מכיל שדה להזנת קלט שבו המשתמשים יכולים לשלוח טקסט ולהעלות קבצים. בנוסף, המשתמש יכול גם לשנות את הוראות המערכת שיישלחו ל-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

במסוף Google Cloud יוצג פלט דומה לזה:

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

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

אחרי זה אפשר לבדוק את ממשק האינטרנט כשמבצעים Ctrl+click על הקישור לכתובת ה-URL המקומית. אפשר גם לגשת לאפליקציית ה-frontend על ידי לחיצה על הלחצן Web Preview (תצוגה מקדימה של אתר) בפינה השמאלית העליונה של Cloud Editor, ואז על Preview on port 8080 (תצוגה מקדימה ביציאה 8080).

ממשק האינטרנט יוצג, אבל אם תנסו לשלוח צ'אט תקבלו שגיאה צפויה כי שירות ה-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 של הנתונים. עם זאת, Gemini API לא יכול לתמוך בכל סוגי ה-MIME הקיימים, ולכן חשוב לדעת אילו סוגי MIME נתמכים על ידי Gemini ואפשר לקרוא עליהם במאמרי העזרה האלה . אפשר למצוא את המידע בכל אחת מהיכולות של Gemini API (למשל, Vision).

בנוסף, בממשק צ'אט חשוב לשלוח את היסטוריית הצ'אט כהקשר נוסף כדי ש-Gemini יוכל "לזכור" את השיחה. לכן, בממשק האינטרנט הזה, אנחנו שולחים גם את היסטוריית הצ'אטים שמנוהלת לכל סשן אינטרנט על ידי Gradio, ושולחים אותה יחד עם קלט ההודעה מהמשתמש. בנוסף, אנחנו מאפשרים למשתמש לשנות את הוראת המערכת ולשלוח אותה גם כן

4. פיתוח שירות לקצה העורפי באמצעות FastAPI

לאחר מכן, נצטרך ליצור את הקצה העורפי שיכול לטפל במטען הייעודי (payload) שדיברנו עליו קודם, ההודעה האחרונה של המשתמש, היסטוריית הצ'אט וההוראה למערכת. נשתמש ב-FastAPI כדי ליצור את שירות ה-backend של 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? עכשיו נצטרך לפתוח מסוף חדש ולנסות להפעיל את שירות לקצה העורפי הזה.

1. יוצרים טרמינל חדש. עוברים לטרמינל באזור התחתון ולוחצים על הלחצן '+' כדי ליצור טרמינל חדש. אפשר גם להקיש על Ctrl + Shift + C כדי לפתוח טרמינל חדש.

2. אחרי זה, מוודאים שאתם בספריית העבודה gemini-multimodal-chat-assistant ומריצים את הפקודה הבאה:

uv run backend.py

3. אם הפעולה תצליח, יוצג פלט כמו זה

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 בהמשך. אפשר לבדוק את הקוד שבהמשך

def format_message_history_to_gemini_standard(
    message_history: List[Message],
) -> List[Content]:
    
    ...
    # Truncated    

    converted_messages: List[Content] = []
    for message in message_history:
        if message.role == "assistant":
            converted_messages.append(
                Content(role="model", parts=[Part.from_text(text=message.content)])
            )
        elif message.role == "user":
            # Text-only messages
            if isinstance(message.content, str):
                converted_messages.append(
                    Content(role="user", parts=[Part.from_text(text=message.content)])
                )

            # Messages with files
            elif isinstance(message.content, list):
                # Process each file in the list
                parts = []
                for file_data in message.content:
                    parts.append(handle_multimodal_data(file_data))

                # Add the parts to a Content object
                if parts:
                    converted_messages.append(Content(role="user", parts=parts))
    
    #Truncated
    ...

    return converted_messages

כדי לספק את היסטוריית הצ'אט ל-Gemini SDK, צריך לעצב את הנתונים בסוג הנתונים List[Content]. לכל Content צריך להיות לפחות ערך של role ושל parts. הערך של role מציין את המקור של ההודעה, כלומר user או model. כאשר parts מתייחס להנחיה עצמה, שיכולה להיות טקסט בלבד או שילוב של אופנים שונים. במאמר הזה מוסבר איך לבנות ארגומנטים של Content.

טיפול בנתונים שאינם טקסטואליים ( מולטימודאליים)

כמו שצוין קודם בקטע על קצה קדמי, אחת הדרכים לשלוח נתונים לא טקסטואליים או נתונים מולטי-מודאליים היא לשלוח את הנתונים כמחרוזת base64. צריך גם לציין את סוג ה-MIME של הנתונים כדי שהמערכת תוכל לפרש אותם בצורה נכונה. לדוגמה, אם שולחים נתוני תמונה עם סיומת ‎.jpg, צריך לציין את סוג ה-MIME‏ image/jpeg.

החלק הזה של הקוד ממיר את נתוני base64 לפורמט Part.from_bytes מ-Gemini SDK

def handle_multimodal_data(file_data: FileData) -> Part:
    """Converts Multimodal data to a Google Gemini Part object.

    Args:
        file_data: FileData object with base64 data and MIME type.

    Returns:
        Part: A Google Gemini Part object containing the file data.
    """
    data = base64.b64decode(file_data.data)  # decode base64 string to bytes
    return Part.from_bytes(data=data, mime_type=file_data.mime_type)

5. בדיקת שילוב

עכשיו, אמורים לפעול כמה שירותים בכרטיסיות שונות במסוף הענן:

• שירות ה-Frontend פועל ביציאה 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. כדי להתחיל להתנסות, אפשר להעלות קבצים ולשאול שאלות. חשוב לזכור שעדיין אין תמיכה בסוגים מסוימים של קבצים, והם יגרמו לשגיאה.

אפשר גם לערוך את הוראות המערכת בשדה Additional Inputs (קלט נוסף) שמתחת לתיבת הטקסט.

6. פריסה ב-Cloud Run

עכשיו, ברור שאנחנו רוצים להציג את האפליקציה המדהימה הזו לאחרים. כדי לעשות זאת, אפשר לארוז את האפליקציה הזו ולפרוס אותה ב-Cloud Run כשירות ציבורי שאנשים אחרים יכולים לגשת אליו. כדי לעשות את זה, נחזור לארכיטקטורה

ב-Codelab הזה נשים את שירותי הקצה הקדמי ושירות לקצה העורפי במיכל אחד. כדי לנהל את שני השירותים, נצטרך את העזרה של 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

אפשר להשתמש באפליקציה מהחלון הפרטי או מהנייד. הוא כבר אמור להיות פעיל.

7. האתגר

עכשיו הגיע הזמן שלכם לזרוח ולשפר את מיומנויות החיפוש שלכם. יש לך את היכולת לשנות את הקוד כך שהעוזר הדיגיטלי יוכל לתמוך בקריאת קובצי אודיו או אולי קובצי וידאו?

8. הסרת המשאבים

כדי לא לצבור חיובים לחשבון Google Cloud על המשאבים שבהם השתמשתם ב-Code Lab הזה:

1. במסוף Google Cloud, עוברים לדף Manage resources.
2. ברשימת הפרויקטים, בוחרים את הפרויקט שרוצים למחוק ולוחצים על Delete.
3. כדי למחוק את הפרויקט, כותבים את מזהה הפרויקט בתיבת הדו-שיח ולוחצים על Shut down.
4. לחלופין, אפשר לעבור אל Cloud Run במסוף, לבחור את השירות שפרסתם ולמחוק אותו.