এই পৃষ্ঠাটি Cloud Translation API অনুবাদ করেছে।

এজেন্ট ডেভেলপমেন্ট কিট সহ মাল্টিমোডাল যাচ্ছেন: জেমিনি 2.5 এর সাথে ব্যক্তিগত খরচ সহকারী, ফায়ারস্টোর এবং ক্লাউড রান, এজেন্ট ডেভেলপমেন্ট কিটের সাথে মাল্টিমোডাল যাচ্ছেন: জেমিনি 2.5, ফায়ারস্টোর এবং ক্লাউড রানের সাথে ব্যক্তিগত ব্যয় সহকারী

১. 📖 ভূমিকা

আপনি কি কখনও আপনার সমস্ত ব্যক্তিগত খরচ পরিচালনা করতে গিয়ে হতাশ এবং অলস বোধ করেছেন? আমিও! আর তাই এই কোডল্যাবে, আমরা জেমিনি ২.৫ দ্বারা চালিত একটি ব্যক্তিগত খরচ ব্যবস্থাপক সহকারী তৈরি করব যা আমাদের জন্য সমস্ত কাজ করে দেবে! আপলোড করা রসিদগুলি পরিচালনা করা থেকে শুরু করে এক কাপ কফি কিনতেই আপনি বেশি খরচ করে ফেলেছেন কিনা তা বিশ্লেষণ করা পর্যন্ত সবকিছুই এটি করবে!

এই অ্যাসিস্ট্যান্টটি একটি চ্যাট ওয়েব ইন্টারফেসের মাধ্যমে ওয়েব ব্রাউজারে ব্যবহার করা যাবে, যেখানে আপনি এর সাথে যোগাযোগ করতে পারবেন, কিছু রসিদের ছবি আপলোড করে অ্যাসিস্ট্যান্টকে সেগুলো সংরক্ষণ করতে বলতে পারবেন, অথবা ফাইলটি পাওয়ার জন্য কিছু রসিদ অনুসন্ধান করে খরচের বিশ্লেষণ করতে চাইতে পারেন। আর এই সবকিছুই গুগল এজেন্ট ডেভেলপমেন্ট কিট ফ্রেমওয়ার্কের উপর ভিত্তি করে তৈরি করা হয়েছে।

অ্যাপ্লিকেশনটি ফ্রন্টএন্ড এবং ব্যাকএন্ড—এই দুটি সার্ভিসে বিভক্ত; যা আপনাকে দ্রুত একটি প্রোটোটাইপ তৈরি করে এর কার্যকারিতা পরখ করার সুযোগ দেয় এবং উভয়কে একীভূত করার জন্য এপিআই কন্ট্রাক্টটি কেমন, তা বুঝতেও সাহায্য করে।

কোডল্যাবের মাধ্যমে, আপনি নিম্নলিখিত ধাপে ধাপে পদ্ধতিটি অনুসরণ করবেন:

আপনার গুগল ক্লাউড প্রজেক্টটি প্রস্তুত করুন এবং এতে প্রয়োজনীয় সকল এপিআই (API) সক্রিয় করুন।
গুগল ক্লাউড স্টোরেজে বাকেট এবং ফায়ারস্টোরে ডেটাবেস সেটআপ করুন।
ফায়ারস্টোর ইনডেক্সিং তৈরি করুন
আপনার কোডিং পরিবেশের জন্য কর্মক্ষেত্র প্রস্তুত করুন।
ADK এজেন্ট সোর্স কোড, টুলস, প্রম্পট ইত্যাদির কাঠামো তৈরি করা।
ADK লোকাল ওয়েব ডেভেলপমেন্ট UI ব্যবহার করে এজেন্টটি পরীক্ষা করা হচ্ছে
Gradio লাইব্রেরি ব্যবহার করে ফ্রন্টএন্ড সার্ভিস - চ্যাট ইন্টারফেসটি তৈরি করুন, যার মাধ্যমে কিছু কোয়েরি পাঠানো এবং রসিদের ছবি আপলোড করা যাবে।
FastAPI ব্যবহার করে ব্যাকএন্ড সার্ভিস – HTTP সার্ভারটি তৈরি করুন, যেখানে আমাদের ADK এজেন্ট কোড, SessionService এবং Artifact Service থাকবে।
ক্লাউড রানে অ্যাপ্লিকেশনটি ডেপ্লয় করার জন্য প্রয়োজনীয় এনভায়রনমেন্ট ভেরিয়েবল ও ফাইলগুলো সেটআপ করুন।
অ্যাপ্লিকেশনটি ক্লাউড রানে স্থাপন করুন

স্থাপত্যের সংক্ষিপ্ত বিবরণ

পূর্বশর্ত

পাইথন নিয়ে কাজ করতে স্বাচ্ছন্দ্যবোধ করি।
HTTP পরিষেবা ব্যবহার করে মৌলিক ফুল-স্ট্যাক আর্কিটেকচার সম্পর্কে ধারণা

আপনি যা শিখবেন

