CrewAI, LangGraph, A2A, এবং ADK ব্যবহার করে এজেন্টদের পরিধি বৃদ্ধি করুন

১. ভূমিকা

এই কোডল্যাবে, আপনি CrewAI , LangGraph , A2A প্রোটোকল এবং ADK (এজেন্ট ডেভেলপমেন্ট কিট) ব্যবহার করে একটি মাল্টি-এজেন্ট অর্কেস্ট্রেশন সিস্টেম তৈরি করতে শিখবেন। আপনি এমন একটি সিস্টেম তৈরি করবেন যেখানে একটি ADK কন্ট্রোল রুম একজন LangGraph প্ল্যানারকে পরিকল্পনার দায়িত্ব অর্পণ করে, যিনি একটি CrewAI এক্সিকিউশন ক্রুকে কাজ পাঠান – এই সবাই A2A-এর মাধ্যমে সংযুক্ত থেকে একটি রিটেইল ইনভেন্টরি রিস্টক পরিস্থিতি সামাল দেবেন।

মাল্টি-এজেন্ট অর্কেস্ট্রেশন বলতে কী বোঝায়?

একটি মাল্টি-এজেন্ট সিস্টেমে , একাধিক বিশেষায়িত এআই এজেন্ট এমন সব কাজ সম্পন্ন করতে সহযোগিতা করে যা একটি একক এজেন্টের পক্ষে করা অত্যন্ত জটিল। একটি একক এজেন্ট সবকিছু করার পরিবর্তে, সমস্যাটিকে বিভিন্ন ভূমিকায় বিভক্ত করা হয় – যেমন একজন পরিকল্পনাকারী এবং একজন নির্বাহক – এবং প্রত্যেকের নিজস্ব সরঞ্জাম ও দক্ষতা থাকে।

এটি মানব সংগঠনগুলোর কার্যপ্রণালীরই প্রতিচ্ছবি: একজন ব্যবস্থাপক বিশ্লেষকদের কাছে কৌশল এবং বিশেষজ্ঞদের কাছে তার বাস্তবায়নের দায়িত্ব অর্পণ করেন। এর সুবিধাগুলোর মধ্যে রয়েছে:

• দায়িত্বের পৃথকীকরণ : প্রতিটি প্রতিনিধি সে যা সবচেয়ে ভালোভাবে করতে পারে, সেদিকেই মনোযোগ দেয়।
• ফ্রেমওয়ার্কের নমনীয়তা : প্রতিটি ভূমিকার জন্য সেরা ফ্রেমওয়ার্কটি ব্যবহার করুন (পরিকল্পনা যুক্তির জন্য LangGraph, টুল সম্পাদনের জন্য CrewAI)।
• পরিমাপযোগ্যতা : সম্পূর্ণ সিস্টেম পরিবর্তন না করেই বিশেষায়িত এজেন্ট যোগ করুন।

দৃশ্যপট

যখন কোনো ব্যবহারকারী "টোকিও অফিসের জন্য ১টি পিক্সেল ৭ ফোন রিস্টক করুন" এর মতো একটি রিস্টক অনুরোধ পাঠান, তখন সিস্টেমটি:

1. ল্যাংগ্রাফ প্ল্যানার অনুরোধটি বিশ্লেষণ করে আইটেম এবং পরিমাণ বের করে।
2. প্ল্যানার কার্য সম্পাদনের দায়িত্ব CrewAI এক্সিকিউশন ক্রু- কে অর্পণ করেন।
3. একজন সোর্সিং স্পেশালিস্ট এজেন্ট টুল ব্যবহার করে প্রোডাক্ট ক্যাটালগ অনুসন্ধান করেন।
4. একজন ক্রয় কর্মকর্তা প্রতিনিধি বাজেট যাচাই করেন এবং সরঞ্জাম ব্যবহার করে ক্রয় আদেশ দেন।
5. ফলাফলটি প্ল্যানারের কাছে ফিরে যায়, যিনি একটি চূড়ান্ত প্রতিবেদন তৈরি করেন।

User Request
    │
    ▼
┌──────────────────────┐
│ ADK Control Room     │  ← Top-level orchestrator, re-plans on failure
│ (BaseAgent)          │
└──────────┬───────────┘
           │ A2A (JSON-RPC)
           ▼
┌──────────────────────┐
│ LangGraph Planner    │  ← Analyzes intent, delegates, reports
│ (State Machine)      │
└──────────┬───────────┘
           │
           ▼
┌──────────────────────┐
│ CrewAI Execution Crew│  ← Runs agents with tools
│  ├─ Sourcing Agent   │  → search_products
│  └─ Procurement Agent│  → check_budget, create_purchase_order
└──────────────────────┘

টেক স্ট্যাক

স্তর	প্রযুক্তি	ভূমিকা
পরিকল্পনা	ল্যাংগ্রাফ	স্টেট মেশিন যা অভিপ্রায় বিশ্লেষণ করে, অনুরোধ রাউট করে এবং রিপোর্ট তৈরি করে।
মৃত্যুদণ্ড	ক্রুএআই	ভূমিকা-ভিত্তিক এজেন্ট যারা ক্রমানুসারে টুলগুলিকে কল করে
এলএলএম	ভার্টেক্স এআই-তে জেমিনি	পাওয়ার্স এজেন্ট যুক্তি এবং সরঞ্জাম নির্বাচন
আন্তঃ-এজেন্ট যোগাযোগ	A2A প্রোটোকল	JSON-RPC 2.0 ব্রিজ, যার মাধ্যমে বিভিন্ন ফ্রেমওয়ার্কের এজেন্টরা একে অপরের সাথে যোগাযোগ করতে পারে।
শীর্ষ-স্তরের অর্কেস্ট্রেটর	ADK (বেসএজেন্ট)	অনুরোধ গ্রহণ করে, A2A-এর মাধ্যমে দায়িত্ব অর্পণ করে, ব্যর্থ হলে পুনর্পরিকল্পনা করে।

