運用服務專員開發套件邁向多模態：使用 Gemini 2.5 的個人支出助理、Firestore 和 Cloud Run

1. 📖 簡介

重要注意事項：

本教學需要一個具有 個有效結算帳戶的 Google Cloud 專案。如果您還沒有帳戶，請依照下列步驟取得可用於本教學的試用計費帳戶：

1. 開啟無痕視窗
2. 如要參加研討會，通常會在研討會中提供兌換入口網站網址，否則您可以前往這個試用帳單帳戶兌換入口網站
3. 使用個人 Gmail 帳戶登入( Google Workspace / 其他電子郵件無法登入！)
4. 按照兌換說明逐步操作，為本教學課程建立新的 Google Cloud 專案

你是否曾經因為懶惰而無法妥善管理所有個人開支？我也是！因此，在這個代碼實驗室中，我們將建立一個個人費用管理助理——由 Gemini 2.5 提供支持，為我們完成所有瑣事！從管理上傳的收據到分析你是否已經花太多錢買咖啡！

你可以透過網頁瀏覽器存取這個助理，以聊天網頁介面的形式與其互動、上傳收據圖片並要求助理儲存，或搜尋收據以取得檔案並進行費用分析。所有這些功能都以 Google Agent Development Kit 架構為基礎

該應用程式本身分為 2 個服務：前端和後端；使您能夠快速建立原型並體驗其感覺，還可以瞭解 API 協定如何將它們整合在一起。

透過程式碼實驗室，您將採用以下逐步方法：

1. 準備好您的 Google Cloud 專案並啟用所有必要的 API。
2. 在 Google Cloud Storage 上設定儲存桶，並在 Firestore 上設定資料庫。
3. 建立 Firestore 索引
4. 設定編碼環境的工作區
5. 建構 ADK 代理程式原始碼、工具、提示符等結構。
6. 使用 ADK 本機 Web 開發 UI 測試代理
7. 使用 Gradio 庫建立前端服務－聊天介面，用於傳送查詢和上傳收據圖片。
8. 使用 FastAPI 建構後端服務 - HTTP 伺服器，其中包含 ADK 代理程式碼、SessionService 和 Artifact Service
9. 管理環境變數並設定將應用程式部署到 Cloud Run 所需的必要文件
10. 將應用程式部署到 Cloud Run

架構總覽

必要條件

• 熟練使用 Python
• 瞭解使用 HTTP 服務的基本全端架構

課程內容

• 使用 Gradio 製作前端網頁原型
• 使用 FastAPI 和 Pydantic 開發後端服務
• 利用 ADK Agent 的多種功能建構其架構
• 工具使用
• 會話和工件管理
• 在傳送至 Gemini 前，使用回呼修改輸入內容
• 利用內建規劃器進行規劃，從而改善任務執行。
• 透過 ADK 本機 Web 介面快速偵錯
• 透過提示工程和使用 ADK 回呼修改 Gemini 要求，剖析及擷取資訊，進而最佳化多模態互動的策略
• 使用 Firestore 作為向量資料庫的代理程式檢索增強生成
• 使用 Pydantic-settings 在 YAML 檔案中管理環境變數
• 使用 Dockerfile 將應用程式部署至 Cloud Run，並透過 YAML 檔案提供環境變數

軟硬體需求

• Chrome 瀏覽器
• Gmail 帳戶
• 已啟用計費功能的 Cloud 專案

本程式碼研究室適用於所有程度的開發人員 (包括初學者)，並在範例應用程式中使用 Python。不過，您不需要具備 Python 知識，也能瞭解本文介紹的概念。

2. 🚀 事前準備

在雲端控制台中選擇活動項目

本程式碼研究室假設您已擁有啟用計費功能的 Google Cloud 專案。如果尚未取得，請按照下列操作說明開始使用。

1. 在 Google Cloud 控制台 的專案選擇器頁面上，選擇或建立一個 Google Cloud 專案。
2. 確認 Cloud 專案已啟用計費功能。學習如何檢查項目是否已啟用計費功能。

準備 Firestore 資料庫

接著，我們也需要建立 Firestore 資料庫。Firestore (原生模式) 是專為自動調整資源配置、發揮高效能及協助開發應用程式所打造的 NoSQL 文件資料庫。此外，這項服務也能做為向量資料庫，支援實驗室的檢索增強生成技術。