Gradio দিয়ে ফ্রন্টএন্ড ওয়েব প্রোটোটাইপিং
FastAPI এবং Pydantic ব্যবহার করে ব্যাকএন্ড পরিষেবা উন্নয়ন
ADK এজেন্টের বিভিন্ন ক্ষমতা ব্যবহার করে এর স্থাপত্য নির্মাণ
সরঞ্জাম ব্যবহার
সেশন এবং আর্টিফ্যাক্ট ম্যানেজমেন্ট
জেমিনিতে পাঠানোর আগে ইনপুট পরিবর্তনের জন্য কলব্যাক ব্যবহার
পরিকল্পনা করার মাধ্যমে কাজের কার্যকারিতা উন্নত করতে BuiltInPlanner ব্যবহার করা
ADK স্থানীয় ওয়েব ইন্টারফেসের মাধ্যমে দ্রুত ডিবাগিং
ADK কলব্যাক ব্যবহার করে প্রম্পট ইঞ্জিনিয়ারিং এবং জেমিনি অনুরোধ পরিবর্তনের মাধ্যমে তথ্য পার্সিং ও পুনরুদ্ধারের দ্বারা মাল্টিমোডাল ইন্টারঅ্যাকশন অপ্টিমাইজ করার কৌশল
ভেক্টর ডেটাবেস হিসেবে ফায়ারস্টোর ব্যবহার করে এজেন্টিক রিট্রিভাল অগমেন্টেড জেনারেশন
Pydantic-settings ব্যবহার করে YAML ফাইলে এনভায়রনমেন্ট ভেরিয়েবল পরিচালনা করুন
Dockerfile ব্যবহার করে Cloud Run-এ অ্যাপ্লিকেশনটি ডেপ্লয় করুন এবং YAML ফাইলের মাধ্যমে এনভায়রনমেন্ট ভেরিয়েবল সরবরাহ করুন।

আপনার যা যা লাগবে

ক্রোম ওয়েব ব্রাউজার
একটি জিমেইল অ্যাকাউন্ট
বিলিং সক্ষম একটি ক্লাউড প্রজেক্ট

সকল স্তরের (শিক্ষানবিশ সহ) ডেভেলপারদের জন্য ডিজাইন করা এই কোডল্যাবটির নমুনা অ্যাপ্লিকেশনে পাইথন ব্যবহার করা হয়েছে। তবে, এখানে উপস্থাপিত ধারণাগুলো বোঝার জন্য পাইথন জ্ঞানের প্রয়োজন নেই।

২. 🚀 শুরু করার আগে

গুরুত্বপূর্ণ দ্রষ্টব্য:

এই টিউটোরিয়ালটির জন্য একটি সক্রিয় বিলিং অ্যাকাউন্টসহ গুগল ক্লাউড প্রজেক্ট প্রয়োজন। যদি আপনার কাছে এখনও এমন কোনো অ্যাকাউন্ট না থাকে, তবে এই টিউটোরিয়ালটির জন্য ব্যবহারযোগ্য একটি ট্রায়াল বিলিং অ্যাকাউন্ট পেতে অনুগ্রহ করে নিচের ধাপগুলো অনুসরণ করুন:

ছদ্মবেশী উইন্ডো খুলুন
আপনি যদি কোনো ওয়ার্কশপ সেশনে যোগদান করেন, তাহলে সাধারণত সেশনেই রিডেম্পশন পোর্টালের ইউআরএল দেওয়া থাকবে, অন্যথায় আপনি এই ট্রায়াল বিলিং অ্যাকাউন্ট রিডেম্পশন পোর্টালে যেতে পারেন।
আপনার ব্যক্তিগত জিমেইল অ্যাকাউন্ট দিয়ে লগ ইন করুন (গুগল ওয়ার্কস্পেস বা অন্য কোনো ইমেল কাজ করবে না!)
এই টিউটোরিয়ালের জন্য পূর্বে দাবি করা ট্রায়াল বিলিং অ্যাকাউন্টের সাথে লিঙ্ক করা একটি নতুন গুগল ক্লাউড প্রজেক্ট সফলভাবে তৈরি করতে ধাপে ধাপে রিডেম্পশন নির্দেশাবলী অনুসরণ করুন।

ক্লাউড কনসোলে সক্রিয় প্রজেক্ট নির্বাচন করুন

এই কোডল্যাবটি ধরে নেয় যে আপনার ইতিমধ্যেই বিলিং চালু করা একটি গুগল ক্লাউড প্রজেক্ট আছে। যদি আপনার এটি এখনও না থাকে, তবে শুরু করার জন্য আপনি নীচের নির্দেশাবলী অনুসরণ করতে পারেন।

গুগল ক্লাউড কনসোলের প্রজেক্ট সিলেক্টর পেজে, একটি গুগল ক্লাউড প্রজেক্ট নির্বাচন করুন বা তৈরি করুন।
আপনার ক্লাউড প্রোজেক্টের জন্য বিলিং চালু আছে কিনা তা নিশ্চিত করুন। কোনো প্রোজেক্টে বিলিং চালু আছে কিনা তা কীভাবে পরীক্ষা করবেন, তা জেনে নিন।

ফায়ারস্টোর ডেটাবেস প্রস্তুত করুন

এরপরে, আমাদের একটি ফায়ারস্টোর ডেটাবেসও তৈরি করতে হবে। নেটিভ মোডে ফায়ারস্টোর হলো একটি NoSQL ডকুমেন্ট ডেটাবেস, যা স্বয়ংক্রিয় স্কেলিং, উচ্চ কর্মক্ষমতা এবং অ্যাপ্লিকেশন তৈরির সুবিধার জন্য নির্মিত। এটি একটি ভেক্টর ডেটাবেস হিসেবেও কাজ করতে পারে, যা আমাদের ল্যাবের জন্য রিট্রিভাল অগমেন্টেড জেনারেশন কৌশল সমর্থন করে।

সার্চ বারে ' firestore' লিখে সার্চ করুন এবং Firestore প্রোডাক্টটিতে ক্লিক করুন।