বাস্তবে দেখুন: যদি উপলব্ধ থাকে, তাহলে https://scale-control-room-761793285222.us-central1.run.app- এ সম্পূর্ণ প্রোডাকশন সিস্টেমটি ব্যবহার করে দেখুন – এটি একটি রিয়েল-টাইম ড্যাশবোর্ড, A2A প্রোটোকল এবং IAM সিকিউরিটির মাধ্যমে এখানে আপনার তৈরি করা বিষয়টিকে আরও প্রসারিত করে।

আপনি যা করবেন

• এজেন্টদের ব্যবহারের জন্য নিজস্ব টুল নির্ধারণ করুন।
• CrewAI দিয়ে বিশেষায়িত এজেন্ট তৈরি করুন।
• LangGraph ব্যবহার করে একটি স্টেট মেশিন প্ল্যানার তৈরি করুন।
• পরিকল্পনাকারী এবং বাস্তবায়নকারী দলের মধ্যে কার্যপ্রবাহ সমন্বয় করুন।
• বিভিন্ন ফ্রেমওয়ার্কের মধ্যে যোগাযোগের জন্য প্ল্যানারটিকে একটি A2A প্রোটোকল সার্ভারের আওতায় আনুন।
• একটি শীর্ষ-স্তরের ADK কন্ট্রোল রুম তৈরি করুন যা A2A-এর মাধ্যমে দায়িত্ব অর্পণ করে এবং ব্যর্থ হলে পুনর্পরিকল্পনা করে।

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

• ক্রোমের মতো একটি ওয়েব ব্রাউজার
• বিলিং সক্ষম একটি গুগল ক্লাউড প্রজেক্ট

এই কোডল্যাবটি সেইসব মধ্যবর্তী স্তরের ডেভেলপারদের জন্য, যারা পাইথন এবং এলএলএম-এর প্রাথমিক ধারণাগুলোর সাথে পরিচিত।

আনুমানিক সময়কাল: ৩৫ মিনিট ।

আনুমানিক খরচ: এই কোডল্যাবে তৈরি রিসোর্সগুলোর খরচ $1-এর কম হওয়া উচিত।

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

একটি গুগল ক্লাউড প্রজেক্ট তৈরি করুন

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

ক্লাউড শেল শুরু করুন

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

1. Google Cloud কনসোলের শীর্ষে থাকা Activate Cloud Shell-এ ক্লিক করুন।
2. ক্লাউড শেলে সংযুক্ত হওয়ার পর, আপনার প্রমাণীকরণ যাচাই করুন:

   gcloud auth list
   

3. আপনার প্রজেক্টটি কনফিগার করা হয়েছে কিনা তা নিশ্চিত করুন:

   gcloud config get project
   

4. আপনার প্রজেক্টটি প্রত্যাশা অনুযায়ী সেট করা না থাকলে, এটি সেট করুন:

   export PROJECT_ID=<YOUR_PROJECT_ID>
   gcloud config set project $PROJECT_ID
   

এপিআই সক্ষম করুন

Vertex AI API সক্রিয় করতে এই কমান্ডটি চালান:

gcloud services enable aiplatform.googleapis.com

দ্রষ্টব্য: ক্লাউড শেল স্বয়ংক্রিয়ভাবে আপনার গুগল ক্লাউড অ্যাকাউন্টের সাথে প্রমাণীকরণ করে। আপনি যদি এই কোডল্যাবটি ক্লাউড শেলের বাইরে চালান, তাহলে ভার্টেক্স এআই-এর সাথে প্রমাণীকরণের জন্য আপনাকে gcloud auth application-default login চালাতে হবে।

আপনার পরিবেশ তৈরি করুন

ক্লাউড শেলে, আপনার প্রোজেক্টের জন্য একটি নতুন ডিরেক্টরি তৈরি করুন এবং সেটির ভেতরে যান:

mkdir scale-agents
cd scale-agents

uv ইনস্টল করুন এবং এটি ব্যবহার করে প্রয়োজনীয় প্যাকেজগুলো ইনস্টল করুন:

curl -LsSf https://astral.sh/uv/install.sh | sh
export PATH="$HOME/.local/bin:$PATH"
uv init --no-workspace --no-readme
rm main.py
sed -i 's/requires-python = ">=3.12"/requires-python = ">=3.12,<3.14"/' pyproject.toml
uv add crewai 'litellm[google]' langgraph 'a2a-sdk>=0.3.25,<0.4' httpx uvicorn 'google-adk>=1.0.0' --prerelease=allow

Vertex AI-এর জন্য এনভায়রনমেন্ট ভেরিয়েবল সেট করুন:

export GOOGLE_CLOUD_PROJECT=$(gcloud config get-value project)
export GOOGLE_CLOUD_LOCATION=us-central1
export GOOGLE_GENAI_USE_VERTEXAI=TRUE

৩. টুল এবং এজেন্ট সংজ্ঞায়িত করুন

একটি মাল্টি-এজেন্ট সিস্টেমে, এজেন্টদের জগতের সাথে মিথস্ক্রিয়া করার জন্য সরঞ্জাম এবং কী করতে হবে তা জানার জন্য নির্দিষ্ট ভূমিকার প্রয়োজন হয়।

scale_agents.py নামে একটি ফাইল তৈরি করুন এবং নিম্নলিখিত কোডটি যোগ করুন। এটি ইম্পোর্ট, মক টুল এবং CrewAI এজেন্টদের সেট আপ করে।

import os
from typing import Optional
from crewai import Agent, Crew, Process, Task, LLM
from crewai.tools import tool
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict

# CrewAI requires this env var even when using Vertex AI
os.environ["OPENAI_API_KEY"] = "NA"

# Set the project ID for Vertex AI
os.environ["VERTEXAI_PROJECT"] = os.getenv("GOOGLE_CLOUD_PROJECT", "")
os.environ["VERTEXAI_LOCATION"] = "us-central1"

# Initialize the LLM to use Vertex AI
llm = LLM(
    model="vertex_ai/gemini-2.5-flash",
    temperature=0.0,
    max_tokens=4096,
)

# --- Step 1: Define Tools ---

