إنشاء "مساعِد متعدد الوسائط" ونشره على السحابة الإلكترونية باستخدام Gemini (Python)

لمحة عن هذا الدرس التطبيقي حول الترميز

0 دقيقة

تاريخ التعديل الأخير: 29 مارس 2025

تأليف: Alvin Prayuda Juniarta Dwiyantoro

1. مقدمة

في هذا الدليل التعليمي حول رموز البرامج، ستُنشئ تطبيقًا في شكل واجهة ويب للمحادثة، يمكنك من خلالها التواصل مع التطبيق وتحميل بعض المستندات أو الصور ومناقشتها. تم تقسيم التطبيق نفسه إلى خدمتَين: الواجهة الأمامية والخلفية، ما يتيح لك إنشاء نموذج أولي سريع وتجربته، وفهم كيفية دمج واجهة برمجة التطبيقات لكلتا الخدمتَين.

من خلال ورشة رموز البرامج، ستطبّق نهجًا خطوة بخطوة على النحو التالي:

1. تجهيز مشروعك على Google Cloud وتفعيل جميع واجهات برمجة التطبيقات المطلوبة فيه
2. إنشاء خدمة الواجهة الأمامية - واجهة المحادثة باستخدام مكتبة Gradio
3. أنشئ خدمة الخلفية، وهي خادم HTTP باستخدام FastAPI الذي سيعيد تنسيق البيانات الواردة وفقًا لمعيار حزمة تطوير البرامج Gemini SDK وسيتيح التواصل مع واجهة برمجة التطبيقات Gemini API.
4. إدارة متغيّرات البيئة وإعداد الملفات المطلوبة لنشر التطبيق على Cloud Run
5. نشر التطبيق على Cloud Run

نظرة عامة على البنية

المتطلبات الأساسية

• الشعور بالارتياح عند العمل مع Gemini API وGoogle Gen AI SDK
• فهم أساسي لبنية الحزمة الكاملة باستخدام خدمة HTTP

ما ستتعرّف عليه

• كيفية استخدام حزمة تطوير البرامج (SDK) في Gemini لإرسال نص وأنواع بيانات أخرى (متعددة الوسائط) وإنشاء ردّ نصي
• كيفية تنظيم سجلّ المحادثات في حزمة Gemini SDK للحفاظ على سياق المحادثة
• إنشاء نماذج أولية لواجهة الويب باستخدام Gradio
• تطوير خدمة الخلفية باستخدام FastAPI وPydantic
• إدارة متغيّرات البيئة في ملف YAML باستخدام Pydantic-settings
• نشر التطبيق على Cloud Run باستخدام Dockerfile وتوفير متغيرات البيئة باستخدام ملف YAML

المتطلبات

• متصفّح الويب Chrome
• حساب Gmail
• مشروع على Cloud تم تفعيل الفوترة فيه

تم تصميم هذا المختبر البرمجي للمطوّرين من جميع المستويات (بما في ذلك المبتدئين)، ويستخدم لغة بايثون في تطبيقه النموذجي. ومع ذلك، لا يُشترط معرفة Python لفهم المفاهيم المعروضة.

2. قبل البدء

إعداد مشروع Cloud في محرِّر Cloud Shell

يفترض هذا الدليل التعليمي للترميز أنّ لديك مشروعًا على Google Cloud مفعّل فيه نظام الفوترة. إذا لم يكن لديك التطبيق بعد، يمكنك اتّباع التعليمات أدناه للبدء.

1. 2في Google Cloud Console، في صفحة أداة اختيار المشاريع، اختَر مشروعًا على 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. فعِّل واجهات برمجة التطبيقات المطلوبة من خلال الأمر الموضَّح أدناه. قد تستغرق هذه العملية بضع دقائق، لذا يُرجى الانتظار.

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

عند تنفيذ الأمر بنجاح، من المفترض أن تظهر لك رسالة مشابهة للرسالة الموضّحة أدناه:

Operation "operations/..." finished successfully.

يمكنك استخدام وحدة التحكّم للبحث عن كل منتج أو استخدام هذا الرابط كبديل لأمر gcloud.

إذا فاتتك أي واجهة برمجة تطبيقات، يمكنك تفعيلها في أي وقت أثناء عملية التنفيذ.

راجِع المستندات لمعرفة أوامر gcloud وكيفية استخدامها.

إعداد دليل العمل للتطبيق