তারপর, Create A Firestore Database বোতামে ক্লিক করুন।
ডাটাবেস আইডি নাম হিসেবে (ডিফল্ট) ব্যবহার করুন এবং স্ট্যান্ডার্ড এডিশন নির্বাচিত রাখুন। এই ল্যাব ডেমোর জন্য, ওপেন সিকিউরিটি রুলস সহ ফায়ারস্টোর নেটিভ ব্যবহার করুন।

আপনি আরও লক্ষ্য করবেন যে এই ডেটাবেসটিতে আসলে ফ্রি-টিয়ার ব্যবহারের সুবিধা রয়েছে, দারুণ! এরপর, ক্রিয়েট ডেটাবেস বাটনে ক্লিক করুন।

এই ধাপগুলো সম্পন্ন করার পর, আপনাকে আপনার তৈরি করা ফায়ারস্টোর ডেটাবেসে রিডাইরেক্ট করা হবে।

ক্লাউড শেল টার্মিনালে ক্লাউড প্রজেক্ট সেটআপ করুন

আপনি ক্লাউড শেল ব্যবহার করবেন, যা গুগল ক্লাউডে চলমান একটি কমান্ড-লাইন পরিবেশ এবং এটি bq-এর সাথে আগে থেকেই লোড করা থাকে। গুগল ক্লাউড কনসোলের শীর্ষে থাকা ‘Activate Cloud Shell’-এ ক্লিক করুন।

ক্লাউড শেলে সংযুক্ত হওয়ার পর, আপনি নিম্নলিখিত কমান্ডটি ব্যবহার করে যাচাই করে নিন যে আপনি ইতিমধ্যেই প্রমাণীকৃত এবং প্রজেক্টটি আপনার প্রজেক্ট আইডিতে সেট করা আছে:

gcloud auth list

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 কমান্ড এবং এর ব্যবহার সম্পর্কে জানতে ডকুমেন্টেশন দেখুন।

গুগল ক্লাউড স্টোরেজ বাকেট প্রস্তুত করুন

এরপর, একই টার্মিনাল থেকে, আপলোড করা ফাইলটি সংরক্ষণের জন্য আমাদের GCS বাকেট প্রস্তুত করতে হবে। বাকেটটি তৈরি করতে নিম্নলিখিত কমান্ডটি চালান। ব্যক্তিগত খরচ সহকারী রসিদের সাথে সম্পর্কিত একটি অনন্য এবং প্রাসঙ্গিক বাকেটের নাম প্রয়োজন হবে, তাই আমরা আপনার প্রজেক্ট আইডির সাথে নিম্নলিখিত বাকেটের নামটি ব্যবহার করব।

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

এটি এই আউটপুটটি দেখাবে

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

আপনি ব্রাউজারের উপরের বাম দিকের নেভিগেশন মেনুতে গিয়ে ক্লাউড স্টোরেজ -> বাকেট নির্বাচন করে এটি যাচাই করতে পারেন।

অনুসন্ধানের জন্য ফায়ারস্টোর ইনডেক্স তৈরি করা

ফায়ারস্টোর মূলত একটি 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-এ গিয়ে (ডিফল্ট) ডাটাবেস ইনস্ট্যান্সে ক্লিক করে এবং নেভিগেশন বার থেকে Indexes নির্বাচন করে তৈরি করা ইনডেক্সটি পরীক্ষা করতে পারেন।

ক্লাউড শেল এডিটর-এ যান এবং অ্যাপ্লিকেশন ওয়ার্কিং ডিরেক্টরি সেটআপ করুন।

এখন, আমরা কোডিং করার জন্য আমাদের কোড এডিটর সেট আপ করতে পারি। এর জন্য আমরা ক্লাউড শেল এডিটর ব্যবহার করব।

ওপেন এডিটর বাটনে ক্লিক করুন, এটি একটি ক্লাউড শেল এডিটর খুলবে, আমরা এখানে আমাদের কোড লিখতে পারি।
এরপর, আমাদের এটাও যাচাই করতে হবে যে শেলটি আপনার সঠিক প্রজেক্ট আইডিতে আগে থেকেই কনফিগার করা আছে কিনা। যদি আপনি টার্মিনালে $ আইকনের আগে ( ) বন্ধনীর ভেতরে কোনো মান দেখতে পান (নিচের স্ক্রিনশটে, মানটি হলো "adk-multimodal-tool" ), তাহলে এই মানটি আপনার সক্রিয় শেল সেশনের জন্য কনফিগার করা প্রজেক্টটি নির্দেশ করে।

প্রদর্শিত মানটি যদি ইতিমধ্যেই সঠিক হয়, তাহলে আপনি পরবর্তী কমান্ডটি এড়িয়ে যেতে পারেন। তবে যদি এটি সঠিক না হয় বা অনুপস্থিত থাকে, তাহলে নিম্নলিখিত কমান্ডটি চালান।

gcloud config set project <YOUR_PROJECT_ID>

এরপর, গিটহাব থেকে এই কোডল্যাবের জন্য টেমপ্লেট ওয়ার্কিং ডিরেক্টরিটি ক্লোন করে নিচের কমান্ডটি চালান। এটি personal-expense-assistant ডিরেক্টরিতে ওয়ার্কিং ডিরেক্টরিটি তৈরি করবে।

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

এরপর, ক্লাউড শেল এডিটরের উপরের অংশে যান এবং File->Open Folder-এ ক্লিক করুন, আপনার ইউজারনেম ডিরেক্টরি এবং personal-expense-assistant ডিরেক্টরিটি খুঁজুন, তারপর OK বোতামে ক্লিক করুন। এটি নির্বাচিত ডিরেক্টরিটিকে প্রধান ওয়ার্কিং ডিরেক্টরি হিসেবে সেট করবে। এই উদাহরণে, ইউজারনেম হলো alvinprayuda , তাই ডিরেক্টরি পাথটি নিচে দেখানো হলো।