@tool("search_products")
def search_products(query: str) -> list:
    """Search for products in the catalog."""
    # Mock product catalog
    products = [
        {"product_id": "pixel-7", "name": "Pixel 7 Phone", "price": 50.0},
        {"product_id": "pixel-8", "name": "Pixel 8 Phone", "price": 80.0},
    ]
    return [p for p in products if query.lower() in p["name"].lower()]

@tool("check_budget")
def check_budget(amount: float) -> dict:
    """Check if a purchase amount is within the budget."""
    limit = 100.0
    if amount <= limit:
        return {"approved": True, "remaining": limit - amount}
    return {"approved": False, "reason": f"Exceeds budget of ${limit}"}

@tool("create_purchase_order")
def create_purchase_order(product_id: str, quantity: int) -> dict:
    """Create a purchase order for a product."""
    return {
        "status": "SUCCESS",
        "po_id": f"PO-{product_id}-{quantity}",
        "message": f"Successfully ordered {quantity} units of {product_id}."
    }

# --- Step 2: Define Agents ---

sourcing_agent = Agent(
    role="Sourcing Specialist",
    goal="Find the best available products that match the intent of the request. You MUST use the search_products tool to look up products -- never make up product data.",
    backstory="You are a veteran procurement specialist with an eye for detail. You always search the catalog before recommending a product.",
    tools=[search_products],
    llm=llm,
    verbose=True,
    allow_delegation=False,
    memory=False,
    reasoning=False,
)

procurement_agent = Agent(
    role="Procurement Officer",
    goal="Validate the purchase against budget constraints and execute the order. You MUST use the check_budget tool before ordering and the create_purchase_order tool to place the order.",
    backstory="You are the gatekeeper of the budget. You always verify budget before placing any order.",
    tools=[check_budget, create_purchase_order],
    llm=llm,
    verbose=True,
    allow_delegation=False,
    memory=False,
    reasoning=False,
)

মূল ধারণা

• @tool ডেকোরেটর : CrewAI এটি ব্যবহার করে সাধারণ পাইথন ফাংশনগুলোকে এমন টুলে পরিণত করে যা LLM বুঝতে ও কল করতে পারে। ফাংশনের টাইপ হিন্টস এবং ডকস্ট্রিং ব্যবহার করে একটি টুল স্কিমা তৈরি করা হয় যা LLM বুঝতে পারে।
• ভূমিকা, লক্ষ্য এবং পটভূমি : এগুলো এজেন্টের ব্যক্তিত্বকে সংজ্ঞায়িত করে এবং তার এলএলএম যুক্তিকে পরিচালিত করে। পটভূমিটি কেবল কথার কথা নয় – "আপনি সবসময় ক্যাটালগ অনুসন্ধান করেন" এই কথাটি এজেন্টকে অলীক উত্তরের পরিবর্তে তার সরঞ্জামগুলো ব্যবহার করতে উৎসাহিত করে।
• reasoning=False : বর্ধিত যুক্তি প্রক্রিয়া নিষ্ক্রিয় করে, ফলে এজেন্ট সরাসরি উত্তর দেওয়ার চেষ্টা না করে সাধারণ টুল-কলিং লুপ অনুসরণ করে।
• allow_delegation=False : অন্য এজেন্টদের কাছে কাজ না পাঠিয়ে প্রতিটি এজেন্টকে তার নিজের জন্য নির্ধারিত টুলের উপর মনোনিবেশ করতে সাহায্য করে।

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

৪. কাজ এবং কর্মীদল নির্ধারণ করুন

এখন টাস্ক তৈরি করে এবং সেগুলোকে একটি ক্রু-এর সাথে সংযুক্ত করার মাধ্যমে এই এজেন্টদের কী করতে হবে তা নির্ধারণ করা যাক।

একই scale_agents.py ফাইলের শেষে নিম্নলিখিত কোডটি যুক্ত করুন:

# --- Step 3: Define Tasks & Crew ---

sourcing_task = Task(
    description="Use the search_products tool to find products matching: '{item_description}'. Return the product_id and price of the best match from the tool results.",
    expected_output="The product_id and price of the best matching product from the search_products tool.",
    agent=sourcing_agent
)

procurement_task = Task(
    description="First, use the check_budget tool to verify the total cost for {quantity} units. Then use the create_purchase_order tool with the product_id and quantity to place the order.",
    expected_output="The purchase order details returned by the create_purchase_order tool.",
    agent=procurement_agent,
    context=[sourcing_task] # This task depends on the output of sourcing_task
)

def run_crew(item_description: str, quantity: int):
    crew = Crew(
        agents=[sourcing_agent, procurement_agent],
        tasks=[sourcing_task, procurement_task],
        process=Process.sequential, # Run tasks in order
        verbose=True,
        memory=False,
        planning=False,
    )
    
    result = crew.kickoff(inputs={
        "item_description": item_description,
        "quantity": quantity
    })
    return result

মূল ধারণা

• টাস্ক কনটেক্সট : context=[sourcing_task] CrewAI-কে জানায় যে প্রকিউরমেন্ট টাস্কটি এগিয়ে যাওয়ার জন্য সোর্সিং টাস্কের আউটপুট প্রয়োজন। প্রকিউরমেন্ট অফিসার কী অর্ডার করবেন সে বিষয়ে সিদ্ধান্ত নেওয়ার আগে সোর্সিং স্পেশালিস্ট কী খুঁজে পেয়েছেন তা দেখতে পারেন।
• প্রক্রিয়া অনুক্রমিক : কাজগুলো তালিকাভুক্ত ক্রমানুসারে সম্পাদিত হয়। এটি গুরুত্বপূর্ণ কারণ সংগ্রহ কাজটি উৎস নির্ধারণের কাজের ফলাফলের উপর নির্ভরশীল – কোন পণ্যটি কিনতে হবে তা জানার আগে আপনি অর্ডার দিতে পারবেন না।
• memory=False / planning=False : এই ডেমোর জন্য এক্সিকিউশনকে সহজ এবং অনুমানযোগ্য রাখতে CrewAI-এর বিল্ট-ইন মেমরি এবং প্ল্যানিং ফিচারগুলো নিষ্ক্রিয় করে।