1. 在搜尋列中搜尋「firestore」，然後點選 Firestore 產品。

2. 然後點選「建立 Firestore 資料庫」按鈕。
3. 使用 (預設) 作為資料庫 ID 名稱，並保持選取 標準版。為了本次實驗室演示，請使用 Firestore Native 和 Open 安全規則。

注意！開放安全規則「不建議」用於正式環境！

4. 您也會發現這個資料庫實際上具有「免費方案用量 YEAY!」接著按一下「建立資料庫」按鈕

完成這些步驟後，系統應該會將您重新導向至剛建立的 Firestore 資料庫

在 Cloud Shell 終端機中設定 Cloud 專案

1. 您將使用 Cloud Shell，這是 Google Cloud 中執行的指令列環境，且已預先載入 bq。點擊 Google Cloud 控制台頂部的「啟動 Cloud Shell」。

2. 連線至 Cloud Shell 後，請使用下列指令檢查您是否已通過驗證，且專案已設為您的專案 ID：

gcloud auth list

3. 在 Cloud Shell 中執行下列指令，確認 gcloud 指令已瞭解您的專案。

gcloud config list project

4. 如果未設定專案，請使用下列指令設定：

gcloud config set project <YOUR_PROJECT_ID>

或者，您也可以在控制台中查看 PROJECT_ID id。

按一下該專案，右側就會顯示所有專案和專案 ID

5. 透過下列指令啟用必要的 API。這可能需要幾分鐘的時間，請耐心等候。

gcloud services enable aiplatform.googleapis.com \
                       firestore.googleapis.com \
                       run.googleapis.com \
                       cloudbuild.googleapis.com \
                       cloudresourcemanager.googleapis.com

成功執行指令後，您應該會看到類似下方的訊息：

Operation "operations/..." finished successfully.

除了使用 gcloud 指令，您也可以透過控制台搜尋各項產品，或使用這個連結。

如果遺漏任何 API，您隨時可以在導入過程中啟用。

如要瞭解 gcloud 指令和用法，請參閱說明文件。

準備 Google Cloud Storage 儲存桶

接著，我們需要從同一個終端機準備 GCS bucket，以便儲存上傳的檔案。執行下列指令來建立 bucket，您需要不重複但與個人支出助理收據相關的 bucket 名稱，因此我們會使用下列 bucket 名稱，並結合您的專案 ID

gsutil mb -l us-central1 gs://personal-expense-{your-project-id}

重要事項：請將 {your-project-id} 的值更新為您自己的專案 ID

輸出內容如下：

Creating gs://personal-expense-{your-project-id}

如要確認，請前往瀏覽器左上方的「導覽選單」，然後選取「Cloud Storage」->「Bucket」。

為搜尋功能建立 Firestore 索引

Firestore 本身是 NoSQL 資料庫，在資料模型方面提供卓越效能和彈性，但在複雜查詢方面則有其限制。由於我們打算使用一些複合多欄位查詢和向量搜尋，因此需要先建立一些索引。如要進一步瞭解詳細資料，請參閱這份說明文件。

請注意！執行下列指令需要幾分鐘的時間。您可以在不同的終端機分頁中執行這些指令，然後放著不管，直到完成為止

1. 執行下列指令，建立索引來支援複合查詢

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)"

2. 並執行這個指令來支援向量搜尋

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)"

如要查看建立的索引，請前往 Cloud 控制台中的 Firestore，點選「(default)」資料庫執行個體，然後選取導覽列上的「Indexes」(索引)。

前往 Cloud Shell 編輯器並設定應用程式工作目錄

現在，我們可以設定程式碼編輯器，進行一些程式設計工作。我們會使用 Cloud Shell 編輯器執行這項操作

1. 按一下「Open Editor」(開啟編輯器) 按鈕，開啟 Cloud Shell 編輯器，即可在此編寫程式碼
2. 接著，我們也需要檢查 Shell 是否已設定為正確的 PROJECT ID。如果終端機的 $ 圖示前有 ( ) 內的值 ( 在下方螢幕截圖中，該值為「adk-multimodal-tool」)，則表示已為進行中的 Shell 工作階段設定專案。

如果顯示的值正確無誤，可以略過下一個指令。但如果該值不正確或遺失，請執行下列指令

gcloud config set project <YOUR_PROJECT_ID>

