1. 📖 Введение
Вы когда-нибудь чувствовали себя разочарованным и слишком ленивым, чтобы управлять всеми своими личными расходами? Я тоже! Поэтому в этой лабораторной работе мы создадим личного помощника по управлению расходами на базе Gemini 2.5, который возьмёт на себя всю рутинную работу! От управления загруженными чеками до анализа того, не потратили ли вы слишком много на кофе!
Этот помощник будет доступен через веб-браузер в виде веб-интерфейса чата, где вы сможете общаться с ним, загружать изображения чеков и просить ассистента сохранить их, или, например, искать чеки, чтобы получить файл и провести анализ расходов. Всё это реализовано на основе фреймворка Google Agent Development Kit.
Само приложение разделено на 2 сервиса: frontend и backend, что позволяет вам быстро создать прототип и попробовать, как он работает, а также понять, как выглядит контракт API для их интеграции.
В ходе выполнения лабораторной работы вы будете применять следующий пошаговый подход:
- Подготовьте свой проект Google Cloud и включите в нем все необходимые API.
- Настройка контейнера в Google Cloud Storage и базы данных в Firestore
- Создать индексацию Firestore
- Настройте рабочее пространство для вашей среды кодирования
- Структурирование исходного кода агента ADK, инструментов, подсказок и т. д.
- Тестирование агента с использованием локального пользовательского интерфейса веб-разработки ADK
- Создайте интерфейс службы frontend — чат-интерфейс с использованием библиотеки Gradio для отправки запросов и загрузки изображений чеков.
- Создайте внутреннюю службу — HTTP-сервер с использованием FastAPI , где находятся код нашего агента ADK, SessionService и Artifact Service.
- Управление переменными среды и настройка необходимых файлов для развертывания приложения в Cloud Run.
- Разверните приложение в Cloud Run
Обзор архитектуры
Предпосылки
- Удобная работа с Python
- Понимание базовой архитектуры полного стека с использованием HTTP-сервиса
Чему вы научитесь
- Прототипирование веб-приложений с помощью Gradio
- Разработка бэкэнд-сервисов с использованием FastAPI и Pydantic
- Архитектура агента ADK с использованием его многочисленных возможностей
- Использование инструмента
- Управление сеансами и артефактами
- Использование обратного вызова для модификации входных данных перед отправкой в Gemini
- Использование BuiltInPlanner для улучшения выполнения задач путем планирования
- Быстрая отладка через локальный веб-интерфейс ADK
- Стратегия оптимизации мультимодального взаимодействия посредством анализа и извлечения информации с помощью оперативной разработки и модификации запросов Gemini с использованием обратного вызова ADK
- Генерация дополненной реальности с использованием Firestore в качестве векторной базы данных
- Управление переменными среды в файле YAML с помощью Pydantic-settings
- Разверните приложение в Cloud Run с помощью Dockerfile и укажите переменные среды в файле YAML.
Что вам понадобится
- веб-браузер Chrome
- Аккаунт Gmail
- Облачный проект с включенным биллингом
Эта лабораторная работа, предназначенная для разработчиков любого уровня (включая новичков), использует Python в качестве примера приложения. Однако для понимания представленных концепций знание Python не требуется.
2. 🚀 Прежде чем начать
Выберите активный проект в облачной консоли.
В этой лабораторной работе предполагается, что у вас уже есть проект Google Cloud с активированной функцией оплаты. Если у вас его ещё нет, следуйте инструкциям ниже, чтобы начать работу.
- В консоли Google Cloud на странице выбора проекта выберите или создайте проект Google Cloud.
- Убедитесь, что для вашего облачного проекта включена функция выставления счетов. Узнайте, как проверить, включена ли функция выставления счетов для проекта .
Подготовить базу данных Firestore
Далее нам потребуется создать базу данных Firestore. Firestore в режиме Native — это документная NoSQL-база данных, созданная для автоматического масштабирования, высокой производительности и простоты разработки приложений. Она также может выступать в качестве векторной базы данных, поддерживающей технологию дополненной генерации (Retrieval Augmented Generation) для нашей лаборатории.
- Введите « firestore» в строке поиска и выберите продукт Firestore.
- Затем нажмите кнопку «Создать базу данных Firestore».
- Используйте (по умолчанию) в качестве идентификатора базы данных и оставьте выбранным Standard Edition . Для этой лабораторной демонстрации используйте Firestore Native with Open security rules.
- Вы также заметите, что эта база данных действительно имеет статус Free-tier Usage. ДА! После этого нажмите кнопку «Создать базу данных».
После этих шагов вы уже должны быть перенаправлены в базу данных Firestore, которую вы только что создали.
Настройка облачного проекта в Cloud Shell Terminal
- Вы будете использовать Cloud Shell — среду командной строки, работающую в Google Cloud и предустановленную вместе с bq. Нажмите «Активировать Cloud Shell» в верхней части консоли Google Cloud.
- После подключения к Cloud Shell вы проверяете, что вы уже прошли аутентификацию и что проекту присвоен ваш идентификатор проекта, с помощью следующей команды:
gcloud auth list
- Выполните следующую команду в Cloud Shell, чтобы подтвердить, что команда gcloud знает о вашем проекте.
gcloud config list project
- Если ваш проект не настроен, используйте следующую команду для его настройки:
gcloud config set project <YOUR_PROJECT_ID>
Кроме того, вы также можете увидеть идентификатор PROJECT_ID в консоли.
Нажмите на нее, и справа вы увидите все данные вашего проекта и его идентификатор.
- Включите необходимые 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 Bucket
Далее, в том же терминале, нам нужно подготовить контейнер GCS для хранения загруженного файла. Выполните следующую команду, чтобы создать контейнер. Потребуется уникальное, но релевантное имя контейнера, соответствующее квитанциям помощника по личным расходам. Поэтому мы используем следующее имя контейнера в сочетании с идентификатором вашего проекта.
gsutil mb -l us-central1 gs://personal-expense-{your-project-id}
Он покажет этот вывод
Creating gs://personal-expense-{your-project-id}
Вы можете убедиться в этом, перейдя в меню навигации в левом верхнем углу браузера и выбрав «Облачное хранилище» -> «Корзина».
Создание индекса Firestore для поиска
Firestore изначально является базой данных NoSQL, которая обеспечивает превосходную производительность и гибкость модели данных, но имеет ограничения при выполнении сложных запросов. Поскольку мы планируем использовать составные запросы по нескольким полям и векторный поиск, нам сначала потребуется создать индекс. Подробнее см. в этой документации.
- Выполните следующую команду, чтобы создать индекс для поддержки составных запросов.
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)"
- И запустите это для поддержки поиска векторов
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)"
Вы можете проверить созданный индекс, посетив Firestore в облачной консоли, щелкнув экземпляр базы данных (по умолчанию) и выбрав «Индексы» на панели навигации.
Перейдите в редактор Cloud Shell и настройте рабочий каталог приложения.
Теперь мы можем настроить наш редактор кода для написания кода. Для этого мы будем использовать Cloud Shell Editor.
- Нажмите кнопку «Открыть редактор», откроется редактор Cloud Shell, где мы можем написать наш код.
- Далее нам также необходимо проверить, настроена ли оболочка на правильный идентификатор ПРОЕКТА , который у вас есть. Если вы видите значение внутри ( ) перед значком $ в терминале (на снимке экрана ниже это значение «adk-multimodal-tool» ), это значение показывает настроенный проект для вашего активного сеанса оболочки.
Если отображаемое значение верно , вы можете пропустить следующую команду . Однако, если оно неверно или отсутствует, выполните следующую команду.
gcloud config set project <YOUR_PROJECT_ID>
- Далее клонируем рабочий каталог шаблона для этой лабораторной работы с Github, выполнив следующую команду. Она создаст рабочий каталог в каталоге personal-expense-assistant.
git clone https://github.com/alphinside/personal-expense-assistant-adk-codelab-starter.git personal-expense-assistant
- После этого перейдите в верхнюю часть редактора Cloud Shell и выберите Файл -> Открыть папку, найдите каталог с вашим именем пользователя и найдите каталог personal-expense-assistant, затем нажмите кнопку ОК . Выбранный каталог станет основным рабочим каталогом. В данном примере имя пользователя — alvinprayuda , поэтому путь к каталогу показан ниже.
Теперь ваш редактор Cloud Shell должен выглядеть так.
Настройка среды
Подготовка виртуальной среды Python
Следующий шаг — подготовка среды разработки. Ваш текущий активный терминал должен находиться в рабочем каталоге personal-expense-assistant . В этой лабораторной работе мы будем использовать Python 3.12 и менеджер проектов UV Python для упрощения создания и управления версиями Python и виртуальными средами.
- Если вы еще не открыли терминал, откройте его, нажав «Терминал» -> «Новый терминал» или используйте сочетание клавиш Ctrl + Shift + C. Это откроет окно терминала в нижней части браузера.
- Теперь инициализируем виртуальную среду с помощью
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",
]
Файлы конфигурации установки
Теперь нам нужно настроить файлы конфигурации для этого проекта. Мы используем 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_LOCATION , BACKEND_URL , DB_COLLECTION_NAME .
Теперь мы можем перейти к следующему шагу: созданию агента, а затем и сервисов.
3. 🚀 Создайте агента с использованием Google ADK и Gemini 2.5.
Введение в структуру каталогов ADK
Давайте начнём с изучения возможностей ADK и способов создания агента. Полную документацию ADK можно найти по этой ссылке . ADK предлагает множество утилит для выполнения команд CLI. Вот некоторые из них:
- Настройте структуру каталога агентов
- Быстро попробуйте взаимодействие через ввод-вывод CLI
- Быстрая настройка локального веб-интерфейса разработки
Теперь создадим структуру каталогов агентов с помощью команды CLI. Выполните следующую команду.
uv run adk create expense_manager_agent
При появлении запроса выберите модель gemini-2.5-flash и бэкенд Vertex AI . Мастер запросит идентификатор и местоположение проекта. Вы можете принять параметры по умолчанию, нажав Enter, или изменить их при необходимости. Только дважды проверьте, что вы используете правильный идентификатор проекта, созданный ранее в этой лабораторной работе. Вывод будет выглядеть следующим образом:
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',
)
Теперь вы можете проверить это, запустив
uv run adk run expense_manager_agent
После завершения тестирования вы можете выйти из агента, введя команду exit или нажав Ctrl+D .
Создание нашего агента по управлению расходами
Давайте создадим нашего агента-менеджера расходов! Откройте файл expenditure_manager_agent / agent.py и скопируйте код ниже, который будет содержать 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,
)
Пояснение кода
Этот сценарий содержит инициализацию нашего агента, где мы инициализируем следующие вещи:
- Установите используемую модель на
gemini-2.5-flash - Настройте описание агента и инструкцию как системную подсказку, которая считывается из
task_prompt.md - Предоставить необходимые инструменты для поддержки функциональности агента
- Включите планирование перед формированием окончательного ответа или его выполнением, используя возможности быстрого мышления Gemini 2.5
- Настройте перехват обратного вызова перед отправкой запроса в Gemini, чтобы ограничить количество отправляемых данных изображения перед выполнением прогноза.
4. 🚀 Настройка инструментов агента
Наш агент по управлению расходами будет обладать следующими возможностями:
- Извлечь данные из изображения чека и сохранить их в файле.
- Точный поиск по данным о расходах
- Контекстный поиск по данным о расходах
Следовательно, нам нужны соответствующие инструменты для поддержки этой функции. Создайте новый файл в каталоге expenditure_manager_agent и назовите его tools.py.
touch expense_manager_agent/tools.py
Откройте expenditure_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
Пояснение кода
При реализации функций этого инструмента мы разрабатываем инструменты вокруг следующих двух основных идей:
- Анализ данных чека и сопоставление с исходным файлом с использованием заполнителя строки идентификатора изображения
[IMAGE-ID <hash-of-image-1>] - Хранение и извлечение данных с использованием базы данных Firestore
Инструмент «store_receipt_data»
Этот инструмент представляет собой инструмент оптического распознавания символов. Он анализирует необходимую информацию из данных изображения, распознает строку идентификатора изображения и сопоставляет их для сохранения в базе данных 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 с целью удаления данных изображений из старого контекста истории разговора (включать только данные изображений из последних трёх взаимодействий с пользователем) для повышения эффективности.
Однако мы по-прежнему хотим, чтобы агент имел доступ к контексту данных изображения при необходимости. Поэтому мы добавляем механизм для добавления строкового идентификатора изображения после каждого байта данных изображения в диалоге. Это поможет агенту связать идентификатор изображения с фактическими данными файла, которые можно использовать как при сохранении изображения, так и при его извлечении. Структура будет выглядеть следующим образом.
<image-byte-data-1> [IMAGE-ID <hash-of-image-1>] <image-byte-data-2> [IMAGE-ID <hash-of-image-2>] And so on..
И даже когда байтовые данные в истории разговоров устаревают, строковый идентификатор остаётся, позволяя получить доступ к данным с помощью инструментов. Пример структуры истории после удаления данных изображения.
[IMAGE-ID <hash-of-image-1>] [IMAGE-ID <hash-of-image-2>] And so on..
Итак, начнём! Создайте новый файл в каталоге expenditure_manager_agent и назовите его callbacks.py.
touch expense_manager_agent/callbacks.py
Откройте файл expenditure_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. Мы также хотим, чтобы агент мог найти и получить для нас нужное изображение чека. Это означает, что нам необходимо правильно передать всю эту информацию в правильной структуре подсказок.
Мы попросим агента структурировать вывод в следующем формате разметки, чтобы проанализировать мыслительный процесс, окончательный ответ и вложение (если таковое имеется).
# 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 уже должен существовать в нашем рабочем каталоге, но нам нужно переместить его в каталог expenditure_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 также позволяет нам использовать пользовательский интерфейс разработки для взаимодействия и наблюдения за происходящим во время взаимодействия. Выполните следующую команду, чтобы запустить локальный сервер пользовательского интерфейса разработки.
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 и выберите «Предварительный просмотр на порту 8080».
Вы увидите следующую веб-страницу, где вы можете выбрать доступных агентов в раскрывающемся списке слева вверху (в нашем случае это expenditure_manager_agent ) и взаимодействовать с ботом. В левом окне вы увидите множество информации о журнале работы агента.
Давайте попробуем что-нибудь сделать! Загрузите эти два примера чеков (источник: наборы данных Hugging Face mousserlane/id_receipt_dataset ). Щёлкните правой кнопкой мыши по каждому изображению и выберите «Сохранить изображение как…» (это загрузит изображение чека), затем загрузите файл боту, нажав на значок «скрепки» , и укажите, что хотите сохранить эти чеки.
После этого попробуйте выполнить следующие запросы для поиска или извлечения файлов.
- «Укажите разбивку расходов и их общую сумму за 2023 год»
- «Дайте мне файл с квитанцией от Indomaret»
При использовании некоторых инструментов вы можете проверить, что происходит в пользовательском интерфейсе разработки.
Посмотрите, как агент реагирует на ваши запросы, и проверьте, соответствует ли он всем правилам, указанным в подсказке внутри task_prompt.py. Поздравляем! Теперь у вас есть полностью рабочий агент разработки.
Теперь пришло время дополнить его соответствующим и приятным пользовательским интерфейсом и возможностями загрузки и скачивания файлов изображений.
8. 🚀 Создание службы Frontend с использованием 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()`.
После этого вы сможете проверить веб-интерфейс, щёлкнув по ссылке локального URL, удерживая клавишу Ctrl . Кроме того, вы можете открыть фронтенд-приложение, нажав кнопку « Предварительный просмотр в Интернете» в правом верхнем углу окна Cloud Editor и выбрав «Предварительный просмотр на порту 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 для отображения каждого компонента с помощью компонента пользовательского интерфейса.
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
Далее нам необходимо создать бэкэнд, который сможет инициализировать наш агент вместе с другими компонентами, чтобы иметь возможность выполнять среду выполнения агента.
Создайте новый файл и назовите его 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)
После этого можно попробовать запустить бэкенд-службу. Помните, что на предыдущем шаге мы правильно запустили фронтенд-службу, теперь нам нужно открыть новый терминал и попробовать запустить эту бэкенд-службу.
- Создайте новый терминал. Перейдите к терминалу в нижней части экрана и найдите кнопку «+», чтобы создать новый терминал. Также можно нажать Ctrl + Shift + C , чтобы открыть новый терминал.
- После этого убедитесь, что вы находитесь в рабочем каталоге personal-expense-assistant, затем выполните следующую команду
uv run backend.py
- В случае успеха будет показан такой вывод:
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 . Вы можете проверить это внутри функции format_user_request_to_adk_content_and_store_artifacts в файле 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,
)
...
Все запросы, которые будут обрабатываться агентом-исполнителем, должны быть отформатированы в формате types.Content type. Внутри функции мы также обрабатываем данные каждого изображения и извлекаем его идентификатор, который затем заменяется заполнителем идентификатора изображения.
Аналогичный механизм используется для загрузки вложений после извлечения идентификаторов изображений с помощью регулярного выражения:
...
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. 🚀 Интеграционный тест
Теперь у вас должно быть запущено несколько служб на разных вкладках облачной консоли:
- Служба frontend запущена на порту 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.
Нажмите кнопку «Веб-просмотр» в верхней части редактора Cloud Shell и выберите «Предварительный просмотр на порту 8080».
Теперь давайте поработаем с помощником!
Загрузите следующие чеки. Диапазон дат этих чеков — 2023–2024 годы, и попросите помощника сохранить/загрузить их.
- Receipt Drive (источник: наборы данных Hugging Face
mousserlane/id_receipt_dataset)
Задавайте разные вопросы
- «Предоставьте мне ежемесячную разбивку расходов на 2023–2024 годы»
- «Покажите мне чек за покупку кофе»
- «Дайте мне файл с квитанцией от Yakiniku Like»
- И т. д
Вот фрагмент успешного взаимодействия
11. 🚀 Развертывание в Cloud Run
Конечно, нам нужен доступ к этому замечательному приложению из любой точки мира. Для этого мы можем упаковать его и развернуть в Cloud Run. В рамках демонстрации этот сервис будет представлен как публичный, к которому смогут получить доступ другие пользователи. Однако имейте в виду, что это не лучший вариант для подобных приложений, поскольку он больше подходит для личных приложений.
В этой лабораторной работе мы разместим как фронтенд-, так и бэкенд-сервисы в одном контейнере. Для управления обоими сервисами нам понадобится Supervisord . Вы можете изучить файл Supervisord.conf и Dockerfile , в котором Supervisord указан как точка входа.
На этом этапе у нас уже есть все файлы, необходимые для развертывания наших приложений в Cloud Run. Давайте развернём их. Перейдите в терминал Cloud Shell и убедитесь, что текущий проект настроен на ваш активный проект. Если нет, используйте команду gcloud configure, чтобы задать идентификатор проекта:
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, просто ответьте Y. Обратите внимание, что мы разрешаем доступ без аутентификации, поскольку это демонстрационное приложение. Рекомендуется использовать соответствующую аутентификацию для ваших корпоративных и производственных приложений.
После завершения развертывания вы должны получить ссылку, похожую на приведенную ниже:
https://personal-expense-assistant-*******.us-central1.run.app
Продолжайте использовать приложение в режиме инкогнито или на мобильном устройстве. Оно уже должно быть доступно.
12. 🎯 Вызов
Теперь пришло время отточить свои навыки исследователя. Сможете ли вы изменить код, чтобы бэкенд мог работать с несколькими пользователями? Какие компоненты нужно обновить?
13. 🧹 Уборка
Чтобы избежать списания средств с вашего аккаунта Google Cloud за ресурсы, используемые в этой лабораторной работе, выполните следующие действия:
- В консоли Google Cloud перейдите на страницу Управление ресурсами .
- В списке проектов выберите проект, который вы хотите удалить, а затем нажмите Удалить .
- В диалоговом окне введите идентификатор проекта, а затем нажмите кнопку «Завершить» , чтобы удалить проект.
- Либо вы можете перейти в Cloud Run на консоли, выбрать службу, которую вы только что развернули, и удалить ее.