৫. ল্যাংগ্রাফ প্ল্যানার তৈরি করুন

বাস্তবায়নকারী দল ‘কীভাবে’ কাজটি করা হবে তা সামলায় – যেমন পণ্য খোঁজা, বাজেট যাচাই করা, অর্ডার দেওয়া। কিন্তু ‘কী’ করা হবে, তা কে ঠিক করে? সেই কাজটিই করে প্ল্যানিং এজেন্ট , যা ল্যাংগ্রাফ (LangGraph) দিয়ে তৈরি।

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

একই scale_agents.py ফাইলের শেষে নিম্নলিখিত কোডটি যুক্ত করুন:

# --- Step 4: Define LangGraph Planner ---

class PlanState(TypedDict):
    objective: str
    item_description: Optional[str]
    quantity_needed: Optional[int]
    execution_result: Optional[str]
    final_report: Optional[str]

def analyze_alert(state: PlanState) -> PlanState:
    """Node 1: Extract intent from the raw objective string."""
    print("--- ANALYZING ALERT ---")
    # In a production app, you would use an LLM here to extract details.
    # For simplicity, we simulate extraction here.
    objective = state["objective"]
    
    # Hardcoded extraction for the demo
    if "Pixel 7" in objective:
        return {
            "item_description": "Pixel 7",
            "quantity_needed": 1,
        }
    return {
        "item_description": "unknown",
        "quantity_needed": 0,
    }

def delegate_to_executor(state: PlanState) -> PlanState:
    """Node 2: Call the CrewAI Execution Crew."""
    print("--- DELEGATING TO CREW ---")
    if state["item_description"] == "unknown":
        return {"execution_result": "Failed: Unknown item"}
        
    result = run_crew(
        item_description=state["item_description"],
        quantity=state["quantity_needed"]
    )
    return {"execution_result": str(result)}

def generate_report(state: PlanState) -> PlanState:
    """Node 3: Synthesize the final outcome."""
    print("--- GENERATING REPORT ---")
    return {
        "final_report": f"Objective handled: {state['objective']}. Result: {state['execution_result']}"
    }

# Build the graph
workflow = StateGraph(PlanState)
workflow.add_node("analyze_alert", analyze_alert)
workflow.add_node("delegate", delegate_to_executor)
workflow.add_node("generate_report", generate_report)

workflow.set_entry_point("analyze_alert")
workflow.add_edge("analyze_alert", "delegate")
workflow.add_edge("delegate", "generate_report")
workflow.add_edge("generate_report", END)

app = workflow.compile()

# --- Main Execution ---
if __name__ == "__main__":
    print("Starting Multi-Agent System...")
    
    initial_state = {
        "objective": "Restock 1 Pixel 7 phones for the Tokyo office"
    }
    
    final_state = app.invoke(initial_state)
    
    print("\n=== FINAL REPORT ===")
    print(final_state["final_report"])

মূল ধারণা

• স্টেটগ্রাফ : স্টেট মেশিনকে সংজ্ঞায়িত করে। PlanState হলো সেই টাইপড স্টেট যা প্রতিটি নোড অনুরোধটি প্রক্রিয়া করার সাথে সাথে জমা হয়।
• নোড : যে ফাংশনগুলো বর্তমান অবস্থা গ্রহণ করে এবং সেটির আপডেট ফেরত দেয়। প্রতিটি নোডের একটিমাত্র দায়িত্ব থাকে – analyze_alert অভিপ্রায় বের করে, delegate_to_executor দলটিকে পরিচালনা করে, generate_report ফলাফলের সারসংক্ষেপ করে।
• এজ (Edges) : নোডগুলোর মধ্যে প্রবাহ নির্ধারণ করে। এই কোডল্যাবে আমরা একটি সরল রৈখিক প্রবাহ ( analyze → delegate → report ) ব্যবহার করি। সম্পূর্ণ ওয়ার্কশপটিতে শর্তসাপেক্ষ রাউটিংয়ের মাধ্যমে এটিকে আরও প্রসারিত করা হয় – যেমন, ধ্বংসাত্মক অনুরোধগুলোকে এক্সিকিউটরের পরিবর্তে একটি নিরাপত্তা পাথে পাঠানো।

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

৬. পরিকল্পনাকারী ও দলকে পরিচালনা করুন

এখন চলুন LangGraph প্ল্যানার এবং CrewAI ক্রু-কে একসাথে পরীক্ষা করে দেখি।

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

uv run python scale_agents.py

গৃহীত পদক্ষেপগুলো নির্দেশ করে এমন আউটপুট আপনি দেখতে পাবেন:

1. সতর্কতা বিশ্লেষণ করা হচ্ছে : LangGraph নোডটি চলছে।
2. ক্রু-কে দায়িত্ব অর্পণ : LangGraph নোডটি CrewAI ক্রু-কে আহ্বান করে।
3. CrewAI-এর কার্যপ্রক্রিয়া : আপনি দেখবেন সোর্সিং স্পেশালিস্ট পণ্যটি খুঁজছেন এবং প্রকিউরমেন্ট অফিসার বাজেট যাচাই করে পারচেজ অর্ডার তৈরি করছেন।
4. চূড়ান্ত প্রতিবেদন : সারসংক্ষেপিত ফলাফল শেষে মুদ্রণ করা হবে।

উদাহরণ আউটপুট (সংক্ষিপ্ত):

Starting Multi-Agent System...
--- ANALYZING ALERT ---
--- DELEGATING TO CREW ---

  Agent: Sourcing Specialist
  Tool: search_products  Args: {'query': 'Pixel 7'}
  Tool Completed  Output: [{'product_id': 'pixel-7', 'name': 'Pixel 7 Phone', 'price': 50.0}]

  Agent: Procurement Officer
  Tool: check_budget  Args: {'amount': 50}
  Tool: create_purchase_order  Args: {'product_id': 'pixel-7', 'quantity': 1}
  Tool Completed  Output: {'status': 'SUCCESS', 'po_id': 'PO-pixel-7-1', ...}

--- GENERATING REPORT ---