1. انقر على الزر Open Editor (فتح المحرِّر)، سيؤدي ذلك إلى فتح محرِّر Cloud Shell، ويمكننا كتابة الرمز البرمجي هنا.
2. تأكَّد من ضبط مشروع Cloud Code في أسفل يمين (شريط الحالة) محرِّر Cloud Shell، كما هو موضّح في الصورة أدناه، وأنّه تم ضبطه على مشروع Google Cloud النشط الذي فعّلت فيه الفوترة. انقر على تفويض إذا طُلب منك ذلك. قد يستغرق الأمر بعض الوقت بعد بدء "محرر Cloud Shell" حتى يظهر الزر Cloud Code - تسجيل الدخول، لذا يُرجى الانتظار. إذا اتّبعت الأمر السابق، قد يشير الزرّ أيضًا مباشرةً إلى مشروعك المفعّل بدلاً من زر تسجيل الدخول.

3. انقر على هذا المشروع النشط في شريط الحالة وانتظِر فتح نافذة Cloud Code المنبثقة. في النافذة المنبثقة، اختَر "طلب جديد".

4. من قائمة التطبيقات، اختَر الذكاء الاصطناعي التوليدي في Gemini، ثم اختَر Gemini API Python.

5. احفظ التطبيق الجديد بالاسم الذي تريده، وفي هذا المثال سنستخدم gemini-multimodal-chat-assistant، ثم انقر على حسنًا.

في هذه المرحلة، من المفترض أن تكون داخل دليل العمل الجديد للتطبيق وأن تظهر لك الملفات التالية.

بعد ذلك، سنعدّ بيئة Python.

إعداد البيئة

إعداد بيئة Python الافتراضية

الخطوة التالية هي إعداد بيئة التطوير. سنستخدم الإصدار 3.12 من لغة بايثون في هذا البرنامج التعليمي، وسنستخدم uv python project manager لتبسيط الحاجة إلى إنشاء إصدار بايثون وبيئة افتراضية وإدارتهما.

1. إذا لم يسبق لك فتح المحطة الطرفية، افتحها بالنقر على المحطة الطرفية -> محطة طرفية جديدة، أو استخدِم 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).

تثبيت التبعيات المطلوبة

سنضيف أيضًا تبعيات حزمة ورشة العمل هذه باستخدام الأمر 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 باستخدام الإعدادات التالية. انقر على ملف->ملف نصي جديد واملأ الملف بالرمز التالي. بعد ذلك، احفظ الملف باسم 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 .

2. بعد ذلك، أنشئ ملف 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

ستظهر لك نتيجة مشابهة لما يلي في وحدة تحكّم السحابة الإلكترونية.

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

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

بعد ذلك، يمكنك التحقّق من واجهة الويب عند الضغط مع الاستمرار على Ctrl والنقر على رابط عنوان URL المحلي. بدلاً من ذلك، يمكنك أيضًا الوصول إلى تطبيق الواجهة الأمامية من خلال النقر على زر معاينة الويب في أعلى يسار "محرر السحابة الإلكترونية"، واختيار معاينة على المنفذ 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.

أنشئ ملفًا جديدًا، وانقر على ملف (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. بعد ذلك، يمكننا محاولة تشغيل خدمة الخلفية. تذكَّر أنّنا في الخطوة السابقة شغّلنا خدمة الواجهة الأمامية، والآن سنحتاج إلى فتح وحدة تحكّم جديدة ومحاولة تشغيل خدمة الخلفية هذه.

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 لتلقّي طلب Chat

في FastAPI، نحدّد المسار باستخدام العنصر الزخرفي app. نستخدم أيضًا Pydantic لتحديد اتّفاقية واجهة برمجة التطبيقات. نحدّد أنّ مسار إنشاء الردّ هو /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
    ...

إعداد تنسيق سجلّ المحادثات في حزمة تطوير البرامج (SDK) لتطبيق Gemini

من الأمور المهمة التي يجب فهمها هي كيفية إعادة هيكلة سجلّ المحادثات حتى يمكن إدراجه كقيمة وسيطة 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]. يجب أن يحتوي كل محتوى على قيمة دور وأجزاء على الأقل. يشير الدور إلى مصدر الرسالة سواء كان مستخدمًا أو نموذجًا. حيث تشير الأجزاء إلى الطلب نفسه، والذي يمكن أن يكون نصًا فقط أو مجموعة من الوسائط المختلفة. اطّلِع على كيفية تنظيم مَعلمات المحتوى بالتفصيل في هذه المستندات.

معالجة البيانات غير النصية ( المتعددة الوسائط)

كما ذكرنا سابقًا في قسم الواجهة الأمامية، تتمثل إحدى طرق إرسال البيانات غير النصية أو المتعددة الوسائط في إرسال البيانات كسلسلة base64. علينا أيضًا تحديد نوع MIME للبيانات حتى يمكن تفسيرها بشكل صحيح، على سبيل المثال، تقديم نوع MIME image/jpeg إذا أرسلنا بيانات الصورة مع اللاحقة ‎.jpg.

يحوّل هذا الجزء من الرمز البرمجي بيانات base64 إلى تنسيق Part.from_bytes من حزمة تطوير البرامج (SDK) لبرنامج Gemini.

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- اختبار الدمج