3. 接著，請從 GitHub 複製本程式碼研究室的範本工作目錄，執行下列指令。系統會在 personal-expense-assistant 目錄中建立工作目錄

git clone https://github.com/alphinside/personal-expense-assistant-adk-codelab-starter.git personal-expense-assistant

4. 完成後，前往 Cloud Shell 編輯器頂端，依序點選「File」->「Open Folder」，找到「username」目錄和「personal-expense-assistant」目錄，然後點選「OK」按鈕。這會將所選目錄設為主要工作目錄。在本範例中，使用者名稱為 alvinprayuda，因此目錄路徑如下所示

現在，Cloud Shell 編輯器應如下所示

環境設定

準備 Python 虛擬環境

下一步是準備開發環境。目前有效的終端機應位於 personal-expense-assistant 工作目錄中。在本程式碼研究室中，我們將使用 Python 3.12，並使用 uv Python 專案管理工具，簡化建立及管理 Python 版本和虛擬環境的需求

1. 如果尚未開啟終端機，請依序點選「Terminal」(終端機) ->「New Terminal」(新增終端機)，或使用 Ctrl + Shift + C 鍵，在瀏覽器底部開啟終端機視窗

2. 現在請使用 uv 初始化虛擬環境，執行下列指令：

cd ~/personal-expense-assistant
uv sync --frozen

這將建立 .venv 目錄並安裝相依性。快速查看 pyproject.toml 文件，即可取得如下所示的依賴項資訊。

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",
]

注意：Python 依附元件會在 pyproject.toml 中宣告，並在 uv.lock 檔案中鎖定版本

注意: 我們不需要手動建立虛擬環境或啟動它，因為 uv 已經處理了所有這些。從這一點開始，標準 python 指令（例如 python main.py）將會被 uv run（例如 uv run main.py）取代。

設定設定檔

現在我們需要為這個專案設定設定檔。我們使用 pydantic-settings 從 YAML 檔案中讀取配置。

我們已在 settings.yaml.example 中提供檔案範本，您需要複製該檔案並重新命名為 settings.yaml。運行此命令創建文件

cp settings.yaml.example settings.yaml

然後，將以下值複製到文件中

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"

重要提示：更新以下值GCLOUD_PROJECT_ID以及STORAGE_BUCKET_NAME使用正確的值

對於本程式碼實驗室，我們將使用 GCLOUD_LOCATION,、BACKEND_URL, 和 DB_COLLECTION_NAME 的預先配置值。

注意：settings.yaml 檔案將從 settings.py 讀取。pydantic-settings 不僅允許我們輕鬆強制執行資料類型，還可以動態地設定配置來源優先權。在本演示中，我們將優先順序設定如下：

1. 從環境變數取得 -> 部署後更改環境變數時非常有用
2. 從 settings.yaml 檔案 -> 在未設定任何環境變數時，初始化首次部署作業時很有幫助
3. 從程式碼初始化 -> 處理預設值

現在我們可以進入下一步，建立代理，然後是服務。

3. 🚀 使用 Google ADK 和 Gemini 2.5 建立代理

ADK 目錄結構簡介

首先，我們來瞭解 ADK 的功能，以及如何建構代理程式。如要查看 ADK 完整說明文件，請前往這個網址。ADK 在執行 CLI 指令時提供許多公用程式。部分範例如下：

• 設定代理程式目錄結構
• 透過 CLI 輸入/輸出快速試用互動功能
• 快速設定本機開發 UI 網頁介面

現在，讓我們使用 CLI 命令建立代理目錄結構。執行下列指令。

uv run adk create expense_manager_agent

系統詢問時，請選擇模型 gemini-2.5-flash 和 Vertex AI 後端。精靈接著會要求您提供專案 ID 和位置。按下 Enter 鍵即可接受預設選項，或視需要變更選項。請再次確認您使用的是稍早在本實驗室中建立的正確專案 ID。輸出內容如下所示：

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

它將建立以下代理目錄結構

expense_manager_agent/
├── __init__.py
├── .env
├── agent.py

如果您檢查 init.py 和 agent.py，會看到這段程式碼

# __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',
)

注意：ADK CLI 必須能辨識目錄結構、init.py 內的代理程式匯入項目，以及 agent.py 中的 root_agent 宣告，因此請務必採用這些結構。

現在你可以透過執行以下命令來測試它：

uv run adk run expense_manager_agent

測試完成後，您可以輸入 exit 或按 Ctrl+D 退出代理程式。