=== FINAL REPORT ===
Objective handled: Restock 1 Pixel 7 phones for the Tokyo office. Result: ...PO-pixel-7-1...SUCCESS...

দ্রষ্টব্য: আপনি আউটপুটে [CrewAIEventsBus] Warning: Event pairing mismatch বার্তা দেখতে পারেন। এগুলি CrewAI-এর অভ্যন্তরীণ ইভেন্ট ট্র্যাকিং থেকে আসা বাহ্যিক সতর্কতা এবং এগুলিকে নিরাপদে উপেক্ষা করা যেতে পারে।

দ্রষ্টব্য: CrewAI ট্রেসিং নিষ্ক্রিয় থাকার বিষয়ে একটি বার্তা প্রদর্শন করতে পারে। এটি শুধুমাত্র তথ্যমূলক এবং নিরাপদে উপেক্ষা করা যেতে পারে।

দ্রষ্টব্য: এই মক ওএমএস-এর বাজেট সীমা $100 । হ্যাপি পাথ সফল হওয়ার জন্য পরিমাণ কম রাখুন (প্রায় ২ ইউনিটের কম)। উদাহরণস্বরূপ, $50 মূল্যের ১টি পিক্সেল ৭ ফোন বাজেট চেক পাস করবে, কিন্তু $150 মূল্যের ৩টি ইউনিট "বাজেট অতিক্রম করেছে" বলে বাতিল হয়ে যাবে।

৭. প্ল্যানারটি একটি A2A সার্ভারে মুড়ে দিন।

ল্যাংগ্রাফ প্ল্যানারটি কাজ করে, কিন্তু এটি একটি পাইথন প্রসেসের মধ্যে আটকে থাকে। এটিকে অন্যান্য এজেন্টদের দ্বারা কলযোগ্য করে তোলার জন্য—যেগুলো হয়তো ভিন্ন ফ্রেমওয়ার্কে লেখা বা ভিন্ন মেশিনে চলছে—আমরা এটিকে একটি A2A (এজেন্ট-টু-এজেন্ট) সার্ভারের মধ্যে রাখি।

A2A হলো একটি JSON-RPC 2.0 ভিত্তিক প্রোটোকল যা এজেন্টদের যোগাযোগের পদ্ধতিকে প্রমিত করে। মূল ধারণাগুলো হলো:

ধারণা	উদ্দেশ্য
এজেন্ট কার্ড	এজেন্টের সক্ষমতা বর্ণনা করে এমন JSON মেটাডেটা (যা /.well-known/agent-card.json ঠিকানায় পরিবেশিত হয়)
message/send	এজেন্টের কাছে টাস্ক পাঠানোর JSON-RPC পদ্ধতি
কাজ	অবস্থাসহ একটি কাজের একক (জমা দেওয়া → চলমান → সম্পন্ন/ব্যর্থ)
প্রত্নবস্তু	একটি কাজের সাথে সংযুক্ত অন্তর্বর্তী এবং চূড়ান্ত আউটপুট

a2a_planner.py নামে একটি নতুন ফাইল তৈরি করুন:

import asyncio
import os
import uvicorn

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.apps import A2AStarletteApplication
from a2a.server.events import EventQueue
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
from a2a.types import (AgentCapabilities, AgentCard, AgentSkill,
                       InternalError, Part, TextPart)
from a2a.utils import new_task
from a2a.utils.errors import ServerError

# Import the LangGraph planner from Step 4
from scale_agents import app as planner_app


class PlannerAgentExecutor(AgentExecutor):
    """Wraps the LangGraph planner as an A2A service."""

    SUPPORTED_CONTENT_TYPES = ["text", "text/plain"]

    async def execute(self, context: RequestContext, event_queue: EventQueue):
        objective = context.get_user_input()

        # Initialize A2A task tracking
        task = context.current_task or new_task(context.message)
        await event_queue.enqueue_event(task)
        updater = TaskUpdater(event_queue, task.id, task.context_id)

        try:
            # Run the LangGraph planner synchronously in a thread
            initial_state = {"objective": objective}
            result = await asyncio.to_thread(planner_app.invoke, initial_state)
            final_report = result.get("final_report", "No report generated.")
        except Exception as e:
            final_report = f"Execution failed: {e}"

        # Send the result back as an artifact
        await updater.add_artifact(
            [Part(root=TextPart(text=final_report))],
            name="orchestration_report"
        )
        await updater.complete()

    async def cancel(self, context, event_queue):
        raise ServerError(error=InternalError(message="Cancel not supported"))


# Define the Agent Card — this is what other agents see
port = int(os.environ.get("PORT", 8080))
agent_card = AgentCard(
    name="Retail-Planner-A2A",
    description="LangGraph planner that delegates logistics tasks to a CrewAI crew.",
    url=f"http://localhost:{port}/",
    version="1.0.0",
    default_input_modes=PlannerAgentExecutor.SUPPORTED_CONTENT_TYPES,
    default_output_modes=PlannerAgentExecutor.SUPPORTED_CONTENT_TYPES,
    capabilities=AgentCapabilities(streaming=False),
    skills=[
        AgentSkill(
            id="plan_logistics",
            name="Plan Logistics",
            description="Analyzes inventory alerts and orchestrates procurement.",
            tags=["logistics", "planning"],
            examples=["Restock 1 Pixel 7 phones for the Tokyo office"],
        )
    ],
)

if __name__ == "__main__":
    executor = PlannerAgentExecutor()
    handler = DefaultRequestHandler(
        agent_executor=executor, task_store=InMemoryTaskStore()
    )
    server = A2AStarletteApplication(
        agent_card=agent_card, http_handler=handler
    )
    print(f"Starting A2A Planner Server on port {port}...")
    uvicorn.run(server.build(), host="0.0.0.0", port=port)

মূল ধারণা