من المفترض أن يكون لديك الآن خدمات متعددة يتم تشغيلها في علامات تبويب مختلفة في وحدة تحكّم السحابة الإلكترونية:

• تشغيل خدمة الواجهة الأمامية على المنفذ 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. يمكنك بدء التجربة من خلال تحميل الملفات وطرح الأسئلة. يُرجى العِلم أنّ أنواع ملفات معيّنة غير متوافقة بعد وستؤدي إلى ظهور خطأ.

يمكنك أيضًا تعديل تعليمات النظام من حقل مدخلات إضافية أسفل مربّع النص.

6- النشر على Cloud Run

الآن، نريد بالتأكيد عرض هذا التطبيق الرائع للآخرين. لإجراء ذلك، يمكننا تجميع هذا التطبيق ونشره على Cloud Run كخدمة عامة يمكن للآخرين الوصول إليها. ولنراجع البنية من أجل إجراء ذلك.

في هذا الدليل التعليمي حول الرموز البرمجية، سنضع كلّ من خدمة الواجهة الأمامية وخدمة الخلفية في حاوية واحدة. سنحتاج إلى مساعدة supervisord لإدارة كلتا الخدمتَين.

أنشئ ملفًا جديدًا، وانقر على ملف->ملف نصي جديد، وانسخ الرمز البرمجي التالي والصقه، ثم احفظه باسم supervisord.conf.

[supervisord]
nodaemon=true
user=root
logfile=/dev/stdout
logfile_maxbytes=0
pidfile=/var/run/supervisord.pid

[program:backend]
command=uv run backend.py
directory=/app
autostart=true
autorestart=true
stdout_logfile=/dev/stdout
stdout_logfile_maxbytes=0
stderr_logfile=/dev/stderr
stderr_logfile_maxbytes=0
startsecs=10
startretries=3

[program:frontend]
command=uv run frontend.py
directory=/app
autostart=true
autorestart=true
stdout_logfile=/dev/stdout
stdout_logfile_maxbytes=0
stderr_logfile=/dev/stderr
stderr_logfile_maxbytes=0
startsecs=10
startretries=3

بعد ذلك، سنحتاج إلى Dockerfile. انقر على ملف->ملف نصي جديد، وانسخ الصيغة التالية والصقها، ثم احفظها باسم Dockerfile.

FROM python:3.12-slim
COPY --from=ghcr.io/astral-sh/uv:0.6.6 /uv /uvx /bin/

RUN apt-get update && apt-get install -y \
    supervisor curl \
    && rm -rf /var/lib/apt/lists/*

ADD . /app
WORKDIR /app

RUN uv sync --frozen

EXPOSE 8080

# Copy supervisord configuration
COPY supervisord.conf /etc/supervisor/conf.d/supervisord.conf

ENV PYTHONUNBUFFERED=1

ENTRYPOINT ["/usr/bin/supervisord", "-c", "/etc/supervisor/conf.d/supervisord.conf"]

في هذه المرحلة، لدينا جميع الملفات اللازمة لنشر تطبيقاتنا على Cloud Run، لننشرها. انتقِل إلى Cloud Shell 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 في المنطقة، عليك الإجابة "نعم" على ذلك. قُل أيضًا "y" عندما يسألك عما إذا كنت تريد السماح بطلبات التفعيل غير المعتمَدة. يُرجى العِلم أنّنا نسمح بالوصول غير المُعتمَد هنا لأنّ هذا تطبيق تجريبي. ننصحك باستخدام المصادقة المناسبة لتطبيقات المؤسسات وتطبيقات الإنتاج.

بعد اكتمال عملية النشر، من المفترض أن يصلك رابط مشابه لما يلي:

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

يمكنك استخدام تطبيقك من نافذة التصفّح المتخفي أو من جهازك الجوّال. من المفترض أن يكون التطبيق متوفّرًا.

7- التحدي

حان الوقت الآن لإبراز مهاراتك في الاستكشاف وتحسينها. هل لديك ما يلزم لتغيير الرمز حتى يتمكّن المساعد من قراءة ملفات الصوت أو ربما ملفات الفيديو؟

8. تَنظيم

لتجنُّب تحصيل رسوم من حسابك على Google Cloud مقابل الموارد المستخدَمة في هذا الدليل التعليمي، اتّبِع الخطوات التالية:

1. في وحدة تحكّم Google Cloud، انتقِل إلى صفحة إدارة الموارد.
2. في قائمة المشاريع، اختَر المشروع الذي تريد حذفه، ثم انقر على حذف.
3. في مربّع الحوار، اكتب رقم تعريف المشروع، ثم انقر على إيقاف لحذف المشروع.
4. بدلاً من ذلك، يمكنك الانتقال إلى Cloud Run في وحدة التحكّم، واختيار الخدمة التي تم نشرها للتو وحذفها.