建構我們的費用管理代理

讓我們來建立費用管理代理程式！開啟 expense_manager_agent/agent.py 文件，並複製以下程式碼，其中將包含 root_agent.

註: 為了更好地管理，我們將說明提示資訊拆分到一個單獨的 Markdown 文件中，稍後會詳細介紹。

# 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,
)

程式碼說明

這個指令碼包含代理程式啟動程序，我們會初始化下列項目：

• 設定要用於 gemini-2.5-flash 的模型
• 將代理程式說明和指令設為系統提示，系統會從 task_prompt.md 讀取這些提示
• 提供支援代理程式功能的必要工具
• 利用 Gemini 2.5 的 Flash 思維功能，在產生最終回應或執行之前進行規劃。
• 在將要求傳送至 Gemini 前，先設定回呼攔截，限制預測前傳送的圖片資料數量

4. 🚀 設定代理工具

我們的費用管理代理程式將具備下列功能：

• 從收據圖片擷取資料，並儲存資料和檔案
• 對費用數據進行精確搜索
• 對費用數據進行上下文搜索

因此，我們需要合適的工具來支援這項功能。在 expense_manager_agent 目錄下建立一個新文件，並將其命名為 tools.py

touch expense_manager_agent/tools.py

開啟 expense_manage_agent/tools.py，然後複製下列程式碼

注意： 函數文件字串非常重要，應該盡可能詳細地描述，以便更好地指導代理何時使用它。盡量使用簡單資料型別作為函數參數型別。

# 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

代碼解釋

在這個工具函式實作中，我們會根據以下 2 個主要概念設計工具：

• 解析收據資料並使用圖像 ID 字串佔位符 [IMAGE-ID <hash-of-image-1>] 映射到原始文件
• 使用 Firestore 資料庫儲存和檢索數據

工具「store_receipt_data」

該工具是光學字元辨識工具，它將從圖像資料中解析所需信息，識別圖像 ID 字串，並將它們映射在一起以儲存在 Firestore 資料庫中。

此外，該工具還使用 text-embedding-004 將收據內容轉換為嵌入，以便將所有元資料和嵌入一起儲存和索引。可透過查詢或情境搜尋擷取資料。

成功執行此工具後，您可以看到收據資料已在 Firestore 資料庫中建立索引，如下所示。

工具「以元資料篩選搜尋收據」

此工具將使用者查詢轉換為元資料查詢過濾器，支援按日期範圍和/或交易總額進行搜尋。它將返回所有匹配的收據數據，在此過程中，我們將刪除嵌入字段，因為代理不需要它來進行上下文理解。

工具「依自然語言查詢搜尋相關收據」

這是我們的檢索增強生成（RAG）工具。我們的代理商能夠設計自己的查詢，從向量資料庫中檢索相關的收據，並且還可以選擇何時使用此工具。允許代理獨立決定是否使用此 RAG 工具並設計自己的查詢，是 Agentic RAG 方法的定義之一。

我們不僅允許它建立自己的查詢，還允許它選擇要檢索多少個相關文件。結合適當的及時工程，例如

# Example prompt

Always filter the result from tool
search_relevant_receipts_by_natural_language_query as the returned 
result may contain irrelevant information

這項工具將能搜尋幾乎所有內容，但由於最鄰近搜尋並非完全比對，因此可能無法傳回所有預期結果。

挑戰！你能想到結合這兩種不同方法進行搜尋的其他方法嗎？

5. 🚀 透過回調修改對話上下文

Google ADK 使我們能夠在各個層面上「攔截」代理程式運行時。您可以在此文件中閱讀更多有關此詳細功能的資訊。在本實驗中，我們利用 before_model_callback 在將請求傳送到 LLM 之前對其進行修改，以移除舊對話歷史上下文中的影像資料（僅包含最近 3 次使用者互動中的影像資料），從而提高效率。

不過，我們仍希望服務專員在需要時，能取得圖片資料的背景資訊。因此，我們在對話中的每個圖片位元組資料後，新增了字串圖片 ID 預留位置的機制。這有助於代理程式將圖片 ID 連結至實際檔案資料，以便在儲存或擷取圖片時使用。結構如下所示

<image-byte-data-1>
[IMAGE-ID <hash-of-image-1>]
<image-byte-data-2>
[IMAGE-ID <hash-of-image-2>]
And so on..

