১. ভূমিকা
এই কোডল্যাবে, আপনি গুগল এজেন্ট ডেভেলপমেন্ট কিট (ADK) ব্যবহার করে একটি অত্যাধুনিক এআই এজেন্ট তৈরি করতে শিখবেন। আমরা একটি স্বাভাবিক বিবর্তনমূলক পথ অনুসরণ করব, যা একটি মৌলিক কথোপকথনমূলক এজেন্ট থেকে শুরু করে ক্রমান্বয়ে বিশেষায়িত সক্ষমতা যুক্ত করবে।
আমরা যে এজেন্টটি তৈরি করছি, সেটি একজন এক্সিকিউটিভ অ্যাসিস্ট্যান্ট, যা আপনাকে ক্যালেন্ডার পরিচালনা, কাজের কথা মনে করিয়ে দেওয়া, গবেষণা করা এবং নোট সংকলনের মতো দৈনন্দিন কাজে সাহায্য করার জন্য ডিজাইন করা হয়েছে। এর সবকিছুই ADK, Gemini এবং Vertex AI ব্যবহার করে একেবারে গোড়া থেকে তৈরি করা হয়েছে।
এই ল্যাবের শেষে আপনার কাছে একটি সম্পূর্ণ কার্যকরী এজেন্ট থাকবে এবং এটিকে আপনার নিজের প্রয়োজন অনুযায়ী সম্প্রসারণ করার জন্য প্রয়োজনীয় জ্ঞানও থাকবে।
পূর্বশর্ত
- পাইথন প্রোগ্রামিং ভাষার প্রাথমিক জ্ঞান
- ক্লাউড রিসোর্স পরিচালনা করার জন্য গুগল ক্লাউড কনসোল সম্পর্কে প্রাথমিক জ্ঞান থাকা প্রয়োজন।
আপনি যা শিখবেন
- এআই এজেন্টদের জন্য গুগল ক্লাউড পরিকাঠামো প্রস্তুত করা হচ্ছে।
- ভার্টেক্স এআই মেমোরি ব্যাংক ব্যবহার করে স্থায়ী দীর্ঘমেয়াদী স্মৃতি বাস্তবায়ন।
- বিশেষায়িত উপ-এজেন্টদের একটি স্তরবিন্যাস গঠন করা।
- বাহ্যিক ডেটাবেস এবং গুগল ওয়ার্কস্পেস ইকোসিস্টেমের একীকরণ।
আপনার যা যা লাগবে
এই ওয়ার্কশপটি সম্পূর্ণভাবে গুগল ক্লাউড শেল-এর মধ্যেই করা যায়, যেখানে সমস্ত প্রয়োজনীয় ডিপেন্ডেন্সি (gcloud CLI, কোড এডিটর, Go, Gemini CLI) আগে থেকেই ইনস্টল করা থাকে।
বিকল্পভাবে , যদি আপনি নিজের মেশিনে কাজ করতে পছন্দ করেন, তাহলে আপনার নিম্নলিখিত জিনিসগুলির প্রয়োজন হবে:
- পাইথন (সংস্করণ ৩.১২ বা উচ্চতর)
- একটি কোড এডিটর বা IDE (যেমন VS Code বা
vim)। - পাইথন এবং
gcloudকমান্ড চালানোর জন্য একটি টার্মিনাল। - সুপারিশকৃত: জেমিনি সিএলআই (Gemini CLI) বা অ্যান্টিগ্র্যাভিটির (Antigravity) মতো একটি কোডিং এজেন্ট।
মূল প্রযুক্তি
আমরা যে প্রযুক্তিগুলো ব্যবহার করব সে সম্পর্কে আরও তথ্য এখানে পেতে পারেন:
২. পরিবেশ সেটআপ
নিম্নলিখিত বিকল্পগুলির মধ্যে একটি বেছে নিন: আপনি যদি এই কোডল্যাবটি আপনার নিজের মেশিনে চালাতে চান তবে 'সেলফ-পেজড এনভায়রনমেন্ট সেটআপ' , অথবা; আপনি যদি এই কোডল্যাবটি সম্পূর্ণরূপে ক্লাউডে চালাতে চান তবে 'স্টার্ট ক্লাউড শেল' ।
স্ব-গতিতে পরিবেশ সেটআপ
- Google Cloud Console- এ সাইন-ইন করুন এবং একটি নতুন প্রজেক্ট তৈরি করুন অথবা বিদ্যমান কোনো প্রজেক্ট পুনরায় ব্যবহার করুন। যদি আপনার আগে থেকে Gmail বা Google Workspace অ্যাকাউন্ট না থাকে, তবে আপনাকে অবশ্যই একটি তৈরি করতে হবে।
- প্রজেক্টের নামটি হলো এই প্রজেক্টের অংশগ্রহণকারীদের প্রদর্শিত নাম। এটি একটি ক্যারেক্টার স্ট্রিং যা গুগল এপিআই ব্যবহার করে না। আপনি যেকোনো সময় এটি আপডেট করতে পারেন।
- প্রজেক্ট আইডি সমস্ত গুগল ক্লাউড প্রজেক্ট জুড়ে অনন্য এবং অপরিবর্তনীয় (একবার সেট করার পর এটি পরিবর্তন করা যায় না)। ক্লাউড কনসোল স্বয়ংক্রিয়ভাবে একটি অনন্য স্ট্রিং তৈরি করে; সাধারণত এটি কী তা নিয়ে আপনার মাথা ঘামানোর দরকার নেই। বেশিরভাগ কোডল্যাবে, আপনাকে আপনার প্রজেক্ট আইডি উল্লেখ করতে হবে (যা সাধারণত
PROJECT_IDহিসাবে চিহ্নিত করা হয়)। তৈরি করা আইডিটি আপনার পছন্দ না হলে, আপনি এলোমেলোভাবে আরেকটি তৈরি করতে পারেন। বিকল্পভাবে, আপনি আপনার নিজের আইডি দিয়ে চেষ্টা করে দেখতে পারেন যে সেটি উপলব্ধ আছে কিনা। এই ধাপের পরে এটি পরিবর্তন করা যাবে না এবং প্রজেক্টের পুরো সময়কাল জুড়ে এটি অপরিবর্তিত থাকবে। - আপনার অবগতির জন্য জানাচ্ছি যে, তৃতীয় একটি ভ্যালু রয়েছে, যা হলো প্রজেক্ট নম্বর , এবং কিছু এপিআই এটি ব্যবহার করে থাকে। ডকুমেন্টেশনে এই তিনটি ভ্যালু সম্পর্কে আরও বিস্তারিত জানুন।
- এরপর, ক্লাউড রিসোর্স/এপিআই ব্যবহার করার জন্য আপনাকে ক্লাউড কনসোলে বিলিং চালু করতে হবে। এই কোডল্যাবটি সম্পন্ন করতে খুব বেশি খরচ হবে না, এমনকি আদৌ কোনো খরচ নাও হতে পারে। এই টিউটোরিয়ালের পর বিলিং এড়াতে রিসোর্সগুলো বন্ধ করার জন্য, আপনি আপনার তৈরি করা রিসোর্সগুলো অথবা প্রজেক্টটি ডিলিট করে দিতে পারেন। নতুন গুগল ক্লাউড ব্যবহারকারীরা ৩০০ মার্কিন ডলারের ফ্রি ট্রায়াল প্রোগ্রামের জন্য যোগ্য।
ক্লাউড শেল শুরু করুন
যদিও গুগল ক্লাউড আপনার ল্যাপটপ থেকে দূরবর্তীভাবে পরিচালনা করা যায়, এই কোডল্যাবে আপনি গুগল ক্লাউড শেল ব্যবহার করবেন, যা ক্লাউডে চালিত একটি কমান্ড লাইন পরিবেশ।
গুগল ক্লাউড কনসোল থেকে, উপরের ডানদিকের টুলবারে থাকা ক্লাউড শেল আইকনটিতে ক্লিক করুন:
পরিবেশটি প্রস্তুত করতে এবং এর সাথে সংযোগ স্থাপন করতে মাত্র কয়েক মুহূর্ত সময় লাগবে। এটি শেষ হলে, আপনি এইরকম কিছু দেখতে পাবেন:
এই ভার্চুয়াল মেশিনটিতে আপনার প্রয়োজনীয় সমস্ত ডেভেলপমেন্ট টুলস লোড করা আছে। এটি একটি স্থায়ী ৫ জিবি হোম ডিরেক্টরি প্রদান করে এবং গুগল ক্লাউডে চলে, যা নেটওয়ার্ক পারফরম্যান্স ও অথেনটিকেশনকে ব্যাপকভাবে উন্নত করে। এই কোডল্যাবে আপনার সমস্ত কাজ একটি ব্রাউজারের মধ্যেই করা যাবে। আপনাকে কিছুই ইনস্টল করতে হবে না।
৩. প্রজেক্ট সেটআপ
কোড লেখার আগে, আমাদের অবশ্যই গুগল ক্লাউডে প্রয়োজনীয় পরিকাঠামো এবং অনুমতিসমূহ প্রস্তুত করে নিতে হবে।
পরিবেশ ভেরিয়েবল সেট করুন
টার্মিনাল খুলুন এবং নিম্নলিখিত এনভায়রনমেন্ট ভেরিয়েবলগুলো সেট করুন:
export PROJECT_ID=`gcloud config get project`
export LOCATION=us-central1
প্রয়োজনীয় এপিআইগুলি সক্রিয় করুন
আপনার এজেন্টের বেশ কয়েকটি গুগল ক্লাউড পরিষেবাতে অ্যাক্সেস প্রয়োজন। সেগুলি সক্রিয় করতে নিম্নলিখিত কমান্ডটি চালান:
gcloud services enable \
aiplatform.googleapis.com \
calendar-json.googleapis.com \
sqladmin.googleapis.com
অ্যাপ্লিকেশনের ডিফল্ট ক্রেডেনশিয়াল দিয়ে প্রমাণীকরণ করুন
আপনার পরিবেশ থেকে গুগল ক্লাউড পরিষেবাগুলির সাথে যোগাযোগ করার জন্য আমাদের অ্যাপ্লিকেশন ডিফল্ট ক্রেডেনশিয়াল (ADC) দিয়ে প্রমাণীকরণ করতে হবে।
আপনার অ্যাপ্লিকেশনের ডিফল্ট ক্রেডেনশিয়ালগুলো সক্রিয় ও হালনাগাদ আছে কিনা তা নিশ্চিত করতে নিম্নলিখিত কমান্ডটি চালান:
gcloud auth application-default login
৪. বেস এজেন্ট তৈরি করুন
এখন, আমাদের সেই ডিরেক্টরিটি প্রস্তুত করতে হবে যেখানে আমরা প্রজেক্টের সোর্স কোড সংরক্ষণ করব:
# setup project directory
mkdir -p adk_ea_codelab && cd adk_ea_codelab
# prepare virtual environment
uv init
# install dependencies
uv add google-adk google-api-python-client tzlocal python-dotenv
uv add cloud-sql-python-connector[pg8000] sqlalchemy
আমরা এজেন্টের পরিচয় এবং প্রাথমিক কথোপকথনের ক্ষমতা নির্ধারণের মাধ্যমে শুরু করি। ADK-তে, Agent ক্লাসটি এজেন্টের ব্যক্তিত্ব এবং তাদের নির্দেশাবলী সংজ্ঞায়িত করে।
এই সময়ে আপনি একজন এজেন্টের নাম নিয়ে ভাবতে পারেন। আমি আমার এজেন্টদের আইডা বা শ্যারনের মতো বিশেষ নাম দিতে পছন্দ করি, কারণ আমার মনে হয় এটি তাদের একটি স্বতন্ত্র "ব্যক্তিত্ব" দিতে সাহায্য করে, তবে আপনি এজেন্টের কাজ অনুযায়ীও তাকে ডাকতে পারেন, যেমন "এক্সিকিউটিভ অ্যাসিস্ট্যান্ট", "ট্র্যাভেল এজেন্ট" বা "কোড এক্সিকিউটর"।
একটি বয়লারপ্লেট এজেন্ট চালু করতে adk create কমান্ডটি চালান:
# replace with your desired agent name
uv run adk create executive_assistant
অনুগ্রহ করে মডেল হিসেবে gemini-2.5-flash এবং ব্যাকএন্ড হিসেবে Vertex AI বেছে নিন। প্রস্তাবিত প্রজেক্ট আইডিটি এই ল্যাবের জন্য আপনার তৈরি করা আইডি কিনা তা পুনরায় যাচাই করুন এবং নিশ্চিত করতে এন্টার চাপুন। গুগল ক্লাউড অঞ্চলের জন্য, আপনি ডিফল্টটি ( us-central1 ) গ্রহণ করতে পারেন। আপনার টার্মিনালটি দেখতে এইরকম হবে:
daniela_petruzalek@cloudshell:~/adk_ea_codelab (your-project-id)$ uv run adk create executive_assistant 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 [your-project-id]: Enter Google Cloud region [us-central1]: Agent created in /home/daniela_petruzalek/adk_ea_codelab/executive_assistant: - .env - __init__.py - agent.py daniela_petruzalek@cloudshell:~/adk_ea_codelab (your-project-id)$
একবার সম্পন্ন হলে, পূর্ববর্তী কমান্ডটি এজেন্টের নামে (যেমন executive_assistant ) একটি ফোল্ডার তৈরি করবে, যার মধ্যে কয়েকটি ফাইল থাকবে, যার মধ্যে agent.py ফাইলটিতে এজেন্টের প্রাথমিক সংজ্ঞা দেওয়া থাকবে:
from google.adk.agents.llm_agent 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 web কমান্ডটি চালিয়ে আপনার ব্রাউজারে ডেভেলপমেন্ট UI খুলতে পারেন। আপনি এইরকম কিছু দেখতে পাবেন:
$ uv run adk web ... INFO: Started server process [1244] INFO: Waiting for application startup. +-----------------------------------------------------------------------------+ | ADK Web Server started | | | | For local testing, access at http://127.0.0.1:8000. | +-----------------------------------------------------------------------------+ INFO: Application startup complete. INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
এই এজেন্টটি বেশ সাধারণ হলেও, এজেন্টটি সম্পাদনা শুরু করার আগে সেটআপটি সঠিকভাবে কাজ করছে কিনা তা নিশ্চিত করতে অন্তত একবার এটি করা দরকার। নিচের স্ক্রিনশটটিতে ডেভেলপমেন্ট UI ব্যবহার করে একটি সাধারণ ইন্টারঅ্যাকশন দেখানো হয়েছে:
এখন, চলুন আমাদের এক্সিকিউটিভ অ্যাসিস্ট্যান্ট পার্সোনা দিয়ে এজেন্ট ডেফিনিশনটি পরিবর্তন করি। নিচের কোডটি কপি করে agent.py ফাইলের ভেতরের কন্টেন্টগুলো প্রতিস্থাপন করুন। এজেন্টের নাম এবং পার্সোনা আপনার পছন্দ অনুযায়ী পরিবর্তন করুন।
from google.adk.agents.llm_agent import Agent
root_agent = Agent(
model='gemini-2.5-flash',
name='executive_assistant',
description='A professional AI Executive Assistant',
instruction='''
You are an elite, warm, and highly efficient AI partner.
Your primary goal is to help the user manage their tasks, schedule, and research.
Always be direct, concise, and high-signal.
''',
)
অনুগ্রহ করে মনে রাখবেন যে, 'name' প্রপার্টিটি এজেন্টের অভ্যন্তরীণ নাম নির্ধারণ করে, যদিও নির্দেশাবলীতে আপনি শেষ ব্যবহারকারীর সাথে যোগাযোগের জন্য তাদের পরিচিতির অংশ হিসেবে একটি সহজবোধ্য নামও দিতে পারেন। অভ্যন্তরীণ নামটি মূলত transfer_to_agent টুল ব্যবহার করে মাল্টি-এজেন্ট সিস্টেমে পর্যবেক্ষণযোগ্যতা এবং হস্তান্তরের জন্য ব্যবহৃত হয়। আপনাকে নিজে এই টুলটি নিয়ে কাজ করতে হবে না, আপনি যখন এক বা একাধিক সাব-এজেন্ট ঘোষণা করবেন, তখন ADK স্বয়ংক্রিয়ভাবে এটি রেজিস্টার করে নেবে।
এইমাত্র তৈরি করা এজেন্টটি চালানোর জন্য, adk web ব্যবহার করুন:
uv run adk web
ব্রাউজারে ADK UI খুলুন এবং আপনার নতুন সহকারীকে স্বাগত জানান!
৫. ভার্টেক্স এআই মেমোরি ব্যাংকের মাধ্যমে স্থায়ী মেমোরি যোগ করুন।
একটি নির্বিঘ্ন ও ব্যক্তিগতকৃত অভিজ্ঞতা প্রদানের জন্য একজন প্রকৃত সহকারীকে অবশ্যই ব্যবহারকারীর পছন্দ এবং পূর্ববর্তী কথোপকথন মনে রাখতে হবে। এই ধাপে, আমরা Vertex AI Agent Engine Memory Bank-কে একীভূত করব, যা Vertex AI-এর একটি বৈশিষ্ট্য এবং এটি ব্যবহারকারীর কথোপকথনের উপর ভিত্তি করে গতিশীলভাবে দীর্ঘমেয়াদী স্মৃতি তৈরি করে।
মেমোরি ব্যাংক আপনার এজেন্টকে একাধিক সেশন জুড়ে অ্যাক্সেসযোগ্য ব্যক্তিগতকৃত তথ্য তৈরি করার সুযোগ দেয়, যা সেশন-মধ্যবর্তী ধারাবাহিকতা স্থাপন করে। নেপথ্যে, এটি একটি সেশনের বার্তাগুলির কালানুক্রমিক ক্রম পরিচালনা করে এবং বর্তমান প্রেক্ষাপটের জন্য এজেন্টকে সবচেয়ে প্রাসঙ্গিক স্মৃতি সরবরাহ করতে সাদৃশ্য অনুসন্ধান পুনরুদ্ধার পদ্ধতি ব্যবহার করতে পারে।
মেমরি পরিষেবা শুরু করুন
ADK দীর্ঘমেয়াদী স্মৃতি সংরক্ষণ এবং পুনরুদ্ধার করতে Vertex AI ব্যবহার করে। আপনাকে আপনার প্রোজেক্টে একটি "মেমরি ইঞ্জিন" ইনিশিয়ালাইজ করতে হবে। এটি মূলত একটি রিজনিং ইঞ্জিন ইনস্ট্যান্স যা মেমরি ব্যাংক হিসেবে কাজ করার জন্য কনফিগার করা হয়েছে।
setup_memory.py নামে নিম্নলিখিত স্ক্রিপ্টটি তৈরি করুন:
setup_memory.py
import vertexai
import os
PROJECT_ID=os.getenv("PROJECT_ID")
LOCATION=os.getenv("LOCATION")
client = vertexai.Client(project=PROJECT_ID, location=LOCATION)
# Create Reasoning Engine for Memory Bank
agent_engine = client.agent_engines.create()
# You will need this resource name to give it to ADK
print(agent_engine.api_resource.name)
এখন মেমরি ব্যাংকের জন্য রিজনিং ইঞ্জিন প্রস্তুত করতে setup_memory.py চালান:
uv run python setup_memory.py
আপনার আউটপুটটি দেখতে এইরকম হওয়া উচিত:
$ uv run python setup.py projects/1234567890/locations/us-central1/reasoningEngines/1234567890
ইঞ্জিন রিসোর্সের নামটি একটি এনভায়রনমেন্ট ভেরিয়েবলে সংরক্ষণ করুন:
export ENGINE_ID="<insert the resource name above>"
এখন আমাদের পারসিস্টেন্ট মেমরি ব্যবহার করার জন্য কোডটি আপডেট করতে হবে। agent.py ফাইলের বিষয়বস্তু নিম্নলিখিত কোড দিয়ে প্রতিস্থাপন করুন:
agent.py
from google.adk.agents.llm_agent import Agent
from google.adk.tools.preload_memory_tool import PreloadMemoryTool
from google.adk.tools.load_memory_tool import load_memory_tool
async def auto_save_session_to_memory_callback(callback_context):
await callback_context._invocation_context.memory_service.add_session_to_memory(
callback_context._invocation_context.session)
# Update root_agent with memory tools and callback
root_agent = Agent(
model='gemini-2.5-flash',
name='executive_assistant',
description='Executive Assistant with Persistent Memory',
instruction='''
You are an elite AI partner with long-term memory.
Use load_memory to find context about the user when needed.
Always be direct, concise, and high-signal.
''',
tools=[PreloadMemoryTool(), load_memory_tool],
after_agent_callback=auto_save_session_to_memory_callback,
)
PreloadMemoryTool স্বয়ংক্রিয়ভাবে অতীতের কথোপকথন থেকে প্রাসঙ্গিক তথ্য প্রতিটি অনুরোধে যুক্ত করে (সাদৃশ্য অনুসন্ধান পদ্ধতির মাধ্যমে), অন্যদিকে load_memory_tool প্রয়োজনে মডেলকে মেমরি ব্যাংক থেকে তথ্য সংগ্রহের জন্য সুস্পষ্টভাবে কোয়েরি করার সুযোগ দেয়। এই সমন্বয় আপনার এজেন্টকে গভীর ও স্থায়ী তথ্য প্রদান করে!
এখন মেমরি সাপোর্ট সহ আপনার এজেন্ট চালু করতে, adk web চালানোর সময় আপনাকে memory_service_uri পাস করতে হবে:
uv run adk web --memory_service_uri="agentengine://$ENGINE_ID"
এজেন্টদেরকে নিজের সম্পর্কে কিছু তথ্য দেওয়ার চেষ্টা করুন, এবং তারপর অন্য একটি সেশনে ফিরে এসে তাদের সম্পর্কে জিজ্ঞাসা করুন। উদাহরণস্বরূপ, আপনার নাম বলুন:
এজেন্ট যে স্মৃতিগুলো সংরক্ষণ করছে, তা আপনি ক্লাউড কনসোলে পরীক্ষা করতে পারেন। 'এজেন্ট ইঞ্জিন'-এর প্রোডাক্ট পেজে যান (সার্চ বার ব্যবহার করুন)।
তারপর আপনার এজেন্ট ইঞ্জিনের নামে ক্লিক করুন (সঠিক অঞ্চল নির্বাচন করেছেন কিনা তা নিশ্চিত করুন):
এবং তারপর মেমোরিজ ট্যাবে যান:
আপনার কিছু স্মৃতি যুক্ত হতে দেখা উচিত।
৬. ওয়েব গবেষণার সক্ষমতা যোগ করুন
উচ্চমানের তথ্য প্রদানের জন্য, আমাদের এজেন্টকে এমন গভীর তদন্ত করতে হয় যা কেবল একটিমাত্র অনুসন্ধানের মধ্যে সীমাবদ্ধ থাকে না। একজন বিশেষায়িত উপ-এজেন্টকে গবেষণার দায়িত্ব অর্পণ করার মাধ্যমে আমরা মূল ব্যক্তির সক্রিয়তা বজায় রাখি, এবং একই সাথে গবেষক নেপথ্যে থেকে জটিল তথ্য সংগ্রহের কাজটি সামলান।
এই ধাপে, আমরা 'গবেষণার গভীরতা' অর্জনের জন্য একটি LoopAgent প্রয়োগ করি—যা এজেন্টকে একটি সম্পূর্ণ চিত্র না পাওয়া পর্যন্ত পুনরাবৃত্তিমূলকভাবে অনুসন্ধান করতে, প্রাপ্ত ফলাফল মূল্যায়ন করতে এবং তার কোয়েরিগুলোকে পরিমার্জন করতে সাহায্য করে। এছাড়াও, আমরা সমস্ত প্রাপ্ত ফলাফলের জন্য ইনলাইন সাইটেশন বাধ্যতামূলক করে প্রযুক্তিগত কঠোরতা নিশ্চিত করি, যা প্রতিটি দাবির সমর্থনে একটি উৎস লিঙ্ক থাকা নিশ্চিত করে।
গবেষণা বিশেষজ্ঞ তৈরি করুন (research.py)
এখানে আমরা গুগল সার্চ টুলসহ একটি বেস এজেন্ট সংজ্ঞায়িত করি এবং এটিকে একটি LoopAgent-এর মধ্যে আবদ্ধ করি। max_iterations প্যারামিটারটি একটি গভর্নর হিসেবে কাজ করে, যা নিশ্চিত করে যে এজেন্টের উপলব্ধিতে কোনো ফাঁক থাকলে এটি অনুসন্ধানটি সর্বোচ্চ ৩ বার পর্যন্ত পুনরাবৃত্তি করবে।
research.py
from google.adk.agents.llm_agent import Agent
from google.adk.agents.loop_agent import LoopAgent
from google.adk.tools.google_search_tool import GoogleSearchTool
from google.adk.tools.tool_context import ToolContext
def exit_loop(tool_context: ToolContext):
"""Call this function ONLY when no further research is needed, signaling the iterative process should end."""
print(f" [Tool Call] exit_loop triggered by {tool_context.agent_name}")
tool_context.actions.escalate = True
# Return empty dict as tools should typically return JSON-serializable output
return {}
# --- RESEARCH LOGIC ---
_research_worker = Agent(
model='gemini-2.5-flash',
name='research_worker',
description='Worker agent that performs a single research step.',
instruction='''
Use google_search to find facts and synthesize them for the user.
Critically evaluate your findings. If the data is incomplete or you need more context, prepare to search again in the next iteration.
You must include the links you found as references in your response, formatting them like citations in a research paper (e.g., [1], [2]).
Use the exit_loop tool to terminate the research early if no further research is needed.
If you need to ask the user for clarifications, call the exit_loop function early to interrupt the research cycle.
''',
tools=[GoogleSearchTool(bypass_multi_tools_limit=True), exit_loop],
)
# The LoopAgent iterates the worker up to 3 times for deeper research
research_agent = LoopAgent(
name='research_specialist',
description='Deep web research specialist.',
sub_agents=[_research_worker],
max_iterations=3,
)
রুট এজেন্ট (agent.py) আপডেট করুন।
research_agent-কে ইম্পোর্ট করুন এবং এটিকে Sharon-এর একটি টুল হিসেবে যুক্ত করুন:
agent.py
from google.adk.agents.llm_agent import Agent
from google.adk.tools.preload_memory_tool import PreloadMemoryTool
from google.adk.tools.load_memory_tool import load_memory_tool
# Import our new sub agent
from .research import research_agent
async def auto_save_session_to_memory_callback(callback_context):
await callback_context._invocation_context.memory_service.add_session_to_memory(
callback_context._invocation_context.session)
# Update root_agent with memory tools and callback
root_agent = Agent(
model='gemini-2.5-flash',
name='executive_assistant',
description='Executive Assistant with persistent memory and research capabilities',
instruction='''
You are an elite AI partner with long-term memory.
1. Use load_memory to recall facts.
2. Delegate research tasks to the research_specialist.
Always be direct, concise, and high-signal.
''',
tools=[PreloadMemoryTool(), load_memory_tool],
sub_agents=[research_agent],
after_agent_callback=auto_save_session_to_memory_callback,
)
রিসার্চ এজেন্টটি পরীক্ষা করার জন্য এডিকে ওয়েব আবার চালু করুন।
uv run adk web --memory_service_uri="agentengine://$ENGINE_ID"
একে একটি সহজ গবেষণামূলক কাজ দিন, যেমন, "কীভাবে একটি ভালো প্রযুক্তি ব্লগ লিখতে হয়?"
আপনি হয়তো লক্ষ্য করেছেন যে, এটি একটি নতুন সেশন হওয়া সত্ত্বেও এজেন্ট আমার নাম মনে রেখেছে। অনুগ্রহ করে "transfer_to_agent" টুল কলটিও লক্ষ্য করুন: এই টুলটিই আমাদের নতুন গবেষণা এজেন্টের কাছে কাজটি হস্তান্তর করে।
এবার, টাস্ক ম্যানেজমেন্ট নিয়ে আলোচনা শুরু করা যাক।
৭. ক্লাউড এসকিউএল-এর মাধ্যমে টাস্ক ম্যানেজমেন্ট যুক্ত করুন
এজেন্টের দীর্ঘমেয়াদী স্মৃতি থাকলেও, এটি করণীয় তালিকার মতো সূক্ষ্ম ও কাঠামোগত ডেটার জন্য উপযুক্ত নয়। কাজের জন্য আমরা একটি প্রচলিত রিলেশনাল ডেটাবেস ব্যবহার করি। আমরা SQLAlchemy এবং একটি Google Cloud SQL (PostgreSQL) ডেটাবেস ব্যবহার করতে যাচ্ছি। কোড লেখার আগে, আমাদের অবশ্যই পরিকাঠামো প্রস্তুত করতে হবে।
অবকাঠামো সরবরাহ করুন
আপনার ডাটাবেস তৈরি করতে এই কমান্ডগুলো চালান। দ্রষ্টব্য: ইনস্ট্যান্স তৈরি হতে প্রায় ৫-১০ মিনিট সময় লাগে। এটি ব্যাকগ্রাউন্ডে চলতে থাকার সময় আপনি পরবর্তী ধাপে এগিয়ে যেতে পারেন।
# 1. Define instance variables
export INSTANCE_NAME="assistant-db"
export USER_EMAIL=$(gcloud config get-value account)
# 2. Create the Cloud SQL instance
gcloud sql instances create $INSTANCE_NAME \
--database-version=POSTGRES_18 \
--tier=db-f1-micro \
--region=us-central1 \
--edition=ENTERPRISE
# 3. Create the database for our tasks
gcloud sql databases create tasks --instance=$INSTANCE_NAME
ডাটাবেস ইনস্ট্যান্স প্রোভিশনিং হতে কয়েক মিনিট সময় লাগবে। এটি শেষ হওয়ার জন্য অপেক্ষা করার সময় এক কাপ কফি বা চা খেয়ে নিতে পারেন, অথবা কোড আপডেট করে নিতে পারেন। তবে ফিরে এসে অ্যাক্সেস কন্ট্রোলের কাজটি শেষ করতে ভুলবেন না!
অ্যাক্সেস নিয়ন্ত্রণ কনফিগার করুন
এখন ডেটাবেসে অ্যাক্সেস পাওয়ার জন্য আপনার ইউজার অ্যাকাউন্টটি কনফিগার করতে হবে। টার্মিনালে নিম্নলিখিত কমান্ডগুলো চালান:
# change this to your favorite password
export DB_PASS="correct-horse-battery-staple"
# Create a regular database user
gcloud sql users create assistant_user \
--instance=$INSTANCE_NAME \
--password=$DB_PASS
পরিবেশ কনফিগারেশন আপডেট করুন
ADK রানটাইমে একটি .env ফাইল থেকে কনফিগারেশন লোড করে। ডাটাবেস সংযোগের বিবরণ দিয়ে আপনার এজেন্টের এনভায়রনমেন্ট আপডেট করুন।
# Retrieve the unique connection name
export DB_CONN=$(gcloud sql instances describe $INSTANCE_NAME --format='value(connectionName)')
# Append configuration to your .env file
cat <<EOF >> executive_assistant/.env
DB_CONNECTION_NAME=$DB_CONN
DB_USER=assistant_user
DB_PASSWORD=$DB_PASS
DB_NAME=tasks
EOF
এবার চলুন কোড পরিবর্তনগুলো করা যাক।
টাস্ক স্পেশালিস্ট (todo.py) তৈরি করুন
রিসার্চ এজেন্টের মতোই, চলুন আমাদের টু-ডু স্পেশালিস্টকে তার নিজস্ব ফাইলে তৈরি করি। todo.py তৈরি করুন:
todo.py
import os
import uuid
import sqlalchemy
from datetime import datetime
from typing import Optional, List
from sqlalchemy import (
Column,
String,
DateTime,
Enum,
select,
delete,
update,
)
from sqlalchemy.orm import declarative_base, Session
from google.cloud.sql.connector import Connector
from google.adk.agents.llm_agent import Agent
# --- DATABASE LOGIC ---
Base = declarative_base()
connector = Connector()
def getconn():
db_connection_name = os.environ.get("DB_CONNECTION_NAME")
db_user = os.environ.get("DB_USER")
db_password = os.environ.get("DB_PASSWORD")
db_name = os.environ.get("DB_NAME", "tasks")
return connector.connect(
db_connection_name,
"pg8000",
user=db_user,
password=db_password,
db=db_name,
)
engine = sqlalchemy.create_engine(
"postgresql+pg8000://",
creator=getconn,
)
class Todo(Base):
__tablename__ = "todos"
id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
title = Column(String(255), nullable=False)
priority = Column(
Enum("high", "medium", "low", name="priority_levels"), nullable=False, default="medium"
)
due_date = Column(DateTime, nullable=True)
status = Column(Enum("pending", "done", name="status_levels"), default="pending")
created_at = Column(DateTime, default=datetime.utcnow)
def init_db():
"""Builds the table if it's missing."""
Base.metadata.create_all(bind=engine)
def add_todo(
title: str, priority: str = "medium", due_date: Optional[str] = None
) -> dict:
"""
Adds a new task to the list.
Args:
title (str): The description of the task.
priority (str): The urgency level. Must be one of: 'high', 'medium', 'low'.
due_date (str, optional): The due date in ISO format (YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS).
Returns:
dict: A dictionary containing the new task's ID and a status message.
"""
init_db()
with Session(engine) as session:
due = datetime.fromisoformat(due_date) if due_date else None
item = Todo(
title=title,
priority=priority.lower(),
due_date=due,
)
session.add(item)
session.commit()
return {"id": item.id, "status": f"Task added ✅"}
def list_todos(status: str = "pending") -> list:
"""
Lists tasks from the database, optionally filtering by status.
Args:
status (str, optional): The status to filter by. 'pending', 'done', or 'all'.
"""
init_db()
with Session(engine) as session:
query = select(Todo)
s_lower = status.lower()
if s_lower != "all":
query = query.where(Todo.status == s_lower)
query = query.order_by(Todo.priority, Todo.created_at)
results = session.execute(query).scalars().all()
return [
{
"id": t.id,
"task": t.title,
"priority": t.priority,
"status": t.status,
}
for t in results
]
def complete_todo(task_id: str) -> str:
"""Marks a specific task as 'done'."""
init_db()
with Session(engine) as session:
session.execute(update(Todo).where(Todo.id == task_id).values(status="done"))
session.commit()
return f"Task {task_id} marked as done."
def delete_todo(task_id: str) -> str:
"""Permanently removes a task from the database."""
init_db()
with Session(engine) as session:
session.execute(delete(Todo).where(Todo.id == task_id))
session.commit()
return f"Task {task_id} deleted."
# --- TODO SPECIALIST AGENT ---
todo_agent = Agent(
model='gemini-2.5-flash',
name='todo_specialist',
description='A specialist agent that manages a structured SQL task list.',
instruction='''
You manage the user's task list using a PostgreSQL database.
- Use add_todo when the user wants to remember something. If no priority is mentioned, mark it as 'medium'.
- Use list_todos to show tasks.
- Use complete_todo to mark a task as finished.
- Use delete_todo to remove a task entirely.
When marking a task as complete or deleting it, if the user doesn't provide the ID,
use list_todos first to find the correct ID for the task they described.
''',
tools=[add_todo, list_todos, complete_todo, delete_todo],
)
উপরের কোডটি মূলত দুটি কাজ করে: ক্লাউড SQL ডেটাবেসের সাথে সংযোগ স্থাপন করা এবং করণীয় কাজের তালিকার সমস্ত সাধারণ কার্যকলাপের জন্য প্রয়োজনীয় টুলস সরবরাহ করা, যার মধ্যে রয়েছে কাজ যোগ করা, মুছে ফেলা এবং সম্পন্ন হিসেবে চিহ্নিত করা।
যেহেতু এই লজিকটি টু-ডু এজেন্টের জন্য খুবই সুনির্দিষ্ট, এবং এক্সিকিউটিভ অ্যাসিস্ট্যান্ট (রুট এজেন্ট)-এর দৃষ্টিকোণ থেকে এই সূক্ষ্ম ব্যবস্থাপনা নিয়ে আমরা অগত্যা চিন্তিত নই, তাই আমরা এই এজেন্টটিকে সাব-এজেন্টের পরিবর্তে একটি ' AgentTool ' হিসেবে প্যাকেজ করব।
AgentTool নাকি সাব-এজেন্ট ব্যবহার করবেন, সেই সিদ্ধান্ত নিতে হলে তাদের কনটেক্সট শেয়ার করার প্রয়োজন আছে কি না, তা বিবেচনা করুন:
- যখন আপনার এজেন্টের রুট এজেন্টের সাথে কনটেক্সট শেয়ার করার প্রয়োজন হয় না, তখন একটি AgentTool ব্যবহার করুন।
- যখন আপনি চান আপনার এজেন্ট রুট এজেন্টের সাথে কনটেক্সট শেয়ার করুক, তখন একটি সাব-এজেন্ট ব্যবহার করুন।
গবেষণা এজেন্টের ক্ষেত্রে কনটেক্সট শেয়ার করা উপকারী হতে পারে, কিন্তু একটি সাধারণ টু-ডু এজেন্টের জন্য তা করার তেমন কোনো সুবিধা নেই।
চলুন agent.py ফাইলে AgentTool প্রয়োগ করি।
রুট এজেন্ট (agent.py) আপডেট করুন।
এখন, আপনার মূল ফাইলে todo_agent ইম্পোর্ট করুন এবং এটিকে একটি টুল হিসেবে সংযুক্ত করুন:
agent.py
import os
from datetime import datetime
from google.adk.agents.llm_agent import Agent
from google.adk.tools.agent_tool import AgentTool
from google.adk.tools.preload_memory_tool import PreloadMemoryTool
from google.adk.tools.load_memory_tool import load_memory_tool
# Import our specialized sub-agents
from .research import research_agent
from .todo import todo_agent
# Callback for persistent memory storage
async def auto_save_session_to_memory_callback(callback_context):
await callback_context._invocation_context.memory_service.add_session_to_memory(
callback_context._invocation_context.session)
# --- ROOT AGENT DEFINITION ---
root_agent = Agent(
model='gemini-2.5-flash',
name='executive_assistant',
description='A professional AI Executive Assistant with memory and specialized tools.',
instruction='''
You are an elite, high-signal AI Executive Assistant.
Your goal is to help the user manage their knowledge, tasks, and research.
## Your Capabilities:
1. Memory: Use load_memory to recall personal facts or past context about the user.
2. Research: Delegate complex web-based investigations to the research_specialist.
3. Tasks: Delegate all to-do list management (adding, listing, or completing tasks) to the todo_specialist.
Always be direct and professional. If a task is successful, provide a brief confirmation.
''',
tools=[
PreloadMemoryTool(),
load_memory_tool,
AgentTool(todo_agent) # Exposes the Todo Specialist as a tool
],
sub_agents=[research_agent], # Exposes the Research Specialist for direct handover
after_agent_callback=auto_save_session_to_memory_callback,
)
নতুন ফিচারটি পরীক্ষা করতে adk web আবার চালান:
uv run adk web --memory_service_uri="agentengine://$ENGINE_ID"
এবং একটি করণীয় তালিকা তৈরি করার চেষ্টা করুন:
৮. ক্যালেন্ডার ব্যবস্থাপনা যোগ করুন
অবশেষে, আমরা গুগল ক্যালেন্ডারের সাথে এটি সংযুক্ত করব যাতে এজেন্ট অ্যাপয়েন্টমেন্টগুলো পরিচালনা করতে পারে। এই কোডল্যাবের জন্য, এজেন্টকে আপনার ব্যক্তিগত ক্যালেন্ডারে অ্যাক্সেস দেওয়ার পরিবর্তে—যা সঠিক উপায়ে না করা হলে সম্ভাব্য বিপজ্জনক হতে পারে—আমরা এজেন্টের পরিচালনার জন্য একটি স্বাধীন ক্যালেন্ডার তৈরি করতে যাচ্ছি।
প্রথমে, আমরা এজেন্টের পরিচয় হিসেবে কাজ করার জন্য একটি নির্দিষ্ট সার্ভিস অ্যাকাউন্ট তৈরি করব। তারপর, আমরা সেই সার্ভিস অ্যাকাউন্টটি ব্যবহার করে প্রোগ্রামগতভাবে এজেন্টের ক্যালেন্ডার তৈরি করব।
পরিষেবা অ্যাকাউন্ট সরবরাহ করুন
আপনার টার্মিনাল খুলুন এবং আইডেন্টিটি তৈরি করতে ও আপনার ব্যক্তিগত অ্যাকাউন্টকে সেটির ছদ্মবেশ ধারণ করার অনুমতি দিতে এই কমান্ডগুলো চালান:
export SA_NAME="ea-agent"
export SA_EMAIL="${SA_NAME}@${PROJECT_ID}.iam.gserviceaccount.com"
# Create the service account
gcloud iam service-accounts create $SA_NAME \
--display-name="Executive Assistant Agent"
# Allow your local user to impersonate it
gcloud iam service-accounts add-iam-policy-binding $SA_EMAIL \
--member="user:$(gcloud config get-value account)" \
--role="roles/iam.serviceAccountTokenCreator"
# Save it to the agent's environment
echo "SERVICE_ACCOUNT_EMAIL=$SA_EMAIL" >> executive_assistant/.env
প্রোগ্রাম্যাটিকভাবে ক্যালেন্ডার তৈরি করুন
সার্ভিস অ্যাকাউন্টকে ক্যালেন্ডার তৈরি করতে বলার জন্য চলুন একটি স্ক্রিপ্ট লিখি। আপনার প্রোজেক্টের রুটে ( setup_memory.py ফাইলের পাশে) setup_calendar.py নামে একটি নতুন ফাইল তৈরি করুন:
setup_calendar.py
import os
import google.auth
from googleapiclient.discovery import build
from google.auth.transport.requests import Request
from google.auth import impersonated_credentials
from dotenv import load_dotenv
load_dotenv('executive_assistant/.env')
SA_EMAIL = os.environ.get("SERVICE_ACCOUNT_EMAIL")
def setup_sa_calendar():
print(f"Authenticating to impersonate {SA_EMAIL}...")
# 1. Base credentials
creds, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"])
creds.refresh(Request())
# 2. Impersonate the Service Account
impersonated = impersonated_credentials.Credentials(
source_credentials=creds,
target_principal=SA_EMAIL,
target_scopes=["https://www.googleapis.com/auth/calendar"],
)
service = build("calendar", "v3", credentials=impersonated)
# 3. Create the calendar
print("Creating independent Service Account calendar...")
calendar = service.calendars().insert(body={
"summary": "AI Assistant (SA Owned)",
"description": "An independent calendar managed purely by the AI."
}).execute()
calendar_id = calendar['id']
# 4. Save the ID
with open("executive_assistant/.env", "a") as f:
f.write(f"\nCALENDAR_ID={calendar_id}\n")
print(f"Setup complete! CALENDAR_ID {calendar_id} added to .env")
if __name__ == "__main__":
setup_sa_calendar()
আপনার টার্মিনাল থেকে স্ক্রিপ্টটি চালান:
uv run python setup_calendar.py
ক্যালেন্ডার বিশেষজ্ঞ তৈরি করুন (calendar.py)
এবার ক্যালেন্ডার বিশেষজ্ঞের দিকে নজর দেওয়া যাক। আমরা এই এজেন্টকে ক্যালেন্ডারের সমস্ত সরঞ্জাম দিয়ে সজ্জিত করব: তালিকাভুক্ত করা, তৈরি করা, আপডেট করা, মুছে ফেলা, এবং এমনকি একটি "কুইক অ্যাড" বৈশিষ্ট্য যা স্বাভাবিক ভাষা বোঝে।
নিচের কোডটি calendar.py ফাইলে কপি করুন।
calendar.py
import os
from datetime import datetime, timedelta, timezone
import google.auth
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from google.adk.agents.llm_agent import Agent
def _get_calendar_service():
"""Build the Google Calendar API service using Service Account Impersonation."""
from google.auth.transport.requests import Request
from google.auth import impersonated_credentials
target_principal = os.environ.get("SERVICE_ACCOUNT_EMAIL")
if not target_principal:
raise ValueError("SERVICE_ACCOUNT_EMAIL environment variable is missing.")
base_scopes = ["https://www.googleapis.com/auth/cloud-platform"]
creds, _ = google.auth.default(scopes=base_scopes)
creds.refresh(Request())
target_scopes = ["https://www.googleapis.com/auth/calendar"]
impersonated = impersonated_credentials.Credentials(
source_credentials=creds,
target_principal=target_principal,
target_scopes=target_scopes,
)
return build("calendar", "v3", credentials=impersonated)
def _format_event(event: dict) -> dict:
"""Format a raw Calendar API event into a clean dict for the LLM."""
start = event.get("start", {})
end = event.get("end", {})
return {
"id": event.get("id"),
"title": event.get("summary", "(No title)"),
"start": start.get("dateTime", start.get("date")),
"end": end.get("dateTime", end.get("date")),
"location": event.get("location", ""),
"description": event.get("description", ""),
"attendees": [
{"email": a["email"], "status": a.get("responseStatus", "unknown")}
for a in event.get("attendees", [])
],
"link": event.get("htmlLink", ""),
"conference_link": (
event.get("conferenceData", {}).get("entryPoints", [{}])[0].get("uri", "")
if event.get("conferenceData")
else ""
),
"status": event.get("status", ""),
}
def list_events(days_ahead: int = 7) -> dict:
"""List upcoming calendar events."""
calendar_id = os.environ.get("CALENDAR_ID")
try:
service = _get_calendar_service()
now = datetime.now(timezone.utc).isoformat()
end = (datetime.now(timezone.utc) + timedelta(days=days_ahead)).isoformat()
events_result = service.events().list(
calendarId=calendar_id, timeMin=now, timeMax=end,
maxResults=50, singleEvents=True, orderBy="startTime"
).execute()
events = events_result.get("items", [])
if not events:
return {"status": "success", "count": 0, "events": []}
return {"status": "success", "count": len(events), "events": [_format_event(e) for e in events]}
except HttpError as e:
return {"status": "error", "message": f"Calendar API error: {e}"}
def create_event(title: str, start_time: str, end_time: str, description: str = "", location: str = "", attendees: str = "", add_google_meet: bool = False) -> dict:
"""Create a new calendar event."""
calendar_id = os.environ.get("CALENDAR_ID")
try:
service = _get_calendar_service()
event_body = {
"summary": title,
"start": {"dateTime": start_time},
"end": {"dateTime": end_time},
}
if description: event_body["description"] = description
if location: event_body["location"] = location
if attendees:
email_list = [e.strip() for e in attendees.split(",") if e.strip()]
event_body["attendees"] = [{"email": e} for e in email_list]
conference_version = 0
if add_google_meet:
event_body["conferenceData"] = {
"createRequest": {"requestId": f"event-{datetime.now().strftime('%Y%m%d%H%M%S')}", "conferenceSolutionKey": {"type": "hangoutsMeet"}}
}
conference_version = 1
event = service.events().insert(calendarId=calendar_id, body=event_body, conferenceDataVersion=conference_version).execute()
return {"status": "success", "message": f"Event created ✅", "event": _format_event(event)}
except HttpError as e:
return {"status": "error", "message": f"Calendar API error: {e}"}
def update_event(event_id: str, title: str = "", start_time: str = "", end_time: str = "", description: str = "") -> dict:
"""Update an existing calendar event."""
calendar_id = os.environ.get("CALENDAR_ID")
try:
service = _get_calendar_service()
patch_body = {}
if title: patch_body["summary"] = title
if start_time: patch_body["start"] = {"dateTime": start_time}
if end_time: patch_body["end"] = {"dateTime": end_time}
if description: patch_body["description"] = description
if not patch_body: return {"status": "error", "message": "No fields to update."}
event = service.events().patch(calendarId=calendar_id, eventId=event_id, body=patch_body).execute()
return {"status": "success", "message": "Event updated ✅", "event": _format_event(event)}
except HttpError as e:
return {"status": "error", "message": f"Calendar API error: {e}"}
def delete_event(event_id: str) -> dict:
"""Delete a calendar event by its ID."""
calendar_id = os.environ.get("CALENDAR_ID")
try:
service = _get_calendar_service()
service.events().delete(calendarId=calendar_id, eventId=event_id).execute()
return {"status": "success", "message": f"Event '{event_id}' deleted ✅"}
except HttpError as e:
return {"status": "error", "message": f"Calendar API error: {e}"}
def quick_add_event(text: str) -> dict:
"""Create an event using natural language (e.g. 'Lunch with Sarah next Monday noon')."""
calendar_id = os.environ.get("CALENDAR_ID")
try:
service = _get_calendar_service()
event = service.events().quickAdd(calendarId=calendar_id, text=text).execute()
return {"status": "success", "message": "Event created from text ✅", "event": _format_event(event)}
except HttpError as e:
return {"status": "error", "message": f"Calendar API error: {e}"}
calendar_agent = Agent(
model='gemini-2.5-flash',
name='calendar_specialist',
description='Manages the user schedule and calendar events.',
instruction='''
You manage the user's Google Calendar.
- Use list_events to check the schedule.
- Use quick_add_event for simple, conversational scheduling requests (e.g., "Lunch tomorrow at noon").
- Use create_event for complex meetings that require attendees, specific durations, or Google Meet links.
- Use update_event to change details of an existing event.
- Use delete_event to cancel or remove an event.
CRITICAL: For update_event and delete_event, you must provide the exact `event_id`.
If the user does not provide the ID, you MUST call list_events first to find the correct `event_id` before attempting the update or deletion.
Always use the current date/time context provided by the root agent to resolve relative dates like "tomorrow".
''',
tools=[list_events, create_event, update_event, delete_event, quick_add_event],
)
রুট এজেন্ট (agent.py) চূড়ান্ত করুন।
নিচের কোড দিয়ে আপনার agent.py ফাইলটি আপডেট করুন:
agent.py
import os
from datetime import datetime
from zoneinfo import ZoneInfo
from google.adk.agents.llm_agent import Agent
from google.adk.tools.agent_tool import AgentTool
from google.adk.tools.preload_memory_tool import PreloadMemoryTool
from google.adk.tools.load_memory_tool import load_memory_tool
# Import all our specialized sub-agents
from .research import research_agent
from .todo import todo_agent
from .calendar import calendar_agent
import tzlocal
# Automatically detect the local system timezone
TIMEZONE = tzlocal.get_localzone_name()
# Callback for persistent memory storage
async def auto_save_session_to_memory_callback(callback_context):
await callback_context._invocation_context.memory_service.add_session_to_memory(
callback_context._invocation_context.session)
# Callback to inject the current time into the prompt
async def setup_agent_context(callback_context, **kwargs):
now = datetime.now(ZoneInfo(TIMEZONE))
callback_context.state["current_time"] = now.strftime("%A, %Y-%m-%d %I:%M %p")
callback_context.state["timezone"] = TIMEZONE
# --- ROOT AGENT DEFINITION ---
root_agent = Agent(
model='gemini-2.5-flash',
name='executive_assistant',
description='A professional AI Executive Assistant with memory and specialized tools.',
instruction='''
You are an elite, high-signal AI Executive Assistant.
Your goal is to help the user manage their knowledge, tasks, research, and schedule.
## Your Capabilities:
1. Memory: Use load_memory to recall personal facts.
2. Research: Delegate complex web investigations to the research_specialist.
3. Tasks: Delegate all to-do list management to the todo_specialist.
4. Scheduling: Delegate all calendar queries to the calendar_specialist.
## 🕒 Current State
- Time: {current_time?}
- Timezone: {timezone?}
Always be direct and professional.
''',
tools=[
PreloadMemoryTool(),
load_memory_tool,
AgentTool(todo_agent),
AgentTool(calendar_agent)
],
sub_agents=[research_agent],
before_agent_callback=[setup_agent_context],
after_agent_callback=[auto_save_session_to_memory_callback],
)
অনুগ্রহ করে লক্ষ্য করুন যে, ক্যালেন্ডার টুলের পাশাপাশি আমরা একটি নতুন 'বিফোর এজেন্ট কলব্যাক' ফাংশনও যুক্ত করেছি: setup_agent_context । এই ফাংশনটি এজেন্টকে বর্তমান তারিখ, সময় এবং টাইমজোন সম্পর্কে অবগত করে, যাতে এটি ক্যালেন্ডারকে আরও দক্ষতার সাথে ব্যবহার করতে পারে। এটি সেশন স্টেট ভেরিয়েবল সেট করার মাধ্যমে কাজ করে, যা স্বল্পমেয়াদী স্থায়িত্বের জন্য ডিজাইন করা এক ভিন্ন ধরনের এজেন্ট মেমরি।
সম্পূর্ণ এজেন্টটি পরীক্ষা করার জন্য শেষবারের মতো adk web চালান!
uv run adk web --memory_service_uri="agentengine://$ENGINE_ID"
আপনি ডেভেলপার UI-এর স্টেট ট্যাবে সেশনের অবস্থা পরিদর্শন করতে পারেন:
এখন আপনার কাছে এমন একজন এজেন্ট আছে যে ক্যালেন্ডারের ইভেন্ট ও করণীয় কাজের তালিকা মনে রাখতে পারে, গবেষণা করতে পারে এবং যার দীর্ঘমেয়াদী স্মৃতিশক্তি রয়েছে!
ল্যাবের কাজ শেষে পরিষ্কার করা
৯. উপসংহার
অভিনন্দন! আপনি ৫টি বিবর্তনমূলক ধাপের মাধ্যমে সফলভাবে একটি বহুমুখী এআই এক্সিকিউটিভ অ্যাসিস্ট্যান্ট ডিজাইন করেছেন।
আমরা যা আলোচনা করেছি
- এআই এজেন্টদের জন্য পরিকাঠামো প্রস্তুত করা হচ্ছে।
- ADK-এর অন্তর্নির্মিত উপাদান ব্যবহার করে স্থায়ী মেমরি এবং বিশেষায়িত সাব-এজেন্ট বাস্তবায়ন করা।
- বাহ্যিক ডেটাবেস এবং প্রোডাক্টিভিটি এপিআই একীভূত করা।
পরবর্তী পদক্ষেপ
আপনি এই প্ল্যাটফর্মের অন্যান্য কোডল্যাবগুলো অন্বেষণ করে, অথবা নিজের মতো করে এক্সিকিউটিভ অ্যাসিস্ট্যান্টটির উন্নতি সাধন করে আপনার শেখার যাত্রা চালিয়ে যেতে পারেন।
উন্নতির জন্য কিছু ধারণার প্রয়োজন হলে আপনি চেষ্টা করে দেখতে পারেন:
- দীর্ঘ কথোপকথনের পারফরম্যান্স অপ্টিমাইজ করতে ইভেন্ট কম্প্যাকশন প্রয়োগ করুন।
- এজেন্টকে আপনার হয়ে নোট নিতে এবং ফাইল হিসেবে সংরক্ষণ করতে দেওয়ার জন্য একটি আর্টিফ্যাক্ট সার্ভিস যোগ করুন।
- Google Cloud Run ব্যবহার করে আপনার এজেন্টকে একটি ব্যাকএন্ড পরিষেবা হিসেবে স্থাপন করুন।
টেস্টিং শেষ হয়ে গেলে, পরিবেশটি পরিষ্কার করতে ভুলবেন না, যাতে আপনার বিলিং অ্যাকাউন্টে কোনো অপ্রত্যাশিত চার্জ যুক্ত না হয়।
কোডিং উপভোগ করুন!