এখন, আপনার ক্লাউড শেল এডিটরটি দেখতে এইরকম হওয়া উচিত।

পরিবেশ সেটআপ

পাইথন ভার্চুয়াল পরিবেশ প্রস্তুত করুন

পরবর্তী ধাপ হলো ডেভেলপমেন্ট এনভায়রনমেন্ট প্রস্তুত করা। আপনার বর্তমান সক্রিয় টার্মিনালটি `personal-expense-assistant` ওয়ার্কিং ডিরেক্টরির ভেতরে থাকা উচিত। আমরা এই কোডল্যাবে পাইথন ৩.১২ ব্যবহার করব এবং পাইথন ভার্সন ও ভার্চুয়াল এনভায়রনমেন্ট তৈরি ও পরিচালনার কাজ সহজ করার জন্য `uv python project manager` ব্যবহার করব।

আপনি যদি এখনও টার্মিনাল না খুলে থাকেন, তাহলে Terminal -> New Terminal- এ ক্লিক করে অথবা 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",
]

সেটআপ কনফিগারেশন ফাইল

এখন আমাদের এই প্রজেক্টের জন্য কনফিগারেশন ফাইল তৈরি করতে হবে। আমরা YAML ফাইল থেকে কনফিগারেশন পড়ার জন্য pydantic-settings ব্যবহার করি।

আমরা 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 এর পূর্ব-নির্ধারিত মানগুলো ব্যবহার করছি।

দ্রষ্টব্য: settings.yaml ফাইলটি settings.py থেকে পড়া হবে। pydantic-settings শুধু আমাদের সহজে ডেটা টাইপ প্রয়োগ করার সুযোগই দেয় না, বরং ডাইনামিকভাবে কনফিগারেশন সোর্সের অগ্রাধিকারও নির্ধারণ করে। এই ডেমোতে, আমরা অগ্রাধিকারটি নিচে যেভাবে দেওয়া আছে সেভাবে সেট করেছি:

এনভায়রনমেন্ট ভেরিয়েবল থেকে -> ডিপ্লয়মেন্টের পরে এনভায়রনমেন্ট ভেরিয়েবল পরিবর্তন করার সময় সহায়ক।
settings.yaml ফাইল থেকে -> প্রথমবার ডেপ্লয়মেন্ট শুরু করার সময় সহায়ক, যখন আগে থেকে কোনো এনভায়রনমেন্ট ভ্যারিয়েবল সেট করা থাকে না।
কোড প্রারম্ভিককরণ থেকে -> ডিফল্ট মান পরিচালনা করুন

এখন আমরা পরবর্তী ধাপে যেতে পারি, অর্থাৎ এজেন্ট এবং তারপর পরিষেবাগুলো তৈরি করতে পারি।

৩. 🚀 গুগল এডিকে এবং জেমিনি ২.৫ ব্যবহার করে এজেন্টটি তৈরি করুন।

ADK ডিরেক্টরি কাঠামোর পরিচিতি

চলুন, ADK কী কী সুবিধা দেয় এবং কীভাবে এজেন্ট তৈরি করতে হয়, তা দিয়ে শুরু করা যাক। ADK-এর সম্পূর্ণ ডকুমেন্টেশন এই URL- এ পাওয়া যাবে। ADK তার CLI কমান্ড এক্সিকিউশনের মাধ্যমে আমাদের অনেক ইউটিলিটি প্রদান করে। সেগুলোর মধ্যে কয়েকটি নিচে দেওয়া হলো:

এজেন্ট ডিরেক্টরি কাঠামো সেটআপ করুন
CLI ইনপুট আউটপুটের মাধ্যমে দ্রুত ইন্টারঅ্যাকশন চেষ্টা করুন
দ্রুত স্থানীয় ডেভেলপমেন্ট UI ওয়েব ইন্টারফেস সেটআপ করুন

এখন, CLI কমান্ড ব্যবহার করে এজেন্ট ডিরেক্টরি কাঠামো তৈরি করা যাক। নিম্নলিখিত কমান্ডটি চালান।

uv run adk create expense_manager_agent

জিজ্ঞাসা করা হলে, gemini-2.5-flash মডেল এবং Vertex AI ব্যাকএন্ড বেছে নিন। এরপর উইজার্ডটি প্রজেক্ট আইডি এবং অবস্থান জানতে চাইবে। আপনি এন্টার চেপে ডিফল্ট অপশনগুলো গ্রহণ করতে পারেন, অথবা প্রয়োজন অনুযায়ী সেগুলো পরিবর্তন করতে পারেন। শুধু নিশ্চিত হয়ে নিন যে আপনি এই ল্যাবে আগে তৈরি করা সঠিক প্রজেক্ট আইডিটিই ব্যবহার করছেন। আউটপুটটি দেখতে এইরকম হবে:

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 চেপে এজেন্ট থেকে বেরিয়ে যেতে পারেন।

আমাদের ব্যয় ব্যবস্থাপক এজেন্ট তৈরি করা

চলুন আমাদের এক্সপেন্স ম্যানেজার এজেন্ট তৈরি করি! expense_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 থেকে পড়া হচ্ছে।
এজেন্টের কার্যকারিতা সমর্থন করার জন্য প্রয়োজনীয় সরঞ্জাম সরবরাহ করুন।
জেমিনি ২.৫ ফ্ল্যাশ থিঙ্কিং সক্ষমতা ব্যবহার করে চূড়ান্ত প্রতিক্রিয়া তৈরি বা তা কার্যকর করার আগে পরিকল্পনা সক্ষম করুন।
পূর্বাভাস দেওয়ার আগে প্রেরিত ইমেজ ডেটার পরিমাণ সীমিত করতে, জেমিনিতে অনুরোধ পাঠানোর পূর্বে কলব্যাক ইন্টারসেপ্ট সেট আপ করুন।