即使對話記錄中的位元組資料過時，字串 ID 仍會保留，方便您使用工具存取資料。移除圖片資料後的記錄結構範例

[IMAGE-ID <hash-of-image-1>]
[IMAGE-ID <hash-of-image-2>]
And so on..

挑戰！您能否找出更有效率的策略？與同儕討論！

立即開始！在 expense_manager_agent 目錄下建立新檔案，並命名為 callbacks.py

touch expense_manager_agent/callbacks.py

開啟 expense_manager_agent/callbacks.py 檔案，然後複製下列程式碼

# 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. 🚀 提示

設計具有複雜互動和功能的代理程式時，我們需要找到足夠好的提示來引導代理程式，讓代理程式的行為符合我們的期望。

我們先前有處理對話記錄中圖片資料的機制，也有可能難以使用的工具，例如 search_relevant_receipts_by_natural_language_query.。我們也希望代理程式能夠搜尋並擷取正確的收據圖片。也就是說，我們必須在適當的提示結構中，正確傳達所有這些資訊

提示：如要進一步瞭解提示工程資源，請參閱這個網頁。

我們會要求代理程式將輸出內容整理成下列 Markdown 格式，以便剖析思考過程、最終回覆和附件 ( 如有)

# 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>]",
      ...
    ]
}

首先，請使用下列提示，達成我們對費用管理代理程式行為的初步期望。task_prompt.md 檔案應已存在於現有的工作目錄中，但我們需要將其移至 expense_manager_agent 目錄下。執行下列指令來移動該檔案

mv task_prompt.md expense_manager_agent/task_prompt.md

注意：請參閱提示，瞭解我們如何避免插入重複資料，以及如何嚴格搜尋類似收據

挑戰！您是否是隱藏的提示大師？分享如何改善這項提示的洞察資料 (或修正一些非預期的回覆 😉)

7. 🚀 測試代理程式

現在我們嘗試透過 CLI 與代理通信，運行以下命令

uv run adk run expense_manager_agent

系統會顯示類似下方的輸出內容，您可輪流與代理程式對話，但只能透過這個介面傳送文字

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: 

🤖試著詢問它的功能：「你可以幫我做什麼？」

現在，除了 CLI 互動之外，ADK 也提供開發 UI，方便我們在互動期間進行互動及檢查。執行下列指令，啟動本機開發 UI 伺服器

uv run adk web --port 8080

這會產生類似下列範例的輸出內容，表示我們已可存取網頁介面

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)

如要檢查，請點選 Cloud Shell 編輯器頂端的「Web Preview」(網頁預覽) 按鈕，然後選取「Preview on port 8080」(透過以下通訊埠預覽：8080)。

您會看到下列網頁，可以在左上方的下拉式按鈕中選取可用的代理程式 ( 在本例中應為 expense_manager_agent)，並與機器人互動。在左側視窗中，您會看到代理程式執行階段的記錄詳細資料

現在來試試一些動作！上傳這 2 張收據範例 ( 來源：Hugging Face 資料集 mousserlane/id_receipt_dataset)。在每張圖片上按一下滑鼠右鍵，然後選擇「另存圖片…」( 系統會下載收據圖片)，然後按一下「迴紋針」圖示，將檔案上傳至機器人，並說明要儲存這些收據

接著，請嘗試下列查詢來搜尋或擷取檔案

• 「請提供 2023 年的支出明細和總額」
• 「Give me receipt file from Indomaret」(給我 Indomaret 的收據檔案)

使用部分工具時，您可以檢查開發 UI 中的情況

查看代理如何回應您，並檢查它是否符合 task_prompt.py 中提示提供的所有規則。恭喜！ 現在您擁有了一個功能齊全的開發代理程式。

現在是時候完善它，使其擁有美觀的用戶介面以及上傳和下載圖像檔案的功能了。

提醒事項！現在我們先終止 adk web 進程，因為我們將重新利用 8080 連接埠為前端服務提供服務。你可以試著按下 Ctrl + C 鍵終止程序

8. 🚀 使用 Gradio 建構前端服務

我們將建立一個如下所示的聊天網頁介面。

它包含一個聊天介面，使用者可以透過輸入欄位發送文字和上傳收據圖像檔案。

我們將使用 Gradio 來建立前端服務。

建立一個新檔案並將其命名為 frontend.py

touch frontend.py

然後複製下列程式碼並儲存

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,
    )

