1. מבוא
בקודלאב הזה תלמדו ליצור אפליקציה בצורת ממשק אינטרנט של צ'אט, שבו תוכלו לתקשר עם האפליקציה, להעלות מסמכים או תמונות ולנהל דיון עליהם. האפליקציה עצמה מופרדת לשני שירותים: חזית (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
- פיתוח שירות לקצה העורפי באמצעות FastAPI ו-Pydantic
- ניהול משתני הסביבה בקובץ YAML באמצעות Pydantic-settings
- פריסת אפליקציה ב-Cloud Run באמצעות Dockerfile ומתן משתני סביבה באמצעות קובץ YAML
מה צריך להכין
- דפדפן האינטרנט Chrome
- חשבון Gmail
- פרויקט ב-Cloud שבו החיוב מופעל
סדנת הקוד הזו מיועדת למפתחים מכל הרמות (כולל למתחילים), והיא כוללת שימוש ב-Python באפליקציה לדוגמה. עם זאת, לא צריך ידע ב-Python כדי להבין את המושגים המוצגים.
2. לפני שמתחילים
הגדרת פרויקט Cloud בעורך Cloud Shell
אנחנו יוצאים מנקודת הנחה שכבר יש לכם פרויקט ב-Google Cloud שבו החיוב מופעל. אם עדיין לא עשיתם זאת, תוכלו לפעול לפי ההוראות שבהמשך כדי להתחיל.
- 2במסוף Google Cloud, בדף לבחירת הפרויקט, בוחרים או יוצרים פרויקט ב-Google Cloud.
- הקפידו לוודא שהחיוב מופעל בפרויקט שלכם ב-Cloud. כך בודקים אם החיוב מופעל בפרויקט
- נשתמש ב-Cloud Shell, סביבת שורת פקודה שפועלת ב-Google Cloud ומגיעה עם bq טעון מראש. לוחצים על Activate Cloud Shell בחלק העליון של מסוף Google Cloud.
- אחרי שמתחברים ל-Cloud Shell, בודקים שכבר בוצע אימות ושהמזהה של הפרויקט מוגדר כפרויקט באמצעות הפקודה הבאה:
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, שבו נוכל לכתוב את הקוד שלנו
- מוודאים שהפרויקט ב-Cloud Code מוגדר בפינה הימנית התחתונה (סרגל הסטטוס) של עורך Cloud Shell, כפי שמודגש בתמונה שבהמשך, והוא מוגדר לפרויקט הפעיל ב-Google Cloud שבו החיוב מופעל. נותנים הרשאה אם מוצגת בקשה לכך. יכול להיות שיחלוף זמן מה אחרי שמפעילים את Cloud Shell Editor עד שהלחצן Cloud Code – Sign In יופיע. יש להמתין. אם כבר פעלתם לפי הפקודה הקודמת, יכול להיות שהלחצן יצביע ישירות על הפרויקט המופעל במקום על לחצן הכניסה
- לוחצים על הפרויקט הפעיל הזה בסרגל הסטטוס וממתינים לפתיחת חלון הקופץ של Cloud Code. בחלון הקופץ, בוחרים באפשרות 'אפליקציה חדשה'.
- מרשימת האפליקציות, בוחרים באפשרות Gemini Generative AI ואז באפשרות Gemini API Python.
- שומרים את האפליקציה החדשה בשם הרצוי. בדוגמה הזו נשתמש בשם gemini-multimodal-chat-assistant, ולוחצים על OK.
בשלב הזה, אתם אמורים להיות כבר בספריית העבודה החדשה של האפליקציה ולראות את הקבצים הבאים
בשלב הבא, נלמד איך להכין את סביבת Python.
הגדרת הסביבה
הכנת סביבת Python וירטואלית
השלב הבא הוא הכנת סביבת הפיתוח. נשתמש ב-Python 3.12 ב-codelab הזה, ונשתמש ב-uv python project manager כדי לפשט את הצורך ביצירה ובניהול של גרסת Python וסביבה וירטואלית.
- אם עדיין לא פתחתם את הטרמינל, לוחצים על Terminal (טרמינל) -> New Terminal (טרמינל חדש) או משתמשים בקיצור המקשים Ctrl + Shift + C.
- מורידים את
uvומתקינים את Python 3.12 באמצעות הפקודה הבאה:
curl -LsSf https://astral.sh/uv/0.6.6/install.sh | sh && \
source $HOME/.local/bin/env && \
uv python install 3.12
- עכשיו נאיץ את הפרויקט ב-Python באמצעות
uv
uv init
- בספרייה יופיעו הקבצים main.py, .python-version ו-pyproject.toml. הקבצים האלה נדרשים כדי לשמור על הפרויקט בספרייה. אפשר לציין את יחסי התלות וההגדרות של Python בקובץ 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!
זה מראה שהפרויקט ב-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
תוכלו לראות שהקטע 'dependencies' בקובץ pyproject.toml יתעדכן כך שישקף את הפקודה הקודמת.
הגדרת קובצי תצורה
עכשיו נצטרך להגדיר קובצי תצורה לפרויקט הזה. קובצי תצורה משמשים לאחסון משתנים דינמיים שאפשר לשנות בקלות במהלך פריסה מחדש. בפרויקט הזה נשתמש בקובצי תצורה מבוססי YAML עם החבילה pydantic-settings, כדי שאפשר יהיה לשלב אותם בקלות עם הפריסה של Cloud Run בהמשך. pydantic-settings היא חבילת Python שיכולה לאכוף בדיקת סוגים בקובצי התצורה.
- יוצרים קובץ בשם settings.yaml עם ההגדרות הבאות. לוחצים על קובץ->קובץ טקסט חדש וממלאים את הקוד הבא. לאחר מכן שומרים אותו בתור 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. המודול הזה ישמש כרשומה פרוגרמטית של ערכי התצורה בקובצי התצורה שלנו. לוחצים על קובץ->קובץ טקסט חדש וממלאים את הקוד הבא. לאחר מכן שומרים אותו בתור 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 בשדה הקלט הנוסף.
נבנה את שירות הקצה באמצעות Gradio. משנים את השם של main.py ל-frontend.py ומחליפים את הקוד באמצעות הקוד הבא
import gradio as gr
import requests
import base64
from pathlib import Path
from typing import List, Dict, Any
from settings import get_settings, DEFAULT_SYSTEM_PROMPT
settings = get_settings()
IMAGE_SUFFIX_MIME_MAP = {
".png": "image/png",
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".heic": "image/heic",
".heif": "image/heif",
".webp": "image/webp",
}
DOCUMENT_SUFFIX_MIME_MAP = {
".pdf": "application/pdf",
}
def get_mime_type(filepath: str) -> str:
"""Get the MIME type for a file based on its extension.
Args:
filepath: Path to the file.
Returns:
str: The MIME type of the file.
Raises:
ValueError: If the file type is not supported.
"""
filepath = Path(filepath)
suffix = filepath.suffix
# modify ".jpg" suffix to ".jpeg" to unify the mime type
suffix = suffix if suffix != ".jpg" else ".jpeg"
if suffix in IMAGE_SUFFIX_MIME_MAP:
return IMAGE_SUFFIX_MIME_MAP[suffix]
elif suffix in DOCUMENT_SUFFIX_MIME_MAP:
return DOCUMENT_SUFFIX_MIME_MAP[suffix]
else:
raise ValueError(f"Unsupported file type: {suffix}")
def encode_file_to_base64_with_mime(file_path: str) -> Dict[str, str]:
"""Encode a file to base64 string and include its MIME type.
Args:
file_path: Path to the file to encode.
Returns:
Dict[str, str]: Dictionary with 'data' and 'mime_type' keys.
"""
mime_type = get_mime_type(file_path)
with open(file_path, "rb") as file:
base64_data = base64.b64encode(file.read()).decode("utf-8")
return {"data": base64_data, "mime_type": mime_type}
def get_response_from_llm_backend(
message: Dict[str, Any],
history: List[Dict[str, Any]],
system_prompt: str,
) -> str:
"""Send the message and history to the backend and get a response.
Args:
message: Dictionary containing the current message with 'text' and optional 'files' keys.
history: List of previous message dictionaries in the conversation.
system_prompt: The system prompt to be sent to the backend.
Returns:
str: The text response from the backend service.
"""
# Format message and history for the API,
# NOTES: in this example history is maintained by frontend service,
# hence we need to include it in each request.
# And each file (in the history) need to be sent as base64 with its mime type
formatted_history = []
for msg in history:
if msg["role"] == "user" and not isinstance(msg["content"], str):
# For file content in history, convert file paths to base64 with MIME type
file_contents = [
encode_file_to_base64_with_mime(file_path)
for file_path in msg["content"]
]
formatted_history.append({"role": msg["role"], "content": file_contents})
else:
formatted_history.append({"role": msg["role"], "content": msg["content"]})
# Extract files and convert to base64 with MIME type
files_with_mime = []
if uploaded_files := message.get("files", []):
for file_path in uploaded_files:
files_with_mime.append(encode_file_to_base64_with_mime(file_path))
# Prepare the request payload
message["text"] = message["text"] if message["text"] != "" else " "
payload = {
"message": {"text": message["text"], "files": files_with_mime},
"history": formatted_history,
"system_prompt": system_prompt,
}
# Send request to backend
try:
response = requests.post(settings.BACKEND_URL, json=payload)
response.raise_for_status() # Raise exception for HTTP errors
result = response.json()
if error := result.get("error"):
return f"Error: {error}"
return result.get("response", "No response received from backend")
except requests.exceptions.RequestException as e:
return f"Error connecting to backend service: {str(e)}"
if __name__ == "__main__":
demo = gr.ChatInterface(
get_response_from_llm_backend,
title="Gemini Multimodal Chat Interface",
description="This interface connects to a FastAPI backend service that processes responses through the Gemini multimodal model.",
type="messages",
multimodal=True,
textbox=gr.MultimodalTextbox(file_count="multiple"),
additional_inputs=[
gr.Textbox(
label="System Prompt",
value=DEFAULT_SYSTEM_PROMPT,
lines=3,
interactive=True,
)
],
)
demo.launch(
server_name="0.0.0.0",
server_port=8080,
)
לאחר מכן, נוכל לנסות להריץ את שירות הקצה באמצעות הפקודה הבאה. חשוב לשנות את שם הקובץ main.py ל-frontend.py.
uv run frontend.py
במסוף Cloud יוצג פלט דומה לזה
* Running on local URL: http://0.0.0.0:8080 To create a public link, set `share=True` in `launch()`.
לאחר מכן תוכלו לבדוק את ממשק האינטרנט על ידי לחיצה על ctrl+click על הקישור של כתובת ה-URL המקומית. לחלופין, אפשר לגשת לאפליקציית הקצה הקדמי גם בלחיצה על הלחצן תצוגה מקדימה באינטרנט בפינה השמאלית העליונה של Cloud Editor, ובחירה באפשרות תצוגה מקדימה ביציאה 8080.
תוצג לכם ממשק האינטרנט, אבל תופיע שגיאה צפויה כשתנסו לשלוח צ'אט בגלל ששירות הקצה העורפי עדיין לא הוגדר.
עכשיו נותנים לשירות לפעול ולא מפסיקים אותו עדיין. בינתיים, אפשר לדון כאן ברכיבי הקוד החשובים
הסבר על הקוד
הקוד לשליחת נתונים מממשק האינטרנט לקצה העורפי נמצא בחלק הזה
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
בשלב הבא נצטרך ליצור את הקצה העורפי שיכול לטפל במטען הייעודי שצוין למעלה, בהודעה האחרונה של המשתמש, בהיסטוריית הצ'אט ובהוראות המערכת. נשתמש ב-FastAPI כדי ליצור את שירות הקצה העורפי של HTTP.
יוצרים קובץ חדש, לוחצים על קובץ->קובץ טקסט חדש,מעתיקים ומדביקים את הקוד הבא ושומרים אותו בשם backend.py.
import base64
from fastapi import FastAPI, Body
from google.genai.types import Content, Part
from google.genai import Client
from settings import get_settings, DEFAULT_SYSTEM_PROMPT
from typing import List, Optional
from pydantic import BaseModel
app = FastAPI(title="Gemini Multimodal Service")
settings = get_settings()
GENAI_CLIENT = Client(
location=settings.VERTEXAI_LOCATION,
project=settings.VERTEXAI_PROJECT_ID,
vertexai=True,
)
GEMINI_MODEL_NAME = "gemini-2.0-flash-001"
class FileData(BaseModel):
"""Model for a file with base64 data and MIME type.
Attributes:
data: Base64 encoded string of the file content.
mime_type: The MIME type of the file.
"""
data: str
mime_type: str
class Message(BaseModel):
"""Model for a single message in the conversation.
Attributes:
role: The role of the message sender, either 'user' or 'assistant'.
content: The text content of the message or a list of file data objects.
"""
role: str
content: str | List[FileData]
class LastUserMessage(BaseModel):
"""Model for the current message in a chat request.
Attributes:
text: The text content of the message.
files: List of file data objects containing base64 data and MIME type.
"""
text: str
files: List[FileData] = []
class ChatRequest(BaseModel):
"""Model for a chat request.
Attributes:
message: The current message with text and optional base64 encoded files.
history: List of previous messages in the conversation.
system_prompt: Optional system prompt to be used in the chat.
"""
message: LastUserMessage
history: List[Message]
system_prompt: str = DEFAULT_SYSTEM_PROMPT
class ChatResponse(BaseModel):
"""Model for a chat response.
Attributes:
response: The text response from the model.
error: Optional error message if something went wrong.
"""
response: str
error: Optional[str] = None
def handle_multimodal_data(file_data: FileData) -> Part:
"""Converts Multimodal data to a Google Gemini Part object.
Args:
file_data: FileData object with base64 data and MIME type.
Returns:
Part: A Google Gemini Part object containing the file data.
"""
data = base64.b64decode(file_data.data) # decode base64 string to bytes
return Part.from_bytes(data=data, mime_type=file_data.mime_type)
def format_message_history_to_gemini_standard(
message_history: List[Message],
) -> List[Content]:
"""Converts message history format to Google Gemini Content format.
Args:
message_history: List of message objects from the chat history.
Each message contains 'role' and 'content' attributes.
Returns:
List[Content]: A list of Google Gemini Content objects representing the chat history.
Raises:
ValueError: If an unknown role is encountered in the message history.
"""
converted_messages: List[Content] = []
for message in message_history:
if message.role == "assistant":
converted_messages.append(
Content(role="model", parts=[Part.from_text(text=message.content)])
)
elif message.role == "user":
# Text-only messages
if isinstance(message.content, str):
converted_messages.append(
Content(role="user", parts=[Part.from_text(text=message.content)])
)
# Messages with files
elif isinstance(message.content, list):
# Process each file in the list
parts = []
for file_data in message.content:
for file_data in message.content:
parts.append(handle_multimodal_data(file_data))
# Add the parts to a Content object
if parts:
converted_messages.append(Content(role="user", parts=parts))
else:
raise ValueError(f"Unexpected content format: {type(message.content)}")
else:
raise ValueError(f"Unknown role: {message.role}")
return converted_messages
@app.post("/chat", response_model=ChatResponse)
async def chat(
request: ChatRequest = Body(...),
) -> ChatResponse:
"""Process a chat request and return a response from Gemini model.
Args:
request: The chat request containing message and history.
Returns:
ChatResponse: The model's response to the chat request.
"""
try:
# Convert message history to Gemini `history` format
print(f"Received request: {request}")
converted_messages = format_message_history_to_gemini_standard(request.history)
# Create chat model
chat_model = GENAI_CLIENT.chats.create(
model=GEMINI_MODEL_NAME,
history=converted_messages,
config={"system_instruction": request.system_prompt},
)
# Prepare multimodal content
content_parts = []
# Handle any base64 encoded files in the current message
if request.message.files:
for file_data in request.message.files:
content_parts.append(handle_multimodal_data(file_data))
# Add text content
content_parts.append(Part.from_text(text=request.message.text))
# Send message to Gemini
response = chat_model.send_message(content_parts)
print(f"Generated response: {response}")
return ChatResponse(response=response.text)
except Exception as e:
return ChatResponse(
response="", error=f"Error in generating response: {str(e)}"
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8081)
אל תשכחו לשמור את הקובץ בתור backend.py. לאחר מכן נוכל לנסות להריץ את שירות הקצה העורפי. חשוב לזכור שבשלב הקודם הפעלנו את שירות הקצה הקדמי. עכשיו נצטרך לפתוח מסוף חדש ולנסות להפעיל את שירות הקצה העורפי.
- יוצרים מסוף חדש. עוברים אל מסוף ה-CLI באזור התחתון ומאתרים את הלחצן '+' כדי ליצור מסוף חדש. לחלופין, אפשר להקיש על 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, מגדירים את המסלול באמצעות ה-decorator 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
אחד הדברים החשובים שצריך להבין הוא איך אפשר לבנות מחדש את היסטוריית הצ'אט כך שניתן יהיה להוסיף אותה כערך של הארגומנט history כשנפעיל את לקוח 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]. לכל תוכן חייבים להיות לפחות ערך תפקיד וחלקים. תפקיד מתייחס למקור ההודעה, בין אם הוא משתמש או מודל. parts מתייחס להנחיה עצמה, והיא יכולה להיות רק טקסט או שילוב של מודלים שונים. במסמכי העזרה מוסבר בפירוט איך לבנות ארגומנטים של תוכן.
טיפול בנתונים שאינם טקסט ( מולטימודיאליים)
כפי שצוין קודם בקטע 'חזית', אחת מהדרכים לשלוח נתונים שאינם טקסט או נתונים מרובי-מודלים היא לשלוח את הנתונים כמחרוזת 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. בדיקת אינטגרציה
עכשיו אמורים לפעול כמה שירותים בכרטיסיות שונות במסוף Cloud:
- שירות הקצה הקדמי פועל ביציאה 8080
* Running on local URL: http://0.0.0.0:8080 To create a public link, set `share=True` in `launch()`.
- שירות לקצה העורפי שפועל ביציאה 8081
INFO: Started server process [xxxxx] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://0.0.0.0:8081 (Press CTRL+C to quit)
נכון לעכשיו, אמורה להיות לך אפשרות לשלוח את המסמכים בצ'אט בצורה חלקה עם Assistant מאפליקציית האינטרנט ביציאה 8080. אתם יכולים להתחיל להתנסות על ידי העלאת קבצים ושליחת שאלות. חשוב לזכור שעדיין אין תמיכה בסוגי קבצים מסוימים, והם יגרמו להצגת הודעת שגיאה.
אפשר גם לערוך את הוראות המערכת מהשדה קלט נוסף שמתחת לתיבת הטקסט.
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 Terminal ומוודאים שהפרויקט הנוכחי מוגדר כפרויקט הפעיל. אם לא, צריך להשתמש בפקודה 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 על המשאבים שבהם השתמשתם בקודלאב הזה, עליכם לפעול לפי השלבים הבאים:
- נכנסים לדף Manage resources במסוף Google Cloud.
- ברשימת הפרויקטים, בוחרים את הפרויקט שרוצים למחוק ולוחצים על Delete.
- כדי למחוק את הפרויקט, כותבים את מזהה הפרויקט בתיבת הדו-שיח ולוחצים על Shut down.
- לחלופין, אפשר לעבור אל Cloud Run במסוף, לבחור את השירות שפרסמתם ולמחוק אותו.