1. Giới thiệu
Bạn có bao giờ cảm thấy khó chịu và lười quản lý tất cả các khoản chi tiêu cá nhân của mình không? Tôi cũng vậy! Vì vậy, trong lớp học lập trình này, chúng ta sẽ xây dựng một trợ lý quản lý chi tiêu cá nhân – được Gemini 2.5 hỗ trợ để làm mọi việc vặt cho chúng ta! Từ việc quản lý các biên nhận đã tải lên đến việc phân tích xem bạn đã chi tiêu quá nhiều để mua cà phê hay chưa!
Bạn có thể truy cập vào trợ lý này thông qua trình duyệt web dưới dạng giao diện web trò chuyện. Trong đó, bạn có thể giao tiếp với trợ lý, tải một số hình ảnh biên nhận lên và yêu cầu trợ lý lưu trữ các hình ảnh đó, hoặc có thể muốn tìm kiếm một số biên nhận để lấy tệp và phân tích một số chi phí. Và tất cả những điều này được xây dựng dựa trên khung Bộ phát triển tác nhân của Google
Bản thân ứng dụng được chia thành 2 dịch vụ: giao diện người dùng và phụ trợ; cho phép bạn tạo một nguyên mẫu nhanh và thử nghiệm cảm giác sử dụng, đồng thời hiểu được hợp đồng API trông như thế nào để tích hợp cả hai dịch vụ này.
Trong lớp học lập trình này, bạn sẽ áp dụng phương pháp từng bước như sau:
- Chuẩn bị dự án Google Cloud và Bật tất cả API bắt buộc trên dự án đó
- Thiết lập bộ chứa trên Google Cloud Storage và cơ sở dữ liệu trên Firestore
- Tạo chỉ mục Firestore
- Thiết lập không gian làm việc cho môi trường lập trình
- Sắp xếp mã nguồn, công cụ, lời nhắc, v.v. của tác nhân ADK.
- Kiểm thử tác nhân bằng giao diện người dùng phát triển web cục bộ của ADK
- Tạo dịch vụ giao diện người dùng – giao diện trò chuyện bằng thư viện Gradio để gửi một số truy vấn và tải hình ảnh biên nhận lên
- Tạo dịch vụ phụ trợ – máy chủ HTTP bằng FastAPI, nơi lưu trữ mã tác nhân ADK, SessionService và Dịch vụ cấu phần phần mềm
- Quản lý các biến môi trường và thiết lập các tệp bắt buộc cần thiết để triển khai ứng dụng cho Cloud Run
- Triển khai ứng dụng trên Cloud Run
Tổng quan về cấu trúc
Điều kiện tiên quyết
- Thoải mái làm việc với Python
- Hiểu biết về cấu trúc toàn stack cơ bản bằng cách sử dụng dịch vụ HTTP
Kiến thức bạn sẽ học được
- Tạo nguyên mẫu web giao diện người dùng bằng Gradio
- Phát triển dịch vụ phụ trợ bằng FastAPI và Pydantic
- Thiết kế Trình đại diện ADK trong khi sử dụng một số tính năng của trình đại diện
- Cách sử dụng công cụ
- Quản lý phiên và cấu phần phần mềm
- Sử dụng lệnh gọi lại để sửa đổi dữ liệu đầu vào trước khi gửi đến Gemini
- Sử dụng BuiltInPlanner để cải thiện việc thực thi tác vụ bằng cách lập kế hoạch
- Gỡ lỗi nhanh qua giao diện web cục bộ của ADK
- Chiến lược tối ưu hoá hoạt động tương tác đa phương thức thông qua việc phân tích cú pháp và truy xuất thông tin thông qua kỹ thuật lời nhắc và sửa đổi yêu cầu Gemini bằng lệnh gọi lại ADK
- Tạo dữ liệu tăng cường truy xuất của tác nhân bằng cách sử dụng Firestore làm cơ sở dữ liệu vectơ
- Quản lý các biến môi trường trong tệp YAML bằng Pydantic-settings
- Triển khai ứng dụng lên Cloud Run bằng Dockerfile và cung cấp các biến môi trường bằng tệp YAML
Bạn cần có
- Trình duyệt web Chrome
- Tài khoản Gmail
- Một dự án trên Google Cloud đã bật tính năng thanh toán
Lớp học lập trình này được thiết kế cho các nhà phát triển ở mọi cấp độ (kể cả người mới bắt đầu), sử dụng Python trong ứng dụng mẫu. Tuy nhiên, bạn không cần phải có kiến thức về Python để hiểu các khái niệm được trình bày.
2. Trước khi bắt đầu
Chọn Dự án đang hoạt động trong Cloud Console
Lớp học lập trình này giả định rằng bạn đã có một dự án trên Google Cloud đã bật tính năng thanh toán. Nếu chưa có, bạn có thể làm theo hướng dẫn bên dưới để bắt đầu.
- Trong Google Cloud Console, trên trang bộ chọn dự án, hãy chọn hoặc tạo một dự án trên Google Cloud.
- Đảm bảo bạn đã bật tính năng thanh toán cho dự án trên Cloud. Tìm hiểu cách kiểm tra xem tính năng thanh toán có được bật trên dự án hay không.
Chuẩn bị cơ sở dữ liệu Firestore
Tiếp theo, chúng ta cũng cần tạo một Cơ sở dữ liệu Firestore. Firestore ở chế độ gốc là một cơ sở dữ liệu dạng tài liệu NoSQL được xây dựng để hỗ trợ việc tự động cấp tài nguyên bổ sung, duy trì hiệu suất cao và tạo điều kiện dễ dàng cho việc phát triển ứng dụng. Cơ sở dữ liệu này cũng có thể đóng vai trò là cơ sở dữ liệu vectơ có thể hỗ trợ kỹ thuật Tạo tăng cường truy xuất cho phòng thí nghiệm của chúng tôi.
- Tìm "firestore" trên thanh tìm kiếm rồi nhấp vào sản phẩm Firestore
- Sau đó, hãy nhấp vào nút Tạo cơ sở dữ liệu Firestore
- Sử dụng (mặc định) làm tên mã cơ sở dữ liệu và chọn Standard Edition (Phiên bản chuẩn). Để minh hoạ cho lớp học này, hãy sử dụng Firestore Native với các quy tắc bảo mật Mở.
- Bạn cũng sẽ nhận thấy rằng cơ sở dữ liệu này thực sự có Free-tier Usage YEAY! (Mức sử dụng miễn phí YEAY!) Sau đó, hãy nhấp vào Nút Tạo cơ sở dữ liệu
Sau các bước này, bạn sẽ được chuyển hướng đến Cơ sở dữ liệu Firestore mà bạn vừa tạo
Thiết lập dự án trên Cloud trong Cloud Shell Terminal
- Bạn sẽ sử dụng Cloud Shell, một môi trường dòng lệnh chạy trong Google Cloud và được tải sẵn bq. Nhấp vào Kích hoạt Cloud Shell ở đầu bảng điều khiển Google Cloud.
- Sau khi kết nối với Cloud Shell, hãy kiểm tra để đảm bảo bạn đã được xác thực và dự án được đặt thành mã dự án của bạn bằng lệnh sau:
gcloud auth list
- Chạy lệnh sau trong Cloud Shell để xác nhận rằng lệnh gcloud biết về dự án của bạn.
gcloud config list project
- Nếu dự án của bạn chưa được đặt, hãy sử dụng lệnh sau để đặt dự án:
gcloud config set project <YOUR_PROJECT_ID>
Ngoài ra, bạn cũng có thể xem mã PROJECT_ID trong bảng điều khiển
Nhấp vào đó, bạn sẽ thấy tất cả dự án và mã dự án ở bên phải
- Bật các API bắt buộc thông qua lệnh hiển thị bên dưới. Quá trình này có thể mất vài phút. Vui lòng kiên nhẫn chờ đợi.
gcloud services enable aiplatform.googleapis.com \
firestore.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
cloudresourcemanager.googleapis.com
Khi thực thi thành công lệnh này, bạn sẽ thấy một thông báo tương tự như thông báo dưới đây:
Operation "operations/..." finished successfully.
Bạn có thể sử dụng bảng điều khiển để tìm kiếm từng sản phẩm hoặc sử dụng đường liên kết này thay cho lệnh gcloud.
Nếu thiếu bất kỳ API nào, bạn luôn có thể bật API đó trong quá trình triển khai.
Tham khảo tài liệu để biết các lệnh và cách sử dụng gcloud.
Chuẩn bị bộ chứa Google Cloud Storage
Tiếp theo, trên cùng một thiết bị đầu cuối, chúng ta cần chuẩn bị bộ chứa GCS để lưu trữ tệp đã tải lên. Chạy lệnh sau để tạo bộ chứa
gsutil mb -l us-central1 gs://personal-expense-assistant-receipts
Kết quả sẽ hiển thị như sau
Creating gs://personal-expense-assistant-receipts/...
Bạn có thể xác minh điều này bằng cách chuyển đến Trình đơn điều hướng ở trên cùng bên trái của trình duyệt rồi chọn Bộ nhớ trên đám mây -> Vùng chứa
Tạo chỉ mục Firestore cho tính năng Tìm kiếm
Firestore là một cơ sở dữ liệu NoSQL gốc, mang lại hiệu suất và tính linh hoạt vượt trội trong mô hình dữ liệu, nhưng có những hạn chế khi nói đến các truy vấn phức tạp. Vì dự định sử dụng một số truy vấn nhiều trường phức hợp và tìm kiếm vectơ, nên trước tiên, chúng ta cần tạo một số chỉ mục. Bạn có thể đọc thêm về thông tin chi tiết trong tài liệu này
- Chạy lệnh sau để tạo chỉ mục hỗ trợ truy vấn phức hợp
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)"
- Và chạy lệnh này để hỗ trợ tìm kiếm vectơ
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)"
Bạn có thể kiểm tra chỉ mục đã tạo bằng cách truy cập Firestore trong bảng điều khiển trên đám mây, rồi nhấp vào phiên bản thể hiện cơ sở dữ liệu (mặc định) và chọn Chỉ mục trên thanh điều hướng
Chuyển đến Trình chỉnh sửa Cloud Shell và Thiết lập thư mục làm việc của ứng dụng
Bây giờ, chúng ta có thể thiết lập trình soạn thảo mã để thực hiện một số thao tác lập trình. Chúng ta sẽ sử dụng Trình chỉnh sửa Cloud Shell để thực hiện việc này
- Nhấp vào nút Open Editor (Mở trình chỉnh sửa), thao tác này sẽ mở một Cloud Shell Editor (Trình chỉnh sửa Cloud Shell), chúng ta có thể viết mã của mình tại đây
- Đảm bảo dự án Cloud Code được đặt ở góc dưới bên trái (thanh trạng thái) của trình chỉnh sửa Cloud Shell, như được làm nổi bật trong hình ảnh bên dưới và được đặt thành dự án Google Cloud đang hoạt động mà bạn đã bật tính năng thanh toán. Uỷ quyền nếu được nhắc. Nếu bạn đã làm theo lệnh trước đó, nút này cũng có thể trỏ trực tiếp đến dự án đã kích hoạt thay vì nút đăng nhập
- Tiếp theo, hãy sao chép thư mục hoạt động của mẫu cho lớp học lập trình này từ GitHub, chạy lệnh sau. Thao tác này sẽ tạo thư mục đang hoạt động trong thư mục personal-expense-assistant
git clone https://github.com/alphinside/personal-expense-assistant-adk-codelab-starter.git personal-expense-assistant
- Sau đó, hãy chuyển đến phần trên cùng của Trình chỉnh sửa Cloud Shell rồi nhấp vào File->Open Folder (Tệp->Mở thư mục), tìm thư mục username (tên người dùng) rồi tìm thư mục personal-expense-assistant (trợ lý chi tiêu cá nhân), sau đó nhấp vào nút OK. Thao tác này sẽ đặt thư mục đã chọn làm thư mục làm việc chính. Trong ví dụ này, tên người dùng là alvinprayuda, do đó, đường dẫn thư mục sẽ hiển thị bên dưới
Bây giờ, Trình chỉnh sửa Cloud Shell của bạn sẽ có dạng như sau
Thiết lập môi trường
Chuẩn bị môi trường ảo Python
Bước tiếp theo là chuẩn bị môi trường phát triển. Cổng hiện đang hoạt động của bạn phải nằm trong thư mục hoạt động personal-expense-assistant. Chúng ta sẽ sử dụng Python 3.12 trong lớp học lập trình này và sử dụng trình quản lý dự án python uv để đơn giản hoá nhu cầu tạo và quản lý phiên bản python cũng như môi trường ảo
- Nếu bạn chưa mở cửa sổ dòng lệnh, hãy mở bằng cách nhấp vào Terminal (Dòng lệnh) -> New Terminal (Dòng lệnh mới) hoặc sử dụng tổ hợp phím Ctrl + Shift + C. Thao tác này sẽ mở một cửa sổ dòng lệnh ở phần dưới cùng của trình duyệt
- Tải
uvxuống và cài đặt python 3.12 bằng lệnh sau
curl -LsSf https://astral.sh/uv/0.6.16/install.sh | sh && \
source $HOME/.local/bin/env && \
uv python install 3.12
- Bây giờ, hãy khởi chạy môi trường ảo bằng
uv, Chạy lệnh này
uv sync --frozen
Thao tác này sẽ tạo thư mục .venv và cài đặt các phần phụ thuộc. Hãy xem nhanh pyproject.toml để biết thông tin về các phần phụ thuộc như sau
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",
]
- Để kiểm thử môi trường ảo, hãy tạo tệp mới main.py và sao chép mã sau
def main():
print("Hello from personal-expense-assistant-adk!")
if __name__ == "__main__":
main()
- Sau đó, hãy chạy lệnh sau
uv run main.py
Bạn sẽ nhận được kết quả như sau
Using CPython 3.12 Creating virtual environment at: .venv Hello from personal-expense-assistant-adk!
Điều này cho thấy dự án python đang được thiết lập đúng cách.
Thiết lập tệp cấu hình
Bây giờ, chúng ta cần thiết lập các tệp cấu hình cho dự án này. Chúng ta sử dụng pydantic-settings để đọc cấu hình từ tệp YAML.
Tạo một tệp có tên settings.yaml với cấu hình sau. Nhấp vào File->New Text File (Tệp->Tệp văn bản mới) rồi điền mã sau. Sau đó, lưu tệp này dưới dạng settings.yaml
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"
Trong lớp học lập trình này, chúng ta sẽ sử dụng các giá trị được định cấu hình trước cho GCLOUD_LOCATION, BACKEND_URL, STORAGE_BUCKET_NAME, DB_COLLECTION_NAME và BACKEND_URL .
Bây giờ, chúng ta có thể chuyển sang bước tiếp theo, tạo tác nhân rồi đến các dịch vụ
3. Tạo tác nhân bằng Google ADK và Gemini 2.5
Giới thiệu về Cấu trúc thư mục ADK
Hãy bắt đầu bằng cách khám phá những gì ADK cung cấp và cách tạo tác nhân. Bạn có thể truy cập vào tài liệu đầy đủ về ADK tại URL này . ADK cung cấp cho chúng ta nhiều tiện ích trong quá trình thực thi lệnh CLI. Sau đây là một số yếu tố :
- Thiết lập cấu trúc thư mục của tác nhân
- Thử tương tác nhanh qua đầu ra đầu vào CLI
- Thiết lập nhanh giao diện web của giao diện người dùng phát triển cục bộ
Bây giờ, hãy tạo cấu trúc thư mục của tác nhân bằng lệnh CLI. Chạy lệnh sau
uv run adk create expense_manager_agent \
--model gemini-2.5-flash-preview-04-17 \
--project {your-project-id} \
--region us-central1
Thao tác này sẽ tạo cấu trúc thư mục của tác nhân như sau
expense_manager_agent/ ├── __init__.py ├── .env ├── agent.py
Và nếu kiểm tra init.py và agent.py, bạn sẽ thấy mã này
# __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',
)
Tạo tác nhân Trình quản lý chi tiêu
Hãy cùng tạo trình quản lý chi tiêu! Mở tệp expense_manager_agent/agent.py rồi sao chép đoạn mã dưới đây. Đoạn mã này sẽ chứa root_agent.
# expense_manager_agent/agent.py
from google.adk.agents import Agent
from expense_manager_agent.tools import (
store_receipt_data,
search_receipts_by_metadata_filter,
search_relevant_receipts_by_natural_language_query,
get_receipt_data_by_image_id,
)
from expense_manager_agent.callbacks import modify_image_data_in_history
import os
from settings import get_settings
from google.adk.planners import BuiltInPlanner
from google.genai import types
SETTINGS = get_settings()
os.environ["GOOGLE_CLOUD_PROJECT"] = SETTINGS.GCLOUD_PROJECT_ID
os.environ["GOOGLE_CLOUD_LOCATION"] = SETTINGS.GCLOUD_LOCATION
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "TRUE"
# Get the code file directory path and read the task prompt file
current_dir = os.path.dirname(os.path.abspath(__file__))
prompt_path = os.path.join(current_dir, "task_prompt.md")
with open(prompt_path, "r") as file:
task_prompt = file.read()
root_agent = Agent(
name="expense_manager_agent",
model="gemini-2.5-flash-preview-04-17",
description=(
"Personal expense agent to help user track expenses, analyze receipts, and manage their financial records"
),
instruction=task_prompt,
tools=[
store_receipt_data,
get_receipt_data_by_image_id,
search_receipts_by_metadata_filter,
search_relevant_receipts_by_natural_language_query,
],
planner=BuiltInPlanner(
thinking_config=types.ThinkingConfig(
thinking_budget=2048,
)
),
before_model_callback=modify_image_data_in_history,
)
Giải thích mã
Tập lệnh này chứa quá trình khởi tạo tác nhân, trong đó chúng ta khởi chạy những nội dung sau:
- Đặt mô hình sẽ được sử dụng thành
gemini-2.5-flash-preview-04-17 - Thiết lập nội dung mô tả và hướng dẫn của trợ lý làm lời nhắc hệ thống đang được đọc từ
task_prompt.md - Cung cấp các công cụ cần thiết để hỗ trợ chức năng của tác nhân
- Cho phép lập kế hoạch trước khi tạo câu trả lời hoặc thực thi cuối cùng bằng các tính năng tư duy nhanh của Gemini 2.5
- Thiết lập tính năng chặn lệnh gọi lại trước khi gửi yêu cầu đến Gemini để giới hạn số lượng dữ liệu hình ảnh được gửi trước khi đưa ra dự đoán
4. Định cấu hình Công cụ của tác nhân
Tác nhân quản lý chi phí của chúng tôi sẽ có các chức năng sau:
- Trích xuất dữ liệu từ hình ảnh biên nhận và lưu trữ dữ liệu cũng như tệp
- Tìm kiếm chính xác trên dữ liệu chi tiêu
- Tìm kiếm theo bối cảnh trên dữ liệu chi tiêu
Do đó, chúng ta cần có các công cụ thích hợp để hỗ trợ chức năng này. Tạo một tệp mới trong thư mục expense_manager_agent và đặt tên là tools.py rồi sao chép mã bên dưới
# 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
Giải thích mã
Trong quá trình triển khai chức năng công cụ này, chúng ta thiết kế các công cụ dựa trên 2 ý tưởng chính sau:
- Phân tích cú pháp dữ liệu biên nhận và liên kết với tệp gốc bằng phần giữ chỗ chuỗi Mã nhận dạng hình ảnh
[IMAGE-ID <hash-of-image-1>] - Lưu trữ và truy xuất dữ liệu bằng cơ sở dữ liệu Firestore
Công cụ "store_receipt_data"
Công cụ này là công cụ Nhận dạng ký tự quang học (OCR), công cụ này sẽ phân tích cú pháp thông tin bắt buộc từ dữ liệu hình ảnh, cùng với việc nhận dạng chuỗi Mã nhận dạng hình ảnh và liên kết các thông tin này với nhau để lưu trữ trong cơ sở dữ liệu Firestore.
Ngoài ra, công cụ này cũng chuyển đổi nội dung của biên nhận thành nội dung nhúng bằng text-embedding-004 để tất cả siêu dữ liệu và nội dung nhúng được lưu trữ và lập chỉ mục cùng nhau. Cho phép truy xuất linh hoạt bằng truy vấn hoặc tìm kiếm theo bối cảnh.
Sau khi thực thi thành công công cụ này, bạn có thể thấy dữ liệu biên nhận đã được lập chỉ mục trong cơ sở dữ liệu Firestore như minh hoạ bên dưới
Công cụ "search_receipts_by_metadata_filter"
Công cụ này chuyển đổi truy vấn của người dùng thành bộ lọc truy vấn siêu dữ liệu hỗ trợ tìm kiếm theo phạm vi ngày và/hoặc tổng số giao dịch. Phương thức này sẽ trả về tất cả dữ liệu hoá đơn đã so khớp, trong đó chúng ta sẽ loại bỏ trường nhúng vì tác nhân không cần trường này để hiểu ngữ cảnh
Công cụ "search_relevant_receipts_by_natural_language_query"
Đây là công cụ Tạo sinh tăng cường truy xuất (RAG) của chúng tôi. Tác nhân của chúng tôi có khả năng thiết kế truy vấn riêng để truy xuất biên nhận có liên quan từ cơ sở dữ liệu vectơ và cũng có thể chọn thời điểm sử dụng công cụ này. Ý tưởng cho phép tác nhân đưa ra quyết định độc lập về việc có sử dụng công cụ RAG này hay không và thiết kế truy vấn riêng là một trong những định nghĩa của phương pháp Agentic RAG.
Chúng ta không chỉ cho phép nó tạo truy vấn riêng mà còn cho phép nó chọn số lượng tài liệu có liên quan mà nó muốn truy xuất. Kết hợp với kỹ thuật câu lệnh phù hợp, ví dụ:
# Example prompt Always filter the result from tool search_relevant_receipts_by_natural_language_query as the returned result may contain irrelevant information
Điều này sẽ giúp công cụ này trở thành một công cụ mạnh mẽ có thể tìm kiếm hầu hết mọi thứ, mặc dù có thể không trả về tất cả kết quả dự kiến do bản chất không chính xác của phương thức tìm kiếm người hàng xóm gần nhất.
5. Sửa đổi ngữ cảnh cuộc trò chuyện thông qua lệnh gọi lại
Google ADK cho phép chúng ta "can thiệp" vào thời gian chạy của tác nhân ở nhiều cấp. Bạn có thể đọc thêm về chức năng chi tiết này trong tài liệu này . Trong lớp học này, chúng ta sử dụng before_model_callback để sửa đổi yêu cầu trước khi gửi đến LLM để xoá dữ liệu hình ảnh trong ngữ cảnh nhật ký trò chuyện cũ ( chỉ bao gồm dữ liệu hình ảnh trong 3 lượt tương tác gần đây nhất của người dùng) để tăng hiệu quả
Tuy nhiên, chúng ta vẫn muốn tác nhân có ngữ cảnh dữ liệu hình ảnh khi cần. Do đó, chúng ta thêm một cơ chế để thêm phần giữ chỗ mã nhận dạng hình ảnh dạng chuỗi sau mỗi dữ liệu byte hình ảnh trong cuộc trò chuyện. Điều này sẽ giúp tác nhân liên kết mã nhận dạng hình ảnh với dữ liệu tệp thực tế của hình ảnh. Dữ liệu này có thể được sử dụng cả tại thời điểm lưu trữ hoặc truy xuất hình ảnh. Cấu trúc sẽ có dạng như sau
<image-byte-data-1> [IMAGE-ID <hash-of-image-1>] <image-byte-data-2> [IMAGE-ID <hash-of-image-2>] And so on..
Và khi dữ liệu byte không còn được dùng trong nhật ký trò chuyện, giá trị nhận dạng chuỗi vẫn còn để cho phép truy cập dữ liệu bằng cách sử dụng công cụ. Ví dụ về cấu trúc nhật ký sau khi xoá dữ liệu hình ảnh
[IMAGE-ID <hash-of-image-1>] [IMAGE-ID <hash-of-image-2>] And so on..
Hãy bắt đầu! Tạo một tệp mới trong thư mục expense_manager_agent và đặt tên là callbacks.py rồi sao chép mã bên dưới
# 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. Câu lệnh
Khi thiết kế một tác nhân có khả năng và khả năng tương tác phức tạp, chúng ta cần tìm một lời nhắc đủ tốt để hướng dẫn tác nhân đó hoạt động theo cách chúng ta muốn.
Trước đây, chúng ta có một cơ chế về cách xử lý dữ liệu hình ảnh trong nhật ký trò chuyện, đồng thời có các công cụ có thể không dễ sử dụng, chẳng hạn như search_relevant_receipts_by_natural_language_query.. Chúng ta cũng muốn nhân viên hỗ trợ có thể tìm kiếm và truy xuất hình ảnh biên nhận chính xác cho chúng ta. Điều này có nghĩa là chúng ta cần truyền tải đúng cách tất cả thông tin này trong một cấu trúc lời nhắc thích hợp
Chúng ta sẽ yêu cầu tác nhân sắp xếp kết quả theo định dạng markdown sau đây để phân tích cú pháp quy trình suy nghĩ, câu trả lời cuối cùng và tệp đính kèm ( nếu có)
# 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>]",
...
]
}
Hãy bắt đầu với lời nhắc sau đây để đạt được kỳ vọng ban đầu của chúng ta về hành vi của tác nhân quản lý chi tiêu. Tệp task_prompt.md phải tồn tại trong thư mục đang hoạt động của chúng ta, nhưng chúng ta cần di chuyển tệp đó vào thư mục expense_manager_agent. Chạy lệnh sau để di chuyển tệp
mv task_prompt.md expense_manager_agent/task_prompt.md
7. Kiểm thử tác nhân
Bây giờ, hãy thử giao tiếp với tác nhân thông qua CLI, hãy chạy lệnh sau
uv run adk run expense_manager_agent
Kết quả sẽ hiển thị như sau, trong đó bạn có thể trò chuyện lần lượt với nhân viên hỗ trợ, tuy nhiên, bạn chỉ có thể gửi văn bản qua giao diện này
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:
Giờ đây, ngoài hoạt động tương tác với CLI, ADK cũng cho phép chúng ta có một giao diện người dùng phát triển để tương tác và kiểm tra những gì đang diễn ra trong quá trình tương tác. Chạy lệnh sau để khởi động máy chủ giao diện người dùng phát triển cục bộ
uv run adk web --port 8080
Thao tác này sẽ tạo ra kết quả như ví dụ sau, tức là chúng ta đã có thể truy cập vào giao diện 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)
Bây giờ, để kiểm tra, hãy nhấp vào nút Xem trước trên web ở khu vực trên cùng của Trình chỉnh sửa Cloud Shell rồi chọn Xem trước trên cổng 8080
Bạn sẽ thấy trang web sau đây, trong đó bạn có thể chọn các tác nhân có sẵn trên nút thả xuống ở trên cùng bên trái ( trong trường hợp này là expense_manager_agent) và tương tác với bot. Bạn sẽ thấy nhiều thông tin về chi tiết nhật ký trong thời gian chạy của tác nhân trong cửa sổ bên trái
Hãy thử một số thao tác! Tải 2 biên nhận mẫu này lên ( nguồn : Tập dữ liệu khuôn mặt ôm mousserlane/id_receipt_dataset ) . Nhấp chuột phải vào từng hình ảnh rồi chọn Save Image as.. (Lưu hình ảnh dưới dạng). ( thao tác này sẽ tải hình ảnh biên nhận xuống), sau đó tải tệp lên bot bằng cách nhấp vào biểu tượng "đoạn trích" và cho biết bạn muốn lưu trữ các biên nhận này
Sau đó, hãy thử các truy vấn sau để tìm kiếm hoặc truy xuất tệp
- "Cung cấp bảng chi tiết về các khoản chi tiêu và tổng chi tiêu trong năm 2023"
- "Gửi cho tôi tệp biên nhận của Indomaret"
Khi sử dụng một số công cụ, bạn có thể kiểm tra những gì đang diễn ra trong giao diện người dùng phát triển
Xem cách tác nhân phản hồi bạn và kiểm tra xem tác nhân có tuân thủ tất cả các quy tắc được cung cấp trong lời nhắc bên trong task_prompt.py hay không. Xin chúc mừng! Bây giờ, bạn đã có một tác nhân phát triển hoàn chỉnh và đang hoạt động.
Bây giờ, đã đến lúc hoàn tất ứng dụng bằng giao diện người dùng phù hợp và đẹp mắt cũng như các chức năng để tải lên và tải tệp hình ảnh xuống.
8. Tạo dịch vụ giao diện người dùng bằng Gradio
Chúng ta sẽ xây dựng một giao diện web trò chuyện như sau
Trang này chứa một giao diện trò chuyện có trường nhập để người dùng gửi văn bản và tải(các) tệp hình ảnh biên nhận lên.
Chúng ta sẽ xây dựng dịch vụ giao diện người dùng bằng Gradio.
Tạo tệp mới, nhấp vào File->New Text File (Tệp->Tệp văn bản mới) rồi đặt tên là frontend.py,sau đó sao chép mã sau và lưu
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,
)
Sau đó, chúng ta có thể thử chạy dịch vụ giao diện người dùng bằng lệnh sau. Đừng quên đổi tên tệp main.py thành frontend.py
uv run frontend.py
Bạn sẽ thấy kết quả tương tự như sau trong bảng điều khiển trên đám mây
* Running on local URL: http://0.0.0.0:8080 To create a public link, set `share=True` in `launch()`.
Sau đó, bạn có thể kiểm tra giao diện web khi nhấn tổ hợp phím ctrl+nhấp vào đường liên kết URL cục bộ. Ngoài ra, bạn cũng có thể truy cập vào ứng dụng giao diện người dùng bằng cách nhấp vào nút Xem trước trên web ở phía trên cùng bên phải của Cloud Editor, rồi chọn Xem trước trên cổng 8080
Bạn sẽ thấy giao diện web, tuy nhiên, bạn sẽ gặp lỗi dự kiến khi cố gắng gửi cuộc trò chuyện do dịch vụ phụ trợ chưa được thiết lập
Bây giờ, hãy để dịch vụ chạy và đừng tắt dịch vụ đó. Chúng ta sẽ chạy dịch vụ phụ trợ trong một thẻ dòng lệnh khác
Giải thích mã
Trong mã giao diện người dùng này, trước tiên, chúng ta cho phép người dùng gửi văn bản và tải nhiều tệp lên. Gradio cho phép chúng ta tạo loại chức năng này bằng phương thức gr.ChatInterface kết hợp với gr.MultimodalTextbox
Bây giờ, trước khi gửi tệp và văn bản đến phần phụ trợ, chúng ta cần tìm hiểu mimetype của tệp vì phần phụ trợ cần tệp đó. Chúng ta cũng cần mã hoá byte tệp hình ảnh thành base64 và gửi cùng với 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
Giản đồ dùng cho hoạt động tương tác giữa giao diện người dùng và phần phụ trợ được xác định trong schema.py. Chúng ta sử dụng BaseModel Pydantic để thực thi việc xác thực dữ liệu trong giản đồ
Khi nhận được phản hồi, chúng tôi đã tách riêng phần quy trình suy nghĩ, phản hồi cuối cùng và tệp đính kèm. Do đó, chúng ta có thể sử dụng thành phần Gradio để hiển thị từng thành phần bằng thành phần giao diện người dùng.
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. Tạo dịch vụ phụ trợ bằng FastAPI
Tiếp theo, chúng ta cần xây dựng phần phụ trợ có thể khởi chạy Tác nhân cùng với các thành phần khác để có thể thực thi thời gian chạy tác nhân.
Tạo tệp mới, nhấp vào File->New Text File (Tệp->Tệp văn bản mới), rồi sao chép và dán mã sau đó lưu dưới dạng backend.py
from expense_manager_agent.agent import root_agent as expense_manager_agent
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.adk.events import Event
from fastapi import FastAPI, Body, Depends
from typing import AsyncIterator
from types import SimpleNamespace
import uvicorn
from contextlib import asynccontextmanager
import asyncio
from utils import (
extract_attachment_ids_and_sanitize_response,
download_image_from_gcs,
extract_thinking_process,
format_user_request_to_adk_content_and_store_artifacts,
)
from schema import ImageData, ChatRequest, ChatResponse
import logger
from google.adk.artifacts import GcsArtifactService
from settings import get_settings
SETTINGS = get_settings()
APP_NAME = "expense_manager_app"
# Application state to hold service contexts
class AppContexts(SimpleNamespace):
"""A class to hold application contexts with attribute access"""
session_service: InMemorySessionService = None
artifact_service: GcsArtifactService = None
expense_manager_agent_runner: Runner = None
# Initialize application state
app_contexts = AppContexts()
@asynccontextmanager
async def lifespan(app: FastAPI):
# Initialize service contexts during application startup
app_contexts.session_service = InMemorySessionService()
app_contexts.artifact_service = GcsArtifactService(
bucket_name=SETTINGS.STORAGE_BUCKET_NAME
)
app_contexts.expense_manager_agent_runner = Runner(
agent=expense_manager_agent, # The agent we want to run
app_name=APP_NAME, # Associates runs with our app
session_service=app_contexts.session_service, # Uses our session manager
artifact_service=app_contexts.artifact_service, # Uses our artifact manager
)
logger.info("Application started successfully")
yield
logger.info("Application shutting down")
# Perform cleanup during application shutdown if necessary
# Helper function to get application state as a dependency
async def get_app_contexts() -> AppContexts:
return app_contexts
# Create FastAPI app
app = FastAPI(title="Personal Expense Assistant API", lifespan=lifespan)
@app.post("/chat", response_model=ChatResponse)
async def chat(
request: ChatRequest = Body(...),
app_context: AppContexts = Depends(get_app_contexts),
) -> ChatResponse:
"""Process chat request and get response from the agent"""
# Prepare the user's message in ADK format and store image artifacts
content = await asyncio.to_thread(
format_user_request_to_adk_content_and_store_artifacts,
request=request,
app_name=APP_NAME,
artifact_service=app_context.artifact_service,
)
final_response_text = "Agent did not produce a final response." # Default
# Use the session ID from the request or default if not provided
session_id = request.session_id
user_id = request.user_id
# Create session if it doesn't exist
if not app_context.session_service.get_session(
app_name=APP_NAME, user_id=user_id, session_id=session_id
):
app_context.session_service.create_session(
app_name=APP_NAME, user_id=user_id, session_id=session_id
)
try:
# Process the message with the agent
# Type annotation: runner.run_async returns an AsyncIterator[Event]
events_iterator: AsyncIterator[Event] = (
app_context.expense_manager_agent_runner.run_async(
user_id=user_id, session_id=session_id, new_message=content
)
)
async for event in events_iterator: # event has type Event
# Key Concept: is_final_response() marks the concluding message for the turn
if event.is_final_response():
if event.content and event.content.parts:
# Extract text from the first part
final_response_text = event.content.parts[0].text
elif event.actions and event.actions.escalate:
# Handle potential errors/escalations
final_response_text = f"Agent escalated: {event.error_message or 'No specific message.'}"
break # Stop processing events once the final response is found
logger.info(
"Received final response from agent", raw_final_response=final_response_text
)
# Extract and process any attachments and thinking process in the response
base64_attachments = []
sanitized_text, attachment_ids = extract_attachment_ids_and_sanitize_response(
final_response_text
)
sanitized_text, thinking_process = extract_thinking_process(sanitized_text)
# Download images from GCS and replace hash IDs with base64 data
for image_hash_id in attachment_ids:
# Download image data and get MIME type
result = await asyncio.to_thread(
download_image_from_gcs,
artifact_service=app_context.artifact_service,
image_hash=image_hash_id,
app_name=APP_NAME,
user_id=user_id,
session_id=session_id,
)
if result:
base64_data, mime_type = result
base64_attachments.append(
ImageData(serialized_image=base64_data, mime_type=mime_type)
)
logger.info(
"Processed response with attachments",
sanitized_response=sanitized_text,
thinking_process=thinking_process,
attachment_ids=attachment_ids,
)
return ChatResponse(
response=sanitized_text,
thinking_process=thinking_process,
attachments=base64_attachments,
)
except Exception as e:
logger.error("Error processing chat request", error_message=str(e))
return ChatResponse(
response="", error=f"Error in generating response: {str(e)}"
)
# Only run the server if this file is executed directly
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8081)
Sau đó, chúng ta có thể thử chạy dịch vụ phụ trợ. Hãy nhớ rằng ở bước trước, chúng ta đã chạy dịch vụ giao diện người dùng, giờ đây, chúng ta cần mở một thiết bị đầu cuối mới và cố gắng chạy dịch vụ phụ trợ này
- Tạo một thiết bị đầu cuối mới. Chuyển đến thiết bị đầu cuối ở khu vực dưới cùng và tìm nút "+" để tạo thiết bị đầu cuối mới. Ngoài ra, bạn có thể nhấn tổ hợp phím Ctrl + Shift + C để mở một thiết bị đầu cuối mới
- Sau đó, hãy đảm bảo rằng bạn đang ở trong thư mục đang hoạt động personal-expense-assistant rồi chạy lệnh sau
uv run backend.py
- Nếu thành công, bạn sẽ thấy kết quả như sau
INFO: Started server process [xxxxx] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://0.0.0.0:8081 (Press CTRL+C to quit)
Giải thích mã
Khởi chạy Trình đại diện ADK, SessionService và ArtifactService
Để chạy tác nhân trong dịch vụ phụ trợ, chúng ta cần tạo một Runner (Trình chạy) chứa cả SessionService và tác nhân của chúng ta. SessionService sẽ quản lý nhật ký và trạng thái cuộc trò chuyện, do đó, khi được tích hợp với Runner, lớp này sẽ cho phép tác nhân của chúng ta nhận được ngữ cảnh cuộc trò chuyện đang diễn ra.
Chúng tôi cũng sử dụng ArtifactService để xử lý tệp đã tải lên. Bạn có thể đọc thêm thông tin chi tiết về Phiên và Cấu phần phần mềm của ADK tại đây
...
@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
...
Trong bản minh hoạ này, chúng ta sử dụng InMemorySessionService và GcsArtifactService để tích hợp với tác nhân Runner. Vì nhật ký trò chuyện được lưu trữ trong bộ nhớ, nên nhật ký này sẽ bị mất khi dịch vụ phụ trợ bị tắt hoặc khởi động lại. Chúng ta khởi tạo các phần phụ thuộc này bên trong vòng đời của ứng dụng FastAPI để được chèn dưới dạng phần phụ thuộc trong tuyến /chat.
Tải hình ảnh lên và tải hình ảnh xuống bằng GcsArtifactService
Tất cả hình ảnh đã tải lên sẽ được GcsArtifactService lưu trữ dưới dạng cấu phần phần mềm, bạn có thể kiểm tra điều này trong hàm format_user_request_to_adk_content_and_store_artifacts bên trong 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,
)
...
Tất cả các yêu cầu sẽ được trình chạy tác nhân xử lý đều cần được định dạng thành loại types.Content. Bên trong hàm, chúng ta cũng xử lý từng dữ liệu hình ảnh và trích xuất mã nhận dạng của dữ liệu đó để thay thế bằng phần giữ chỗ Mã nhận dạng hình ảnh.
Cơ chế tương tự được dùng để tải tệp đính kèm xuống sau khi trích xuất mã nhận dạng hình ảnh bằng biểu thức chính quy:
...
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. Kiểm thử tích hợp
Bây giờ, bạn sẽ có nhiều dịch vụ chạy trong các thẻ khác nhau trên Google Cloud Console:
- Dịch vụ giao diện người dùng chạy ở cổng 8080
* Running on local URL: http://0.0.0.0:8080 To create a public link, set `share=True` in `launch()`.
- Dịch vụ phụ trợ chạy ở cổng 8081
INFO: Started server process [xxxxx] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://0.0.0.0:8081 (Press CTRL+C to quit)
Ở trạng thái hiện tại, bạn có thể tải hình ảnh biên nhận lên và trò chuyện liền mạch với trợ lý từ ứng dụng web trên cổng 8080.
Nhấp vào nút Xem trước trên web ở khu vực trên cùng của Trình chỉnh sửa Cloud Shell rồi chọn Xem trước trên cổng 8080
Bây giờ, hãy tương tác với trợ lý!
Tải các biên nhận sau đây xuống. Phạm vi ngày của dữ liệu biên nhận này là từ năm 2023 đến năm 2024 và yêu cầu trợ lý lưu trữ/tải dữ liệu đó lên
- Receipt Drive ( nguồn dữ liệu khuôn mặt ôm
mousserlane/id_receipt_dataset)
Đặt nhiều câu hỏi
- "Cho tôi biết bảng chi tiết chi phí hằng tháng trong giai đoạn 2023-2024"
- "Cho tôi xem biên nhận giao dịch mua cà phê"
- "Cho tôi tệp biên nhận của Yakiniku Like"
- V.v.
Sau đây là một số đoạn mã tương tác thành công
11. Triển khai lên Cloud Run
Tất nhiên, chúng ta muốn truy cập vào ứng dụng tuyệt vời này ở bất cứ đâu. Để làm như vậy, chúng ta có thể đóng gói ứng dụng này và triển khai ứng dụng đó lên Cloud Run. Để minh hoạ, dịch vụ này sẽ được hiển thị dưới dạng dịch vụ công khai mà người khác có thể truy cập. Tuy nhiên, hãy lưu ý rằng đây không phải là phương pháp hay nhất cho loại ứng dụng này vì phương pháp này phù hợp hơn với các ứng dụng cá nhân
Trong lớp học lập trình này, chúng ta sẽ đặt cả dịch vụ giao diện người dùng và dịch vụ phụ trợ vào 1 vùng chứa. Chúng ta sẽ cần đến sự trợ giúp của supervisord để quản lý cả hai dịch vụ. Bạn có thể kiểm tra tệp supervisord.conf và kiểm tra Dockerfile mà chúng tôi đặt supervisord làm điểm truy cập.
Tại thời điểm này, chúng ta đã có tất cả các tệp cần thiết để triển khai ứng dụng trên Cloud Run. Hãy triển khai ứng dụng. Chuyển đến Cloud Shell Terminal và đảm bảo dự án hiện tại được định cấu hình thành dự án đang hoạt động, nếu không, bạn phải sử dụng lệnh gcloud configure để đặt mã dự án:
gcloud config set project [PROJECT_ID]
Sau đó, hãy chạy lệnh sau để triển khai ứng dụng đó lên Cloud Run.
gcloud run deploy personal-expense-assistant \
--source . \
--port=8080 \
--allow-unauthenticated \
--env-vars-file=settings.yaml \
--memory 1024Mi \
--region us-central1
Nếu bạn được nhắc xác nhận việc tạo một kho đăng ký cấu phần phần mềm cho kho lưu trữ docker, bạn chỉ cần trả lời Có. Xin lưu ý rằng chúng tôi đang cho phép quyền truy cập chưa xác thực tại đây vì đây là ứng dụng minh hoạ. Bạn nên sử dụng phương thức xác thực phù hợp cho các ứng dụng doanh nghiệp và ứng dụng phát hành chính thức.
Sau khi triển khai xong, bạn sẽ nhận được một đường liên kết tương tự như bên dưới:
https://personal-expense-assistant-*******.us-central1.run.app
Hãy tiếp tục sử dụng ứng dụng của bạn trong cửa sổ Ẩn danh hoặc trên thiết bị di động. Trang web đã được xuất bản.
12. Thách thức
Giờ là lúc bạn tỏa sáng và trau dồi kỹ năng khám phá. Bạn có đủ kiến thức để thay đổi mã để phần phụ trợ có thể phục vụ nhiều người dùng không? Những thành phần nào cần được cập nhật?
13. Dọn dẹp
Để tránh bị tính phí cho tài khoản Google Cloud của bạn đối với các tài nguyên dùng trong lớp học lập trình này, hãy làm theo các bước sau:
- Trong Google Cloud Console, hãy chuyển đến trang Quản lý tài nguyên.
- Trong danh sách dự án, hãy chọn dự án mà bạn muốn xoá, rồi nhấp vào Xoá.
- Trong hộp thoại, hãy nhập mã dự án, sau đó nhấp vào Shut down (Tắt) để xoá dự án.
- Ngoài ra, bạn có thể chuyển đến Cloud Run trên bảng điều khiển, chọn dịch vụ bạn vừa triển khai rồi xoá.