接著，我們可以嘗試執行下列指令來執行前端服務。別忘了將 main.py 檔案重新命名為 frontend.py

uv run frontend.py

您將在雲端控制台中看到類似這樣的輸出。

* Running on local URL:  http://0.0.0.0:8080

To create a public link, set `share=True` in `launch()`.

之後，您可以查看網頁介面。按住 Ctrl 鍵並點選本地 URL 連結。或者，您也可以點擊以下連結存取前端應用程式：網頁預覽點擊雲端編輯器右上角的按鈕，然後選擇埠 8080 預覽

您將看到網頁介面，但由於後端服務尚未設置，嘗試提交聊天時會收到 預期錯誤。

現在，讓服務繼續運行，暫時不要終止它。我們將在另一個終端標籤頁中運行後端服務。

代碼解釋

在這個前端程式碼中，首先要讓使用者能夠傳送文字和上傳多個檔案。Gradio 讓我們可以結合使用 gr.ChatInterface 方法和 gr.MultimodalTextbox 來建立這個功能。

現在，在將檔案和文字傳送至後端之前，我們需要找出檔案的 MIME 類型，因為後端需要這項資訊。我們也需要將圖片檔案位元編碼為 base64，並與 MIME 類型一併傳送。

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

用於前端 - 後端互動的結構定義是在 schema.py 中定義。我們使用 Pydantic BaseModel 在結構定義中強制執行資料驗證

收到回覆時，我們已將思考過程、最終回覆和附件分開。因此，我們可以利用 Gradio 元件，透過 UI 元件顯示每個元件。

class ChatResponse(BaseModel):
    """Model for a chat response.

    Attributes:
        response: The text response from the model.
        thinking_process: Optional thinking process of the model.
        attachments: List of image data to be displayed to the user.
        error: Optional error message if something went wrong.
    """

    response: str
    thinking_process: str = ""
    attachments: List[ImageData] = []
    error: Optional[str] = None

9. 🚀 使用 FastAPI 建置後端服務

接著，我們需要建構後端，以便初始化 Agent 和其他元件，執行 Agent 執行階段。

建立新檔案，並命名為 backend.py

touch 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
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)

之後我們就可以嘗試執行後端服務了。請記住，在上一個步驟中我們已經成功地運行了前端服務，現在我們需要開啟一個新的終端並嘗試執行後端服務。

1. 建立新終端機。前往底部的終端機，然後找到「+」按鈕，即可建立新的終端機。或者，您也可以按下 Ctrl + Shift + C 開啟新的終端機

2. 接著，請確認您位於工作目錄 personal-expense-assistant，然後執行下列指令

uv run backend.py

3. 如果成功，輸出結果會如下所示

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)

程式碼說明

初始化 ADK Agent、SessionService 和 ArtifactService

如要在後端服務中執行代理程式，我們需要建立 Runner，同時採用 SessionService 和我們的代理程式。SessionService 會管理對話記錄和狀態，因此與 Runner 整合後，我們的代理程式就能接收進行中對話的脈絡。

我們也會使用 ArtifactService 處理上傳的檔案。如要進一步瞭解 ADK 工作階段和構件，請參閱這篇文章和這篇文章。

...

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialize service contexts during application startup
    app_contexts.session_service = InMemorySessionService()
    app_contexts.artifact_service = GcsArtifactService(
        bucket_name=SETTINGS.STORAGE_BUCKET_NAME
    )
    app_contexts.expense_manager_agent_runner = Runner(
        agent=expense_manager_agent,  # The agent we want to run
        app_name=APP_NAME,  # Associates runs with our app
        session_service=app_contexts.session_service,  # Uses our session manager
        artifact_service=app_contexts.artifact_service,  # Uses our artifact manager
    )

    logger.info("Application started successfully")
    yield
    logger.info("Application shutting down")
    # Perform cleanup during application shutdown if necessary

...

在本示範中，我們會使用 InMemorySessionService 和 GcsArtifactService，與代理程式 Runner 整合。由於對話記錄會儲存在記憶體中，因此一旦後端服務終止或重新啟動，記錄就會遺失。我們會在 FastAPI 應用程式生命週期中初始化這些項目，以便在 /chat 路由中做為依附元件插入。

使用 GcsArtifactService 上傳及下載圖片