• এজেন্ট কার্ড : এটি /.well-known/agent-card.json এ পরিবেশিত হয় – যেকোনো এজেন্ট এই URL-টি ফেচ করে জানতে পারেন এই সার্ভারটি কী কাজ করে। এতে এজেন্টের দক্ষতা, সমর্থিত কন্টেন্ট টাইপ এবং সক্ষমতার তালিকা থাকে।
• AgentExecutor.execute() : এটিই একমাত্র মেথড যা আপনাকে ইমপ্লিমেন্ট করতে হবে। এটি আগত রিকোয়েস্ট গ্রহণ করে, আপনার এজেন্ট লজিক (এখানে, ল্যাংগ্রাফ প্ল্যানার) চালায় এবং ফলাফল আর্টিফ্যাক্ট হিসেবে ফেরত পাঠায়।
• TaskUpdater : টাস্কের জীবনচক্র পরিচালনা করে – add_artifact() অন্তর্বর্তী/চূড়ান্ত আউটপুট পাঠায়, complete() টাস্কটিকে সম্পন্ন হিসেবে চিহ্নিত করে। A2A লাইব্রেরিটি সমস্ত JSON-RPC অভ্যন্তরীণ কার্যাবলী পরিচালনা করে।

টার্মিনালে A2A সার্ভারটি চালু করে পরীক্ষা করুন:

uv run python a2a_planner.py

আরেকটি ক্লাউড শেল ট্যাব খুলুন (বর্তমান ট্যাবের পাশে থাকা + চিহ্নে ক্লিক করুন) এবং এজেন্ট কার্ডটি পরিবেশিত হয়েছে কিনা তা যাচাই করুন:

cd ~/scale-agents
curl http://localhost:8080/.well-known/agent-card.json | python3 -m json.tool

আপনি এজেন্ট কার্ডের JSON দেখতে পাবেন। পরবর্তী ধাপের জন্য প্রথম টার্মিনালে A2A সার্ভারটি চালু রাখুন।

৮. এডিকে কন্ট্রোল রুম তৈরি করুন।

স্ট্যাকের শীর্ষে রয়েছে কন্ট্রোল রুম , যা ADK (গুগলের এজেন্ট ডেভেলপমেন্ট কিট) দিয়ে তৈরি। এটি ব্যবহারকারীর অনুরোধ গ্রহণ করে, A2A-এর মাধ্যমে প্ল্যানারের কাছে দায়িত্ব অর্পণ করে, ফলাফল মূল্যায়ন করে এবং – সবচেয়ে গুরুত্বপূর্ণভাবে – ব্যর্থতার ক্ষেত্রে পুনঃপরিকল্পনার কাজটি পরিচালনা করে (CUJ 2)।

ADK, BaseAgent , LlmAgent , এবং InMemoryRunner মতো এজেন্ট প্রিমিটিভ সরবরাহ করে। আমরা কাস্টম অর্কেস্ট্রেশন লজিক—যেমন A2A কল, রিপোর্ট ক্লাসিফিকেশন, এবং একটি LlmAgent সাব-এজেন্টের সাহায্যে ডাইনামিক রি-প্ল্যানিং—লেখার জন্য BaseAgent সাবক্লাস করি।

control_room.py নামে একটি নতুন ফাইল তৈরি করুন:

import asyncio
import uuid
import os
import httpx

from google.adk.agents import BaseAgent, LlmAgent, RunConfig
from google.adk.agents.invocation_context import InvocationContext
from google.adk.events import Event
from google.adk.runners import InMemoryRunner
from google.genai import types
from typing import AsyncGenerator

A2A_SERVER_URL = os.environ.get("PLANNER_AGENT_URL", "http://127.0.0.1:8080")


def _classify_report(report: str) -> tuple[bool, bool]:
    """Return (is_success, should_retry) for a planner report."""
    normalized = (report or "").replace("*", "").strip().lower()

    success_markers = [
        "status: success", "'status': 'success'",
        "outcome: success", "po_id", "successfully ordered",
    ]
    retryable_markers = ["not found", "discontinued", "no inventory",
                         "unknown item"]
    terminal_markers = [
        "status: failed", "over budget", "not issued",
        "procurement failed",
    ]

    if any(m in normalized for m in terminal_markers):
        return False, False      # Failed, don't retry
    if any(m in normalized for m in retryable_markers):
        return False, True       # Failed, but retryable
    if any(m in normalized for m in success_markers):
        return True, False       # Success!
    return False, False          # Unknown → treat as failure


class ControlRoomAgent(BaseAgent):
    """Top-level orchestrator: delegates via A2A, re-plans on failure."""

    async def _run_async_impl(
        self, ctx: InvocationContext
    ) -> AsyncGenerator[Event, None]:
        # Extract user input from session events
        user_msg = ""
        if ctx.session and ctx.session.events:
            for evt in reversed(ctx.session.events):
                if evt.content and evt.content.role == "user":
                    user_msg = evt.content.parts[0].text
                    break

        max_attempts = 2
        current_objective = user_msg
        final_report = "No report returned."

        for attempt in range(1, max_attempts + 1):
            print(f"\n--- Attempt {attempt}: Calling A2A Planner ---")
            print(f"    Objective: {current_objective}")

            # Build the A2A JSON-RPC request
            payload = {
                "jsonrpc": "2.0",
                "id": f"req-{attempt}",
                "method": "message/send",
                "params": {
                    "message": {
                        "message_id": str(uuid.uuid4()),
                        "parts": [{"text": current_objective}],
                        "role": "user"
                    }
                }
            }

            try:
                async with httpx.AsyncClient(timeout=300.0) as client:
                    resp = await client.post(
                        f"{A2A_SERVER_URL}/", json=payload
                    )
                    data = resp.json()
                    if "error" in data:
                        final_report = data["error"].get("message", "Unknown A2A error")
                    elif "result" in data:
                        artifacts = data["result"].get("artifacts", [])
                        if artifacts and "parts" in artifacts[-1]:
                            parts = artifacts[-1]["parts"]
                            if parts and "text" in parts[0]:
                                final_report = parts[0]["text"]
            except Exception as e:
                final_report = f"Connection error: {e}"

            print(f"\n--- Report ---\n{final_report}\n")
            is_success, should_retry = _classify_report(final_report)

            if is_success:
                yield Event(
                    author=self.name,
                    invocation_id=ctx.invocation_id,
                    content=types.Content(
                        role="model",
                        parts=[types.Part.from_text(text=final_report)]
                    ),
                )
                return

            # --- Re-planning (CUJ 2) ---
            if should_retry and attempt < max_attempts:
                print("--- Re-planning with LLM ---")
                replanner = LlmAgent(
                    name=f"replanner_{attempt}",
                    model="gemini-2.5-flash",
                    instruction=(
                        "You are a strategic re-planner. The previous request "
                        "failed. Rewrite the objective to be broader or more "
                        "likely to succeed. Output ONLY the new objective text."
                    ),
                )

                # Run the re-planner as a sub-agent
                child_ctx = InvocationContext(
                    invocation_id=f"{ctx.invocation_id}_replan_{attempt}",
                    agent=replanner,
                    session=ctx.session,
                    session_service=ctx.session_service,
                    run_config=ctx.run_config or RunConfig(),
                )
                child_ctx.user_content = types.Content(
                    role="user",
                    parts=[types.Part.from_text(text=(
                        f"Original Objective: {current_objective}\n"
                        f"Failure Reason: {final_report}\n"
                        "Please broaden the search."
                    ))]
                )

                new_objective = ""
                async for event in replanner.run_async(child_ctx):
                    if event.content and event.content.parts:
                        for part in event.content.parts:
                            if hasattr(part, "text") and part.text:
                                new_objective += part.text

                current_objective = new_objective.strip()
                print(f"New objective: {current_objective}")
                continue

            # Terminal failure (not retryable)
            if not should_retry:
                yield Event(
                    author=self.name,
                    invocation_id=ctx.invocation_id,
                    content=types.Content(
                        role="model",
                        parts=[types.Part.from_text(text=f"FAILED: {final_report}")]
                    ),
                )
                return

        # Max attempts exhausted
        yield Event(
            author=self.name,
            invocation_id=ctx.invocation_id,
            content=types.Content(
                role="model",
                parts=[types.Part.from_text(
                    text=f"FAILED after {max_attempts} attempts: {final_report}"
                )]
            ),
        )


