1. 소개
이 Codelab에서는 채팅 웹 인터페이스 형태의 애플리케이션을 빌드합니다. 이 인터페이스에서 애플리케이션과 소통하고, 문서나 이미지를 업로드하고, 이에 관해 논의할 수 있습니다. 애플리케이션 자체는 프런트엔드와 백엔드라는 두 서비스로 분리되어 있으므로 빠른 프로토타입을 빌드하고 어떤 느낌인지 시도해 볼 수 있으며, API 계약이 어떻게 보이는지 이해하여 두 서비스를 통합할 수도 있습니다.
Codelab을 통해 다음과 같이 단계별 접근 방식을 사용합니다.
- Google Cloud 프로젝트를 준비하고 필요한 API를 모두 사용 설정합니다.
- Gradio 라이브러리를 사용하여 프런트엔드 서비스(채팅 인터페이스) 빌드
- FastAPI를 사용하여 백엔드 서비스(HTTP 서버)를 빌드합니다. 이 서비스는 수신 데이터를 Gemini SDK 표준으로 다시 포맷하고 Gemini API와의 통신을 지원합니다.
- 환경 변수를 관리하고 애플리케이션을 Cloud Run에 배포하는 데 필요한 파일을 설정합니다.
- Cloud Run에 애플리케이션 배포
아키텍처 개요
기본 요건
- Gemini API 및 Google 생성형 AI SDK를 사용하는 데 능숙함
- HTTP 서비스를 사용한 기본 풀 스택 아키텍처에 대한 이해
학습할 내용
- Gemini SDK를 사용하여 텍스트 및 기타 데이터 유형 (멀티모달)을 제출하고 텍스트 응답을 생성하는 방법
- 대화 컨텍스트를 유지하기 위해 채팅 기록을 Gemini SDK로 구조화하는 방법
- Gradio를 사용한 프런트엔드 웹 프로토타입 제작
- FastAPI 및 Pydantic을 사용한 백엔드 서비스 개발
- Pydantic-settings로 YAML 파일의 환경 변수 관리
- Dockerfile을 사용하여 Cloud Run에 애플리케이션을 배포하고 YAML 파일로 환경 변수 제공
필요한 항목
- Chrome 웹브라우저
- Gmail 계정
- 결제가 사용 설정된 Cloud 프로젝트
이 Codelab은 초보자를 포함한 모든 수준의 개발자를 대상으로 하며 샘플 애플리케이션에서 Python을 사용합니다. 하지만 제시된 개념을 이해하는 데 Python 지식이 필요하지는 않습니다.
2. 시작하기 전에
Cloud Shell 편집기에서 클라우드 프로젝트 설정하기
이 Codelab에서는 결제가 사용 설정된 Google Cloud 프로젝트가 이미 있다고 가정합니다. 아직 설치하지 않았다면 아래 안내에 따라 시작하세요.
- 2Google Cloud 콘솔의 프로젝트 선택기 페이지에서 Google Cloud 프로젝트를 선택하거나 만듭니다.
- Cloud 프로젝트에 결제가 사용 설정되어 있는지 확인합니다. 프로젝트에 결제가 사용 설정되어 있는지 확인하는 방법을 알아보세요 .
- bq가 미리 로드되어 제공되는 Google Cloud에서 실행되는 명령줄 환경인 Cloud Shell을 사용합니다. Google Cloud 콘솔 상단에서 Cloud Shell 활성화를 클릭합니다.
- Cloud Shell에 연결되면 다음 명령어를 사용하여 이미 인증되었고 프로젝트가 프로젝트 ID로 설정되었는지 확인합니다.
gcloud auth list
- Cloud Shell에서 다음 명령어를 실행하여 gcloud 명령어가 프로젝트를 알고 있는지 확인합니다.
gcloud config list project
- 프로젝트가 설정되지 않은 경우 다음 명령어를 사용하여 설정합니다.
gcloud config set project <YOUR_PROJECT_ID>
또는 콘솔에서 PROJECT_ID ID를 확인할 수도 있습니다.
클릭하면 오른쪽에 모든 프로젝트와 프로젝트 ID가 표시됩니다.
- 아래에 표시된 명령어를 통해 필수 API를 사용 설정합니다. 몇 분 정도 걸릴 수 있으니 잠시 기다려 주세요.
gcloud services enable aiplatform.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
cloudresourcemanager.googleapis.com
명령어가 성공적으로 실행되면 아래와 유사한 메시지가 표시됩니다.
Operation "operations/..." finished successfully.
gcloud 명령의 대안은 콘솔을 통해 각 제품을 검색하거나 이 링크를 사용하는 것입니다.
API가 누락된 경우 구현 과정에서 언제든지 사용 설정할 수 있습니다.
gcloud 명령어 및 사용법은 문서를 참조하세요.
애플리케이션 작업 디렉터리 설정
- 편집기 열기 버튼을 클릭하면 Cloud Shell 편집기가 열립니다. 여기에 코드를 작성할 수 있습니다.
- 아래 이미지에 강조 표시된 것처럼 Cloud Shell 편집기의 왼쪽 하단 (상태 표시줄)에 Cloud Code 프로젝트가 설정되어 있고 결제가 사용 설정된 활성 Google Cloud 프로젝트로 설정되어 있는지 확인합니다. 메시지가 표시되면 승인을 클릭합니다. Cloud Shell 편집기를 초기화한 후 Cloud Code - 로그인 버튼이 표시되기까지 시간이 걸릴 수 있으니 잠시 기다려 주세요. 이전 명령어를 이미 따른 경우 버튼이 로그인 버튼 대신 활성화된 프로젝트를 직접 가리킬 수도 있습니다.
- 상태 표시줄에서 활성 프로젝트를 클릭하고 Cloud Code 팝업이 열릴 때까지 기다립니다. 팝업에서 '새 애플리케이션'을 선택합니다.
- 애플리케이션 목록에서 Gemini 생성형 AI를 선택한 다음 Gemini API Python을 선택합니다.
- 원하는 이름으로 새 애플리케이션을 저장합니다. 이 예에서는 gemini-multimodal-chat-assistant를 사용합니다. 그런 다음 OK를 클릭합니다.
이 시점에서 새 애플리케이션 작업 디렉터리 내에 있어야 하며 다음 파일이 표시됩니다.
다음으로 Python 환경을 준비합니다.
환경 설정
Python 가상 환경 준비
다음 단계는 개발 환경을 준비하는 것입니다. 이 Codelab에서는 Python 3.12를 사용하고 uv python 프로젝트 관리자를 사용하여 Python 버전과 가상 환경을 만들고 관리할 필요성을 간소화합니다.
- 터미널을 아직 열지 않은 경우 터미널 -> 새 터미널을 클릭하여 열거나 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
- 이제
uv를 사용하여 Python 프로젝트를 초기화해 보겠습니다.
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)로 대체됩니다.
필수 종속 항목 설치
uv 명령어를 사용하여 이 Codelab 패키지 종속 항목도 추가합니다. 다음 명령어를 실행합니다.
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 'dependencies' 섹션이 이전 명령어를 반영하도록 업데이트됩니다.
설정 구성 파일
이제 이 프로젝트의 구성 파일을 설정해야 합니다. 구성 파일은 재배포 시 쉽게 변경할 수 있는 동적 변수를 저장하는 데 사용됩니다. 이 프로젝트에서는 나중에 Cloud Run 배포와 쉽게 통합할 수 있도록 pydantic-settings 패키지와 함께 YAML 기반 구성 파일을 사용합니다. pydantic-settings는 구성 파일의 유형 검사를 강제할 수 있는 Python 패키지입니다.
- 다음 구성으로 settings.yaml이라는 파일을 만듭니다. 파일->새 텍스트 파일을 클릭하고 다음 코드로 채웁니다. 그런 다음 settings.yaml로 저장합니다.
VERTEXAI_LOCATION: "us-central1"
VERTEXAI_PROJECT_ID: "{YOUR-PROJECT-ID}"
BACKEND_URL: "http://localhost:8081/chat"
Google Cloud 프로젝트를 만들 때 선택한 내용에 따라 VERTEXAI_PROJECT_ID 값을 업데이트하세요. 이 Codelab에서는 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
클라우드 콘솔에 다음과 비슷한 출력이 표시됩니다.
* Running on local URL: http://0.0.0.0:8080 To create a public link, set `share=True` in `launch()`.
그런 다음 로컬 URL 링크를 Ctrl+클릭하면 웹 인터페이스를 확인할 수 있습니다. 또는 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 유형을 모두 지원할 수 없으므로 이 문서에서 Gemini가 지원하는 MIME 유형을 읽을 수 있는지 확인하는 것이 중요합니다 . 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로 저장하는 것을 잊지 마세요. 그런 다음 백엔드 서비스를 실행해 볼 수 있습니다. 이전 단계에서 프런트엔드 서비스를 올바르게 실행했으므로 이제 새 터미널을 열고 이 백엔드 서비스를 실행해야 합니다.
- 새 터미널을 만듭니다. 하단 영역의 터미널로 이동하여 '+' 버튼을 찾아 새 터미널을 만듭니다. 또는 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에서는 app 데코레이터를 사용하여 경로를 정의합니다. 또한 Pydantic을 사용하여 API 계약을 정의합니다. 대답을 생성하는 경로는 POST 메서드가 있는 /chat 경로에 있다고 지정합니다. 다음 코드에 선언된 기능
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 Chat History Format 준비
이해해야 할 중요한 사항 중 하나는 나중에 Gemini 클라이언트를 초기화할 때 history 인수 값으로 삽입할 수 있도록 채팅 기록을 재구성하는 방법입니다. 아래 코드를 검사할 수 있습니다.
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 유형도 지정해야 합니다. 예를 들어 .jpg 접미사가 있는 이미지 데이터를 보내는 경우 image/jpeg MIME 유형을 제공합니다.
이 코드 부분은 base64 데이터를 Gemini SDK의 Part.from_bytes 형식으로 변환합니다.
def handle_multimodal_data(file_data: FileData) -> Part:
"""Converts Multimodal data to a Google Gemini Part object.
Args:
file_data: FileData object with base64 data and MIME type.
Returns:
Part: A Google Gemini Part object containing the file data.
"""
data = base64.b64decode(file_data.data) # decode base64 string to bytes
return Part.from_bytes(data=data, mime_type=file_data.mime_type)
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)
현재 상태에서는 포트 8080의 웹 애플리케이션에서 어시스턴트와 채팅할 때 문서를 원활하게 보낼 수 있습니다. 파일을 업로드하고 질문을 던져 실험을 시작해 보세요. 일부 파일 형식은 아직 지원되지 않으며 오류가 발생합니다.
텍스트 상자 아래의 추가 입력 필드에서 시스템 안내를 수정할 수도 있습니다.
6. Cloud Run에 배포
이제 이 멋진 앱을 다른 사람들에게도 보여주고 싶습니다. 이렇게 하려면 이 애플리케이션을 패키징하고 다른 사용자가 액세스할 수 있는 공개 서비스로 Cloud Run에 배포하면 됩니다. 이를 위해 아키텍처를 다시 살펴보겠습니다.
이 Codelab에서는 프런트엔드와 백엔드 서비스를 모두 하나의 컨테이너에 넣습니다. 두 서비스를 모두 관리하려면 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 터미널로 이동하여 현재 프로젝트가 활성 프로젝트로 구성되어 있는지 확인합니다. 그렇지 않은 경우 gcloud configure 명령어를 사용하여 프로젝트 ID를 설정해야 합니다.
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. 삭제
이 Codelab에서 사용한 리소스의 비용이 Google Cloud 계정에 청구되지 않도록 하려면 다음 단계를 따르세요.