৪. 🚀 এজেন্ট টুলস কনফিগার করা

আমাদের ব্যয় ব্যবস্থাপক এজেন্টের নিম্নলিখিত সক্ষমতা থাকবে:

রসিদের ছবি থেকে ডেটা বের করুন এবং ডেটা ও ফাইলটি সংরক্ষণ করুন।
ব্যয়ের তথ্যের উপর সঠিক অনুসন্ধান
ব্যয়ের তথ্যের উপর প্রাসঙ্গিক অনুসন্ধান

সুতরাং, এই কার্যকারিতা সমর্থন করার জন্য আমাদের উপযুক্ত সরঞ্জাম প্রয়োজন। 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

কোডের ব্যাখ্যা

এই টুলস ফাংশন বাস্তবায়নে আমরা এই দুটি প্রধান ধারণাকে কেন্দ্র করে টুলগুলো ডিজাইন করি:

ইমেজ আইডি স্ট্রিং প্লেসহোল্ডার [IMAGE-ID <hash-of-image-1>] ব্যবহার করে রসিদের ডেটা পার্স করুন এবং মূল ফাইলের সাথে ম্যাপ করুন।
ফায়ারস্টোর ডাটাবেস ব্যবহার করে ডেটা সংরক্ষণ এবং পুনরুদ্ধার

টুল "রিসিপ্ট_ডেটা"

এই টুলটি হলো অপটিক্যাল ক্যারেক্টার রিকগনিশন টুল; এটি ইমেজ ডেটা থেকে প্রয়োজনীয় তথ্য পার্স করার পাশাপাশি ইমেজ আইডি স্ট্রিং শনাক্ত করবে এবং সেগুলোকে একত্রিত করে ফায়ারস্টোর ডেটাবেসে সংরক্ষণের জন্য ম্যাপ করবে।

এছাড়াও, এই টুলটি text-embedding-004 ব্যবহার করে রসিদের বিষয়বস্তুকে এমবেডিং-এ রূপান্তর করে, যাতে সমস্ত মেটাডেটা এবং এমবেডিং একসাথে সংরক্ষিত ও সূচিবদ্ধ হয়। এর ফলে কোয়েরি অথবা প্রাসঙ্গিক অনুসন্ধানের মাধ্যমে এটি পুনরুদ্ধার করার সুবিধা পাওয়া যায়।

এই টুলটি সফলভাবে চালানোর পর, আপনি দেখতে পাবেন যে রসিদের ডেটা ইতিমধ্যেই ফায়ারস্টোর ডেটাবেসে নীচে দেখানো ছবির মতো ইনডেক্স করা হয়েছে।

টুল "search_receipts_by_metadata_filter"

এই টুলটি ব্যবহারকারীর কোয়েরিকে একটি মেটাডেটা কোয়েরি ফিল্টারে রূপান্তরিত করে, যা তারিখের পরিসর এবং/অথবা মোট লেনদেন অনুযায়ী অনুসন্ধান সমর্থন করে। এটি মিলে যাওয়া সমস্ত রসিদের ডেটা ফেরত দেবে, এবং এই প্রক্রিয়ায় আমরা এমবেডিং ফিল্ডটি বাদ দিয়ে দেব, কারণ প্রাসঙ্গিক বোঝার জন্য এজেন্টের এটির প্রয়োজন নেই।

টুল "search_relevant_receipts_by_natural_language_query"

এটি আমাদের রিট্রিভাল অগমেন্টেড জেনারেশন (RAG) টুল। আমাদের এজেন্টের ভেক্টর ডেটাবেস থেকে প্রাসঙ্গিক রসিদ পুনরুদ্ধার করার জন্য নিজস্ব কোয়েরি ডিজাইন করার ক্ষমতা রয়েছে এবং এটি কখন এই টুলটি ব্যবহার করবে তাও বেছে নিতে পারে। এজেন্ট এই RAG টুলটি ব্যবহার করবে কি না এবং নিজস্ব কোয়েরি ডিজাইন করবে কি না, সে বিষয়ে তাকে স্বাধীন সিদ্ধান্ত নেওয়ার সুযোগ দেওয়ার ধারণাটিই হলো এজেন্টিক RAG পদ্ধতির অন্যতম সংজ্ঞা।

আমরা এটিকে শুধু নিজস্ব কোয়েরি তৈরি করার সুযোগই দিই না, বরং এটি কতগুলো প্রাসঙ্গিক ডকুমেন্ট পুনরুদ্ধার করতে চায়, তাও নির্বাচন করার সুযোগ দিই। যথাযথ প্রম্পট ইঞ্জিনিয়ারিংয়ের সাথে মিলিত হয়ে, যেমন

# Example prompt

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

এটি এই টুলটিকে একটি শক্তিশালী টুলে পরিণত করবে যা প্রায় যেকোনো কিছু অনুসন্ধান করতে সক্ষম, যদিও নিকটতম প্রতিবেশী অনুসন্ধানের অ-সুনির্দিষ্ট প্রকৃতির কারণে এটি সমস্ত প্রত্যাশিত ফলাফল নাও দিতে পারে।

৫. 🚀 কলব্যাকের মাধ্যমে কথোপকথনের প্রেক্ষাপট পরিবর্তন