async def main():
    prompt = "Restock 1 Pixel 7 phones for the Tokyo office"
    print(f"Starting Control Room with: {prompt}\n")

    agent = ControlRoomAgent(name="control_room")
    runner = InMemoryRunner(app_name="control_room", agent=agent)
    session = await runner.session_service.create_session(
        app_name="control_room", user_id="admin"
    )
    content = types.Content(
        role="user", parts=[types.Part.from_text(text=prompt)]
    )

    async for event in runner.run_async(
        user_id="admin", session_id=session.id, new_message=content
    ):
        if event.content and event.content.parts:
            for part in event.content.parts:
                if hasattr(part, "text") and part.text:
                    print(f"Result: {part.text}")

    print("\n=== CONTROL ROOM COMPLETE ===")


if __name__ == "__main__":
    asyncio.run(main())

মূল ধারণা

• BaseAgent : কাস্টম এজেন্টের জন্য ADK প্রিমিটিভ। আপনি এটিকে সাবক্লাস করে এবং _run_async_impl ওভাররাইড করে যেকোনো অ্যাসিঙ্ক অর্কেস্ট্রেশন লজিক লিখতে পারেন – এখানে, A2A কল + ক্লাসিফাই + রি-প্ল্যান লুপ।
• A2A JSON-RPC কল : কন্ট্রোল রুম httpx ব্যবহার করে প্ল্যানারের A2A সার্ভারে একটি স্ট্যান্ডার্ড message/send রিকোয়েস্ট পাঠায় এবং চূড়ান্ত রিপোর্টটি বের করার জন্য JSON-RPC রেসপন্সটি পার্স করে।
• _classify_report() : একটি সাধারণ কীওয়ার্ড-ভিত্তিক শ্রেণিবিন্যাস যা রিপোর্টের টেক্সট থেকে সাফল্য, পুনরায় চেষ্টার যোগ্য ব্যর্থতা, অথবা চূড়ান্ত ব্যর্থতা নির্ধারণ করে। এটি পুনঃপরিকল্পনা চক্রকে চালনা করে।
• সাব-এজেন্ট আহ্বান : পুনরায় পরিকল্পনা করার জন্য, কন্ট্রোল রুম একটি LlmAgent তৈরি করে এবং একটি চাইল্ড InvocationContext কনস্ট্রাক্ট করে ও replanner.run_async(child_ctx) কল করার মাধ্যমে এটিকে রান করে। এটি আপনাকে কাস্টম অর্কেস্ট্রেশন লজিকের ভিতরে ডায়নামিকভাবে LLM এজেন্ট চালু করার সুযোগ দেয়।
• InMemoryRunner : এটি একটি ইন-মেমরি সেশন স্টোর ব্যবহার করে এজেন্টকে স্থানীয়ভাবে চালায়। প্রোডাকশনে, Vertex AI Agent Engine-এ ডেপ্লয় করার জন্য আপনি adk deploy ব্যবহার করবেন।

৯. ফুল স্ট্যাক চালান

এখন চলুন সম্পূর্ণ তিন-স্তরীয় সিস্টেমটি পরীক্ষা করে দেখি: ADK কন্ট্রোল রুম → A2A → LangGraph প্ল্যানার → CrewAI Crew।

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

cd ~/scale-agents
export GOOGLE_CLOUD_PROJECT=$(gcloud config get-value project)
export GOOGLE_CLOUD_LOCATION=us-central1
export GOOGLE_GENAI_USE_VERTEXAI=TRUE
uv run python control_room.py

আপনার সম্পূর্ণ অর্কেস্ট্রেশন প্রবাহটি দেখা উচিত:

1. কন্ট্রোল রুম অনুরোধটি গ্রহণ করে এবং A2A সার্ভারে একটি message/send JSON-RPC কল পাঠায়।
2. A2A সার্ভার অনুরোধটি গ্রহণ করে এবং LangGraph প্ল্যানারকে আহ্বান করে।
3. LangGraph প্ল্যানার অনুরোধটি বিশ্লেষণ করে CrewAI ক্রুদের কাছে দায়িত্ব অর্পণ করে।
4. CrewAI ক্রু সোর্সিং এবং প্রকিউরমেন্ট এজেন্টদের পরিচালনা করে।
5. ফলাফলটি সরাসরি কন্ট্রোল রুমে ফিরে যায়।

