1. Giới thiệu
Trong lớp học lập trình này, bạn sẽ xây dựng một ứng dụng dưới dạng giao diện web trò chuyện, trong đó bạn có thể giao tiếp, tải một số tài liệu hoặc hình ảnh lên và thảo luận về chúng. Bản thân ứng dụng được chia thành 2 dịch vụ: giao diện người dùng và phụ trợ; cho phép bạn tạo một nguyên mẫu nhanh và thử nghiệm cảm giác sử dụng, đồng thời hiểu được hợp đồng API trông như thế nào để tích hợp cả hai dịch vụ này.
Trong lớp học lập trình này, bạn sẽ áp dụng phương pháp từng bước như sau:
- Chuẩn bị dự án Google Cloud và Bật tất cả API bắt buộc trên dự án đó
- Tạo dịch vụ giao diện người dùng – giao diện trò chuyện bằng thư viện Gradio
- Tạo dịch vụ phụ trợ – máy chủ HTTP bằng FastAPI. Công cụ này sẽ định dạng lại dữ liệu đến theo tiêu chuẩn SDK Gemini và cho phép giao tiếp với API Gemini
- Quản lý các biến môi trường và thiết lập các tệp bắt buộc cần thiết để triển khai ứng dụng cho Cloud Run
- Triển khai ứng dụng trên Cloud Run
Tổng quan về cấu trúc
Điều kiện tiên quyết
- Dễ dàng làm việc với Gemini API và SDK AI thế hệ mới của Google
- Hiểu biết về cấu trúc toàn stack cơ bản bằng cách sử dụng dịch vụ HTTP
Kiến thức bạn sẽ học được
- Cách sử dụng SDK Gemini để gửi văn bản và loại dữ liệu khác (đa phương thức) cũng như tạo câu trả lời bằng văn bản
- Cách sắp xếp nhật ký trò chuyện vào SDK Gemini để duy trì ngữ cảnh cuộc trò chuyện
- Tạo bản minh hoạ web giao diện người dùng bằng Gradio
- Phát triển dịch vụ phụ trợ bằng FastAPI và Pydantic
- Quản lý các biến môi trường trong tệp YAML bằng Pydantic-settings
- Triển khai ứng dụng lên Cloud Run bằng Dockerfile và cung cấp các biến môi trường bằng tệp YAML
Bạn cần có
- Trình duyệt web Chrome
- Tài khoản Gmail
- Một dự án trên Google Cloud đã bật tính năng thanh toán
Lớp học lập trình này được thiết kế cho các nhà phát triển ở mọi cấp độ (bao gồm cả người mới bắt đầu), sử dụng Python trong ứng dụng mẫu. Tuy nhiên, bạn không cần phải có kiến thức về Python để hiểu các khái niệm được trình bày.
2. Trước khi bắt đầu
Thiết lập Dự án trên đám mây trong Trình chỉnh sửa Cloud Shell
Lớp học lập trình này giả định rằng bạn đã có một dự án trên Google Cloud đã bật tính năng thanh toán. Nếu chưa có, bạn có thể làm theo hướng dẫn bên dưới để bắt đầu.
- 2Trong Google Cloud Console, trên trang bộ chọn dự án, hãy chọn hoặc tạo một dự án trên Google Cloud.
- Đảm bảo bạn đã bật tính năng thanh toán cho dự án trên Cloud. Tìm hiểu cách kiểm tra xem tính năng thanh toán có được bật trên dự án hay không .
- Bạn sẽ sử dụng Cloud Shell, một môi trường dòng lệnh chạy trong Google Cloud và được tải sẵn bq. Nhấp vào Kích hoạt Cloud Shell ở đầu bảng điều khiển Google Cloud.
- Sau khi kết nối với Cloud Shell, hãy kiểm tra để đảm bảo bạn đã được xác thực và dự án được đặt thành mã dự án của bạn bằng lệnh sau:
gcloud auth list
- Chạy lệnh sau trong Cloud Shell để xác nhận rằng lệnh gcloud biết về dự án của bạn.
gcloud config list project
- Nếu dự án của bạn chưa được đặt, hãy sử dụng lệnh sau để đặt dự án:
gcloud config set project <YOUR_PROJECT_ID>
Ngoài ra, bạn cũng có thể xem mã PROJECT_ID trong bảng điều khiển
Nhấp vào đó, bạn sẽ thấy tất cả dự án và mã dự án ở bên phải
- Bật các API bắt buộc thông qua lệnh hiển thị bên dưới. Quá trình này có thể mất vài phút. Vui lòng kiên nhẫn chờ đợi.
gcloud services enable aiplatform.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
cloudresourcemanager.googleapis.com
Khi thực thi thành công lệnh này, bạn sẽ thấy một thông báo tương tự như thông báo dưới đây:
Operation "operations/..." finished successfully.
Bạn có thể sử dụng bảng điều khiển để tìm kiếm từng sản phẩm hoặc sử dụng đường liên kết này thay cho lệnh gcloud.
Nếu thiếu bất kỳ API nào, bạn luôn có thể bật API đó trong quá trình triển khai.
Tham khảo tài liệu để biết các lệnh và cách sử dụng gcloud.
Thiết lập thư mục hoạt động của ứng dụng
- Nhấp vào nút Open Editor (Mở trình chỉnh sửa), thao tác này sẽ mở Cloud Shell Editor (Trình chỉnh sửa Cloud Shell), chúng ta có thể viết mã của mình tại đây
- Đảm bảo dự án Cloud Code được đặt ở góc dưới bên trái (thanh trạng thái) của trình chỉnh sửa Cloud Shell, như được làm nổi bật trong hình ảnh bên dưới và được đặt thành dự án Google Cloud đang hoạt động mà bạn đã bật tính năng thanh toán. Uỷ quyền nếu được nhắc. Sau khi khởi chạy Trình chỉnh sửa Cloud Shell, có thể mất một lúc thì nút Cloud Code – Sign In (Mã trên đám mây – Đăng nhập) mới xuất hiện. Vui lòng kiên nhẫn chờ đợi. Nếu bạn đã làm theo lệnh trước đó, nút này cũng có thể trỏ trực tiếp đến dự án đã kích hoạt thay vì nút đăng nhập
- Nhấp vào dự án đang hoạt động đó trên thanh trạng thái và đợi cửa sổ bật lên Cloud Code mở ra. Trong cửa sổ bật lên, hãy chọn "New Application" (Ứng dụng mới).
- Trong danh sách ứng dụng, hãy chọn AI tạo sinh Gemini, sau đó chọn Gemini API Python
- Lưu ứng dụng mới với tên mà bạn muốn, trong ví dụ này chúng ta sẽ sử dụng gemini-multimodal-chat-assistant , sau đó nhấp vào OK
Tại thời điểm này, bạn đã ở trong thư mục hoạt động của ứng dụng mới và thấy các tệp sau
Tiếp theo, chúng ta sẽ chuẩn bị môi trường python
Thiết lập môi trường
Chuẩn bị môi trường ảo Python
Bước tiếp theo là chuẩn bị môi trường phát triển. Chúng ta sẽ sử dụng Python 3.12 trong lớp học lập trình này và sẽ sử dụng trình quản lý dự án python uv để đơn giản hoá nhu cầu tạo và quản lý phiên bản python cũng như môi trường ảo
- Nếu bạn chưa mở cửa sổ dòng lệnh, hãy mở bằng cách nhấp vào Terminal (Dòng lệnh) -> New Terminal (Dòng lệnh mới) hoặc sử dụng tổ hợp phím Ctrl + Shift + C
- Tải
uvxuống và cài đặt python 3.12 bằng lệnh sau
curl -LsSf https://astral.sh/uv/0.6.6/install.sh | sh && \
source $HOME/.local/bin/env && \
uv python install 3.12
- Bây giờ, hãy khởi chạy dự án Python bằng
uv
uv init
- Bạn sẽ thấy main.py, .python-version và pyproject.toml được tạo trong thư mục. Bạn cần các tệp này để duy trì dự án trong thư mục. Bạn có thể chỉ định các phần phụ thuộc và cấu hình Python trong pyproject.toml và .python-version đã chuẩn hoá phiên bản Python dùng cho dự án này. Để đọc thêm về vấn đề này, bạn có thể xem tài liệu này
main.py .python-version pyproject.toml
- Để kiểm thử, hãy ghi đè main.py vào mã sau
def main():
print("Hello from gemini-multimodal-chat-assistant!")
if __name__ == "__main__":
main()
- Sau đó, hãy chạy lệnh sau
uv run main.py
Bạn sẽ nhận được kết quả như sau
Using CPython 3.12 Creating virtual environment at: .venv Hello from gemini-multimodal-chat-assistant!
Điều này cho thấy dự án python đang được thiết lập đúng cách. Chúng ta không cần tạo môi trường ảo theo cách thủ công vì uv đã xử lý môi trường đó. Vì vậy, từ thời điểm này, lệnh python chuẩn (ví dụ: python main.py ) sẽ được thay thế bằng uv run (ví dụ: uv run main.py ).
Cài đặt các phần phụ thuộc bắt buộc
Chúng ta cũng sẽ thêm các phần phụ thuộc gói lớp học lập trình này bằng lệnh uv. Chạy lệnh sau
uv add google-genai==1.5.0 \
gradio==5.20.1 \
pydantic==2.10.6 \
pydantic-settings==2.8.1 \
pyyaml==6.0.2
Bạn sẽ thấy phần "dependencies" ("phần phụ thuộc") của pyproject.toml sẽ được cập nhật để phản ánh lệnh trước đó
Thiết lập tệp cấu hình
Bây giờ, chúng ta cần thiết lập các tệp cấu hình cho dự án này. Tệp cấu hình được dùng để lưu trữ các biến động có thể dễ dàng thay đổi khi triển khai lại. Trong dự án này, chúng ta sẽ sử dụng các tệp cấu hình dựa trên YAML với gói pydantic-settings để có thể dễ dàng tích hợp với quá trình triển khai Cloud Run sau này. pydantic-settings là một gói python có thể thực thi việc kiểm tra kiểu cho các tệp cấu hình.
- Tạo một tệp có tên settings.yaml với cấu hình sau. Nhấp vào File->New Text File (Tệp->Tệp văn bản mới) rồi điền mã sau. Sau đó, lưu tệp này dưới dạng settings.yaml
VERTEXAI_LOCATION: "us-central1"
VERTEXAI_PROJECT_ID: "{YOUR-PROJECT-ID}"
BACKEND_URL: "http://localhost:8081/chat"
Vui lòng cập nhật các giá trị cho VERTEXAI_PROJECT_ID theo những gì bạn đã chọn trong khi tạo Dự án Google Cloud. Trong lớp học lập trình này, chúng ta sẽ sử dụng các giá trị được định cấu hình trước cho VERTEXAI_LOCATION và BACKEND_URL .
- Sau đó, hãy tạo tệp python settings.py, mô-đun này sẽ đóng vai trò là mục nhập có lập trình cho các giá trị cấu hình trong tệp cấu hình của chúng ta. Nhấp vào File->New Text File (Tệp->Tệp văn bản mới) rồi điền mã sau. Sau đó, lưu tệp này dưới dạng settings.py. Bạn có thể thấy trong mã rằng chúng ta đã đặt rõ ràng tệp có tên settings.yaml là tệp sẽ được đọc
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()
Các cấu hình này cho phép chúng ta linh hoạt cập nhật thời gian chạy. Trong lần triển khai đầu tiên, chúng ta sẽ dựa vào cấu hình settings.yaml để có cấu hình mặc định đầu tiên. Sau đó, chúng ta có thể linh hoạt cập nhật các biến môi trường thông qua bảng điều khiển và triển khai lại khi đặt các biến môi trường ở mức độ ưu tiên cao hơn so với cấu hình YAML mặc định
Bây giờ, chúng ta có thể chuyển sang bước tiếp theo, xây dựng các dịch vụ
3. Tạo dịch vụ giao diện người dùng bằng Gradio
Chúng ta sẽ xây dựng một giao diện web trò chuyện như sau
Thành phần này chứa một trường nhập để người dùng gửi văn bản và tải tệp lên. Ngoài ra, người dùng cũng có thể ghi đè hướng dẫn hệ thống sẽ được gửi đến API Gemini trong trường dữ liệu đầu vào bổ sung
Chúng ta sẽ xây dựng dịch vụ giao diện người dùng bằng Gradio. Đổi tên main.py thành frontend.py và ghi đè mã bằng mã sau
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,
)
Sau đó, chúng ta có thể thử chạy dịch vụ giao diện người dùng bằng lệnh sau. Đừng quên đổi tên tệp main.py thành frontend.py
uv run frontend.py
Bạn sẽ thấy kết quả tương tự như sau trong bảng điều khiển trên đám mây
* Running on local URL: http://0.0.0.0:8080 To create a public link, set `share=True` in `launch()`.
Sau đó, bạn có thể kiểm tra giao diện web khi nhấn tổ hợp phím ctrl+nhấp vào đường liên kết URL cục bộ. Ngoài ra, bạn cũng có thể truy cập vào ứng dụng giao diện người dùng bằng cách nhấp vào nút Xem trước trên web ở phía trên cùng bên phải của Cloud Editor rồi chọn Xem trước trên cổng 8080
Bạn sẽ thấy giao diện web, tuy nhiên, bạn sẽ gặp lỗi dự kiến khi cố gắng gửi cuộc trò chuyện do dịch vụ phụ trợ chưa được thiết lập
Bây giờ, hãy để dịch vụ chạy và đừng tắt dịch vụ đó. Trong thời gian chờ đợi, chúng ta có thể thảo luận về các thành phần mã quan trọng tại đây
Giải thích mã
Mã để gửi dữ liệu từ giao diện web đến phần phụ trợ nằm ở phần này
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
...
Khi muốn gửi dữ liệu đa phương thức đến Gemini và cho phép truy cập dữ liệu giữa các dịch vụ, chúng ta có thể sử dụng một cơ chế là chuyển đổi dữ liệu thành loại dữ liệu base64 như đã khai báo trong mã. Chúng ta cũng cần khai báo loại MIME của dữ liệu. Tuy nhiên, Gemini API không thể hỗ trợ tất cả các loại MIME hiện có. Do đó, điều quan trọng là bạn phải biết những loại MIME nào được Gemini hỗ trợ mà bạn có thể đọc trong tài liệu này. Bạn có thể tìm thấy thông tin trong từng chức năng của API Gemini (ví dụ: Vision)
Ngoài ra, trong giao diện trò chuyện, bạn cũng cần gửi nhật ký trò chuyện dưới dạng ngữ cảnh bổ sung để cung cấp cho Gemini "bộ nhớ" về cuộc trò chuyện. Vì vậy, trong giao diện web này, chúng ta cũng gửi nhật ký trò chuyện do Gradio quản lý cho mỗi phiên web và gửi nhật ký đó cùng với thông báo do người dùng nhập. Ngoài ra, chúng tôi cũng cho phép người dùng sửa đổi hướng dẫn hệ thống và gửi hướng dẫn đó
4. Tạo dịch vụ phụ trợ bằng FastAPI
Tiếp theo, chúng ta cần xây dựng phần phụ trợ có thể xử lý tải trọng đã thảo luận trước đó, tin nhắn gần đây nhất của người dùng, nhật ký trò chuyện và hướng dẫn hệ thống. Chúng ta sẽ sử dụng FastAPI để tạo dịch vụ phụ trợ HTTP.
Tạo tệp mới, nhấp vào File->New Text File (Tệp->Tệp văn bản mới), rồi sao chép và dán mã sau, sau đó lưu dưới dạng 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)
Đừng quên lưu tệp này dưới dạng backend.py . Sau đó, chúng ta có thể thử chạy dịch vụ phụ trợ. Hãy nhớ rằng ở bước trước, chúng ta đã chạy dịch vụ giao diện người dùng, giờ đây, chúng ta cần mở một thiết bị đầu cuối mới và thử chạy dịch vụ phụ trợ này
- Tạo một thiết bị đầu cuối mới. Chuyển đến thiết bị đầu cuối ở khu vực dưới cùng rồi tìm nút "+" để tạo thiết bị đầu cuối mới. Ngoài ra, bạn có thể nhấn tổ hợp phím Ctrl + Shift + C để mở một thiết bị đầu cuối mới
- Sau đó, hãy đảm bảo rằng bạn đang ở thư mục đang hoạt động gemini-multimodal-chat-assistant rồi chạy lệnh sau
uv run backend.py
- Nếu thành công, bạn sẽ thấy kết quả như sau
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)
Giải thích mã
Xác định tuyến HTTP để nhận yêu cầu trò chuyện
Trong FastAPI, chúng ta xác định tuyến bằng cách sử dụng trình trang trí app. Chúng ta cũng sử dụng Pydantic để xác định hợp đồng API. Chúng ta chỉ định rằng tuyến để tạo phản hồi nằm trong tuyến /chat bằng phương thức POST. Các chức năng này được khai báo trong mã sau
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
...
Chuẩn bị định dạng nhật ký trò chuyện của SDK Gemini
Một trong những điều quan trọng cần hiểu là cách chúng ta có thể tái cấu trúc nhật ký trò chuyện để có thể chèn nhật ký đó dưới dạng giá trị đối số history khi khởi chạy ứng dụng Gemini sau này. Bạn có thể kiểm tra mã dưới đây
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
Để cung cấp nhật ký trò chuyện vào SDK Gemini, chúng ta cần định dạng dữ liệu ở loại dữ liệu List[Content]. Mỗi Content (Nội dung) phải có ít nhất một giá trị role (vai trò) và parts (phần). role (vai trò) đề cập đến nguồn của thông báo, cho dù đó là user (người dùng) hay model (mô hình). Trong đó, parts (phần) đề cập đến chính câu lệnh, có thể chỉ là văn bản hoặc kết hợp nhiều phương thức. Xem cách định cấu trúc đối số Content (Nội dung) trong tài liệu này
Xử lý dữ liệu không phải văn bản ( đa phương thức)
Như đã đề cập trước đó trong phần giao diện người dùng, một trong những cách gửi dữ liệu không phải văn bản hoặc dữ liệu đa phương thức là gửi dữ liệu dưới dạng chuỗi base64. Chúng ta cũng cần chỉ định loại MIME cho dữ liệu để có thể diễn giải chính xác dữ liệu đó, ví dụ: cung cấp loại MIME image/jpeg nếu chúng ta gửi dữ liệu hình ảnh có hậu tố .jpg.
Phần mã này chuyển đổi dữ liệu base64 thành định dạng Part.from_bytes từ 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. Kiểm thử tích hợp
Bây giờ, bạn sẽ có nhiều dịch vụ chạy trong các thẻ khác nhau trên Google Cloud Console:
- Dịch vụ giao diện người dùng chạy ở cổng 8080
* Running on local URL: http://0.0.0.0:8080 To create a public link, set `share=True` in `launch()`.
- Dịch vụ phụ trợ chạy ở cổng 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)
Ở trạng thái hiện tại, bạn có thể gửi tài liệu trong cuộc trò chuyện với trợ lý từ ứng dụng web trên cổng 8080 một cách liền mạch. Bạn có thể bắt đầu thử nghiệm bằng cách tải tệp lên và đặt câu hỏi! Hãy lưu ý rằng một số loại tệp chưa được hỗ trợ và sẽ gây ra Lỗi.
Bạn cũng có thể chỉnh sửa hướng dẫn hệ thống trong trường Additional Inputs (Dữ liệu đầu vào bổ sung) bên dưới hộp văn bản
6. Triển khai lên Cloud Run
Tất nhiên, chúng ta muốn giới thiệu ứng dụng tuyệt vời này cho những người khác. Để làm như vậy, chúng ta có thể đóng gói ứng dụng này và triển khai ứng dụng đó lên Cloud Run dưới dạng một dịch vụ công khai mà người khác có thể truy cập. Để làm được điều đó, hãy xem lại cấu trúc
Trong lớp học lập trình này, chúng ta sẽ đặt cả dịch vụ giao diện người dùng và dịch vụ phụ trợ vào 1 vùng chứa. Chúng ta sẽ cần đến sự trợ giúp của supervisord để quản lý cả hai dịch vụ.
Tạo tệp mới, nhấp vào File->New Text File (Tệp->Tệp văn bản mới), rồi sao chép và dán mã sau, sau đó lưu dưới dạng 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
Tiếp theo, chúng ta sẽ cần Dockerfile, Nhấp vào File->New Text File (Tệp->Tệp văn bản mới), rồi sao chép và dán mã sau, sau đó lưu dưới dạng 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"]
Tại thời điểm này, chúng ta đã có tất cả các tệp cần thiết để triển khai ứng dụng trên Cloud Run. Hãy triển khai ứng dụng. Chuyển đến Cloud Shell Terminal và đảm bảo dự án hiện tại được định cấu hình thành dự án đang hoạt động của bạn, nếu không, bạn phải sử dụng lệnh gcloud configure để đặt mã dự án:
gcloud config set project [PROJECT_ID]
Sau đó, hãy chạy lệnh sau để triển khai ứng dụng đó lên Cloud Run.
gcloud run deploy --source . \
--env-vars-file settings.yaml \
--port 8080 \
--region us-central1
Thao tác này sẽ nhắc bạn nhập tên cho dịch vụ của mình, giả sử là "gemini-multimodal-chat-assistant". Vì chúng ta có Dockerfile trong thư mục hoạt động của ứng dụng, nên tệp này sẽ tạo vùng chứa Docker và đẩy vùng chứa đó vào Cấu phần phần mềm. Thao tác này cũng sẽ nhắc bạn rằng sẽ tạo kho lưu trữ Cấu phần phần mềm trong khu vực, hãy trả lời "Có" cho câu hỏi này. Ngoài ra, hãy nói "y" khi ứng dụng hỏi xem bạn có muốn cho phép các lệnh gọi chưa được xác thực hay không. Xin lưu ý rằng chúng tôi đang cho phép quyền truy cập chưa xác thực tại đây vì đây là ứng dụng minh hoạ. Bạn nên sử dụng phương thức xác thực phù hợp cho các ứng dụng doanh nghiệp và ứng dụng phát hành chính thức.
Sau khi triển khai xong, bạn sẽ nhận được một đường liên kết tương tự như bên dưới:
https://gemini-multimodal-chat-assistant-*******.us-central1.run.app
Hãy tiếp tục sử dụng ứng dụng của bạn trong cửa sổ Ẩn danh hoặc trên thiết bị di động. Trang web đã được xuất bản.
7. Thách thức
Giờ là lúc bạn tỏa sáng và trau dồi kỹ năng khám phá. Bạn có đủ điều kiện để thay đổi mã để trợ lý có thể hỗ trợ đọc tệp âm thanh hoặc có thể là tệp video không?
8. Dọn dẹp
Để tránh bị tính phí cho tài khoản Google Cloud của bạn đối với các tài nguyên dùng trong lớp học lập trình này, hãy làm theo các bước sau:
- Trong Google Cloud Console, hãy chuyển đến trang Quản lý tài nguyên.
- Trong danh sách dự án, hãy chọn dự án mà bạn muốn xoá, rồi nhấp vào Xoá.
- Trong hộp thoại, hãy nhập mã dự án, sau đó nhấp vào Shut down (Tắt) để xoá dự án.
- Ngoài ra, bạn có thể chuyển đến Cloud Run trên bảng điều khiển, chọn dịch vụ bạn vừa triển khai rồi xoá.