所有上傳的圖片都會由 GcsArtifactService 儲存為構件，您可以在 utils.py 內的 format_user_request_to_adk_content_and_store_artifacts 函式中查看。

...    

# 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,
)

...

代理程式執行器處理的所有要求都必須格式化為 types.Content 類型。在函式中，我們也會處理每個圖片資料，並擷取其 ID，以取代圖片 ID 預留位置。

注意：GcsArtifactService 提供結構化 Blob 儲存機制，並依 app_name、session_id 和 user_id 分組。這項功能可輕鬆擴充多個應用程式、工作階段或使用者的代理程式應用程式

使用規則運算式擷取圖片 ID 後，下載附件的機制也類似：

...
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. 🚀 整合測試

現在，您應該會在不同的 Cloud 控制台分頁中執行多項服務：

• 前端服務在通訊埠 8080 執行

* Running on local URL:  http://0.0.0.0:8080

To create a public link, set `share=True` in `launch()`.

• 後端服務在通訊埠 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)

目前，您應該能夠透過 8080 連接埠上的 Web 應用程式上傳收據圖片並與助手無縫聊天。

按一下 Cloud Shell 編輯器頂端的「Web Preview」(網頁預覽) 按鈕，然後選取「Preview on port 8080」(透過以下通訊埠預覽：8080)。

現在，讓我們與助理互動！

警告！繼續操作前，請避免一次上傳過多收據圖片 (一次最多 2 到 3 張)。系統會引發 429 速率限制錯誤，您必須等待幾分鐘後再試一次。上傳的圖片越多，回應時間就越長。

下載下列收據。這些收據資料的日期範圍介於 2023 年至 2024 年，並要求助理儲存/上傳這些資料

• Receipt Drive ( 來源：Hugging Face 資料集 mousserlane/id_receipt_dataset )

問各種各樣的問題

• 「Give me monthly expense breakdown during 2023-2024」(提供 2023 年至 2024 年的每月支出明細)
• 「顯示咖啡交易的收據」
• 「Give me receipt file from Yakiniku Like」(給我 Yakiniku Like 的收據檔案)
• ETC

以下是成功互動的程式碼片段

11. 🚀 部署至 Cloud Run

當然，我們希望隨時隨地都能存取這個絕佳的應用程式。為此，我們可以封裝這個應用程式，並部署至 Cloud Run。在本示範中，這項服務會公開，供他人存取。不過請注意，這類應用程式不適合採用這種做法，因為這比較適合個人應用程式

在本程式碼研究室中，我們將前端和後端服務都放在 1 個容器中。我們需要 supervisord 的協助，才能管理這兩項服務。您可以檢查 supervisord.conf 檔案，並查看我們將 supervisord 設為進入點的 Dockerfile。

此時，我們已備妥將應用程式部署至 Cloud Run 的所有必要檔案，現在就來部署吧。前往 Cloud Shell 終端機，確認目前專案已設為有效專案。如果不是，請使用 gcloud configure 指令設定專案 ID：

gcloud config set project [PROJECT_ID]

接著，執行下列指令將其部署至 Cloud Run。

gcloud run deploy personal-expense-assistant \
                  --source . \
                  --port=8080 \
                  --allow-unauthenticated \
                  --env-vars-file=settings.yaml \
                  --memory 1024Mi \
                  --region us-central1

如果系統提示您確認要為 Docker 存放區建立 Artifact Registry，請回答 Y。請注意，我們允許未經驗證的存取要求，因為這是示範應用程式。建議您為企業和正式版應用程式使用適當的驗證方式。

部署完成後，您會取得類似下方的連結：

https://personal-expense-assistant-*******.us-central1.run.app

請繼續在無痕視窗或行動裝置上使用應用程式。這項功能應該已經上線。

12. 🎯 挑戰

現在輪到你大顯身手，磨練你的探索技巧了。您是否具備修改程式碼以使後端能夠支援多個使用者的能力？需要更新哪些組件？

13. 🧹 清理

為避免因本程式碼實驗室中使用的資源而產生 Google Cloud 帳戶費用，請按照以下步驟操作：

1. 在 Google Cloud 控制台中，前往 管理資源 頁面。
2. 在專案清單中選取要刪除的專案，然後點按「刪除」。
3. 在對話方塊中，輸入項目 ID，然後按一下 關閉 刪除項目。
4. 或者，您也可以前往控制台的「Cloud Run」，選取剛部署的服務並刪除。