গুরুত্বপূর্ণ ব্যবহারকারী যাত্রা (CUJs)

এই সিনারিওগুলো নিয়ে পরীক্ষা করার জন্য control_room.py তে prompt স্ট্রিংটি পরিবর্তন করে দেখুন:

সিইউজে	প্রম্পট	কী ঘটে
১. শুভ পথ	Restock 1 Pixel 7 phones for the Tokyo office	অনুসন্ধান -> বাজেট যাচাই -> ক্রয় আদেশ (সফল)। শুরু থেকে শেষ পর্যন্ত কাজ করে।
২. পুনর্পরিকল্পনা	Order 1 unit of the discontinued XR-7000 Quantum Holographic Display	প্ল্যানারটি "ব্যর্থ: অজানা আইটেম" বার্তাটি ফেরত দেয়। কন্ট্রোল রুম এটি শনাক্ত করে এবং অনুসন্ধানের পরিধি বাড়ানোর জন্য একটি LlmAgent রি-প্ল্যানার চালু করে। উভয় চেষ্টাই ব্যর্থ হয় (হার্ডকোডেড প্ল্যানারটি শুধুমাত্র "পিক্সেল ৭" শনাক্ত করতে পারে), কিন্তু আপনি সম্পূর্ণ রি-প্ল্যানিং প্রক্রিয়াটি কার্যকর হতে দেখবেন।

CUJ 2 (পুনঃপরিকল্পনা) পরীক্ষা করার জন্য, control_room.py এর prompt পরিবর্তন করে নিচের মতো করুন:

prompt = "Order 1 unit of the discontinued XR-7000 Quantum Holographic Display"

হার্ডকোডেড প্ল্যানারটি এই আইটেমটি চিনতে পারবে না এবং "ব্যর্থ: অজানা আইটেম" ফেরত দেবে। কন্ট্রোল রুম এই ব্যর্থতা শনাক্ত করবে, ডায়নামিকভাবে একটি LlmAgent রি-প্ল্যানার তৈরি করবে এবং আরও বড় একটি উদ্দেশ্য নিয়ে পুনরায় চেষ্টা করবে। যেহেতু প্ল্যানারটি শুধুমাত্র "পিক্সেল ৭" চিনতে পারে, তাই এই পুনঃচেষ্টাও ব্যর্থ হবে – কিন্তু আপনি সম্পূর্ণ রি-প্ল্যানিং প্রক্রিয়াটি কার্যকর হতে দেখবেন। FAILED after 2 attempts: ...

১০. পরিষ্কার করুন

আপনার গুগল ক্লাউড অ্যাকাউন্টে চলমান চার্জ এড়াতে, আপনি এই কোডল্যাবের সময় তৈরি করা রিসোর্সগুলি মুছে ফেলতে পারেন। আপনি কেবল আপনার তৈরি করা ডিরেক্টরিটি সরিয়ে ফেলতে পারেন:

cd ..
rm -rf scale-agents

১১. অভিনন্দন

অভিনন্দন! আপনি CrewAI, LangGraph, A2A Protocol, এবং ADK ব্যবহার করে সফলভাবে একটি মাল্টি-এজেন্ট অর্কেস্ট্রেশন সিস্টেম তৈরি করেছেন।

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

• CrewAI-এর @tool ডেকোরেটর ব্যবহার করে এজেন্টদের জন্য টুল কীভাবে নির্ধারণ করবেন।
• স্বতন্ত্র ভূমিকা, সরঞ্জাম এবং লক্ষ্য সহ বিশেষায়িত এজেন্ট কীভাবে তৈরি করা যায়।
• কীভাবে এজেন্টদেরকে কার্যনির্ভরশীলতাসহ একটি ধারাবাহিক দলে সংযুক্ত করা যায়।
• LangGraph ব্যবহার করে কীভাবে একটি স্টেট মেশিন প্ল্যানার তৈরি করা যায় যা কর্মীদের মধ্যে কাজ ভাগ করে দেয়।
• AgentCard এবং AgentExecutor ব্যবহার করে প্ল্যানারকে কীভাবে একটি A2A পরিষেবা হিসাবে প্রকাশ করা যায়।
• কিভাবে একটি কাস্টম ADK BaseAgent তৈরি করা যায় যা A2A-এর মাধ্যমে দায়িত্ব অর্পণ করে এবং ব্যর্থ হলে একটি LlmAgent সাব-এজেন্টকে আহ্বান করে পুনরায় পরিকল্পনা করে।
• কেন বিভিন্ন ফ্রেমওয়ার্কে পরিকল্পনা, বাস্তবায়ন এবং সমন্বয়কে পৃথক রাখলে আপনি মডুলারিটি এবং স্থিতিস্থাপকতা লাভ করেন।

আরও এগিয়ে যাওয়া

সম্পূর্ণ কর্মশালাটি এই সিস্টেমটিকে নিম্নলিখিত বিষয়গুলির মাধ্যমে আরও প্রসারিত করে:

• রিয়েল-টাইম ড্যাশবোর্ড – একাধিক এজেন্টের অগ্রগতি দেখার জন্য SSE স্ট্রিমিং
• আইডেন্টিটি শিল্ড – এটি একটি IAM-ভিত্তিক নিরাপত্তা ব্যবস্থা যা অবকাঠামোগত স্তরে ক্ষতিকর কার্যকলাপ প্রতিরোধ করে, প্রম্পট স্তরে নয়।
• ভার্টেক্স এআই এজেন্ট ইঞ্জিন – adk deploy ব্যবহার করে পরিচালিত ক্লাউড অবকাঠামোতে ADK এজেন্ট স্থাপন করুন

রেফারেন্স নথি

• ক্রুএআই ডকুমেন্টেশন
• ল্যাংগ্রাফ ডকুমেন্টেশন
• ভার্টেক্স এআই ডকুমেন্টেশন
• A2A প্রোটোকল
• ADK ডকুমেন্টেশন