1. 📖 Giới thiệu
Bạn có bao giờ cảm thấy khó chịu và quá lười biếng để quản lý tất cả các chi phí cá nhân của mình không? Tôi cũng vậy! Vì vậy, trong codelab này, chúng ta sẽ xây dựng một trợ lý quản lý chi phí cá nhân - được hỗ trợ bởi Gemini 2.5 để làm mọi việc vặt thay chúng ta! Từ việc quản lý biên nhận đã tải lên cho đến phân tích xem bạn đã chi quá nhiều tiền để mua cà phê hay chưa!
Bạn có thể truy cập trợ lý này thông qua trình duyệt web dưới dạng giao diện trò chuyện trên web, trong đó bạn có thể giao tiếp với trợ lý, tải lên một số hình ảnh biên lai và yêu cầu trợ lý lưu trữ chúng hoặc có thể muốn tìm kiếm một số biên lai để lấy tệp và thực hiện một số phân tích chi phí. Tất cả những điều này được xây dựng dựa trên khung Google Agent Development Kit
Bản thân ứng dụng được chia thành 2 dịch vụ: frontend và backend; cho phép bạn xây dựng một nguyên mẫu nhanh và dùng thử xem nó như thế nào, đồng thời hiểu được hợp đồng API trông như thế nào để tích hợp cả hai.
Thông qua codelab, 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 bucket 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
- Cấu trúc 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ộ ADK
- Xây dựng 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
- Xây dựng dịch vụ phụ trợ - máy chủ HTTP sử dụng FastAPI nơi chứa mã tác nhân ADK, SessionService và Artifact Service của chúng tôi
- Quản lý các biến môi trường và thiết lập các tệp cần thiết để triển khai ứng dụng lên Cloud Run
- Triển khai ứng dụng lê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 cơ bản của ngăn xếp đầy đủ bằng cách sử dụng dịch vụ HTTP
Kiến thức bạn sẽ học được
- Tạo mẫu web frontend với Gradio
- Phát triển dịch vụ phụ trợ bằng FastAPI và Pydantic
- Thiết kế ADK Agent trong khi sử dụng nhiều khả năng của nó
- 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 hiện nhiệm vụ bằng cách lập kế hoạch
- Gỡ lỗi nhanh thông qua giao diện web cục bộ 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 tạo câu lệnh và sửa đổi yêu cầu Gemini bằng lệnh gọi lại ADK
- Tạo thông tin tăng cường dựa trên hoạt động truy xuất bằng tác nhân 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 với tệp YAML
Bạn cần có
- Trình duyệt web Chrome
- Một tài khoản Gmail
- Một dự án đám mây có hỗ trợ thanh toán
Lớp học lập trình này được thiết kế cho 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 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ạn 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 chọn dự án, hãy chọn hoặc tạo một dự á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 một 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ế độ Native 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. Đây cũng có thể là một cơ sở dữ liệu vectơ có thể hỗ trợ kỹ thuật Tạo thông tin tăng cường để truy xuất cho phòng thí nghiệm của chúng tôi.
- Tìm kiếm "firestore" trên thanh tìm kiếm và nhấp vào sản phẩm Firestore
- Sau đó, 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à giữ nguyên lựa chọn Standard Edition (Phiên bản tiêu chuẩn). Để minh hoạ trong phòng thí nghiệm 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! (Sử dụng cấp miễn phí YEAY!) Sau đó, 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 thiết bị đầu cuối Cloud Shell
- 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 bằng 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 xem bạn đã được xác thực chưa và dự án đã được đặt theo ID dự án của bạn chưa 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 thiết lập, hãy sử dụng lệnh sau để thiết lập:
gcloud config set project <YOUR_PROJECT_ID>
Ngoài ra, bạn cũng có thể xem id PROJECT_ID trong bảng điều khiển
Nhấp vào đó và bạn sẽ thấy toàn bộ dự án của mình và ID dự án ở bên phải
- Bật các API bắt buộc thông qua lệnh bên dưới. Quá trình này có thể mất vài phút, vì vậy, 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 hiện lệnh thành công, bạn sẽ thấy thông báo tương tự như thông báo bên dưới:
Operation "operations/..." finished successfully.
Bạn có thể thay thế lệnh gcloud bằng cách tìm kiếm từng sản phẩm trên bảng điều khiển hoặc sử dụng đường liên kết này.
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 gcloud và cách sử dụng.
Chuẩn bị Google Cloud Storage Bucket
Tiếp theo, từ cùng một thiết bị đầu cuối, chúng ta sẽ 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. Bạn sẽ cần một tên bộ chứa duy nhất nhưng phù hợp với biên nhận của trợ lý chi tiêu cá nhân. Do đó, chúng ta sẽ sử dụng tên bộ chứa sau đây kết hợp với mã dự án của bạn
gsutil mb -l us-central1 gs://personal-expense-{your-project-id}
Nó sẽ hiển thị đầu ra này
Creating gs://personal-expense-{your-project-id}
Bạn có thể xác minh điều này bằng cách vào Menu điều hướng ở góc trên bên trái của trình duyệt và chọn Cloud Storage -> Bucket
Tạo chỉ mục Firestore cho tìm kiếm
Firestore là cơ sở dữ liệu NoSQL gốc, cung cấp hiệu suất và tính linh hoạt vượt trội trong mô hình dữ liệu, nhưng có hạn chế khi xử lý các truy vấn phức tạp. Khi chúng ta dự định sử dụng một số truy vấn đa trường hợp hợp và tìm kiếm vectơ, trước tiên chúng ta cần tạo một số chỉ mục. Bạn có thể đọc thêm 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ợ các truy vấn hợp chất
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 cái 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 đám mây và nhấp vào phiên bả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 Cloud Shell Editor 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 chỉnh sửa mã để thực hiện một số việc liên quan đến mã hoá. Chúng ta sẽ sử dụng Cloud Shell Editor cho 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ở Cloud Shell Editor. Chúng ta có thể viết mã tại đây
- Tiếp theo, chúng ta cũng cần kiểm tra xem shell đã được định cấu hình thành PROJECT ID chính xác mà bạn có hay chưa. Nếu thấy có giá trị bên trong ( ) trước biểu tượng $ trong thiết bị đầu cuối ( trong ảnh chụp màn hình bên dưới, giá trị là "adk-multimodal-tool"), thì giá trị này cho biết dự án đã định cấu hình cho phiên shell đang hoạt động của bạn.
Nếu giá trị xuất hiện đã chính xác, bạn có thể bỏ qua lệnh tiếp theo. Tuy nhiên, nếu không chính xác hoặc bị thiếu, hãy chạy lệnh sau
gcloud config set project <YOUR_PROJECT_ID>
- Tiếp theo, hãy sao chép thư mục làm việc của mẫu cho lớp học lập trình này từ GitHub bằng cách 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 Cloud Shell Editor 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) và 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ẽ xuất hiện bên dưới
Giờ đây, Cloud Shell Editor 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. Thiết bị đầu cuối đang hoạt động hiện tại của bạn phải nằm trong thư mục làm việc 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 uv python để đơ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 (Cửa sổ dòng lệnh) -> New Terminal (Cửa sổ 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
- Bây giờ, hãy khởi tạo môi trường ảo bằng
uv, chạy các lệnh này
cd ~/personal-expense-assistant
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. Xem nhanh pyproject.toml sẽ cung cấp cho bạn thông tin về các phụ thuộc được hiển thị như thế này
dependencies = [
"datasets>=3.5.0",
"google-adk==1.18",
"google-cloud-firestore>=2.20.1",
"gradio>=5.23.1",
"pydantic>=2.10.6",
"pydantic-settings[yaml]>=2.8.1",
]
Thiết lập tệp cấu hình
Bây giờ chúng ta sẽ cần thiết lập các tệp cấu hình cho dự án này. Chúng tôi sử dụng pydantic-settings để đọc cấu hình từ tệp YAML.
Chúng tôi đã cung cấp mẫu tệp bên trong settings.yaml.example , chúng tôi sẽ cần sao chép tệp và đổi tên thành settings.yaml. Chạy lệnh này để tạo tệp
cp settings.yaml.example settings.yaml
Sau đó, sao chép giá trị sau vào tệp
GCLOUD_LOCATION: "us-central1"
GCLOUD_PROJECT_ID: "your-project-id"
BACKEND_URL: "http://localhost:8081/chat"
STORAGE_BUCKET_NAME: "personal-expense-{your-project-id}"
DB_COLLECTION_NAME: "personal-expense-assistant-receipts"
Đối với codelab này, chúng ta sẽ sử dụng các giá trị được cấu hình sẵn cho GCLOUD_LOCATION, BACKEND_URL, và DB_COLLECTION_NAME .
Giờ đây, chúng ta có thể chuyển sang bước tiếp theo, xây dựng tác nhân rồi đến các dịch vụ
3. 🚀 Xây dựng Agent 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 tính năng mà 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ố ví dụ :
- Thiết lập cấu trúc thư mục tác nhân
- Nhanh chóng thử tương tác thông qua đầu vào và đầu ra của 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 tác nhân bằng lệnh CLI. Chạy lệnh sau.
uv run adk create expense_manager_agent
Khi được hỏi, hãy chọn mô hình gemini-2.5-flash và phần phụ trợ Vertex AI. Sau đó, trình hướng dẫn sẽ yêu cầu bạn nhập mã dự án và vị trí. Bạn có thể chấp nhận các lựa chọn mặc định bằng cách nhấn phím Enter hoặc thay đổi các lựa chọn đó nếu cần. Hãy kiểm tra kỹ để đảm bảo bạn đang sử dụng đúng mã dự án đã tạo trước đó trong phòng thí nghiệm này. Kết quả đầu ra sẽ có dạng như sau:
Choose a model for the root agent: 1. gemini-2.5-flash 2. Other models (fill later) Choose model (1, 2): 1 1. Google AI 2. Vertex AI Choose a backend (1, 2): 2 You need an existing Google Cloud account and project, check out this link for details: https://google.github.io/adk-docs/get-started/quickstart/#gemini---google-cloud-vertex-ai Enter Google Cloud project ID [going-multimodal-lab]: Enter Google Cloud region [us-central1]: Agent created in /home/username/personal-expense-assistant/expense_manager_agent: - .env - __init__.py - agent.py
Thao tác này sẽ tạo cấu trúc thư mục tác nhân sau
expense_manager_agent/ ├── __init__.py ├── .env ├── agent.py
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',
name='root_agent',
description='A helpful assistant for user questions.',
instruction='Answer user questions to the best of your knowledge',
)
Giờ đây, bạn có thể kiểm thử bằng cách chạy
uv run adk run expense_manager_agent
Bất cứ khi nào kiểm thử xong, bạn có thể thoát khỏi tác nhân bằng cách nhập exit hoặc nhấn tổ hợp phím Ctrl+D.
Xây dựng Đại lý Quản lý Chi phí của Chúng tôi
Hãy cùng xây dựng tác nhân quản lý chi phí! Mở tệp expense_manager_agent/agent.py rồi sao chép đoạn mã bên dưới. Đ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",
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 phần khởi tạo nhân viên hỗ trợ, trong đó chúng ta khởi tạo những điều sau:
- Đặt mô hình sẽ được dùng thành
gemini-2.5-flash - Thiết lập mô tả và hướng dẫn của tác nhân 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 phản hồi hoặc thực hiện cuối cùng bằng cách sử dụng khả năng tư duy Flash của Gemini 2.5
- Thiết lập chặn 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. 🚀 Cấu hình Công cụ Đại lý
Đại lý quản lý chi phí của chúng tôi sẽ có những khả năng sau:
- Trích xuất dữ liệu từ hình ảnh biên lai và lưu trữ dữ liệu và tệp
- Tìm kiếm chính xác trên dữ liệu chi phí
- Tìm kiếm theo ngữ cảnh trên dữ liệu chi phí
Do đó, chúng ta cần các công cụ phù 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
touch expense_manager_agent/tools.py
Mở expense_manage_agent/tools.py, sau đó 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 tôi thiết kế các công cụ xung quanh 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 cách sử dụ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, nó sẽ phân tích thông tin cần thiết từ dữ liệu hình ảnh, cùng với việc nhận dạng chuỗi ID hình ảnh và ánh xạ chúng lại 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 dữ liệu nhúng bằng cách sử dụng text-embedding-004 để tất cả siêu dữ liệu và dữ liệu nhúng được lưu trữ và lập chỉ mục cùng nhau. Cho phép linh hoạt tìm kiếm thông tin bằng truy vấn hoặc tìm kiếm theo ngữ cảnh.
Sau khi thực hiện thành công công cụ này, bạn có thể thấy dữ liệu biên lai đã được lập chỉ mục trong cơ sở dữ liệu Firestore như hiển thị 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 giao dịch. Thao tác này sẽ trả về tất cả dữ liệu biên nhận trùng khớp, trong đó chúng tôi 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 dựa trên truy xuất (RAG) của chúng tôi. Trợ lý của chúng tôi có khả năng thiết kế truy vấn riêng để truy xuất các 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. Khái niệm 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à tự thiết kế truy vấn là một trong những định nghĩa về phương pháp RAG dựa trên tác nhân.
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 tạo 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ẽ biến công cụ này thành một công cụ mạnh mẽ có khả năng tìm kiếm hầu hết mọi thứ, mặc dù nó có thể không trả về tất cả các kết quả mong đợi do tính chất không chính xác của tìm kiếm hàng xóm gần nhất.
5. 🚀 Thay đổi ngữ cảnh hội thoại thông qua lệnh gọi lại
Google ADK cho phép chúng ta "chặn" thời gian chạy của tác nhân ở nhiều cấp độ. Bạn có thể đọc thêm về khả năng chi tiết này trong tài liệu này . Trong phòng thí nghiệm này, chúng ta sẽ sử dụng before_model_callback để sửa đổi yêu cầu trước khi gửi đến LLM nhằm xoá dữ liệu hình ảnh trong bối cảnh nhật ký cuộc 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 tôi vẫn muốn nhân viên hỗ trợ có được bối cảnh dữ liệu hình ảnh khi cần. Do đó, chúng ta sẽ 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 mã nhận dạng đó. Dữ liệu này có thể được sử dụng cả khi 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 trở nên lỗi thời trong nhật ký trò chuyện, giá trị nhận dạng chuỗi vẫn còn đó để vẫn cho phép truy cập vào dữ liệu nhờ việc 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
touch expense_manager_agent/callbacks.py
Mở tệp expense_manager_agent/callbacks.py, sau đó sao chép đoạn 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. 🚀 Lời nhắc
Để thiết kế một trợ lý ảo có khả năng tương tác và có nhiều chức năng phức tạp, chúng ta cần tìm một câu lệnh đủ tốt để hướng dẫn trợ lý ảo đó sao cho nó có thể hoạt động theo cách chúng ta muốn.
Trước đây, chúng tôi có cơ chế về cách xử lý dữ liệu hình ảnh trong lịch sử hội thoại và cũng 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 tôi cũng muốn tác nhân có thể tìm kiếm và truy xuất hình ảnh biên lai chính xác cho chúng tôi. Điều này có nghĩa là chúng ta cần truyền tải tất cả thông tin này một cách phù hợp trong cấu trúc câu lệnh phù hợp
Chúng tôi sẽ yêu cầu trợ lý sắp xếp đầu ra theo định dạng markdown sau đây để phân tích 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>]",
...
]
}
Chúng ta hãy bắt đầu bằng 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 phí. Tệp task_prompt.md phải tồn tại trong thư mục làm việc hiện tại của chúng ta, nhưng chúng ta cần di chuyển tệp này vào thư mục expense_manager_agent. Chạy lệnh sau để di chuyển nó
mv task_prompt.md expense_manager_agent/task_prompt.md
7. 🚀 Kiểm tra tác nhân
Bây giờ chúng ta 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
Công cụ này sẽ cho ra kết quả như thế này, trong đó bạn có thể trò chuyện lần lượt với trợ lý ảo, tuy nhiên, bạn chỉ có thể gửi văn bản thông 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:
Hiện nay, bên cạnh tương tác CLI, ADK còn cho phép chúng ta có 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
Nó sẽ tạo ra kết quả như ví dụ sau, nghĩa 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 Web Preview (Xem trước trên web) ở khu vực trên cùng của Cloud Shell Editor rồi chọn Preview on port 8080 (Xem trước trên cổng 8080)
Bạn sẽ thấy trang web sau, nơi bạn có thể chọn các tác nhân có sẵn trên nút thả xuống ở góc trên bên trái ( trong trường hợp của chúng tôi 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 ở cửa sổ bên trái
Hãy thử một số hành động nhé! Tải lên 2 biên lai mẫu này ( nguồn : Bộ 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 Lưu hình ảnh dưới dạng... ( điều này sẽ tải xuống hình ảnh biên lai), sau đó tải tệp lên bot bằng cách nhấp vào biểu tượng "clip" và nói rằng bạn muốn lưu trữ các biên lai này
Sau đó hãy thử các truy vấn sau để thực hiện một số tìm kiếm hoặc truy xuất tệp
- "Cung cấp bảng phân tích chi phí và tổng chi phí trong năm 2023"
- "Đưa cho tôi file biên lai từ 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 với bạn và kiểm tra xem tác nhân đó có tuân thủ mọi 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.
Bây giờ là lúc hoàn thiện nó với giao diện người dùng đẹp mắt và phù hợp cùng khả năng tải lên và tải xuống tệp hình ảnh.
8. 🚀 Xây dựng 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 có dạng như sau
Nó bao gồm một giao diện trò chuyện với trường nhập liệu để người dùng gửi văn bản và tải lên tệp hình ảnh biên lai.
Chúng tôi sẽ xây dựng dịch vụ giao diện người dùng bằng Gradio.
Tạo tệp mới và đặt tên là frontend.py
touch frontend.py
sau đó sao chép mã sau và lưu lại
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ụ frontend 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 đầu ra tương tự như thế này trong bảng điều khiển đám mây của bạn
* 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 bạn ctrl+nhấp vào liên kết URL cục bộ. Ngoài ra, bạn cũng có thể truy cập ứ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 bên phải của Trình chỉnh sửa đám mây và 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 nó ngay. Chúng tôi sẽ chạy dịch vụ phụ trợ trong một tab thiết bị đầu cuối khác
Giải thích mã
Trong mã giao diện này, trước tiên chúng tôi cho phép người dùng gửi văn bản và tải lên nhiều tệp. Gradio cho phép chúng ta tạo ra 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 xác định MIMETYPE của tệp theo nhu cầu của phần phụ trợ. Chúng ta cũng cần mã hóa byte tệp hình ảnh thành base64 và gửi nó 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
Sơ đồ được sử dụng cho tương tác giữa giao diện người dùng và giao diện quản trị được định nghĩa trong schema.py. Chúng tôi sử dụng Pydantic BaseModel để thực thi xác thực dữ liệu trong lược đồ
Khi nhận được phản hồi, chúng tôi đã tách riêng phần nào là 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. 🚀 Xây dựng dịch vụ Backend bằng FastAPI
Tiếp theo, chúng ta sẽ cần xây dựng phần phụ trợ có thể khởi tạo Agent cùng với các thành phần khác để có thể thực thi thời gian chạy của Agent.
Tạo tệp mới và đặt tên là backend.py
touch backend.py
Và sao chép mã sau
from expense_manager_agent.agent import root_agent as expense_manager_agent
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.adk.events import Event
from fastapi import FastAPI, Body, Depends
from typing import AsyncIterator
from types import SimpleNamespace
import uvicorn
from contextlib import asynccontextmanager
from utils import (
extract_attachment_ids_and_sanitize_response,
download_image_from_gcs,
extract_thinking_process,
format_user_request_to_adk_content_and_store_artifacts,
)
from schema import ImageData, ChatRequest, ChatResponse
import logger
from google.adk.artifacts import GcsArtifactService
from settings import get_settings
SETTINGS = get_settings()
APP_NAME = "expense_manager_app"
# Application state to hold service contexts
class AppContexts(SimpleNamespace):
"""A class to hold application contexts with attribute access"""
session_service: InMemorySessionService = None
artifact_service: GcsArtifactService = None
expense_manager_agent_runner: Runner = None
# Initialize application state
app_contexts = AppContexts()
@asynccontextmanager
async def lifespan(app: FastAPI):
# Initialize service contexts during application startup
app_contexts.session_service = InMemorySessionService()
app_contexts.artifact_service = GcsArtifactService(
bucket_name=SETTINGS.STORAGE_BUCKET_NAME
)
app_contexts.expense_manager_agent_runner = Runner(
agent=expense_manager_agent, # The agent we want to run
app_name=APP_NAME, # Associates runs with our app
session_service=app_contexts.session_service, # Uses our session manager
artifact_service=app_contexts.artifact_service, # Uses our artifact manager
)
logger.info("Application started successfully")
yield
logger.info("Application shutting down")
# Perform cleanup during application shutdown if necessary
# Helper function to get application state as a dependency
async def get_app_contexts() -> AppContexts:
return app_contexts
# Create FastAPI app
app = FastAPI(title="Personal Expense Assistant API", lifespan=lifespan)
@app.post("/chat", response_model=ChatResponse)
async def chat(
request: ChatRequest = Body(...),
app_context: AppContexts = Depends(get_app_contexts),
) -> ChatResponse:
"""Process chat request and get response from the agent"""
# Prepare the user's message in ADK format and store image artifacts
content = await format_user_request_to_adk_content_and_store_artifacts(
request=request,
app_name=APP_NAME,
artifact_service=app_context.artifact_service,
)
final_response_text = "Agent did not produce a final response." # Default
# Use the session ID from the request or default if not provided
session_id = request.session_id
user_id = request.user_id
# Create session if it doesn't exist
if not await app_context.session_service.get_session(
app_name=APP_NAME, user_id=user_id, session_id=session_id
):
await app_context.session_service.create_session(
app_name=APP_NAME, user_id=user_id, session_id=session_id
)
try:
# Process the message with the agent
# Type annotation: runner.run_async returns an AsyncIterator[Event]
events_iterator: AsyncIterator[Event] = (
app_context.expense_manager_agent_runner.run_async(
user_id=user_id, session_id=session_id, new_message=content
)
)
async for event in events_iterator: # event has type Event
# Key Concept: is_final_response() marks the concluding message for the turn
if event.is_final_response():
if event.content and event.content.parts:
# Extract text from the first part
final_response_text = event.content.parts[0].text
elif event.actions and event.actions.escalate:
# Handle potential errors/escalations
final_response_text = f"Agent escalated: {event.error_message or 'No specific message.'}"
break # Stop processing events once the final response is found
logger.info(
"Received final response from agent", raw_final_response=final_response_text
)
# Extract and process any attachments and thinking process in the response
base64_attachments = []
sanitized_text, attachment_ids = extract_attachment_ids_and_sanitize_response(
final_response_text
)
sanitized_text, thinking_process = extract_thinking_process(sanitized_text)
# Download images from GCS and replace hash IDs with base64 data
for image_hash_id in attachment_ids:
# Download image data and get MIME type
result = await download_image_from_gcs(
artifact_service=app_context.artifact_service,
image_hash=image_hash_id,
app_name=APP_NAME,
user_id=user_id,
session_id=session_id,
)
if result:
base64_data, mime_type = result
base64_attachments.append(
ImageData(serialized_image=base64_data, mime_type=mime_type)
)
logger.info(
"Processed response with attachments",
sanitized_response=sanitized_text,
thinking_process=thinking_process,
attachment_ids=attachment_ids,
)
return ChatResponse(
response=sanitized_text,
thinking_process=thinking_process,
attachments=base64_attachments,
)
except Exception as e:
logger.error("Error processing chat request", error_message=str(e))
return ChatResponse(
response="", error=f"Error in generating response: {str(e)}"
)
# Only run the server if this file is executed directly
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8081)
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 đúng dịch vụ frontend, bây giờ chúng ta sẽ cần mở terminal mới và thử chạy dịch vụ backend này
- Tạo một thiết bị đầu cuối mới. Điều hướng đến thiết bị đầu cuối của bạn ở khu vực bên dưới và tìm nút "+" để tạo thiết bị đầu cuối mới. Ngoài ra, bạn có thể sử dụng Ctrl + Shift + C để mở terminal mới
- Sau đó, hãy đảm bảo rằng bạn đang ở trong thư mục làm việc personal-expense-assistant rồi chạy lệnh sau
uv run backend.py
- Nếu thành công, kết quả sẽ 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 tạo ADK Agent, SessionService và ArtifactService
Để chạy tác nhân trong dịch vụ phụ trợ, chúng ta sẽ cần tạo một Runner để sử dụng 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 trò chuyện. Do đó, khi được tích hợp với Runner, dịch vụ này sẽ giúp tác nhân của chúng tôi có khả năng nhận được ngữ cảnh của các 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ề ADK Session và Artifacts 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 demo này, chúng tôi sử dụng InMemorySessionService và GcsArtifactService để tích hợp với tác nhân Runner. Vì lịch sử trò chuyện được lưu trữ trong bộ nhớ nên nó sẽ bị mất khi dịch vụ phụ trợ bị tắt hoặc khởi động lại. Chúng ta sẽ khởi tạo các đối tượng này 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 đường /chat.
Tải lên và tải xuống hình ảnh với GcsArtifactService
Tất cả hình ảnh được tải lên sẽ được lưu trữ dưới dạng cấu phần phần mềm bằng GcsArtifactService, bạn có thể kiểm tra cấu phần phần mềm 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 xử lý bởi trình chạy tác nhân cần được định dạng thành kiểu types.Content. Bên trong hàm, chúng tôi cũng xử lý từng dữ liệu hình ảnh và trích xuất ID của hình ảnh đó để thay thế bằng ID hình ảnh giữ chỗ.
Cơ chế tương tự được sử dụng để tải xuống các tệp đính kèm sau khi trích xuất ID hình ảnh bằng regex:
...
sanitized_text, attachment_ids = extract_attachment_ids_and_sanitize_response(
final_response_text
)
sanitized_text, thinking_process = extract_thinking_process(sanitized_text)
# Download images from GCS and replace hash IDs with base64 data
for image_hash_id in attachment_ids:
# Download image data and get MIME type
result = await asyncio.to_thread(
download_image_from_gcs,
artifact_service=app_context.artifact_service,
image_hash=image_hash_id,
app_name=APP_NAME,
user_id=user_id,
session_id=session_id,
)
...
10. 🚀 Kiểm thử tích hợp
Giờ đây, bạn sẽ có nhiều dịch vụ chạy trong các thẻ bảng điều khiển đám mây khác nhau:
- 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 lai 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 và chọn Xem trước trên cổng 8080
Bây giờ chúng ta hãy tương tác với trợ lý nhé!
Tải xuống các biên lai sau. 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: tập dữ liệu Hugging Face
mousserlane/id_receipt_dataset)
Hỏi nhiều thứ khác nhau
- "Cho tôi biết chi tiết chi phí hàng tháng trong năm 2023-2024"
- "Cho tôi xem biên lai giao dịch cà phê"
- "Give me receipt file from Yakiniku Like" (Gửi cho tôi tệp biên nhận của Yakiniku Like)
- Etc
Sau đây là một đoạn 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 ở mọi nơi. Để làm như vậy, chúng ta có thể đóng gói ứng dụng này và triển khai nó trên Cloud Run. Để minh hoạ, dịch vụ này sẽ được cung cấp dưới dạng một 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ì nó 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 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.
Đến đây, chúng ta đã có tất cả các tệp cần thiết để triển khai ứng dụng lê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 cho dự án đang hoạt động của bạn. Nếu không, bạn phải 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 này vào 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 sổ đăng ký hiện vật cho kho lưu trữ docker, hãy trả lời Y. Xin lưu ý rằng chúng tôi đang cho phép truy cập chưa xác thực tại đây vì đây là một ứ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 công khai.
Sau khi quá trình triển khai hoàn tất, bạn sẽ nhận được một đường liên kết tương tự như đường liên kết 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. Nội dung đó đã được phát trực tiếp.
12. 🎯 Thách thức
Giờ là lúc bạn thể hiện và trau dồi kỹ năng khám phá của mình. Bạn có đủ khả năng để thay đổi mã để phần phụ trợ có thể đáp ứng 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 phát sinh chi phí cho tài khoản Google Cloud của bạn đối với các tài nguyên được sử dụng trong codelab này, hãy làm theo các bước sau:
- Trong bảng điều khiển Google Cloud, hãy đi tới 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 xóa, sau đó nhấp vào Xóa.
- Trong hộp thoại, hãy nhập mã dự án rồi nhấp vào 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á.