Google ADK আমাদেরকে বিভিন্ন স্তরে এজেন্ট রানটাইম "ইন্টারসেপ্ট" করার সুযোগ দেয়। এই বিস্তারিত সক্ষমতা সম্পর্কে আপনি এই ডকুমেন্টেশনে আরও পড়তে পারেন। এই ল্যাবে, আমরা দক্ষতার জন্য LLM-এ পাঠানোর আগে অনুরোধটি পরিবর্তন করতে before_model_callback ব্যবহার করি, যাতে পুরোনো কথোপকথনের ইতিহাসের প্রেক্ষাপট থেকে ছবির ডেটা মুছে ফেলা যায় (শুধুমাত্র শেষ ৩টি ব্যবহারকারীর ইন্টারঅ্যাকশনের ছবির ডেটা অন্তর্ভুক্ত থাকে)।

তবে, আমরা এখনও চাই যে প্রয়োজনে এজেন্টের কাছে ছবির ডেটার প্রেক্ষাপটটি থাকুক। এজন্য আমরা কথোপকথনে প্রতিটি ছবির বাইট ডেটার পরে একটি স্ট্রিং ইমেজ আইডি প্লেসহোল্ডার যোগ করার একটি ব্যবস্থা যুক্ত করেছি। এটি এজেন্টকে ছবির আইডিকে তার আসল ফাইল ডেটার সাথে সংযুক্ত করতে সাহায্য করবে, যা ছবি সংরক্ষণ বা পুনরুদ্ধার উভয় সময়েই ব্যবহার করা যাবে। কাঠামোটি দেখতে এইরকম হবে।

<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..

চলুন শুরু করা যাক! 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

৬. 🚀 প্রম্পট

জটিল মিথস্ক্রিয়া ও সক্ষমতা সম্পন্ন একটি এজেন্ট ডিজাইন করার জন্য, আমাদের এমন একটি উপযুক্ত নির্দেশিকা খুঁজে বের করতে হয় যা এজেন্টটিকে পথ দেখাবে, যাতে এটি আমাদের ইচ্ছামত আচরণ করতে পারে।

পূর্বে আমাদের কাছে কথোপকথনের ইতিহাসে ছবির ডেটা পরিচালনা করার একটি ব্যবস্থা ছিল, এবং এমন কিছু টুলও ছিল যা ব্যবহার করা হয়তো ততটা সহজ ছিল না, যেমন 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 ফাইলটি আমাদের বর্তমান ওয়ার্কিং ডিরেক্টরিতে আগে থেকেই থাকার কথা, কিন্তু আমাদের এটিকে expense_manager_agent ডিরেক্টরির অধীনে সরাতে হবে। এটিকে সরানোর জন্য নিম্নলিখিত কমান্ডটি চালান।

mv task_prompt.md expense_manager_agent/task_prompt.md

৭. 🚀 এজেন্টকে পরীক্ষা করা হচ্ছে

এখন 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)

এখন, এটি পরীক্ষা করার জন্য, আপনার ক্লাউড শেল এডিটরের উপরের অংশে থাকা ওয়েব প্রিভিউ বোতামে ক্লিক করুন এবং পোর্ট ৮০৮০-তে প্রিভিউ নির্বাচন করুন।

আপনি নিম্নলিখিত ওয়েব পেজটি দেখতে পাবেন, যেখানে আপনি উপরের বাম দিকের ড্রপ-ডাউন বোতাম থেকে উপলব্ধ এজেন্টদের (আমাদের ক্ষেত্রে এটি expense_manager_agent হওয়া উচিত) নির্বাচন করতে এবং বটটির সাথে ইন্টারঅ্যাক্ট করতে পারবেন। এজেন্ট চালু থাকার সময়কার লগ সংক্রান্ত অনেক তথ্য আপনি বাম দিকের উইন্ডোতে দেখতে পাবেন।

চলুন কিছু কাজ করে দেখি! এই দুটি নমুনা রসিদ আপলোড করুন ( উৎস: Hugging face datasets mousserlane/id_receipt_dataset )। প্রতিটি ছবির উপর রাইট ক্লিক করে ' Save Image as..' বিকল্পটি বেছে নিন (এতে রসিদের ছবিটি ডাউনলোড হবে), তারপর "ক্লিপ" আইকনে ক্লিক করে ফাইলটি বটে আপলোড করুন এবং জানান যে আপনি এই রসিদগুলো সংরক্ষণ করতে চান।

এরপর অনুসন্ধান বা ফাইল পুনরুদ্ধারের জন্য নিম্নলিখিত কোয়েরিগুলো ব্যবহার করে দেখুন।

২০২৩ সালের ব্যয়ের বিস্তারিত বিবরণ এবং এর মোট পরিমাণ উল্লেখ করুন।
আমাকে ইন্ডোমারেট থেকে রসিদ ফাইলটি দিন।

কিছু টুল ব্যবহার করার সময়, আপনি ডেভেলপমেন্ট UI-তে কী ঘটছে তা পর্যবেক্ষণ করতে পারেন।

এজেন্টটি আপনাকে কীভাবে সাড়া দেয় তা দেখুন এবং task_prompt.py ফাইলের ভেতরের প্রম্পটে দেওয়া সমস্ত নিয়ম এটি মেনে চলছে কি না তা যাচাই করুন। অভিনন্দন! এখন আপনার কাছে একটি সম্পূর্ণ কার্যকরী ডেভেলপমেন্ট এজেন্ট রয়েছে।

এখন সময় এসেছে একটি যথাযথ ও সুন্দর ইউজার ইন্টারফেস এবং ইমেজ ফাইল আপলোড ও ডাউনলোড করার সুবিধা দিয়ে এটিকে সম্পূর্ণ করার।

