1. 📖 Pengantar
Pernahkah Anda merasa frustrasi dan terlalu malas untuk mengelola semua pengeluaran pribadi Anda? 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 terlalu banyak mengeluarkan 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 analisis pengeluaran. Semua ini dibangun di atas framework Google Agent Development Kit
Aplikasi itu sendiri dibagi menjadi 2 layanan: frontend dan backend; sehingga Anda dapat membuat prototipe cepat dan mencoba tampilannya, serta memahami tampilan kontrak API untuk mengintegrasikan keduanya.
Melalui codelab ini, Anda akan menggunakan pendekatan langkah demi langkah sebagai berikut:
- Siapkan proyek Google Cloud Anda dan Aktifkan semua API yang diperlukan di dalamnya
- Menyiapkan bucket di Google Cloud Storage dan database di Firestore
- Buat Pengindeksan Firestore
- Menyiapkan ruang kerja untuk lingkungan coding Anda
- Menyusun kode sumber agen ADK, alat, perintah, dll.
- Menguji agen menggunakan UI Pengembangan Web lokal ADK
- Buat layanan frontend - antarmuka chat menggunakan library Gradio, untuk mengirim beberapa kueri dan mengupload gambar tanda terima
- Bangun layanan backend - server HTTP menggunakan FastAPI yang merupakan tempat kode agen ADK, SessionService, dan Layanan Artefak berada
- Kelola variabel lingkungan dan siapkan file yang diperlukan untuk menyebarkan aplikasi ke Cloud Run
- Terapkan aplikasi ke Cloud Run
Ringkasan Arsitektur
Prasyarat
- Nyaman bekerja dengan Python
- Pemahaman tentang arsitektur full-stack dasar menggunakan layanan HTTP
Yang akan Anda pelajari
- Pembuatan prototipe web frontend dengan Gradio
- Pengembangan layanan backend dengan FastAPI dan Pydantic
- Merancang ADK Agent sambil memanfaatkan beberapa kemampuannya
- Penggunaan alat
- Manajemen Sesi dan Artefak
- Penggunaan callback untuk modifikasi input sebelum dikirim ke Gemini
- Memanfaatkan BuiltInPlanner untuk meningkatkan pelaksanaan tugas dengan melakukan perencanaan
- Debugging cepat melalui antarmuka web lokal ADK
- Strategi untuk mengoptimalkan interaksi multimoda melalui penguraian dan pengambilan informasi melalui rekayasa prompt dan modifikasi permintaan Gemini menggunakan panggilan balik ADK
- Agentic Retrieval Augmented Generation menggunakan Firestore sebagai Database Vektor
- Kelola variabel lingkungan dalam file YAML dengan Pydantic-settings
- Terapkan aplikasi ke Cloud Run menggunakan Dockerfile dan berikan 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 Anda memulai
Pilih Project Aktif di Konsol Cloud
Codelab ini mengasumsikan bahwa Anda sudah memiliki proyek Google Cloud dengan penagihan yang diaktifkan. Jika Anda belum memilikinya, Anda dapat mengikuti petunjuk di bawah ini 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.
Siapkan Basis Data Firestore
Selanjutnya, kita juga perlu membuat Database Firestore. Firestore dalam mode Asli adalah basis data dokumen NoSQL yang dibuat untuk penskalaan otomatis, kinerja tinggi, dan kemudahan pengembangan aplikasi. Ia juga dapat bertindak sebagai basis data vektor yang dapat mendukung teknik Retrieval Augmented Generation untuk lab kami.
- Cari "firestore" di bilah pencarian, lalu klik produk Firestore
- Kemudian, klik tombol Buat Database Firestore
- Gunakan (default) sebagai nama ID basis data dan biarkan Edisi Standar dipilih. Untuk demo lab ini, gunakan aturan keamanan Firestore Native dengan Open.
- Anda juga akan melihat bahwa basis data ini sebenarnya memiliki Penggunaan Tingkat Gratis YEAY! Setelah itu, klik tombol Buat Database
Setelah langkah-langkah ini, Anda seharusnya sudah diarahkan ke Database Firestore yang baru saja Anda buat
Menyiapkan Proyek Cloud di Terminal Cloud Shell
- Anda akan menggunakan Cloud Shell, lingkungan baris perintah yang berjalan di Google Cloud yang sudah dimuat sebelumnya dengan bq. Klik Aktifkan Cloud Shell di bagian atas konsol Google Cloud.
- Setelah terhubung ke Cloud Shell, Anda dapat memeriksa bahwa Anda sudah diautentikasi dan project sudah ditetapkan ke project ID Anda menggunakan perintah berikut:
gcloud auth list
- Jalankan perintah berikut di Cloud Shell untuk mengonfirmasi bahwa perintah gcloud mengetahui tentang proyek 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 tersebut dan Anda akan melihat semua project dan project ID di sisi kanan
- Aktifkan API yang diperlukan melalui perintah yang ditunjukkan di bawah ini. Ini mungkin memakan 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 dijalankan, Anda akan melihat pesan seperti yang ditunjukkan 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 selalu dapat mengaktifkannya selama implementasi.
Baca dokumentasi untuk mempelajari perintah gcloud dan penggunaannya.
Siapkan Bucket Penyimpanan Google Cloud
Berikutnya, dari terminal yang sama, kita perlu menyiapkan bucket GCS untuk menyimpan berkas yang diunggah. Jalankan perintah berikut untuk membuat bucket, nama bucket yang unik namun relevan yang relevan dengan tanda terima asisten pengeluaran pribadi akan dibutuhkan, oleh karena itu kami akan menggunakan nama bucket berikut yang dikombinasikan dengan id proyek Anda
gsutil mb -l us-central1 gs://personal-expense-{your-project-id}
Ini akan menampilkan output ini
Creating gs://personal-expense-{your-project-id}
Anda dapat memverifikasi ini dengan membuka Menu Navigasi di kiri atas browser dan pilih Penyimpanan Cloud -> Bucket
Membuat Indeks Firestore untuk Penelusuran
Firestore pada dasarnya adalah basis data NoSQL, yang menawarkan kinerja dan fleksibilitas unggul dalam model data, tetapi memiliki keterbatasan jika menyangkut kueri yang kompleks. Karena kami berencana untuk memanfaatkan beberapa kueri multi-bidang gabungan dan penelusuran vektor, kami perlu membuat beberapa indeks terlebih dahulu. Anda dapat membaca lebih lanjut tentang detailnya di dokumentasi ini
- Jalankan perintah berikut untuk membuat indeks untuk 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 yang ini untuk mendukung pencarian 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 mengunjungi Firestore di konsol cloud dan klik instance database (default) lalu pilih Indeks di bilah navigasi
Buka Editor Cloud Shell dan Siapkan Direktori Kerja Aplikasi
Sekarang, kita dapat mengatur editor kode kita untuk melakukan beberapa hal pengkodean. Kami akan menggunakan Cloud Shell Editor untuk ini
- Klik tombol Buka Editor, ini akan membuka Editor Cloud Shell, kita dapat menulis kode kita di sini
- Selanjutnya, kita juga perlu memeriksa apakah shell sudah dikonfigurasi ke PROJECT ID yang benar yang Anda miliki. Jika Anda melihat ada nilai di dalam ( ) sebelum ikon $ di terminal ( pada screenshot di bawah, nilainya adalah "adk-multimodal-tool"), nilai ini menunjukkan project yang dikonfigurasi untuk sesi shell aktif Anda.
Jika nilai yang ditampilkan sudah benar, Anda dapat melewati perintah berikutnya. Namun jika tidak benar atau hilang, jalankan perintah berikut
gcloud config set project <YOUR_PROJECT_ID>
- Selanjutnya, clone direktori kerja template untuk codelab ini dari GitHub dengan menjalankan perintah berikut. Perintah 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, 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
Pengaturan Lingkungan
Siapkan Lingkungan Virtual Python
Langkah berikutnya adalah menyiapkan lingkungan pengembangan. Terminal aktif Anda saat ini harus berada di dalam direktori kerja personal-expense-assistant. Kami akan menggunakan Python 3.12 dalam codelab ini dan kami akan menggunakan uv python project manager untuk menyederhanakan kebutuhan dalam membuat dan mengelola versi python dan lingkungan virtual.
- Jika Anda belum membuka terminal, buka dengan mengklik Terminal -> Terminal Baru , atau gunakan Ctrl + Shift + C , ini akan membuka jendela terminal di bagian bawah browser
- Sekarang , mari kita lakukan inisialisasi lingkungan virtual menggunakan
uv. Jalankan perintah ini
cd ~/personal-expense-assistant
uv sync --frozen
Tindakan ini akan membuat direktori .venv dan menginstal dependensi. Pratinjau singkat tentang pyproject.toml akan memberi Anda informasi tentang dependensi yang ditampilkan seperti ini
dependencies = [
"datasets>=3.5.0",
"google-adk==1.18",
"google-cloud-firestore>=2.20.1",
"gradio>=5.23.1",
"pydantic>=2.10.6",
"pydantic-settings[yaml]>=2.8.1",
]
Menyiapkan File Konfigurasi
Sekarang kita perlu menyiapkan file konfigurasi untuk project ini. Kita menggunakan pydantic-settings untuk membaca konfigurasi dari file YAML.
Kami telah menyediakan templat file di dalam settings.yaml.example , kami perlu menyalin file tersebut dan mengganti namanya menjadi settings.yaml. Jalankan perintah ini untuk membuat file
cp settings.yaml.example settings.yaml
Kemudian salin nilai berikut ke dalam file
GCLOUD_LOCATION: "us-central1"
GCLOUD_PROJECT_ID: "your-project-id"
BACKEND_URL: "http://localhost:8081/chat"
STORAGE_BUCKET_NAME: "personal-expense-{your-project-id}"
DB_COLLECTION_NAME: "personal-expense-assistant-receipts"
Untuk codelab ini, kita akan menggunakan nilai yang telah dikonfigurasi sebelumnya untuk GCLOUD_LOCATION, BACKEND_URL, dan DB_COLLECTION_NAME .
Sekarang kita dapat melanjutkan ke langkah berikutnya, yaitu membangun agen dan kemudian layanan
3. 🚀 Bangun Agen menggunakan Google ADK dan Gemini 2.5
Pengantar Struktur Direktori ADK
Mari kita mulai dengan mempelajari apa yang ditawarkan ADK dan cara membuat agen. Dokumentasi lengkap ADK dapat diakses di URL ini . ADK menawarkan banyak utilitas dalam eksekusi perintah CLI-nya. Beberapa di antaranya adalah sebagai berikut :
- Menyiapkan struktur direktori agen
- Mencoba interaksi dengan cepat melalui input output 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
Saat ditanya, pilih model gemini-2.5-flash dan backend Vertex AI. Wizard kemudian akan menanyakan ID dan lokasi proyek. Anda dapat menerima pilihan default dengan menekan enter, atau mengubahnya seperlunya. Periksa kembali apakah Anda menggunakan ID proyek yang benar yang dibuat sebelumnya di lab ini. Outputnya akan terlihat seperti ini:
Choose a model for the root agent: 1. gemini-2.5-flash 2. Other models (fill later) Choose model (1, 2): 1 1. Google AI 2. Vertex AI Choose a backend (1, 2): 2 You need an existing Google Cloud account and project, check out this link for details: https://google.github.io/adk-docs/get-started/quickstart/#gemini---google-cloud-vertex-ai Enter Google Cloud project ID [going-multimodal-lab]: Enter Google Cloud region [us-central1]: Agent created in /home/username/personal-expense-assistant/expense_manager_agent: - .env - __init__.py - agent.py
Ini akan membuat struktur direktori agen berikut
expense_manager_agent/ ├── __init__.py ├── .env ├── agent.py
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',
name='root_agent',
description='A helpful assistant for user questions.',
instruction='Answer user questions to the best of your knowledge',
)
Sekarang Anda dapat mengujinya dengan menjalankan
uv run adk run expense_manager_agent
Setiap kali Anda selesai menguji, Anda dapat keluar dari agen dengan mengetik exit atau menekan Ctrl+D.
Membangun Agen Manajer Biaya Kami
Mari kita bangun agen manajer pengeluaran kita! Buka file expense_manager_agent/agent.py dan salin kode di bawah ini 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",
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 inisiasi agen tempat kita menginisialisasi hal-hal berikut:
- Menetapkan model yang akan digunakan ke
gemini-2.5-flash - Siapkan deskripsi dan instruksi agen sebagai perintah sistem yang sedang dibaca dari
task_prompt.md - Menyediakan alat yang diperlukan untuk mendukung fungsionalitas agen
- Aktifkan perencanaan sebelum menghasilkan respons atau eksekusi akhir menggunakan kemampuan berpikir Gemini 2.5 Flash
- Siapkan intersepsi panggilan balik 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:
- Ekstrak data dari gambar tanda terima dan simpan data dan file
- Pencarian tepat pada data pengeluaran
- Pencarian kontekstual pada data pengeluaran
Oleh karena itu kita memerlukan alat yang tepat untuk mendukung fungsi ini. Buat file baru di direktori expense_manager_agent dan beri nama tools.py
touch expense_manager_agent/tools.py
Buka expense_manage_agent/tools.py, lalu salin kode di bawah ini
# 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 implementasi fungsi alat ini, kami merancang alat berdasarkan 2 ide utama berikut:
- Mengurai data tanda terima dan memetakannya 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 Pengenalan Karakter Optik, alat ini akan mengurai informasi yang diperlukan dari data gambar, bersamaan dengan pengenalan string ID Gambar dan memetakannya bersama-sama untuk disimpan dalam basis data Firestore.
Selain itu, alat ini juga mengubah konten tanda terima menjadi penyematan menggunakan text-embedding-004 sehingga semua metadata dan penyematan disimpan dan diindeks bersama. Memungkinkan fleksibilitas untuk diambil baik dengan kueri maupun penelusuran kontekstual.
Setelah berhasil menjalankan alat ini, Anda dapat melihat bahwa data tanda terima sudah diindeks dalam database Firestore seperti yang ditunjukkan di bawah
Alat "search_receipts_by_metadata_filter"
Alat ini mengonversi kueri pengguna menjadi filter kueri metadata yang mendukung penelusuran menurut rentang tanggal dan/atau total transaksi. Ini akan mengembalikan semua data penerimaan yang cocok, di mana dalam prosesnya kami akan membuang bidang 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. Gagasan untuk mengizinkan agen membuat keputusan secara independen, apakah akan menggunakan alat RAG ini atau tidak, dan mendesain kuerinya sendiri adalah salah satu definisi pendekatan RAG Berbasis Agen.
Kami tidak hanya mengizinkannya membuat kuerinya sendiri, tetapi juga mengizinkannya memilih berapa banyak dokumen relevan yang ingin diambil. Dikombinasikan dengan rekayasa cepat 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 alat hebat yang mampu mencari hampir apa saja, meskipun mungkin tidak mengembalikan semua hasil yang diharapkan karena sifat pencarian tetangga terdekat yang tidak tepat.
5. 🚀 Modifikasi Konteks Percakapan melalui Panggilan Balik
Google ADK memungkinkan kita untuk "mencegat" runtime agen di berbagai tingkatan. Anda dapat membaca lebih lanjut tentang kemampuan terperinci ini dalam dokumentasi ini . Di lab ini, kami menggunakan before_model_callback untuk mengubah permintaan sebelum dikirim ke LLM untuk menghapus data gambar dalam konteks riwayat percakapan lama ( hanya menyertakan data gambar dalam 3 interaksi pengguna terakhir) demi efisiensi
Namun, kami tetap ingin agen memiliki konteks data gambar jika diperlukan. Oleh karena itu, kami 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 saat 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..
Selain itu, saat data byte menjadi tidak berlaku lagi dalam histori percakapan, ID string masih ada untuk tetap memungkinkan 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
touch expense_manager_agent/callbacks.py
Buka file expense_manager_agent/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
Mendesain agen dengan interaksi dan kemampuan yang kompleks mengharuskan kita menemukan petunjuk 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 untuk kami. Ini berarti kita perlu menyampaikan semua informasi ini dengan benar dalam struktur prompt yang tepat
Kami akan meminta agen untuk menyusun output ke dalam format penurunan harga 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 terhadap 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, coba berkomunikasi dengan agen melalui CLI, jalankan perintah berikut
uv run adk run expense_manager_agent
Output akan menampilkan seperti ini, tempat Anda dapat melakukan percakapan secara bergiliran dengan agen, 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
Ini akan memunculkan output seperti contoh berikut, berarti kita sudah bisa 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 Cloud Shell Editor Anda, lalu pilih Preview on port 8080
Anda akan melihat halaman web berikut tempat Anda dapat memilih agen yang tersedia pada tombol drop down kiri atas ( dalam kasus kami seharusnya 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! Unggah 2 contoh tanda terima ini ( sumber : Kumpulan data wajah berpelukan mousserlane/id_receipt_dataset) . Klik kanan pada setiap gambar dan pilih Simpan Gambar sebagai.. ( ini akan mengunduh gambar tanda terima), lalu unggah file ke bot dengan mengklik ikon "klip" dan katakan bahwa Anda ingin menyimpan tanda terima ini
Setelah itu coba query berikut untuk melakukan pencarian atau pengambilan file
- "Berikan rincian pengeluaran dan totalnya selama tahun 2023"
- "Give me receipt file from Indomaret" (Beri saya 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 ia mematuhi semua aturan yang diberikan dalam perintah di dalam task_prompt.py. Selamat! Sekarang Anda memiliki agen pengembangan yang berfungsi lengkap.
Sekarang saatnya melengkapinya dengan UI yang tepat dan bagus serta kemampuan untuk mengunggah dan mengunduh file gambar.
8. 🚀 Bangun Layanan Frontend menggunakan Gradio
Kami akan membangun antarmuka web obrolan yang terlihat seperti ini
Berisi antarmuka obrolan dengan kolom masukan bagi pengguna untuk mengirim teks dan mengunggah file gambar tanda terima.
Kami akan membangun layanan frontend menggunakan Gradio.
Buat file baru dan beri nama frontend.py
touch 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 serupa dengan ini di konsol cloud Anda
* 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 Anda ctrl+klik tautan URL lokal. Sebagai alternatif, Anda juga dapat mengakses aplikasi frontend dengan mengklik tombol Pratinjau Web di sisi kanan atas Editor Cloud, lalu pilih Pratinjau pada port 8080
Anda akan melihat antarmuka web, namun Anda akan mendapatkan kesalahan yang diharapkan saat mencoba mengirimkan obrolan karena layanan backend belum disiapkan
Sekarang, biarkan layanannya berjalan dan jangan menghentikannya dulu. Kami akan menjalankan layanan backend di tab terminal lain
Penjelasan Kode
Dalam kode frontend ini, pertama-tama kita memungkinkan pengguna untuk mengirim teks dan mengunggah beberapa file. Gradio memungkinkan kita membuat fungsi semacam ini dengan metode gr.ChatInterface yang dikombinasikan dengan gr.MultimodalTextbox
Sekarang sebelum mengirim berkas dan teks ke backend, kita perlu mencari tahu mimetype berkas sebagaimana dibutuhkan oleh backend. Kita juga perlu mengodekan byte file gambar ke dalam base64 dan mengirimkannya bersama 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 dan backend didefinisikan dalam schema.py. Kami menggunakan Pydantic BaseModel untuk menerapkan validasi data dalam skema
Saat menerima respon, kita sudah memisahkan bagian mana yang merupakan proses berpikir, respon akhir, dan keterikatan. Dengan demikian kita dapat memanfaatkan 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. 🚀 Bangun Layanan Backend menggunakan FastAPI
Berikutnya, kita perlu membangun backend yang dapat menginisialisasi Agen kita bersama dengan komponen lain agar dapat menjalankan agen runtime.
Buat file baru dan beri nama backend.py
touch backend.py
Dan salin kode berikut
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
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 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 await app_context.session_service.get_session(
app_name=APP_NAME, user_id=user_id, session_id=session_id
):
await 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 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. Ingatlah bahwa pada langkah sebelumnya kita menjalankan layanan frontend dengan benar, sekarang kita perlu membuka terminal baru dan mencoba menjalankan layanan backend ini
- Buat terminal baru. Arahkan ke terminal Anda 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 maka akan muncul output 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
Menginisialisasi Agen ADK, SessionService, dan ArtifactService
Untuk menjalankan agen di layanan backend, kita perlu membuat Runner yang mengambil SessionService dan agen kita. SessionService akan mengelola riwayat dan status percakapan, oleh karena itu ketika diintegrasikan dengan Runner, ia akan memberi agen kita kemampuan untuk menerima konteks percakapan yang sedang berlangsung.
Kami juga memanfaatkan ArtifactService untuk menangani berkas yang diunggah. Anda dapat membaca detail lebih lanjut tentang Sesi dan Artefak ADK di sini
...
@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, kami menggunakan InMemorySessionService dan GcsArtifactService untuk diintegrasikan dengan agen kami Runner. Karena riwayat percakapan disimpan dalam memori, riwayat tersebut akan hilang setelah layanan backend dihentikan atau dimulai ulang. Kami menginisialisasinya di dalam siklus hidup aplikasi FastAPI untuk disuntikkan sebagai dependensi dalam rute /chat.
Mengunggah dan Mengunduh Gambar dengan GcsArtifactService
Semua gambar yang diunggah 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 agen runner, perlu diformat ke dalam tipe types.Content. Di dalam fungsi tersebut, kami juga memproses setiap data gambar dan mengekstrak ID-nya untuk digantikan dengan placeholder ID Gambar.
Mekanisme serupa digunakan untuk mengunduh lampiran setelah mengekstrak ID gambar menggunakan regex:
...
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. 🚀 Tes Integrasi
Sekarang, Anda harus menjalankan beberapa layanan di tab konsol cloud yang berbeda:
- Layanan frontend dijalankan pada port 8080
* Running on local URL: http://0.0.0.0:8080 To create a public link, set `share=True` in `launch()`.
- Layanan backend dijalankan pada 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 kondisi saat ini, Anda seharusnya dapat mengunggah gambar tanda terima dan mengobrol lancar dengan asisten dari aplikasi web pada port 8080.
Klik tombol Pratinjau Web di area atas Editor Cloud Shell Anda dan pilih Pratinjau pada port 8080
Sekarang mari kita berinteraksi dengan asisten!
Unduh tanda terima berikut. Rentang tanggal data penerimaan ini berada di antara tahun 2023-2024 dan minta asisten untuk menyimpan/mengunggahnya
- Receipt Drive ( sumber Dataset wajah berpelukan
mousserlane/id_receipt_dataset)
Menanyakan berbagai hal
- "Berikan saya rincian pengeluaran bulanan selama tahun 2023-2024"
- "Tunjukkan struk transaksi kopi"
- "Berikan saya file struk 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 mengemas aplikasi ini dan men-deploy-nya ke Cloud Run. Untuk tujuan demo ini, layanan ini akan diekspos sebagai layanan publik yang dapat diakses oleh orang lain. Namun, perlu diingat bahwa ini bukan praktik terbaik untuk jenis aplikasi ini karena lebih cocok untuk aplikasi pribadi
Dalam codelab ini, kita akan menempatkan layanan frontend dan backend dalam 1 container. Kita akan memerlukan bantuan supervisord untuk mengelola kedua layanan. Anda dapat memeriksa file supervisord.conf dan memeriksa Dockerfile yang menetapkan supervisord sebagai entrypoint.
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 menyetel project ID:
gcloud config set project [PROJECT_ID]
Kemudian, jalankan perintah berikut untuk menyebarkannya 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 tanpa autentikasi di sini karena ini adalah aplikasi demo. Sebaiknya gunakan autentikasi yang sesuai untuk aplikasi produksi dan perusahaan Anda.
Setelah deployment selesai, Anda akan mendapatkan link yang mirip dengan di bawah ini:
https://personal-expense-assistant-*******.us-central1.run.app
Lanjutkan penggunaan aplikasi Anda dari jendela Samaran atau perangkat seluler Anda. Fitur ini seharusnya sudah aktif.
12. 🎯 Tantangan
Sekaranglah saatnya Anda untuk mengasah dan mengasah keterampilan eksplorasi Anda. Apakah Anda memiliki kemampuan untuk mengubah kode sehingga backend dapat menampung banyak pengguna? Komponen apa yang perlu diperbarui?
13. 🧹 Bersihkan
Untuk menghindari tagihan ke akun Google Cloud Anda atas sumber daya yang digunakan dalam codelab ini, ikuti langkah-langkah berikut:
- Di konsol Google Cloud, buka halaman Kelola sumber daya.
- Dalam daftar project, pilih project yang ingin Anda hapus, lalu klik Delete.
- Dalam dialog, ketik ID proyek, lalu klik Matikan untuk menghapus proyek.
- Atau, Anda dapat membuka Cloud Run di konsol, memilih layanan yang baru saja Anda deploy, lalu menghapusnya.