1. Giriş
Kişisel harcamalarınız konusunda bunaldığınız ve bunları yönetmek için yeterince tembel olduğunuz oldu mu? Ben de. Bu nedenle, bu kod laboratuvarında, tüm işleri bizim için yapacak Gemini 2.5 destekli bir kişisel harcama yöneticisi asistanı oluşturacağız. Yüklenen makbuzları yönetmekten, kahve için çok fazla para harcayıp harcamadığınızı analiz etmeye kadar birçok işlemi yapabilirsiniz.
Bu asistana web tarayıcısı üzerinden sohbet web arayüzü şeklinde erişilebilir. Bu arayüzde asistanla iletişim kurabilir, makbuz resimleri yükleyebilir ve asistandan bunları saklamasını isteyebilir ya da dosyayı almak ve harcama analizi yapmak için bazı makbuzları arayabilirsiniz. Tüm bunlar Google Agent Development Kit çerçevesinin üzerine inşa edilmiştir.
Uygulamanın kendisi 2 hizmete ayrılmıştır: ön uç ve arka uç. Bu sayede hızlı bir prototip oluşturabilir, nasıl bir deneyim sunduğunu deneyebilir ve API sözleşmesinin ikisini de entegre etmek için nasıl göründüğünü anlayabilirsiniz.
Codelab'de aşağıdaki gibi adım adım bir yaklaşım uygulayacaksınız:
- Google Cloud projenizi hazırlayın ve projenizdeki gerekli tüm API'leri etkinleştirin
- Google Cloud Storage'da paket ve Firestore'da veritabanı oluşturma
- Firestore dizine ekleme oluşturma
- Kodlama ortamınız için çalışma alanı oluşturma
- ADK aracısı kaynak kodunu, araçlarını, istemlerini vb. yapılandırma
- ADK yerel web geliştirme kullanıcı arayüzünü kullanarak aracıyı test etme
- Sorgu göndermek ve makbuz resimleri yüklemek için Gradio kitaplığını kullanarak kullanıcı arayüzü hizmetini - sohbet arayüzünü oluşturun
- ADK aracısı kodumuz, SessionService ve Artifact Service'in bulunduğu FastAPI'yi kullanarak arka uç hizmetini (HTTP sunucusu) oluşturun.
- Uygulamayı Cloud Run'a dağıtmak için gereken ortam değişkenlerini yönetin ve gerekli dosyaları ayarlayın
- Uygulamayı Cloud Run'a dağıtma
Mimariye Genel Bakış
Ön koşullar
- Python ile rahatça çalışabilme
- HTTP hizmetini kullanan temel tam yığın mimarisini anlama
Neler öğreneceksiniz?
- Gradio ile ön uç web prototipi oluşturma
- FastAPI ve Pydantic ile arka uç hizmeti geliştirme
- ADK aracının çeşitli özelliklerini kullanırken ADK aracının mimarisini oluşturma
- Araç kullanımı
- Oturum ve Yapı Yönetimi
- Gemini'ye gönderilmeden önce giriş değişikliği için geri çağırma kullanımı
- Planlama yaparak görev yürütmeyi iyileştirmek için BuiltInPlanner'ı kullanma
- ADK yerel web arayüzü üzerinden hızlı hata ayıklama
- ADK geri çağırma işlevi kullanılarak istem mühendisliği ve Gemini istek değişikliği aracılığıyla bilgi ayrıştırma ve alma yoluyla çoklu formatlı etkileşimi optimize etme stratejisi
- Vektör veritabanı olarak Firestore'u kullanan vekil getirmeyle desteklenen artırılmış oluşturma
- Pydantic-settings ile YAML dosyasındaki ortam değişkenlerini yönetme
- Dockerfile kullanarak uygulamayı Cloud Run'a dağıtma ve YAML dosyası ile ortam değişkenleri sağlama
Gerekenler
- Chrome web tarayıcısı
- Gmail hesabı
- Faturalandırmanın etkin olduğu bir Cloud projesi
Her seviyeden geliştirici (yeni başlayanlar dahil) için tasarlanan bu kod laboratuvarının örnek uygulamasında Python kullanılır. Ancak sunulan kavramları anlamak için Python bilgisine sahip olmanız gerekmez.
2. Başlamadan önce
Cloud Console'da Etkin Projeyi Seçme
Bu kod laboratuvarında, faturalandırmanın etkin olduğu bir Google Cloud projeniz olduğu varsayılmaktadır. Henüz yoksa başlamak için aşağıdaki talimatları uygulayabilirsiniz.
- Google Cloud Console'daki proje seçici sayfasında bir Google Cloud projesi seçin veya oluşturun.
- Cloud projeniz için faturalandırmanın etkinleştirildiğinden emin olun. Projede faturalandırmanın etkin olup olmadığını nasıl kontrol edeceğinizi öğrenin.
Firestore veritabanını hazırlama
Ardından, bir Firestore veritabanı da oluşturmamız gerekir. Yerel moddaki Firestore, otomatik ölçeklendirme, yüksek performans ve uygulama geliştirme kolaylığı için oluşturulmuş bir NoSQL belge veritabanıdır. Ayrıca, laboratuvarımız için Retrieval Augmented Generation tekniğini destekleyebilecek bir vektör veritabanı olarak da kullanılabilir.
- Arama çubuğunda "firestore" ifadesini arayın ve Firestore ürününü tıklayın.
- Ardından Firestore Veritabanı Oluştur düğmesini tıklayın.
- Veritabanı kimliği adı olarak (varsayılan) seçeneğini kullanın ve Standart Sürüm'ü seçili bırakın. Bu laboratuvar denemesi için Açık güvenlik kurallarıyla Firestore Native'ı kullanın.
- Ayrıca bu veritabanında Ücretsiz Katman Kullanımı YEAY! ifadesinin yer aldığını da göreceksiniz. Ardından Veritabanı Oluştur Düğmesi'ni tıklayın.
Bu adımlardan sonra, yeni oluşturduğunuz Firestore veritabanına yönlendirilirsiniz.
Cloud Shell Terminal'de Cloud projesi oluşturma
- Google Cloud'da çalışan ve bq ile önceden yüklenmiş bir komut satırı ortamı olan Cloud Shell'i kullanacaksınız. Google Cloud Console'un üst kısmından Cloud Shell'i etkinleştir'i tıklayın.
- Cloud Shell'e bağlandıktan sonra aşağıdaki komutu kullanarak kimliğinizin doğrulanıp doğrulanmadığını ve projenin proje kimliğinize ayarlanıp ayarlanmadığını kontrol edin:
gcloud auth list
- gcloud komutunun projeniz hakkında bilgi sahibi olduğunu onaylamak için Cloud Shell'de aşağıdaki komutu çalıştırın.
gcloud config list project
- Projeniz ayarlanmadıysa ayarlamak için aşağıdaki komutu kullanın:
gcloud config set project <YOUR_PROJECT_ID>
Alternatif olarak, PROJECT_ID kimliğini konsolda da görebilirsiniz.
Bu düğmeyi tıkladığınızda projenizin tamamı ve proje kimliği sağ tarafta gösterilir.
- Aşağıda gösterilen komutu kullanarak gerekli API'leri etkinleştirin. Bu işlem birkaç dakika sürebilir. Lütfen bekleyin.
gcloud services enable aiplatform.googleapis.com \
firestore.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
cloudresourcemanager.googleapis.com
Komut başarıyla yürütüldüğünde aşağıdakine benzer bir mesaj görürsünüz:
Operation "operations/..." finished successfully.
gcloud komutunun alternatifi, her ürünü arayarak veya bu bağlantıyı kullanarak konsoldan geçmektir.
Atlanan bir API varsa uygulama sırasında istediğiniz zaman etkinleştirebilirsiniz.
gcloud komutları ve kullanımı için belgelere bakın.
Google Cloud Storage paketini hazırlama
Ardından, aynı terminalden, yüklenen dosyayı depolamak için GCS paketini hazırlamamız gerekir. Paketi oluşturmak için aşağıdaki komutu çalıştırın
gsutil mb -l us-central1 gs://personal-expense-assistant-receipts
Bu çıkışı gösterir
Creating gs://personal-expense-assistant-receipts/...
Bunu doğrulamak için tarayıcının sol üst kısmındaki Gezinme menüsüne gidip Cloud Storage -> Paket'i seçin.
Arama için Firestore dizini oluşturma
Firestore, veri modelinde üstün performans ve esneklik sunan, doğal olarak NoSQL bir veritabanıdır ancak karmaşık sorgular söz konusu olduğunda sınırlamaları vardır. Bazı bileşik çok alan sorguları ve vektör araması kullanmayı planladığımız için önce bazı dizinler oluşturmamız gerekir. Ayrıntılar hakkında daha fazla bilgiyi bu dokümanda bulabilirsiniz.
- Bileşik sorguları desteklemek için dizin oluşturmak üzere aşağıdaki komutu çalıştırın
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)"
- Vektör aramayı desteklemek için bunu çalıştırın.
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)"
Oluşturulan dizini kontrol etmek için Cloud Console'da Firestore'u ziyaret edin, (varsayılan) veritabanı örneğini tıklayın ve gezinme çubuğundan Dizinler'i seçin.
Cloud Shell Düzenleyici'ye gidin ve uygulama çalışma dizini ayarlayın
Şimdi kod düzenleyicimizi kodlama işlemleri yapmak için ayarlayabiliriz. Bunun için Cloud Shell Düzenleyici'yi kullanacağız.
- Düzenleyiciyi aç düğmesini tıklayın. Bu işlemle Cloud Shell Düzenleyici açılır. Kodumuzu buraya yazabiliriz
- Cloud Code projesinin, Cloud Shell düzenleyicisinin sol alt köşesinde (durum çubuğu) aşağıdaki resimde vurgulandığı gibi ayarlandığından ve faturalandırmanın etkinleştirildiği etkin Google Cloud projesine ayarlandığından emin olun. İstenirse Yetkilendir'i tıklayın. Önceki komutu zaten uyguladıysanız düğme, oturum açma düğmesi yerine doğrudan etkinleştirilen projenize de yönlendirebilir.
- Ardından, bu kod laboratuvarının şablon çalışma dizinini GitHub'dan klonlayalım. Aşağıdaki komutu çalıştırın. Çalışma dizini, personal-expense-assistant dizininde oluşturulur.
git clone https://github.com/alphinside/personal-expense-assistant-adk-codelab-starter.git personal-expense-assistant
- Ardından Cloud Shell Düzenleyici'nin üst bölümüne gidin ve Dosya->Klasör Aç'ı tıklayın, kullanıcı adınız dizinini ve kişisel-harcama-asistanı dizinini bulun,ardından Tamam düğmesini tıklayın. Bu işlem, seçilen dizini ana çalışma dizini yapar. Bu örnekte kullanıcı adı alvinprayuda olduğundan dizin yolu aşağıda gösterilmiştir.
Cloud Shell Düzenleyiciniz şu şekilde görünecektir:
Ortam Kurulumu
Python sanal ortamını hazırlama
Sonraki adım, geliştirme ortamını hazırlamaktır. Mevcut etkin terminaliniz personal-expense-assistant çalışma dizininde olmalıdır. Bu kod laboratuvarında Python 3.12'yi kullanacağız ve Python sürümü ile sanal ortamı oluşturma ve yönetme ihtiyacını basitleştirmek için uv python proje yöneticisini kullanacağız.
- Terminali henüz açmadıysanız Terminal -> Yeni Terminal'i tıklayarak veya Ctrl + Üst Karakter + C tuşlarına basarak açın. Bu işlem, tarayıcının alt kısmında bir terminal penceresi açar.
uv'yi indirin ve aşağıdaki komutla python 3.12'yi yükleyin
curl -LsSf https://astral.sh/uv/0.6.16/install.sh | sh && \
source $HOME/.local/bin/env && \
uv python install 3.12
- Şimdi
uvkullanarak sanal ortamı başlatalım. Bu komutu çalıştırın.
uv sync --frozen
Bu işlem, .venv dizinini oluşturur ve bağımlılıkları yükler. pyproject.toml dosyasını hızlıca gözden geçirerek aşağıdaki gibi gösterilen bağımlılıklarla ilgili bilgi edinebilirsiniz.
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",
]
- Sanal ortamı test etmek için main.py adlı yeni bir dosya oluşturun ve aşağıdaki kodu kopyalayın
def main():
print("Hello from personal-expense-assistant-adk!")
if __name__ == "__main__":
main()
- Ardından aşağıdaki komutu çalıştırın
uv run main.py
Aşağıda gösterildiği gibi bir çıkış alırsınız.
Using CPython 3.12 Creating virtual environment at: .venv Hello from personal-expense-assistant-adk!
Bu, python projesinin doğru şekilde ayarlandığını gösterir.
Yapılandırma dosyalarını ayarlama
Şimdi bu proje için yapılandırma dosyaları oluşturmamız gerekiyor. YAML dosyasından yapılandırmayı okumak için pydantic-settings'i kullanırız.
Aşağıdaki yapılandırmayı içeren settings.yaml adlı bir dosya oluşturun. Dosya->Yeni Metin Dosyası'nı tıklayın ve dosyayı aşağıdaki kodla doldurun. Ardından dosyayı settings.yaml olarak kaydedin.
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"
Bu codelab'de GCLOUD_LOCATION, BACKEND_URL, STORAGE_BUCKET_NAME, DB_COLLECTION_NAME ve BACKEND_URL için önceden yapılandırılmış değerleri kullanacağız .
Şimdi bir sonraki adıma, aracıyı ve ardından hizmetleri oluşturmaya geçebiliriz.
3. Google ADK ve Gemini 2.5'i kullanarak temsilciyi oluşturma
ADK Dizin Yapısına Giriş
ADK'nın sunduğu avantajları ve aracıyı nasıl oluşturacağınızı inceleyerek başlayalım. ADK'nın tam dokümanına bu URL üzerinden erişebilirsiniz . ADK, KSA komut yürütme özelliğiyle bize birçok yardımcı program sunar. Bunlardan bazıları şunlardır :
- Temsilci dizin yapısını ayarlama
- KSA giriş çıkışı üzerinden etkileşimi hızlıca deneme
- Yerel geliştirme kullanıcı arayüzü web arayüzünü hızlıca ayarlama
Şimdi, CLI komutunu kullanarak temsilci dizini yapısını oluşturalım. Aşağıdaki komutu çalıştırın
uv run adk create expense_manager_agent \
--model gemini-2.5-flash-preview-04-17 \
--project {your-project-id} \
--region us-central1
Aşağıdaki aracı dizini yapısı oluşturulur:
expense_manager_agent/ ├── __init__.py ├── .env ├── agent.py
init.py ve agent.py dosyalarını incelerseniz bu kodu görürsünüz.
# __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',
)
Gider Yöneticisi Temsilcimizi Oluşturma
Harcama yöneticisi temsilcimizi oluşturalım. expense_manager_agent/agent.py dosyasını açın ve root_agent öğesini içeren aşağıdaki kodu kopyalayın.
# 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,
)
Kod Açıklaması
Bu komut dosyası, aşağıdakileri başlattığımız temsilci başlatma işlemimizi içerir:
- Kullanılacak modeli
gemini-2.5-flash-preview-04-17olarak ayarlayın - Temsilci açıklamasını ve talimatlarını,
task_prompt.mdkaynağından okunan sistem istemi olarak ayarlayın - Temsilci işlevini desteklemek için gerekli araçları sağlayın
- Gemini 2.5 Flash Thinking özelliklerini kullanarak nihai yanıtı veya yürütmeyi oluşturmadan önce planlamayı etkinleştirme
- Tahmin yapmadan önce gönderilen görüntü verilerinin sayısını sınırlamak için Gemini'ye istek göndermeden önce geri çağırma işlemini kesintiye uğratmayı ayarlama
4. Temsilci araçlarını yapılandırma
Harcama yöneticisi temsilcimiz aşağıdaki işlemleri yapabilir:
- Makbuz resminden veri ayıklama ve verileri ve dosyayı depolama
- Harcama verilerinde tam arama
- Harcama verilerinde bağlama dayalı arama
Bu nedenle, bu işlevi desteklemek için uygun araçlara ihtiyacımız var. expense_manager_agent dizininde yeni bir dosya oluşturun ve dosyayı tools.py olarak adlandırın. Ardından aşağıdaki kodu kopyalayın.
# 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
Kod Açıklaması
Bu araç işlevini uygularken araçları şu 2 ana fikir etrafında tasarladık:
- Resim kimliği dize yer tutucusu
[IMAGE-ID <hash-of-image-1>]kullanarak makbuz verilerini ayrıştırma ve orijinal dosyayla eşleme - Firestore veritabanını kullanarak veri depolama ve alma
"store_receipt_data" aracı
Bu araç, optik karakter tanıma aracıdır. Resim kimliği dizesini tanıyarak resim verilerinden gerekli bilgileri ayrıştırır ve Firestore veritabanında depolanacak şekilde birlikte eşler.
Ayrıca bu araç, tüm meta verilerin ve yerleştirilmenin birlikte depolanması ve dizine eklenmesi için makbuzun içeriğini text-embedding-004 kullanarak yerleştirilmeye dönüştürür. Sorgu veya bağlamsal aramayla alınabilecek esneklik sağlar.
Bu aracı başarıyla çalıştırdıktan sonra, makbuz verilerinin Firestore veritabanında aşağıdaki gibi dizine eklendiğini görebilirsiniz.
"search_receipts_by_metadata_filter" aracı
Bu araç, kullanıcı sorgusunu tarih aralığına ve/veya toplam işleme göre aramayı destekleyen bir meta veri sorgu filtresine dönüştürür. Eşleşen tüm makbuz verilerini döndürür. Bu süreçte, müşteri temsilcisinin bağlamsal anlayış için ihtiyaç duymadığından yerleştirme alanını kaldırırız.
"search_relevant_receipts_by_natural_language_query" aracı
Bu, Almayla Artırılmış Üretim (RAG) aracımızdır. Temsilcimiz, vektör veritabanından alakalı makbuzları almak için kendi sorgusunu tasarlayabilir ve bu aracı ne zaman kullanacağını da seçebilir. Temsilcinin bu RAG aracını kullanıp kullanmayacağı ve kendi sorgusunu tasarlayacağı konusunda bağımsız karar vermesine izin verme fikri, temsilci RAG yaklaşımının tanımlarından biridir.
Yalnızca kendi sorgusunu oluşturmasına değil, aynı zamanda almak istediği alakalı doküman sayısını seçmesine de izin veriyoruz. Uygun bir istem mühendisliğiyle birlikte kullanıldığında ör.
# Example prompt Always filter the result from tool search_relevant_receipts_by_natural_language_query as the returned result may contain irrelevant information
Bu, aracı neredeyse her şeyi arayabilecek güçlü bir araç haline getirir. Ancak en yakın komşu aramasının tam olmaması nedeniyle beklenen tüm sonuçları döndürmeyebilir.
5. Geri Çağırma İşlemi ile Sohbet Bağlamında Değişiklik Yapma
Google ADK, aracı çalışma süresini çeşitli düzeylerde "kesmemize" olanak tanır. Bu ayrıntılı özellik hakkında daha fazla bilgiyi bu dokümanda bulabilirsiniz . Bu laboratuvarda, verimlilik sağlamak için eski sohbet geçmişi bağlamındaki resim verilerini kaldırmak ( yalnızca son 3 kullanıcı etkileşimindeki resim verilerini dahil etmek) amacıyla isteği LLM'ye gönderilmeden önce değiştirmek için before_model_callback işlevini kullanırız.
Ancak temsilcinin gerektiğinde görüntü verisi bağlamına sahip olmasını isteriz. Bu nedenle, sohbetteki her resim bayt verisinin ardından bir dize resim kimliği yer tutucu eklemek için bir mekanizma ekleriz. Bu, temsilcinin resim kimliğini gerçek dosya verilerine bağlamasına yardımcı olur. Bu veriler hem resim saklama hem de alma sırasında kullanılabilir. Yapı şu şekilde görünür
<image-byte-data-1> [IMAGE-ID <hash-of-image-1>] <image-byte-data-2> [IMAGE-ID <hash-of-image-2>] And so on..
Görüşme geçmişinde bayt verileri geçersiz hale geldiğinde, araç kullanımıyla veri erişimini etkinleştirmek için dize tanımlayıcısı hâlâ mevcuttur. Resim verileri kaldırıldıktan sonra örnek geçmiş yapısı
[IMAGE-ID <hash-of-image-1>] [IMAGE-ID <hash-of-image-2>] And so on..
Başlayalım! expense_manager_agent dizininde callbacks.py adlı yeni bir dosya oluşturun ve aşağıdaki kodu kopyalayın
# 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. İstem
Karmaşık etkileşim ve özelliklere sahip bir temsilci tasarlamak için, temsilciye rehberlik edecek ve istediğimiz şekilde davranmasını sağlayacak yeterince iyi bir istem bulmamız gerekir.
Daha önce, sohbet geçmişindeki resim verilerinin nasıl işleneceğine dair bir mekanizmamız vardı ve kullanımı kolay olmayabilecek araçlara da sahiptik. Örneğin, search_relevant_receipts_by_natural_language_query. Ayrıca müşteri temsilcisinin doğru makbuz resmini bulup bize iletebilmesini istiyoruz. Bu nedenle, tüm bu bilgileri uygun bir istem yapısında doğru şekilde iletmemiz gerekiyor.
Düşünme sürecini, nihai yanıtı ve eki ( varsa) ayrıştırmak için müşteri temsilcisinden çıktıyı aşağıdaki markdown biçiminde yapılandırmasını isteyeceğiz.
# 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>]",
...
]
}
Harcama yöneticisi temsilcisinin davranışıyla ilgili ilk beklentimizi karşılamak için aşağıdaki istemle başlayalım. task_prompt.md dosyası mevcut çalışma dizinimizde zaten mevcuttur ancak expense_manager_agent dizinine taşımamız gerekir. Taşımak için aşağıdaki komutu çalıştırın
mv task_prompt.md expense_manager_agent/task_prompt.md
7. Temsilciyi test etme
Şimdi, aracıyla CLI üzerinden iletişim kurmayı deneyelim. Aşağıdaki komutu çalıştırın.
uv run adk run expense_manager_agent
Bu arayüzde, müşteri temsilcisiyle sırayla sohbet edebileceğiniz aşağıdaki gibi bir çıkış gösterilir. Ancak bu arayüz üzerinden yalnızca metin gönderebilirsiniz.
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:
ADK artık CLI etkileşiminin yanı sıra etkileşimde bulunmak ve etkileşim sırasında neler olduğunu incelemek için bir geliştirme kullanıcı arayüzüne de sahip olmamızı sağlıyor. Yerel geliştirme kullanıcı arayüzü sunucusunu başlatmak için aşağıdaki komutu çalıştırın
uv run adk web --port 8080
Aşağıdaki örnekte gösterildiği gibi bir çıktı oluşturur. Bu, web arayüzüne zaten erişebildiğimiz anlamına gelir.
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)
Şimdi, bunu kontrol etmek için Cloud Shell düzenleyicinizin üst kısmındaki Web Önizlemesi düğmesini tıklayın ve 8080 bağlantı noktasında önizle'yi seçin.
Sol üstteki açılır menü düğmesinden müsait temsilcileri seçebileceğiniz ( bizim durumumuzda expense_manager_agent olmalıdır) ve botla etkileşim kurabileceğiniz aşağıdaki web sayfasını görürsünüz. Sol pencerede, aracı çalışırken günlük ayrıntıları hakkında birçok bilgi görürsünüz.
Bazı işlemleri deneyelim. Aşağıdaki 2 örnek makbuzu ( kaynak : Hugging Face veri kümeleri mousserlane/id_receipt_dataset) yükleyin. Her bir resmi sağ tıklayıp Resmi farklı kaydet'i seçin. ( Bu işlem makbuz resmini indirir.) Ardından "klip" simgesini tıklayarak dosyayı bota yükleyin ve bu makbuzları saklamak istediğinizi söyleyin.
Ardından, arama veya dosya getirme işlemi yapmak için aşağıdaki sorguları deneyin
- "2023'teki harcamaların dökümü ve toplam tutarı verin"
- "İndomaret'ten makbuz dosyasını gönder"
Bazı araçları kullanırken geliştirme kullanıcı arayüzünde neler olduğunu inceleyebilirsiniz.
Temsilcinin size nasıl yanıt verdiğini görün ve task_prompt.py dosyasında istemde verilen tüm kurallara uyup uymadığını kontrol edin. Tebrikler! Artık çalışan bir geliştirme aracınız var.
Şimdi, resim dosyasını yükleyip indirme özellikleriyle birlikte uygun ve güzel bir kullanıcı arayüzüyle tamamlamanın zamanı geldi.
8. Gradio'yu kullanarak ön uç hizmeti oluşturma
Aşağıdaki gibi görünen bir sohbet web arayüzü oluşturacağız.
Kullanıcıların metin gönderip makbuz resim dosyalarını yükleyebileceği bir giriş alanı içeren bir sohbet arayüzü içerir.
Ön uç hizmetini Gradio'yu kullanarak oluşturacağız.
Yeni bir dosya oluşturun, Dosya->Yeni Metin Dosyası'nı tıklayın ve dosyayı frontend.py olarak adlandırın. Ardından aşağıdaki kodu kopyalayıp kaydedin.
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,
)
Ardından, ön uç hizmetini aşağıdaki komutla çalıştırmayı deneyebiliriz. main.py dosyasını frontend.py olarak yeniden adlandırmayı unutmayın.
uv run frontend.py
Cloud Console'da buna benzer bir çıkış görürsünüz.
* Running on local URL: http://0.0.0.0:8080 To create a public link, set `share=True` in `launch()`.
Ardından, yerel URL bağlantısını ctrl+tıkladığınızda web arayüzünü kontrol edebilirsiniz. Alternatif olarak, Cloud Düzenleyici'nin sağ üst tarafındaki Web Önizlemesi düğmesini tıklayıp 8080 bağlantı noktasında önizle'yi seçerek de kullanıcı arayüzü uygulamasına erişebilirsiniz.
Web arayüzünü görürsünüz ancak henüz ayarlanmamış arka uç hizmeti nedeniyle sohbet göndermeye çalışırken beklenen hata alırsınız.
Şimdi hizmetin çalışmasını bekleyin ve henüz sonlandırmayın. Arka uç hizmetini başka bir terminal sekmesinde çalıştıracağız.
Kod Açıklaması
Bu kullanıcı arayüzü kodunda, önce kullanıcının metin göndermesini ve birden fazla dosya yüklemesini sağlarız. Gradio, gr.MultimodalTextbox ile birlikte gr.ChatInterface yöntemini kullanarak bu tür işlevler oluşturmamıza olanak tanır.
Şimdi, dosyayı ve metni arka uca göndermeden önce, arka uç tarafından ihtiyaç duyulduğu için dosyanın mimetype'sini bulmamız gerekiyor. Ayrıca resim dosyası baytını base64 olarak kodlayıp mimetype ile birlikte göndermemiz gerekir.
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
Ön uç - arka uç etkileşimi için kullanılan şema schema.py dosyasında tanımlanır. Şemada veri doğrulamasını zorunlu kılmak için Pydantic BaseModel'den yararlanırız.
Yanıtı aldığımızda, hangi kısmın düşünme süreci, nihai yanıt ve ek olduğunu zaten ayırırız. Bu nedenle, her bileşeni kullanıcı arayüzü bileşeniyle görüntülemek için Gradio bileşenini kullanabiliriz.
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. FastAPI'yi kullanarak arka uç hizmeti oluşturma
Ardından, temsilci çalışma zamanını çalıştırabilmek için temsilcimizi diğer bileşenlerle birlikte başlatabilecek arka uç oluşturmamız gerekir.
Yeni bir dosya oluşturun, Dosya->Yeni Metin Dosyası'nı tıklayın ve aşağıdaki kodu kopyalayıp yapıştırın,ardından backend.py olarak kaydedin.
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)
Ardından arka uç hizmetini çalıştırmayı deneyebiliriz. Önceki adımda ön uç hizmetini doğru şekilde çalıştırdığımızı unutmayın. Şimdi yeni bir terminal açmamız ve bu arka uç hizmetini çalıştırmaya çalışmamız gerekiyor.
- Yeni bir terminal oluşturun. Alt kısımdaki terminalinize gidin ve yeni bir terminal oluşturmak için "+" düğmesini bulun. Alternatif olarak, yeni bir terminal açmak için Ctrl + Üst Karakter + C tuşlarına da basabilirsiniz.
- Ardından, personal-expense-assistant çalışma dizininde olduğunuzdan emin olun ve aşağıdaki komutu çalıştırın
uv run backend.py
- İşlem başarılı olursa aşağıdaki gibi bir çıkış gösterilir.
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)
Kod Açıklaması
ADK aracısı, SessionService ve ArtifactService'i başlatma
Temsilciyi arka uç hizmetinde çalıştırmak için hem SessionService'i hem de temsilcimizi alan bir Runner oluşturmamız gerekir. SessionService, sohbet geçmişini ve durumunu yönetir. Bu nedenle, Runner ile entegre edildiğinde temsilcimize devam eden sohbetlerin bağlamını alma olanağı sunar.
Yüklenen dosyayı işlemek için ArtifactService'i de kullanırız. ADK Oturumu ve Öğeler hakkında daha fazla bilgiyi burada bulabilirsiniz.
...
@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
...
Bu demoda, aracımız Runner ile entegrasyon için InMemorySessionService ve GcsArtifactService'i kullanıyoruz. Görüşme geçmişi bellekte depolandığından, arka uç hizmeti kapatıldığında veya yeniden başlatıldığında kaybolur. Bunları, /chat rotasına bağımlılık olarak eklenmek üzere FastAPI uygulama yaşam döngüsü içinde başlatırız.
GcsArtifactService ile resim yükleme ve indirme
Yüklenen tüm resimler GcsArtifactService tarafından yapı olarak depolanır. Bunu utils.py içindeki format_user_request_to_adk_content_and_store_artifacts işlevinde kontrol edebilirsiniz.
...
# 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,
)
...
Agent Runner tarafından işlenecek tüm isteklerin types.Content türüne göre biçimlendirilmesi gerekir. İşlevin içinde, her resim verisini de işler ve kimliğini, bir resim kimliği yer tutucusuyla değiştirilmek üzere ayıklıyoruz.
Resim kimlikleri normal ifade kullanılarak ayıklandıktan sonra ekleri indirmek için benzer bir mekanizma kullanılır:
...
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. Entegrasyon Testi
Artık farklı Cloud Console sekmelerinde birden fazla hizmet çalıştırmış olmanız gerekir:
- 8080 bağlantı noktasında çalışan ön uç hizmeti
* Running on local URL: http://0.0.0.0:8080 To create a public link, set `share=True` in `launch()`.
- Arka uç hizmeti 8081 numaralı bağlantı noktasında çalışır.
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)
Şu anda, makbuz resimlerinizi yükleyebilir ve 8080 bağlantı noktasındaki web uygulamasından asistanla sorunsuz bir şekilde sohbet edebilirsiniz.
Cloud Shell düzenleyicinizin üst kısmındaki Web Önizlemesi düğmesini tıklayın ve 8080 bağlantı noktasında önizle'yi seçin.
Şimdi Asistan ile etkileşime geçelim.
Aşağıdaki makbuzları indirin. Bu makbuz verilerinin tarih aralığı 2023-2024 yılları arasındadır ve asistandan bu verileri depolamasını/yüklemesini isteyin.
- Receipt Drive ( kaynak Hugging Face veri kümeleri
mousserlane/id_receipt_dataset)
Çeşitli sorular sorabilirsiniz.
- "2023-2024 arasındaki aylık harcama dökümünü göster"
- "Kahve işleminin makbuzunu göster"
- "Yakiniku Like'ın makbuz dosyasını gönder"
- Vb.
Aşağıda başarılı bir etkileşim snippet'i verilmiştir.
11. Cloud Run'a dağıtma
Elbette bu muhteşem uygulamaya her yerden erişmek isteriz. Bunun için bu uygulamayı paketleyip Cloud Run'a dağıtabiliriz. Bu demo için bu hizmet, diğer kullanıcıların erişebileceği herkese açık bir hizmet olarak gösterilecektir. Ancak bu, kişisel uygulamalar için daha uygun olduğundan bu tür uygulamalarda en iyi uygulama olmadığını unutmayın.
Bu codelab'de hem ön uç hem de arka uç hizmetini 1 kapsayıcıya yerleştireceğiz. Her iki hizmeti de yönetmek için gözetimli kullanıcının yardımına ihtiyacımız var. supervisord.conf dosyasını inceleyebilir ve supervisord'ı giriş noktası olarak ayarladığımız Dockerfile dosyasını kontrol edebilirsiniz.
Bu aşamada, uygulamalarımızı Cloud Run'a dağıtmak için gereken tüm dosyalara sahibiz. Şimdi dağıtalım. Cloud Shell Terminal'e gidin ve mevcut projenin etkin projenizle yapılandırıldığından emin olun. Aksi takdirde, proje kimliğini ayarlamak için gcloud configure komutunu kullanmanız gerekir:
gcloud config set project [PROJECT_ID]
Ardından, Cloud Run'a dağıtmak için aşağıdaki komutu çalıştırın.
gcloud run deploy personal-expense-assistant \
--source . \
--port=8080 \
--allow-unauthenticated \
--env-vars-file=settings.yaml \
--memory 1024Mi \
--region us-central1
Docker deposu için bir yapı kayıt defteri oluşturulduğunu onaylamanız istenirse E yanıtını vermeniz yeterlidir. Bu bir demo uygulama olduğu için kimlik doğrulaması yapılmayan erişime izin verildiğini unutmayın. Kuruluş ve üretim uygulamalarınızda uygun kimlik doğrulamayı kullanmanız önerilir.
Dağıtım tamamlandığında aşağıdakine benzer bir bağlantı alırsınız:
https://personal-expense-assistant-*******.us-central1.run.app
Uygulamanızı gizli pencereden veya mobil cihazınızdan kullanmaya devam edebilirsiniz. Bu özellik zaten kullanıma sunulmuştur.
12. Zorluk
Şimdi keşif becerilerinizi geliştirme ve öne çıkma zamanı. Arka uçta birden fazla kullanıcının bulunabilmesi için kodu değiştirmek için gerekenlere sahip misiniz? Hangi bileşenlerin güncellenmesi gerekiyor?
13. Temizleme
Bu kod laboratuvarında kullanılan kaynaklar için Google Cloud hesabınızın ücretlendirilmesini istemiyorsanız şu adımları uygulayın:
- Google Cloud Console'da Kaynakları yönetin sayfasına gidin.
- Proje listesinde, silmek istediğiniz projeyi seçin ve ardından Sil'i tıklayın.
- İletişim kutusuna proje kimliğini yazın ve projeyi silmek için Kapat'ı tıklayın.
- Alternatif olarak, konsolda Cloud Run'a gidip yeni dağıttığınız hizmeti seçip silebilirsiniz.