1. Pengantar
Pernahkah Anda merasa frustrasi dan terlalu malas untuk mengelola semua pengeluaran pribadi? Saya juga! Oleh karena itu, dalam codelab ini, kita akan membuat asisten pengelola pengeluaran pribadi - yang didukung oleh Gemini 2.5 untuk melakukan semua tugas untuk kita. Mulai dari mengelola tanda terima yang diupload hingga menganalisis apakah Anda sudah menghabiskan terlalu banyak uang untuk membeli kopi.
Asisten ini akan dapat diakses melalui browser web dalam bentuk antarmuka web chat, tempat Anda dapat berkomunikasi dengannya, mengupload beberapa gambar tanda terima, dan meminta asisten untuk menyimpannya, atau mungkin ingin menelusuri beberapa tanda terima untuk mendapatkan file dan melakukan beberapa analisis pengeluaran. Semua ini dibuat berdasarkan framework Google Agent Development Kit
Aplikasi itu sendiri dibagi menjadi 2 layanan: frontend dan backend; yang memungkinkan Anda membuat prototipe cepat dan mencoba rasanya, serta memahami tampilan kontrak API untuk mengintegrasikan keduanya.
Melalui codelab ini, Anda akan menggunakan pendekatan langkah demi langkah sebagai berikut:
- Siapkan project Google Cloud Anda dan Aktifkan semua API yang diperlukan di dalamnya
- Menyiapkan bucket di Google Cloud Storage dan database di Firestore
- Membuat Pengindeksan Firestore
- Menyiapkan ruang kerja untuk lingkungan coding Anda
- Menyusun kode sumber, alat, perintah, dll. agen ADK
- Menguji agen menggunakan UI Pengembangan Web lokal ADK
- Buat antarmuka chat layanan frontend menggunakan library Gradio, untuk mengirim beberapa kueri dan mengupload gambar tanda terima
- Buat layanan backend - server HTTP menggunakan FastAPI yang menjadi tempat kode agen ADK, SessionService, dan Layanan Artefak berada
- Mengelola variabel lingkungan dan menyiapkan file yang diperlukan untuk men-deploy aplikasi ke Cloud Run
- Men-deploy aplikasi ke Cloud Run
Ringkasan Arsitektur
Prasyarat
- Nyaman menggunakan Python
- Pemahaman tentang arsitektur full-stack dasar menggunakan layanan HTTP
Yang akan Anda pelajari
- Membuat prototipe web frontend dengan Gradio
- Pengembangan layanan backend dengan FastAPI dan Pydantic
- Mendesain Agen ADK sambil memanfaatkan beberapa kemampuannya
- Penggunaan alat
- Pengelolaan Sesi dan Artefak
- Penggunaan callback untuk modifikasi input sebelum dikirim ke Gemini
- Menggunakan BuiltInPlanner untuk meningkatkan eksekusi tugas dengan melakukan perencanaan
- Proses debug cepat melalui antarmuka web lokal ADK
- Strategi untuk mengoptimalkan interaksi multimodal melalui penguraian dan pengambilan informasi melalui rekayasa perintah dan modifikasi permintaan Gemini menggunakan callback ADK
- Retrieval-Augmented Generation Agen menggunakan Firestore sebagai Database Vektor
- Mengelola variabel lingkungan dalam file YAML dengan Pydantic-settings
- Men-deploy aplikasi ke Cloud Run menggunakan Dockerfile dan memberikan variabel lingkungan dengan file YAML
Yang Anda butuhkan
- Browser web Chrome
- Akun Gmail
- Project Cloud dengan penagihan diaktifkan
Codelab ini, yang dirancang untuk developer dari semua level (termasuk pemula), menggunakan Python dalam aplikasi contohnya. Namun, pengetahuan Python tidak diperlukan untuk memahami konsep yang disajikan.
2. Sebelum memulai
Memilih Project Aktif di Konsol Cloud
Codelab ini mengasumsikan bahwa Anda sudah memiliki project Google Cloud dengan penagihan yang diaktifkan. Jika belum memilikinya, Anda dapat mengikuti petunjuk di bawah untuk memulai.
- Di Konsol Google Cloud, di halaman pemilih project, pilih atau buat project Google Cloud.
- Pastikan penagihan diaktifkan untuk project Cloud Anda. Pelajari cara memeriksa apakah penagihan telah diaktifkan pada suatu project.
Menyiapkan Database Firestore
Selanjutnya, kita juga perlu membuat Database Firestore. Firestore dalam mode Native adalah database dokumen NoSQL yang dibuat untuk penskalaan otomatis, performa tinggi, dan kemudahan pengembangan aplikasi. Database ini juga dapat berfungsi sebagai database vektor yang dapat mendukung teknik Retrieval Augmented Generation untuk lab kita.
- Telusuri "firestore" di kotak penelusuran, lalu klik produk Firestore
- Kemudian, klik tombol Create A Firestore Database
- Gunakan (default) sebagai nama ID database dan biarkan Standard Edition dipilih. Untuk demo lab ini, gunakan Firestore Native dengan aturan keamanan Terbuka.
- Anda juga akan melihat bahwa database ini sebenarnya memiliki Penggunaan Paket gratis YEAY! Setelah itu, klik Tombol Create Database
Setelah langkah-langkah ini, Anda akan dialihkan ke Database Firestore yang baru saja Anda buat
Menyiapkan Project Cloud di Terminal Cloud Shell
- Anda akan menggunakan Cloud Shell, lingkungan command line yang berjalan di Google Cloud yang telah dilengkapi dengan bq. Klik Aktifkan Cloud Shell di bagian atas konsol Google Cloud.
- Setelah terhubung ke Cloud Shell, Anda akan memeriksa apakah Anda sudah diautentikasi dan project ditetapkan ke project ID Anda menggunakan perintah berikut:
gcloud auth list
- Jalankan perintah berikut di Cloud Shell untuk mengonfirmasi bahwa perintah gcloud mengetahui project Anda.
gcloud config list project
- Jika project Anda belum ditetapkan, gunakan perintah berikut untuk menetapkannya:
gcloud config set project <YOUR_PROJECT_ID>
Atau, Anda juga dapat melihat ID PROJECT_ID di konsol
Klik project ID dan Anda akan melihat semua project dan project ID di sisi kanan
- Aktifkan API yang diperlukan melalui perintah yang ditampilkan di bawah. Tindakan ini mungkin memerlukan waktu beberapa menit, jadi harap bersabar.
gcloud services enable aiplatform.googleapis.com \
firestore.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
cloudresourcemanager.googleapis.com
Setelah perintah berhasil dieksekusi, Anda akan melihat pesan yang mirip dengan yang ditampilkan di bawah ini:
Operation "operations/..." finished successfully.
Alternatif untuk perintah gcloud adalah melalui konsol dengan menelusuri setiap produk atau menggunakan link ini.
Jika ada API yang terlewat, Anda dapat mengaktifkannya kapan saja selama proses penerapan.
Baca dokumentasi untuk mempelajari perintah gcloud dan penggunaannya.
Menyiapkan Bucket Google Cloud Storage
Selanjutnya, dari terminal yang sama, kita harus menyiapkan bucket GCS untuk menyimpan file yang diupload. Jalankan perintah berikut untuk membuat bucket
gsutil mb -l us-central1 gs://personal-expense-assistant-receipts
Tindakan ini akan menampilkan output ini
Creating gs://personal-expense-assistant-receipts/...
Anda dapat memverifikasinya dengan membuka Menu Navigasi di kiri atas browser dan memilih Cloud Storage -> Bucket
Membuat Indeks Firestore untuk Penelusuran
Firestore adalah database NoSQL secara native, yang menawarkan performa dan fleksibilitas yang unggul dalam model data, tetapi memiliki batasan terkait kueri yang kompleks. Karena kita berencana untuk menggunakan beberapa kueri multi-kolom gabungan dan penelusuran vektor, kita harus membuat beberapa indeks terlebih dahulu. Anda dapat membaca detail selengkapnya di dokumentasi ini
- Jalankan perintah berikut untuk membuat indeks guna mendukung kueri gabungan
gcloud firestore indexes composite create \
--collection-group=personal-expense-assistant-receipts \
--field-config field-path=total_amount,order=ASCENDING \
--field-config field-path=transaction_time,order=ASCENDING \
--field-config field-path=__name__,order=ASCENDING \
--database="(default)"
- Dan jalankan kode ini untuk mendukung penelusuran vektor
gcloud firestore indexes composite create \
--collection-group="personal-expense-assistant-receipts" \
--query-scope=COLLECTION \
--field-config field-path="embedding",vector-config='{"dimension":"768", "flat": "{}"}' \
--database="(default)"
Anda dapat memeriksa indeks yang dibuat dengan membuka Firestore di konsol cloud, lalu mengklik instance database (default) dan memilih Indexes di menu navigasi
Buka Cloud Shell Editor dan Siapkan Direktori Kerja Aplikasi
Sekarang, kita dapat menyiapkan editor kode untuk melakukan beberapa hal terkait coding. Kita akan menggunakan Cloud Shell Editor untuk ini
- Klik tombol Open Editor, tindakan ini akan membuka Cloud Shell Editor, kita dapat menulis kode di sini
- Pastikan project Cloud Code ditetapkan di pojok kiri bawah (status bar) editor Cloud Shell, seperti yang ditandai pada gambar di bawah dan ditetapkan ke project Google Cloud aktif tempat Anda mengaktifkan penagihan. Authorize jika diminta. Jika Anda sudah mengikuti perintah sebelumnya, tombol tersebut juga dapat mengarah langsung ke project yang diaktifkan, bukan tombol login
- Selanjutnya, mari kita clone direktori kerja template untuk codelab ini dari GitHub, jalankan perintah berikut. Tindakan ini akan membuat direktori kerja di direktori personal-expense-assistant
git clone https://github.com/alphinside/personal-expense-assistant-adk-codelab-starter.git personal-expense-assistant
- Setelah itu, buka bagian atas Editor Cloud Shell dan klik File->Open Folder, temukan direktori username Anda, lalu temukan direktori personal-expense-assistant, lalu klik tombol OK. Tindakan ini akan menjadikan direktori yang dipilih sebagai direktori kerja utama. Dalam contoh ini, nama penggunanya adalah alvinprayuda, sehingga jalur direktori ditampilkan di bawah
Sekarang, Editor Cloud Shell Anda akan terlihat seperti ini
Penyiapan Lingkungan
Menyiapkan Lingkungan Virtual Python
Langkah berikutnya adalah menyiapkan lingkungan pengembangan. Terminal aktif Anda saat ini harus berada di dalam direktori kerja personal-expense-assistant. Kita akan menggunakan Python 3.12 dalam codelab ini dan kita akan menggunakan pengelola project uv python untuk menyederhanakan kebutuhan pembuatan dan pengelolaan versi python serta lingkungan virtual
- Jika Anda belum membuka terminal, buka dengan mengklik Terminal -> New Terminal , atau gunakan Ctrl + Shift + C , yang akan membuka jendela terminal di bagian bawah browser
- Download
uvdan instal python 3.12 dengan perintah berikut
curl -LsSf https://astral.sh/uv/0.6.16/install.sh | sh && \
source $HOME/.local/bin/env && \
uv python install 3.12
- Sekarang, mari kita lakukan inisialisasi lingkungan virtual menggunakan
uv, Jalankan perintah ini
uv sync --frozen
Tindakan ini akan membuat direktori .venv dan menginstal dependensi. Sekilas tentang pyproject.toml akan memberi Anda informasi tentang dependensi yang ditampilkan seperti ini
dependencies = [
"datasets>=3.5.0",
"google-adk>=0.2.0",
"google-cloud-firestore>=2.20.1",
"gradio>=5.23.1",
"pydantic>=2.10.6",
"pydantic-settings[yaml]>=2.8.1",
]
- Untuk menguji virtual env, buat file baru main.py dan salin kode berikut
def main():
print("Hello from personal-expense-assistant-adk!")
if __name__ == "__main__":
main()
- Kemudian, jalankan perintah berikut
uv run main.py
Anda akan mendapatkan output seperti yang ditunjukkan di bawah
Using CPython 3.12 Creating virtual environment at: .venv Hello from personal-expense-assistant-adk!
Ini menunjukkan bahwa project python sedang disiapkan dengan benar.
Menyiapkan File Konfigurasi
Sekarang kita harus menyiapkan file konfigurasi untuk project ini. Kita menggunakan pydantic-settings untuk membaca konfigurasi dari file YAML.
Buat file bernama settings.yaml dengan konfigurasi berikut. Klik File->New Text File dan isi dengan kode berikut. Kemudian, simpan sebagai settings.yaml
GCLOUD_LOCATION: "us-central1"
GCLOUD_PROJECT_ID: "your_gcloud_project_id"
BACKEND_URL: "http://localhost:8081/chat"
STORAGE_BUCKET_NAME: "personal-expense-assistant-receipts"
DB_COLLECTION_NAME: "personal-expense-assistant-receipts"
Untuk codelab ini, kita akan menggunakan nilai yang telah dikonfigurasi sebelumnya untuk GCLOUD_LOCATION, BACKEND_URL, STORAGE_BUCKET_NAME, DB_COLLECTION_NAME dan BACKEND_URL .
Sekarang kita dapat melanjutkan ke langkah berikutnya, membuat agen, lalu layanan
3. Mem-build Agen menggunakan Google ADK dan Gemini 2.5
Pengantar Struktur Direktori ADK
Mari kita mulai dengan menjelajahi apa yang ditawarkan ADK dan cara mem-build agen. Dokumentasi lengkap ADK dapat diakses di URL ini . ADK menawarkan banyak utilitas dalam eksekusi perintah CLI-nya. Beberapa di antaranya adalah :
- Menyiapkan struktur direktori agen
- Mencoba interaksi dengan cepat melalui output input CLI
- Menyiapkan antarmuka web UI pengembangan lokal dengan cepat
Sekarang, mari kita buat struktur direktori agen menggunakan perintah CLI. Jalankan perintah berikut
uv run adk create expense_manager_agent \
--model gemini-2.5-flash-preview-04-17 \
--project {your-project-id} \
--region us-central1
Tindakan ini akan membuat struktur direktori agen berikut
expense_manager_agent/ ├── __init__.py ├── .env ├── agent.py
Dan jika Anda memeriksa init.py dan agent.py, Anda akan melihat kode ini
# __init__.py
from . import agent
# agent.py
from google.adk.agents import Agent
root_agent = Agent(
model='gemini-2.5-flash-preview-04-17',
name='root_agent',
description='A helpful assistant for user questions.',
instruction='Answer user questions to the best of your knowledge',
)
Mem-build Agen Pengelola Pengeluaran
Mari kita buat agen pengelola pengeluaran. Buka file expense_manager_agent/agent.py dan salin kode di bawah yang akan berisi root_agent.
# expense_manager_agent/agent.py
from google.adk.agents import Agent
from expense_manager_agent.tools import (
store_receipt_data,
search_receipts_by_metadata_filter,
search_relevant_receipts_by_natural_language_query,
get_receipt_data_by_image_id,
)
from expense_manager_agent.callbacks import modify_image_data_in_history
import os
from settings import get_settings
from google.adk.planners import BuiltInPlanner
from google.genai import types
SETTINGS = get_settings()
os.environ["GOOGLE_CLOUD_PROJECT"] = SETTINGS.GCLOUD_PROJECT_ID
os.environ["GOOGLE_CLOUD_LOCATION"] = SETTINGS.GCLOUD_LOCATION
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "TRUE"
# Get the code file directory path and read the task prompt file
current_dir = os.path.dirname(os.path.abspath(__file__))
prompt_path = os.path.join(current_dir, "task_prompt.md")
with open(prompt_path, "r") as file:
task_prompt = file.read()
root_agent = Agent(
name="expense_manager_agent",
model="gemini-2.5-flash-preview-04-17",
description=(
"Personal expense agent to help user track expenses, analyze receipts, and manage their financial records"
),
instruction=task_prompt,
tools=[
store_receipt_data,
get_receipt_data_by_image_id,
search_receipts_by_metadata_filter,
search_relevant_receipts_by_natural_language_query,
],
planner=BuiltInPlanner(
thinking_config=types.ThinkingConfig(
thinking_budget=2048,
)
),
before_model_callback=modify_image_data_in_history,
)
Penjelasan Kode
Skrip ini berisi inisialisasi agen tempat kita melakukan inisialisasi hal-hal berikut:
- Menetapkan model yang akan digunakan ke
gemini-2.5-flash-preview-04-17 - Siapkan deskripsi dan petunjuk agen sebagai perintah sistem yang sedang dibaca dari
task_prompt.md - Menyediakan alat yang diperlukan untuk mendukung fungsi agen
- Mengaktifkan perencanaan sebelum membuat respons atau eksekusi akhir menggunakan kemampuan pemikiran Gemini 2.5 Flash
- Menyiapkan intersepsi callback sebelum mengirim permintaan ke Gemini untuk membatasi jumlah data gambar yang dikirim sebelum membuat prediksi
4. Mengonfigurasi Alat Agen
Agen pengelola pengeluaran kami akan memiliki kemampuan berikut:
- Mengekstrak data dari gambar tanda terima dan menyimpan data serta file
- Penelusuran persis pada data pengeluaran
- Penelusuran kontekstual pada data pengeluaran
Oleh karena itu, kita memerlukan alat yang sesuai untuk mendukung fungsi ini. Buat file baru di direktori expense_manager_agent dan beri nama tools.py, lalu salin kode di bawah
# expense_manager_agent/tools.py
import datetime
from typing import Dict, List, Any
from google.cloud import firestore
from google.cloud.firestore_v1.vector import Vector
from google.cloud.firestore_v1 import FieldFilter
from google.cloud.firestore_v1.base_query import And
from google.cloud.firestore_v1.base_vector_query import DistanceMeasure
from settings import get_settings
from google import genai
SETTINGS = get_settings()
DB_CLIENT = firestore.Client(
project=SETTINGS.GCLOUD_PROJECT_ID
) # Will use "(default)" database
COLLECTION = DB_CLIENT.collection(SETTINGS.DB_COLLECTION_NAME)
GENAI_CLIENT = genai.Client(
vertexai=True, location=SETTINGS.GCLOUD_LOCATION, project=SETTINGS.GCLOUD_PROJECT_ID
)
EMBEDDING_DIMENSION = 768
EMBEDDING_FIELD_NAME = "embedding"
INVALID_ITEMS_FORMAT_ERR = """
Invalid items format. Must be a list of dictionaries with 'name', 'price', and 'quantity' keys."""
RECEIPT_DESC_FORMAT = """
Store Name: {store_name}
Transaction Time: {transaction_time}
Total Amount: {total_amount}
Currency: {currency}
Purchased Items:
{purchased_items}
Receipt Image ID: {receipt_id}
"""
def sanitize_image_id(image_id: str) -> str:
"""Sanitize image ID by removing any leading/trailing whitespace."""
if image_id.startswith("[IMAGE-"):
image_id = image_id.split("ID ")[1].split("]")[0]
return image_id.strip()
def store_receipt_data(
image_id: str,
store_name: str,
transaction_time: str,
total_amount: float,
purchased_items: List[Dict[str, Any]],
currency: str = "IDR",
) -> str:
"""
Store receipt data in the database.
Args:
image_id (str): The unique identifier of the image. For example IMAGE-POSITION 0-ID 12345,
the ID of the image is 12345.
store_name (str): The name of the store.
transaction_time (str): The time of purchase, in ISO format ("YYYY-MM-DDTHH:MM:SS.ssssssZ").
total_amount (float): The total amount spent.
purchased_items (List[Dict[str, Any]]): A list of items purchased with their prices. Each item must have:
- name (str): The name of the item.
- price (float): The price of the item.
- quantity (int, optional): The quantity of the item. Defaults to 1 if not provided.
currency (str, optional): The currency of the transaction, can be derived from the store location.
If unsure, default is "IDR".
Returns:
str: A success message with the receipt ID.
Raises:
Exception: If the operation failed or input is invalid.
"""
try:
# In case of it provide full image placeholder, extract the id string
image_id = sanitize_image_id(image_id)
# Check if the receipt already exists
doc = get_receipt_data_by_image_id(image_id)
if doc:
return f"Receipt with ID {image_id} already exists"
# Validate transaction time
if not isinstance(transaction_time, str):
raise ValueError(
"Invalid transaction time: must be a string in ISO format 'YYYY-MM-DDTHH:MM:SS.ssssssZ'"
)
try:
datetime.datetime.fromisoformat(transaction_time.replace("Z", "+00:00"))
except ValueError:
raise ValueError(
"Invalid transaction time format. Must be in ISO format 'YYYY-MM-DDTHH:MM:SS.ssssssZ'"
)
# Validate items format
if not isinstance(purchased_items, list):
raise ValueError(INVALID_ITEMS_FORMAT_ERR)
for _item in purchased_items:
if (
not isinstance(_item, dict)
or "name" not in _item
or "price" not in _item
):
raise ValueError(INVALID_ITEMS_FORMAT_ERR)
if "quantity" not in _item:
_item["quantity"] = 1
# Create a combined text from all receipt information for better embedding
result = GENAI_CLIENT.models.embed_content(
model="text-embedding-004",
contents=RECEIPT_DESC_FORMAT.format(
store_name=store_name,
transaction_time=transaction_time,
total_amount=total_amount,
currency=currency,
purchased_items=purchased_items,
receipt_id=image_id,
),
)
embedding = result.embeddings[0].values
doc = {
"receipt_id": image_id,
"store_name": store_name,
"transaction_time": transaction_time,
"total_amount": total_amount,
"currency": currency,
"purchased_items": purchased_items,
EMBEDDING_FIELD_NAME: Vector(embedding),
}
COLLECTION.add(doc)
return f"Receipt stored successfully with ID: {image_id}"
except Exception as e:
raise Exception(f"Failed to store receipt: {str(e)}")
def search_receipts_by_metadata_filter(
start_time: str,
end_time: str,
min_total_amount: float = -1.0,
max_total_amount: float = -1.0,
) -> str:
"""
Filter receipts by metadata within a specific time range and optionally by amount.
Args:
start_time (str): The start datetime for the filter (in ISO format, e.g. 'YYYY-MM-DDTHH:MM:SS.ssssssZ').
end_time (str): The end datetime for the filter (in ISO format, e.g. 'YYYY-MM-DDTHH:MM:SS.ssssssZ').
min_total_amount (float): The minimum total amount for the filter (inclusive). Defaults to -1.
max_total_amount (float): The maximum total amount for the filter (inclusive). Defaults to -1.
Returns:
str: A string containing the list of receipt data matching all applied filters.
Raises:
Exception: If the search failed or input is invalid.
"""
try:
# Validate start and end times
if not isinstance(start_time, str) or not isinstance(end_time, str):
raise ValueError("start_time and end_time must be strings in ISO format")
try:
datetime.datetime.fromisoformat(start_time.replace("Z", "+00:00"))
datetime.datetime.fromisoformat(end_time.replace("Z", "+00:00"))
except ValueError:
raise ValueError("start_time and end_time must be strings in ISO format")
# Start with the base collection reference
query = COLLECTION
# Build the composite query by properly chaining conditions
# Notes that this demo assume 1 user only,
# need to refactor the query for multiple user
filters = [
FieldFilter("transaction_time", ">=", start_time),
FieldFilter("transaction_time", "<=", end_time),
]
# Add optional filters
if min_total_amount != -1:
filters.append(FieldFilter("total_amount", ">=", min_total_amount))
if max_total_amount != -1:
filters.append(FieldFilter("total_amount", "<=", max_total_amount))
# Apply the filters
composite_filter = And(filters=filters)
query = query.where(filter=composite_filter)
# Execute the query and collect results
search_result_description = "Search by Metadata Results:\n"
for doc in query.stream():
data = doc.to_dict()
data.pop(
EMBEDDING_FIELD_NAME, None
) # Remove embedding as it's not needed for display
search_result_description += f"\n{RECEIPT_DESC_FORMAT.format(**data)}"
return search_result_description
except Exception as e:
raise Exception(f"Error filtering receipts: {str(e)}")
def search_relevant_receipts_by_natural_language_query(
query_text: str, limit: int = 5
) -> str:
"""
Search for receipts with content most similar to the query using vector search.
This tool can be use for user query that is difficult to translate into metadata filters.
Such as store name or item name which sensitive to string matching.
Use this tool if you cannot utilize the search by metadata filter tool.
Args:
query_text (str): The search text (e.g., "coffee", "dinner", "groceries").
limit (int, optional): Maximum number of results to return (default: 5).
Returns:
str: A string containing the list of contextually relevant receipt data.
Raises:
Exception: If the search failed or input is invalid.
"""
try:
# Generate embedding for the query text
result = GENAI_CLIENT.models.embed_content(
model="text-embedding-004", contents=query_text
)
query_embedding = result.embeddings[0].values
# Notes that this demo assume 1 user only,
# need to refactor the query for multiple user
vector_query = COLLECTION.find_nearest(
vector_field=EMBEDDING_FIELD_NAME,
query_vector=Vector(query_embedding),
distance_measure=DistanceMeasure.EUCLIDEAN,
limit=limit,
)
# Execute the query and collect results
search_result_description = "Search by Contextual Relevance Results:\n"
for doc in vector_query.stream():
data = doc.to_dict()
data.pop(
EMBEDDING_FIELD_NAME, None
) # Remove embedding as it's not needed for display
search_result_description += f"\n{RECEIPT_DESC_FORMAT.format(**data)}"
return search_result_description
except Exception as e:
raise Exception(f"Error searching receipts: {str(e)}")
def get_receipt_data_by_image_id(image_id: str) -> Dict[str, Any]:
"""
Retrieve receipt data from the database using the image_id.
Args:
image_id (str): The unique identifier of the receipt image. For example, if the placeholder is
[IMAGE-ID 12345], the ID to use is 12345.
Returns:
Dict[str, Any]: A dictionary containing the receipt data with the following keys:
- receipt_id (str): The unique identifier of the receipt image.
- store_name (str): The name of the store.
- transaction_time (str): The time of purchase in UTC.
- total_amount (float): The total amount spent.
- currency (str): The currency of the transaction.
- purchased_items (List[Dict[str, Any]]): List of items purchased with their details.
Returns an empty dictionary if no receipt is found.
"""
# In case of it provide full image placeholder, extract the id string
image_id = sanitize_image_id(image_id)
# Query the receipts collection for documents with matching receipt_id (image_id)
# Notes that this demo assume 1 user only,
# need to refactor the query for multiple user
query = COLLECTION.where(filter=FieldFilter("receipt_id", "==", image_id)).limit(1)
docs = list(query.stream())
if not docs:
return {}
# Get the first matching document
doc_data = docs[0].to_dict()
doc_data.pop(EMBEDDING_FIELD_NAME, None)
return doc_data
Penjelasan Kode
Dalam penerapan fungsi alat ini, kami mendesain alat berdasarkan 2 ide utama berikut:
- Mengurai data tanda terima dan pemetaan ke file asli menggunakan placeholder string ID Gambar
[IMAGE-ID <hash-of-image-1>] - Menyimpan dan mengambil data menggunakan database Firestore
Alat "store_receipt_data"
Alat ini adalah alat Optical Character Recognition, yang akan mengurai informasi yang diperlukan dari data gambar, bersama dengan mengenali string ID Gambar dan memetakan keduanya untuk disimpan di database Firestore.
Selain itu, alat ini juga mengonversi konten tanda terima menjadi penyematan menggunakan text-embedding-004 sehingga semua metadata dan penyematan disimpan dan diindeks bersama. Memungkinkan fleksibilitas untuk diambil baik melalui kueri maupun penelusuran kontekstual.
Setelah berhasil menjalankan alat ini, Anda dapat melihat bahwa data tanda terima sudah diindeks di database Firestore seperti yang ditunjukkan di bawah ini
Alat "search_receipts_by_metadata_filter"
Alat ini mengonversi kueri pengguna menjadi filter kueri metadata yang mendukung penelusuran berdasarkan rentang tanggal dan/atau total transaksi. Tindakan ini akan menampilkan semua data tanda terima yang cocok, dan dalam prosesnya, kami akan menghapus kolom penyematan karena tidak diperlukan oleh agen untuk pemahaman kontekstual
Alat "search_relevant_receipts_by_natural_language_query"
Ini adalah alat Retrieval Augmented Generation (RAG) kami. Agen kami memiliki kemampuan untuk mendesain kuerinya sendiri guna mengambil tanda terima yang relevan dari database vektor dan juga dapat memilih kapan harus menggunakan alat ini. Ide untuk mengizinkan keputusan independen dari agen, baik untuk menggunakan alat RAG ini maupun tidak, dan mendesain kueri sendiri adalah salah satu definisi pendekatan Agentic RAG.
Kami tidak hanya mengizinkannya membuat kueri sendiri, tetapi juga mengizinkannya memilih jumlah dokumen relevan yang ingin diambil. Dikombinasikan dengan rekayasa perintah yang tepat, misalnya
# Example prompt Always filter the result from tool search_relevant_receipts_by_natural_language_query as the returned result may contain irrelevant information
Hal ini akan menjadikan alat ini sebagai alat canggih yang dapat menelusuri hampir semua hal, meskipun mungkin tidak menampilkan semua hasil yang diharapkan karena sifat penelusuran tetangga terdekat yang tidak persis.
5. Modifikasi Konteks Percakapan melalui Callback
Google ADK memungkinkan kita "mencegat" runtime agen di berbagai tingkat. Anda dapat membaca lebih lanjut kemampuan mendetail ini di dokumentasi ini . Di lab ini, kita menggunakan before_model_callback untuk mengubah permintaan sebelum dikirim ke LLM untuk menghapus data gambar dalam konteks histori percakapan lama ( hanya menyertakan data gambar dalam 3 interaksi pengguna terakhir) untuk efisiensi
Namun, kita tetap ingin agen memiliki konteks data gambar jika diperlukan. Oleh karena itu, kita menambahkan mekanisme untuk menambahkan placeholder ID gambar string setelah setiap data byte gambar dalam percakapan. Hal ini akan membantu agen menautkan ID gambar ke data file sebenarnya yang dapat digunakan pada waktu penyimpanan atau pengambilan gambar. Strukturnya akan terlihat seperti ini
<image-byte-data-1> [IMAGE-ID <hash-of-image-1>] <image-byte-data-2> [IMAGE-ID <hash-of-image-2>] And so on..
Dan saat data byte tidak digunakan lagi dalam histori percakapan, ID string masih ada untuk tetap mengaktifkan akses data dengan bantuan penggunaan alat. Contoh struktur histori setelah data gambar dihapus
[IMAGE-ID <hash-of-image-1>] [IMAGE-ID <hash-of-image-2>] And so on..
Ayo mulai! Buat file baru di direktori expense_manager_agent dan beri nama callbacks.py, lalu salin kode di bawah
# expense_manager_agent/callbacks.py
import hashlib
from google.genai import types
from google.adk.agents.callback_context import CallbackContext
from google.adk.models.llm_request import LlmRequest
def modify_image_data_in_history(
callback_context: CallbackContext, llm_request: LlmRequest
) -> None:
# The following code will modify the request sent to LLM
# We will only keep image data in the last 3 user messages using a reverse and counter approach
# Count how many user messages we've processed
user_message_count = 0
# Process the reversed list
for content in reversed(llm_request.contents):
# Only count for user manual query, not function call
if (content.role == "user") and (content.parts[0].function_response is None):
user_message_count += 1
modified_content_parts = []
# Check any missing image ID placeholder for any image data
# Then remove image data from conversation history if more than 3 user messages
for idx, part in enumerate(content.parts):
if part.inline_data is None:
modified_content_parts.append(part)
continue
if (
(idx + 1 >= len(content.parts))
or (content.parts[idx + 1].text is None)
or (not content.parts[idx + 1].text.startswith("[IMAGE-ID "))
):
# Generate hash ID for the image and add a placeholder
image_data = part.inline_data.data
hasher = hashlib.sha256(image_data)
image_hash_id = hasher.hexdigest()[:12]
placeholder = f"[IMAGE-ID {image_hash_id}]"
# Only keep image data in the last 3 user messages
if user_message_count <= 3:
modified_content_parts.append(part)
modified_content_parts.append(types.Part(text=placeholder))
else:
# Only keep image data in the last 3 user messages
if user_message_count <= 3:
modified_content_parts.append(part)
# This will modify the contents inside the llm_request
content.parts = modified_content_parts
6. Perintah
Untuk mendesain agen dengan interaksi dan kemampuan yang kompleks, kita harus menemukan perintah yang cukup baik untuk memandu agen agar dapat berperilaku sesuai keinginan kita.
Sebelumnya, kami memiliki mekanisme tentang cara menangani data gambar dalam histori percakapan, dan juga memiliki alat yang mungkin tidak mudah digunakan, seperti search_relevant_receipts_by_natural_language_query. Kami juga ingin agen dapat menelusuri dan mengambil gambar tanda terima yang benar kepada kami. Artinya, kita perlu menyampaikan semua informasi ini dengan benar dalam struktur perintah yang tepat
Kami akan meminta agen untuk menyusun output ke dalam format markdown berikut untuk mengurai proses berpikir, respons akhir, dan lampiran ( jika ada)
# THINKING PROCESS
Thinking process here
# FINAL RESPONSE
Response to the user here
Attachments put inside json block
{
"attachments": [
"[IMAGE-ID <hash-id-1>]",
"[IMAGE-ID <hash-id-2>]",
...
]
}
Mari kita mulai dengan perintah berikut untuk mencapai ekspektasi awal kita tentang perilaku agen pengelola pengeluaran. File task_prompt.md seharusnya sudah ada di direktori kerja yang ada, tetapi kita perlu memindahkannya ke direktori expense_manager_agent. Jalankan perintah berikut untuk memindahkannya
mv task_prompt.md expense_manager_agent/task_prompt.md
7. Menguji Agen
Sekarang, mari kita coba berkomunikasi dengan agen melalui CLI, jalankan perintah berikut
uv run adk run expense_manager_agent
Tindakan ini akan menampilkan output seperti ini, tempat Anda dapat melakukan chat dengan agen secara bergantian, tetapi Anda hanya dapat mengirim teks melalui antarmuka ini
Log setup complete: /tmp/agents_log/agent.xxxx_xxx.log To access latest log: tail -F /tmp/agents_log/agent.latest.log Running agent root_agent, type exit to exit. user: hello [root_agent]: Hello there! How can I help you today? user:
Sekarang, selain interaksi CLI, ADK juga memungkinkan kita memiliki UI pengembangan untuk berinteraksi dan memeriksa apa yang terjadi selama interaksi. Jalankan perintah berikut untuk memulai server UI pengembangan lokal
uv run adk web --port 8080
Tindakan ini akan menghasilkan output seperti contoh berikut, yang berarti kita sudah dapat mengakses antarmuka web
INFO: Started server process [xxxx] INFO: Waiting for application startup. +-----------------------------------------------------------------------------+ | ADK Web Server started | | | | For local testing, access at http://localhost:8080. | +-----------------------------------------------------------------------------+ INFO: Application startup complete. INFO: Uvicorn running on http://0.0.0.0:8080 (Press CTRL+C to quit)
Sekarang, untuk memeriksanya, klik tombol Web Preview di area atas Editor Cloud Shell Anda, lalu pilih Preview on port 8080
Anda akan melihat halaman web berikut tempat Anda dapat memilih agen yang tersedia di tombol drop-down kiri atas ( dalam kasus ini, agennya adalah expense_manager_agent) dan berinteraksi dengan bot. Anda akan melihat banyak informasi tentang detail log selama runtime agen di jendela kiri
Mari kita coba beberapa tindakan. Upload 2 contoh tanda terima ini ( sumber : set data Hugging Face mousserlane/id_receipt_dataset ) . Klik kanan pada setiap gambar dan pilih Save Image as.. ( tindakan ini akan mendownload gambar tanda terima), lalu upload file ke bot dengan mengklik ikon "clip" dan menyatakan bahwa Anda ingin menyimpan tanda terima ini
Setelah itu, coba kueri berikut untuk melakukan beberapa penelusuran atau pengambilan file
- "Berikan perincian pengeluaran dan totalnya selama tahun 2023"
- "Berikan file tanda terima dari Indomaret"
Saat menggunakan beberapa alat, Anda dapat memeriksa apa yang terjadi di UI pengembangan
Lihat bagaimana agen merespons Anda dan periksa apakah agen mematuhi semua aturan yang diberikan dalam perintah di dalam task_prompt.py. Selamat! Sekarang Anda memiliki agen pengembangan yang lengkap dan berfungsi.
Sekarang saatnya menyelesaikannya dengan UI dan kemampuan yang tepat dan bagus untuk mengupload dan mendownload file gambar.
8. Mem-build Layanan Frontend menggunakan Gradio
Kita akan membuat antarmuka web chat yang terlihat seperti ini
Halaman ini berisi antarmuka chat dengan kolom input bagi pengguna untuk mengirim teks dan mengupload file gambar tanda terima.
Kita akan mem-build layanan frontend menggunakan Gradio.
Buat file baru, Klik File->New Text File, dan beri nama frontend.py, lalu salin kode berikut dan simpan
import mimetypes
import gradio as gr
import requests
import base64
from typing import List, Dict, Any
from settings import get_settings
from PIL import Image
import io
from schema import ImageData, ChatRequest, ChatResponse
SETTINGS = get_settings()
def encode_image_to_base64_and_get_mime_type(image_path: str) -> ImageData:
"""Encode a file to base64 string and get MIME type.
Reads an image file and returns the base64-encoded image data and its MIME type.
Args:
image_path: Path to the image file to encode.
Returns:
ImageData object containing the base64 encoded image data and its MIME type.
"""
# Read the image file
with open(image_path, "rb") as file:
image_content = file.read()
# Get the mime type
mime_type = mimetypes.guess_type(image_path)[0]
# Base64 encode the image
base64_data = base64.b64encode(image_content).decode("utf-8")
# Return as ImageData object
return ImageData(serialized_image=base64_data, mime_type=mime_type)
def decode_base64_to_image(base64_data: str) -> Image.Image:
"""Decode a base64 string to PIL Image.
Converts a base64-encoded image string back to a PIL Image object
that can be displayed or processed further.
Args:
base64_data: Base64 encoded string of the image.
Returns:
PIL Image object of the decoded image.
"""
# Decode the base64 string and convert to PIL Image
image_data = base64.b64decode(base64_data)
image_buffer = io.BytesIO(image_data)
image = Image.open(image_buffer)
return image
def get_response_from_llm_backend(
message: Dict[str, Any],
history: List[Dict[str, Any]],
) -> List[str | gr.Image]:
"""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.
Returns:
List containing text response and any image attachments from the backend service.
"""
# Extract files and convert to base64
image_data = []
if uploaded_files := message.get("files", []):
for file_path in uploaded_files:
image_data.append(encode_image_to_base64_and_get_mime_type(file_path))
# Prepare the request payload
payload = ChatRequest(
text=message["text"],
files=image_data,
session_id="default_session",
user_id="default_user",
)
# Send request to backend
try:
response = requests.post(SETTINGS.BACKEND_URL, json=payload.model_dump())
response.raise_for_status() # Raise exception for HTTP errors
result = ChatResponse(**response.json())
if result.error:
return [f"Error: {result.error}"]
chat_responses = []
if result.thinking_process:
chat_responses.append(
gr.ChatMessage(
role="assistant",
content=result.thinking_process,
metadata={"title": "🧠 Thinking Process"},
)
)
chat_responses.append(gr.ChatMessage(role="assistant", content=result.response))
if result.attachments:
for attachment in result.attachments:
image_data = attachment.serialized_image
chat_responses.append(gr.Image(decode_base64_to_image(image_data)))
return chat_responses
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="Personal Expense Assistant",
description="This assistant can help you to store receipts data, find receipts, and track your expenses during certain period.",
type="messages",
multimodal=True,
textbox=gr.MultimodalTextbox(file_count="multiple", file_types=["image"]),
)
demo.launch(
server_name="0.0.0.0",
server_port=8080,
)
Setelah itu, kita dapat mencoba menjalankan layanan frontend dengan perintah berikut. Jangan lupa untuk mengganti nama file main.py menjadi frontend.py
uv run frontend.py
Anda akan melihat output yang mirip dengan ini di konsol cloud
* Running on local URL: http://0.0.0.0:8080 To create a public link, set `share=True` in `launch()`.
Setelah itu, Anda dapat memeriksa antarmuka web saat ctrl+mengklik link URL lokal. Atau, Anda juga dapat mengakses aplikasi frontend dengan mengklik tombol Web Preview di sisi kanan atas Cloud Editor, lalu pilih Preview on port 8080
Anda akan melihat antarmuka web, tetapi akan mendapatkan error yang diharapkan saat mencoba mengirim chat karena layanan backend belum disiapkan
Sekarang, biarkan layanan berjalan dan jangan hentikan dulu. Kita akan menjalankan layanan backend di tab terminal lain
Penjelasan Kode
Dalam kode frontend ini, pertama-tama kita mengaktifkan pengguna untuk mengirim teks dan mengupload beberapa file. Gradio memungkinkan kita membuat jenis fungsi ini dengan metode gr.ChatInterface yang digabungkan dengan gr.MultimodalTextbox
Sekarang, sebelum mengirim file dan teks ke backend, kita perlu mengetahui mimetype file seperti yang diperlukan oleh backend. Kita juga perlu mengenkode byte file gambar ke dalam base64 dan mengirimkannya bersama dengan mimetype.
class ImageData(BaseModel):
"""Model for image data with hash identifier.
Attributes:
serialized_image: Optional Base64 encoded string of the image content.
mime_type: MIME type of the image.
"""
serialized_image: str
mime_type: str
Skema yang digunakan untuk interaksi frontend - backend ditentukan di schema.py. Kita menggunakan BaseModel Pydantic untuk menerapkan validasi data dalam skema
Saat menerima respons, kami sudah memisahkan bagian mana yang merupakan proses berpikir, respons akhir, dan lampiran. Dengan demikian, kita dapat menggunakan komponen Gradio untuk menampilkan setiap komponen dengan komponen UI.
class ChatResponse(BaseModel):
"""Model for a chat response.
Attributes:
response: The text response from the model.
thinking_process: Optional thinking process of the model.
attachments: List of image data to be displayed to the user.
error: Optional error message if something went wrong.
"""
response: str
thinking_process: str = ""
attachments: List[ImageData] = []
error: Optional[str] = None
9. Mem-build Layanan Backend menggunakan FastAPI
Selanjutnya, kita harus mem-build backend yang dapat menginisialisasi Agen bersama dengan komponen lainnya agar dapat menjalankan runtime agen.
Buat file baru, Klik File->New Text File, lalu salin dan tempel kode berikut, lalu simpan sebagai backend.py
from expense_manager_agent.agent import root_agent as expense_manager_agent
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.adk.events import Event
from fastapi import FastAPI, Body, Depends
from typing import AsyncIterator
from types import SimpleNamespace
import uvicorn
from contextlib import asynccontextmanager
import asyncio
from utils import (
extract_attachment_ids_and_sanitize_response,
download_image_from_gcs,
extract_thinking_process,
format_user_request_to_adk_content_and_store_artifacts,
)
from schema import ImageData, ChatRequest, ChatResponse
import logger
from google.adk.artifacts import GcsArtifactService
from settings import get_settings
SETTINGS = get_settings()
APP_NAME = "expense_manager_app"
# Application state to hold service contexts
class AppContexts(SimpleNamespace):
"""A class to hold application contexts with attribute access"""
session_service: InMemorySessionService = None
artifact_service: GcsArtifactService = None
expense_manager_agent_runner: Runner = None
# Initialize application state
app_contexts = AppContexts()
@asynccontextmanager
async def lifespan(app: FastAPI):
# Initialize service contexts during application startup
app_contexts.session_service = InMemorySessionService()
app_contexts.artifact_service = GcsArtifactService(
bucket_name=SETTINGS.STORAGE_BUCKET_NAME
)
app_contexts.expense_manager_agent_runner = Runner(
agent=expense_manager_agent, # The agent we want to run
app_name=APP_NAME, # Associates runs with our app
session_service=app_contexts.session_service, # Uses our session manager
artifact_service=app_contexts.artifact_service, # Uses our artifact manager
)
logger.info("Application started successfully")
yield
logger.info("Application shutting down")
# Perform cleanup during application shutdown if necessary
# Helper function to get application state as a dependency
async def get_app_contexts() -> AppContexts:
return app_contexts
# Create FastAPI app
app = FastAPI(title="Personal Expense Assistant API", lifespan=lifespan)
@app.post("/chat", response_model=ChatResponse)
async def chat(
request: ChatRequest = Body(...),
app_context: AppContexts = Depends(get_app_contexts),
) -> ChatResponse:
"""Process chat request and get response from the agent"""
# Prepare the user's message in ADK format and store image artifacts
content = await asyncio.to_thread(
format_user_request_to_adk_content_and_store_artifacts,
request=request,
app_name=APP_NAME,
artifact_service=app_context.artifact_service,
)
final_response_text = "Agent did not produce a final response." # Default
# Use the session ID from the request or default if not provided
session_id = request.session_id
user_id = request.user_id
# Create session if it doesn't exist
if not app_context.session_service.get_session(
app_name=APP_NAME, user_id=user_id, session_id=session_id
):
app_context.session_service.create_session(
app_name=APP_NAME, user_id=user_id, session_id=session_id
)
try:
# Process the message with the agent
# Type annotation: runner.run_async returns an AsyncIterator[Event]
events_iterator: AsyncIterator[Event] = (
app_context.expense_manager_agent_runner.run_async(
user_id=user_id, session_id=session_id, new_message=content
)
)
async for event in events_iterator: # event has type Event
# Key Concept: is_final_response() marks the concluding message for the turn
if event.is_final_response():
if event.content and event.content.parts:
# Extract text from the first part
final_response_text = event.content.parts[0].text
elif event.actions and event.actions.escalate:
# Handle potential errors/escalations
final_response_text = f"Agent escalated: {event.error_message or 'No specific message.'}"
break # Stop processing events once the final response is found
logger.info(
"Received final response from agent", raw_final_response=final_response_text
)
# Extract and process any attachments and thinking process in the response
base64_attachments = []
sanitized_text, attachment_ids = extract_attachment_ids_and_sanitize_response(
final_response_text
)
sanitized_text, thinking_process = extract_thinking_process(sanitized_text)
# Download images from GCS and replace hash IDs with base64 data
for image_hash_id in attachment_ids:
# Download image data and get MIME type
result = await asyncio.to_thread(
download_image_from_gcs,
artifact_service=app_context.artifact_service,
image_hash=image_hash_id,
app_name=APP_NAME,
user_id=user_id,
session_id=session_id,
)
if result:
base64_data, mime_type = result
base64_attachments.append(
ImageData(serialized_image=base64_data, mime_type=mime_type)
)
logger.info(
"Processed response with attachments",
sanitized_response=sanitized_text,
thinking_process=thinking_process,
attachment_ids=attachment_ids,
)
return ChatResponse(
response=sanitized_text,
thinking_process=thinking_process,
attachments=base64_attachments,
)
except Exception as e:
logger.error("Error processing chat request", error_message=str(e))
return ChatResponse(
response="", error=f"Error in generating response: {str(e)}"
)
# Only run the server if this file is executed directly
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8081)
Setelah itu, kita dapat mencoba menjalankan layanan backend. Ingat bahwa pada langkah sebelumnya kita menjalankan layanan frontend dengan benar, sekarang kita harus membuka terminal baru dan mencoba menjalankan layanan backend ini
- Buat terminal baru. Buka terminal di area bawah dan temukan tombol "+" untuk membuat terminal baru. Atau, Anda dapat menekan Ctrl + Shift + C untuk membuka terminal baru
- Setelah itu, pastikan Anda berada di direktori kerja personal-expense-assistant, lalu jalankan perintah berikut
uv run backend.py
- Jika berhasil, output akan ditampilkan seperti ini
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)
Penjelasan Kode
Melakukan inisialisasi ADK Agent, SessionService, dan ArtifactService
Untuk menjalankan agen di layanan backend, kita harus membuat Runner yang menggunakan SessionService dan agen kita. SessionService, SessionService akan memberi agen kemampuan untuk menerima konteks percakapan yang sedang berlangsung.
Kita juga menggunakan ArtifactService untuk menangani file yang diupload. Anda dapat membaca detail selengkapnya di sini tentang Sesi dan Artefak ADK
...
@asynccontextmanager
async def lifespan(app: FastAPI):
# Initialize service contexts during application startup
app_contexts.session_service = InMemorySessionService()
app_contexts.artifact_service = GcsArtifactService(
bucket_name=SETTINGS.STORAGE_BUCKET_NAME
)
app_contexts.expense_manager_agent_runner = Runner(
agent=expense_manager_agent, # The agent we want to run
app_name=APP_NAME, # Associates runs with our app
session_service=app_contexts.session_service, # Uses our session manager
artifact_service=app_contexts.artifact_service, # Uses our artifact manager
)
logger.info("Application started successfully")
yield
logger.info("Application shutting down")
# Perform cleanup during application shutdown if necessary
...
Dalam demo ini, kita menggunakan InMemorySessionService dan GcsArtifactService untuk diintegrasikan dengan Runner agen kita. Karena histori percakapan disimpan di memori, histori tersebut akan hilang setelah layanan backend dihentikan atau dimulai ulang. Kita melakukan inisialisasi di dalam siklus proses aplikasi FastAPI untuk dimasukkan sebagai dependensi di rute /chat.
Mengupload dan Mendownload Gambar dengan GcsArtifactService
Semua gambar yang diupload akan disimpan sebagai artefak oleh GcsArtifactService, Anda dapat memeriksanya di dalam fungsi format_user_request_to_adk_content_and_store_artifacts di dalam utils.py
...
# Prepare the user's message in ADK format and store image artifacts
content = await asyncio.to_thread(
format_user_request_to_adk_content_and_store_artifacts,
request=request,
app_name=APP_NAME,
artifact_service=app_context.artifact_service,
)
...
Semua permintaan yang akan diproses oleh runner agen harus diformat ke dalam jenis types.Content. Di dalam fungsi, kita juga memproses setiap data gambar dan mengekstrak ID-nya untuk diganti dengan placeholder ID Gambar.
Mekanisme serupa digunakan untuk mendownload lampiran setelah mengekstrak ID gambar menggunakan ekspresi reguler:
...
sanitized_text, attachment_ids = extract_attachment_ids_and_sanitize_response(
final_response_text
)
sanitized_text, thinking_process = extract_thinking_process(sanitized_text)
# Download images from GCS and replace hash IDs with base64 data
for image_hash_id in attachment_ids:
# Download image data and get MIME type
result = await asyncio.to_thread(
download_image_from_gcs,
artifact_service=app_context.artifact_service,
image_hash=image_hash_id,
app_name=APP_NAME,
user_id=user_id,
session_id=session_id,
)
...
10. Pengujian Integrasi
Sekarang, Anda akan memiliki beberapa layanan yang berjalan di tab konsol cloud yang berbeda:
- Layanan frontend berjalan di port 8080
* Running on local URL: http://0.0.0.0:8080 To create a public link, set `share=True` in `launch()`.
- Layanan backend berjalan di port 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)
Pada status saat ini, Anda seharusnya dapat mengupload gambar tanda terima dan melakukan chat dengan lancar dengan asisten dari aplikasi web di port 8080.
Klik tombol Web Preview di area atas Editor Cloud Shell Anda, lalu pilih Preview on port 8080
Sekarang, mari kita lakukan beberapa interaksi dengan asisten.
Download tanda terima berikut. Rentang tanggal data tanda terima ini adalah antara tahun 2023-2024 dan minta asisten untuk menyimpan/menguploadnya
- Receipt Drive ( set data Hugging Face sumber
mousserlane/id_receipt_dataset)
Mengajukan berbagai hal
- "Beri saya perincian pengeluaran bulanan selama 2023-2024"
- "Tampilkan tanda terima untuk transaksi kopi"
- "Berikan file tanda terima dari Yakiniku Like"
- Dst.
Berikut beberapa cuplikan interaksi yang berhasil
11. Men-deploy ke Cloud Run
Sekarang, tentu saja kita ingin mengakses aplikasi luar biasa ini dari mana saja. Untuk melakukannya, kita dapat memaketkan aplikasi ini dan men-deploy-nya ke Cloud Run. Untuk demo ini, layanan ini akan diekspos sebagai layanan publik yang dapat diakses oleh orang lain. Namun, perlu diingat bahwa ini bukan praktik terbaik untuk aplikasi semacam ini karena lebih cocok untuk aplikasi pribadi
Dalam codelab ini, kita akan menempatkan layanan frontend dan backend dalam 1 penampung. Kita memerlukan bantuan supervisord untuk mengelola kedua layanan tersebut. Anda dapat memeriksa file supervisord.conf dan memeriksa Dockerfile yang menetapkan supervisord sebagai titik entri.
Pada tahap ini, kita sudah memiliki semua file yang diperlukan untuk men-deploy aplikasi ke Cloud Run. Mari kita deploy. Buka Terminal Cloud Shell dan pastikan project saat ini dikonfigurasi ke project aktif Anda. Jika tidak, Anda harus menggunakan perintah gcloud configure untuk menetapkan project ID:
gcloud config set project [PROJECT_ID]
Kemudian, jalankan perintah berikut untuk men-deploynya ke Cloud Run.
gcloud run deploy personal-expense-assistant \
--source . \
--port=8080 \
--allow-unauthenticated \
--env-vars-file=settings.yaml \
--memory 1024Mi \
--region us-central1
Jika Anda diminta untuk mengonfirmasi pembuatan registry artefak untuk repositori Docker, cukup jawab Y. Perhatikan bahwa kami mengizinkan akses yang tidak diautentikasi di sini karena ini adalah aplikasi demo. Rekomendasinya adalah menggunakan autentikasi yang sesuai untuk aplikasi perusahaan dan produksi Anda.
Setelah deployment selesai, Anda akan mendapatkan link yang mirip dengan link di bawah ini:
https://personal-expense-assistant-*******.us-central1.run.app
Gunakan aplikasi Anda dari jendela Samaran atau perangkat seluler. Video tersebut seharusnya sudah ditayangkan.
12. Tantangan
Sekarang saatnya Anda bersinar dan mengasah keterampilan eksplorasi Anda. Apakah Anda memiliki kemampuan untuk mengubah kode agar backend dapat mengakomodasi beberapa pengguna? Komponen apa yang perlu diperbarui?
13. Pembersihan
Agar tidak menimbulkan biaya pada akun Google Cloud Anda untuk resource yang digunakan dalam codelab ini, ikuti langkah-langkah berikut:
- Di konsol Google Cloud, buka halaman Manage resources.
- Dalam daftar project, pilih project yang ingin Anda hapus, lalu klik Delete.
- Pada dialog, ketik project ID, lalu klik Shut down untuk menghapus project.
- Atau, Anda dapat membuka Cloud Run di konsol, memilih layanan yang baru saja di-deploy, lalu menghapusnya.