৮. 🚀 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+click করে ওয়েব ইন্টারফেসটি দেখতে পারেন। বিকল্পভাবে, আপনি ক্লাউড এডিটরের উপরের ডানদিকে থাকা ওয়েব প্রিভিউ বোতামে ক্লিক করে এবং পোর্ট 8080-এ প্রিভিউ নির্বাচন করে ফ্রন্টএন্ড অ্যাপ্লিকেশনটি অ্যাক্সেস করতে পারেন।

আপনি ওয়েব ইন্টারফেসটি দেখতে পাবেন, কিন্তু ব্যাকএন্ড পরিষেবাটি এখনও সেট আপ না হওয়ার কারণে চ্যাট জমা দেওয়ার চেষ্টা করার সময় একটি প্রত্যাশিত ত্রুটি পাবেন।

এখন, সার্ভিসটি চলতে দিন এবং এখনই বন্ধ করবেন না। আমরা অন্য একটি টার্মিনাল ট্যাবে ব্যাকএন্ড সার্ভিসটি চালাব।

কোডের ব্যাখ্যা

এই ফ্রন্টএন্ড কোডে, প্রথমে আমরা ব্যবহারকারীকে টেক্সট পাঠাতে এবং একাধিক ফাইল আপলোড করার সুবিধা দিই। Gradio আমাদেরকে gr.ChatInterface মেথডের সাথে gr.MultimodalTextbox যুক্ত করে এই ধরনের কার্যকারিতা তৈরি করার সুযোগ দেয়।

এখন ব্যাকএন্ডে ফাইল এবং টেক্সট পাঠানোর আগে, আমাদের ফাইলটির মাইমটাইপ বের করতে হবে, কারণ ব্যাকএন্ডের জন্য এটি প্রয়োজন। এছাড়াও, আমাদের ইমেজ ফাইলের বাইটকে বেস৬৪-এ এনকোড করে মাইমটাইপের সাথে পাঠাতে হবে।

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 ব্যবহার করি।

প্রতিক্রিয়া পাওয়ার সময়, আমরা ইতিমধ্যেই এর মধ্যে চিন্তন প্রক্রিয়া, চূড়ান্ত প্রতিক্রিয়া এবং সংযুক্তি আলাদা করে ফেলি। ফলে, আমরা UI কম্পোনেন্টের সাহায্যে প্রতিটি উপাদান প্রদর্শন করতে 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

৯. 🚀 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 এজেন্ট, সেশনসার্ভিস এবং আর্টিফ্যাক্টসার্ভিস শুরু করা হচ্ছে

ব্যাকএন্ড সার্ভিসে এজেন্টটি চালানোর জন্য আমাদের একটি রানার তৈরি করতে হবে, যা সেশনসার্ভিস এবং আমাদের এজেন্ট উভয়কেই গ্রহণ করবে। সেশনসার্ভিস কথোপকথনের ইতিহাস ও অবস্থা পরিচালনা করবে, তাই রানারের সাথে সংযুক্ত হলে এটি আমাদের এজেন্টকে চলমান কথোপকথনের প্রেক্ষাপট গ্রহণ করার ক্ষমতা দেবে।

আমরা আপলোড করা ফাইলটি পরিচালনা করার জন্য 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 ব্যবহার করেছি। যেহেতু কথোপকথনের ইতিহাস মেমরিতে সংরক্ষিত থাকে, তাই ব্যাকএন্ড সার্ভিসটি বন্ধ বা পুনরায় চালু করা হলে তা হারিয়ে যাবে। /chat রুটে ডিপেন্ডেন্সি হিসেবে ইনজেক্ট করার জন্য আমরা FastAPI অ্যাপ্লিকেশন লাইফসাইকেলের ভিতরে এগুলোকে ইনিশিয়ালাইজ করি।

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

...

এজেন্ট রানার দ্বারা প্রক্রিয়াকৃত হবে এমন সমস্ত অনুরোধকে Content type-এ ফরম্যাট করতে হবে। ফাংশনের ভিতরে, আমরা প্রতিটি ছবির ডেটাও প্রক্রিয়া করি এবং একটি Image 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,
    )
...

১০. 🚀 ইন্টিগ্রেশন টেস্ট

এখন, আপনার বিভিন্ন ক্লাউড কনসোল ট্যাবে একাধিক পরিষেবা চালু থাকা উচিত:

ফ্রন্টএন্ড সার্ভিসটি ৮০৮০ পোর্টে চালু আছে।

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

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

ব্যাকএন্ড পরিষেবা ৮০৮১ পোর্টে চালু আছে।

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)

বর্তমান অবস্থায়, আপনি ওয়েব অ্যাপ্লিকেশন থেকে পোর্ট ৮০৮০-তে আপনার রসিদের ছবি আপলোড করতে এবং সহকারীর সাথে নির্বিঘ্নে চ্যাট করতে পারবেন।

আপনার ক্লাউড শেল এডিটরের উপরের অংশে থাকা ওয়েব প্রিভিউ বোতামে ক্লিক করুন এবং পোর্ট ৮০৮০-তে প্রিভিউ নির্বাচন করুন।

এবার চলুন অ্যাসিস্ট্যান্টের সাথে কিছু আলাপচারিতা করা যাক!

নিম্নলিখিত রসিদগুলি ডাউনলোড করুন। এই রসিদগুলির তথ্য ২০২৩-২০২৪ সালের মধ্যে হতে হবে এবং সহকারীকে এটি সংরক্ষণ/আপলোড করতে বলুন।

রসিদ ড্রাইভ ( উৎস হাগিং ফেস ডেটাসেট mousserlane/id_receipt_dataset )

বিভিন্ন জিনিস জিজ্ঞাসা করুন

২০২৩-২০২৪ সালের মাসিক খরচের বিস্তারিত বিবরণ দিন।
কফি লেনদেনের রসিদটি আমাকে দেখান।
"আমাকে ইয়াকিনিকু লাইক থেকে রসিদ ফাইলটি দিন"
ইত্যাদি

এখানে সফল কথোপকথনের কিছু অংশ তুলে ধরা হলো।

১১. 🚀 ক্লাউড রান-এ ডেপ্লয় করা

এখন, অবশ্যই আমরা এই চমৎকার অ্যাপটি যেকোনো জায়গা থেকে ব্যবহার করতে চাই। তা করার জন্য, আমরা এই অ্যাপ্লিকেশনটিকে প্যাকেজ করে ক্লাউড রান-এ ডেপ্লয় করতে পারি। এই ডেমোর জন্য, এই সার্ভিসটিকে একটি পাবলিক সার্ভিস হিসেবে প্রকাশ করা হবে যা অন্যরা ব্যবহার করতে পারবে। তবে, মনে রাখবেন যে এই ধরনের অ্যাপ্লিকেশনের জন্য এটি সর্বোত্তম পন্থা নয়, কারণ এটি ব্যক্তিগত অ্যাপ্লিকেশনের জন্য বেশি উপযুক্ত।

এই কোডল্যাবে আমরা ফ্রন্টএন্ড এবং ব্যাকএন্ড উভয় সার্ভিস একটি কন্টেইনারে রাখব। উভয় সার্ভিস পরিচালনা করার জন্য আমাদের supervisord- এর সাহায্য লাগবে। আপনি supervisord.conf ফাইলটি দেখতে পারেন এবং Dockerfile-টিও পরীক্ষা করতে পারেন, যেখানে আমরা supervisord-কে এন্ট্রিপয়েন্ট হিসেবে সেট করেছি।

এই পর্যায়ে, আমাদের অ্যাপ্লিকেশনগুলো ক্লাউড রান-এ ডেপ্লয় করার জন্য প্রয়োজনীয় সমস্ত ফাইল ইতিমধ্যেই রয়েছে, চলুন এটি ডেপ্লয় করা যাক। ক্লাউড শেল টার্মিনালে যান এবং নিশ্চিত করুন যে বর্তমান প্রজেক্টটি আপনার সক্রিয় প্রজেক্ট হিসাবে কনফিগার করা আছে, যদি না থাকে তবে প্রজেক্ট আইডি সেট করার জন্য আপনাকে gcloud configure কমান্ডটি ব্যবহার করতে হবে:

gcloud config set project [PROJECT_ID]

এরপর, এটিকে ক্লাউড রান-এ ডেপ্লয় করতে নিম্নলিখিত কমান্ডটি চালান।

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

ডকার রিপোজিটরির জন্য একটি আর্টিফ্যাক্ট রেজিস্ট্রি তৈরির বিষয়টি স্বীকার করতে বলা হলে, শুধু 'Y' উত্তর দিন। উল্লেখ্য যে, এটি একটি ডেমো অ্যাপ্লিকেশন হওয়ায় আমরা এখানে প্রমাণীকরণবিহীন অ্যাক্সেসের অনুমতি দিচ্ছি। আপনার এন্টারপ্রাইজ এবং প্রোডাকশন অ্যাপ্লিকেশনগুলোর জন্য উপযুক্ত প্রমাণীকরণ ব্যবহার করার পরামর্শ দেওয়া হচ্ছে।

ডেপ্লয়মেন্ট সম্পন্ন হলে, আপনি নীচের মতো একটি লিঙ্ক পাবেন:

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

ইনকগনিটো উইন্ডো বা আপনার মোবাইল ডিভাইস থেকে অ্যাপ্লিকেশনটি ব্যবহার করুন। এটি ইতিমধ্যে চালু হয়ে যাওয়ার কথা।

১২. 🎯 চ্যালেঞ্জ

এখনই আপনার দক্ষতা দেখানোর এবং অন্বেষণের ক্ষমতাকে আরও শাণিত করার সময়। ব্যাকএন্ডকে একাধিক ব্যবহারকারীর উপযোগী করে তোলার জন্য কোড পরিবর্তন করার মতো যোগ্যতা কি আপনার আছে? কোন কোন উপাদান আপডেট করা প্রয়োজন?

১৩. 🧹 পরিষ্কার করা

এই কোডল্যাবে ব্যবহৃত রিসোর্সগুলির জন্য আপনার গুগল ক্লাউড অ্যাকাউন্টে চার্জ হওয়া এড়াতে, এই ধাপগুলি অনুসরণ করুন:

গুগল ক্লাউড কনসোলে, রিসোর্স পরিচালনা (Manage resources) পৃষ্ঠায় যান।
প্রজেক্ট তালিকা থেকে, আপনি যে প্রজেক্টটি মুছতে চান সেটি নির্বাচন করুন এবং তারপর ডিলিট বোতামে ক্লিক করুন।
ডায়ালগ বক্সে প্রজেক্ট আইডি টাইপ করুন এবং তারপর প্রজেক্টটি মুছে ফেলার জন্য 'শাট ডাউন'-এ ক্লিক করুন।
বিকল্পভাবে, আপনি কনসোলে Cloud Run- এ গিয়ে, এইমাত্র ডেপ্লয় করা সার্ভিসটি নির্বাচন করে ডিলিট করে দিতে পারেন।