1. परिचय
एजेंट-टू-एजेंट (A2A) प्रोटोकॉल को एआई एजेंट के बीच कम्यूनिकेशन को स्टैंडर्ड बनाने के लिए डिज़ाइन किया गया है. खास तौर पर, उन एजेंट के लिए जिन्हें बाहरी सिस्टम में डिप्लॉय किया जाता है. इससे पहले, इस तरह के प्रोटोकॉल टूल के लिए बनाए गए थे. इन्हें मॉडल कॉन्टेक्स्ट प्रोटोकॉल (एमसीपी) कहा जाता है. यह एलएलएम को डेटा और संसाधनों से कनेक्ट करने का एक नया स्टैंडर्ड है. A2A, एमसीपी के साथ मिलकर काम करता है. A2A का फ़ोकस किसी दूसरी समस्या पर होता है, जबकि एमसीपी का फ़ोकस एजेंट को टूल और डेटा से कनेक्ट करने की जटिलता को कम करने पर होता है. वहीं, A2A का फ़ोकस इस बात पर होता है कि एजेंट को उनकी स्वाभाविक क्षमताओं के साथ मिलकर काम करने के लिए कैसे तैयार किया जाए. इस सुविधा की मदद से, एजेंट टूल के तौर पर काम करने के बजाय एजेंट (या उपयोगकर्ता) के तौर पर बातचीत कर सकते हैं. उदाहरण के लिए, कोई सामान ऑर्डर करते समय, दोनों पक्षों के बीच बातचीत करने की सुविधा चालू करना.
A2A को एमसीपी के साथ काम करने के लिए बनाया गया है. आधिकारिक दस्तावेज़ में यह सुझाव दिया गया है कि ऐप्लिकेशन, टूल के लिए एमसीपी और एजेंट के लिए A2A का इस्तेमाल करें. एजेंट को AgentCard के तौर पर दिखाया जाता है ( हम इसके बारे में बाद में बात करेंगे ). इसके बाद, फ़्रेमवर्क अपने उपयोगकर्ता, रिमोट एजेंट, और अन्य एजेंट के साथ कम्यूनिकेट करने के लिए A2A का इस्तेमाल कर सकते हैं.
इस डेमो में, हम Python SDK का इस्तेमाल करके, A2A को लागू करने का तरीका बताएंगे. हम ऐसे इस्तेमाल के उदाहरण पर विचार करेंगे जिसमें हमारे पास निजी खरीदारी की कंसीयर्ज सेवा हो. इससे हमें बर्गर और पिज़्ज़ा बेचने वाले एजेंटों से संपर्क करने में मदद मिल सकती है, ताकि हम अपने ऑर्डर को मैनेज कर सकें.
A2A, क्लाइंट-सर्वर के सिद्धांत का इस्तेमाल करता है. यहां A2A का सामान्य फ़्लो दिया गया है, जो हमें इस डेमो में दिखेगा
- A2A क्लाइंट, सबसे पहले ऐक्सेस किए जा सकने वाले सभी A2A सर्वर एजेंट कार्ड की खोज करेगा. इसके बाद, कनेक्शन क्लाइंट बनाने के लिए, इसकी जानकारी का इस्तेमाल करेगा
- ज़रूरत पड़ने पर, A2A क्लाइंट, A2A सर्वर को एक मैसेज भेजेगा. सर्वर इस मैसेज को पूरा किए जाने वाले टास्क के तौर पर देखेगा. अगर A2A क्लाइंट पर पुश नोटिफ़िकेशन पाने वाले यूआरएल को कॉन्फ़िगर किया गया है और A2A सर्वर इसका इस्तेमाल कर सकता है, तो सर्वर, क्लाइंट पर सूचना पाने वाले एंडपॉइंट को टास्क की प्रोग्रेस की स्थिति भी पब्लिश कर पाएगा
- टास्क पूरा होने के बाद, A2A सर्वर, A2A क्लाइंट को जवाब वाला आर्टफ़ैक्ट भेजेगा
कोडलैब के ज़रिए, आपको यहां दिया गया तरीका अपनाना होगा:
- अपना Google Cloud प्रोजेक्ट तैयार करें और उस पर सभी ज़रूरी एपीआई चालू करें
- कोडिंग एनवायरमेंट के लिए वर्कस्पेस सेट अप करना
- बर्गर और पिज़्ज़ा एजेंट सेवाओं के लिए, एनवायरमेंट वैरिएबल तैयार करें और उन्हें स्थानीय तौर पर आज़माएं
- बर्गर और पिज़्ज़ा एजेंट को Cloud Run पर डिप्लॉय करें
- A2A सर्वर कैसे सेट अप किया गया, इस बारे में जानकारी देखना
- कॉन्शिएर्ज खरीदने के लिए, एनवायरमेंट वैरिएबल तैयार करें और उन्हें स्थानीय तौर पर आज़माएं
- Agent Engine में खरीदारी से जुड़ी सलाह देने वाले एजेंट को डिप्लॉय करना
- लोकल इंटरफ़ेस के ज़रिए एजेंट इंजन से कनेक्ट करना
- A2A क्लाइंट को कैसे सेट अप किया गया और उसकी डेटा मॉडलिंग के बारे में जानकारी देखें
- A2A क्लाइंट और सर्वर के बीच पेलोड और इंटरैक्शन की जांच करना
आर्किटेक्चर की खास जानकारी
हम इस सेवा के लिए, यहां दिया गया आर्किटेक्चर डिप्लॉय करेंगे
हम दो ऐसी सेवाएं डिप्लॉय करेंगे जो A2A सर्वर के तौर पर काम करेंगी. पहली सेवा, बर्गर एजेंट ( CrewAI एजेंट फ़्रेमवर्क पर आधारित) और दूसरी सेवा, पिज़्ज़ा एजेंट ( Langgraph एजेंट फ़्रेमवर्क पर आधारित). उपयोगकर्ता सिर्फ़ खरीदारी करने वाले कंसीयर्ज से सीधे तौर पर इंटरैक्ट करेगा. इसे एजेंट डेवलपमेंट किट (एडीके) फ़्रेमवर्क का इस्तेमाल करके चलाया जाएगा. यह A2A क्लाइंट के तौर पर काम करेगा.
इनमें से हर एजेंट का अपना एनवायरमेंट होगा और वे खुद ही डिप्लॉय होंगे.
ज़रूरी शर्तें
- Python का इस्तेमाल करने में सहज
- एचटीटीपी सेवा का इस्तेमाल करके, बुनियादी फ़ुल-स्टैक आर्किटेक्चर के बारे में जानकारी
आपको क्या सीखने को मिलेगा
- A2A सर्वर का मुख्य स्ट्रक्चर
- A2A क्लाइंट का मुख्य स्ट्रक्चर
- एजेंट सेवा को Cloud Run पर डिप्लॉय करना
- एजेंट इंजन में एजेंट सेवा को डिप्लॉय करना
- A2A क्लाइंट, A2A सर्वर से कैसे कनेक्ट होता है
- बिना स्ट्रीमिंग वाले कनेक्शन पर अनुरोध और जवाब का स्ट्रक्चर
आपको इन चीज़ों की ज़रूरत होगी
- Chrome वेब ब्राउज़र
- Gmail खाता
- ऐसा Cloud प्रोजेक्ट जिसमें बिलिंग की सुविधा चालू हो
यह कोडलैब, सभी लेवल के डेवलपर के लिए बनाया गया है. इसमें शुरुआती डेवलपर भी शामिल हैं. इसमें सैंपल ऐप्लिकेशन में Python का इस्तेमाल किया गया है. हालांकि, यहां दिए गए कॉन्सेप्ट को समझने के लिए, Python के बारे में जानकारी होना ज़रूरी नहीं है.
2. शुरू करने से पहले
Cloud Console में चालू प्रोजेक्ट चुनना
इस कोडलैब में यह माना गया है कि आपके पास पहले से ही बिलिंग की सुविधा वाला Google Cloud प्रोजेक्ट है. अगर आपके पास अब तक यह सुविधा नहीं है, तो इसे इस्तेमाल करने के लिए नीचे दिए गए निर्देशों का पालन करें.
- Google Cloud Console में, प्रोजेक्ट चुनने वाले पेज पर जाकर, Google Cloud प्रोजेक्ट चुनें या बनाएं.
- पक्का करें कि आपके Cloud प्रोजेक्ट के लिए बिलिंग चालू हो. किसी प्रोजेक्ट के लिए बिलिंग चालू है या नहीं, यह देखने का तरीका जानें.
Cloud Shell टर्मिनल में क्लाउड प्रोजेक्ट सेट अप करना
- आपको Cloud Shell का इस्तेमाल करना होगा. यह Google Cloud में चलने वाला कमांड-लाइन एनवायरमेंट है. इसमें bq पहले से लोड होता है. Google Cloud Console में सबसे ऊपर मौजूद, Cloud Shell चालू करें पर क्लिक करें. अगर आपसे अनुमति देने के लिए कहा जाता है, तो अनुमति दें पर क्लिक करें
- Cloud Shell से कनेक्ट होने के बाद, यह देखने के लिए कि आपकी पुष्टि हो चुकी है और प्रोजेक्ट को आपके प्रोजेक्ट आईडी पर सेट किया गया है, इस कमांड का इस्तेमाल करें:
gcloud auth list
- यह पुष्टि करने के लिए कि gcloud कमांड को आपके प्रोजेक्ट के बारे में पता है, Cloud Shell में यह कमांड चलाएं.
gcloud config list project
- अगर आपका प्रोजेक्ट सेट नहीं है, तो इसे सेट करने के लिए इस निर्देश का इस्तेमाल करें:
gcloud config set project <YOUR_PROJECT_ID>
इसके अलावा, PROJECT_ID आईडी को कंसोल में भी देखा जा सकता है
इस पर क्लिक करने से, आपको अपने सभी प्रोजेक्ट और उनके आईडी दाईं ओर दिखेंगे
- नीचे दिए गए निर्देश का इस्तेमाल करके, ज़रूरी एपीआई चालू करें. इसमें कुछ मिनट लग सकते हैं. इसलिए, कृपया इंतज़ार करें.
gcloud services enable aiplatform.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
cloudresourcemanager.googleapis.com
कमांड के सही तरीके से काम करने पर, आपको यहां दिखाए गए मैसेज जैसा मैसेज दिखेगा:
Operation "operations/..." finished successfully.
gcloud कमांड के बजाय, कंसोल का इस्तेमाल करके भी ऐसा किया जा सकता है. इसके लिए, हर प्रॉडक्ट को खोजें या इस लिंक का इस्तेमाल करें.
अगर कोई एपीआई छूट जाता है, तो उसे लागू करने के दौरान कभी भी चालू किया जा सकता है.
gcloud कमांड और उनके इस्तेमाल के बारे में जानने के लिए, दस्तावेज़ देखें.
Cloud Shell Editor पर जाएं और ऐप्लिकेशन की वर्किंग डायरेक्ट्री सेट अप करें
अब हम कोडिंग से जुड़े कुछ काम करने के लिए, कोड एडिटर सेट अप कर सकते हैं. इसके लिए, हम Cloud Shell Editor का इस्तेमाल करेंगे
- 'एडिटर खोलें' बटन पर क्लिक करें. इससे Cloud Shell Editor खुल जाएगा. यहां हम अपना कोड लिख सकते हैं
- पक्का करें कि Cloud Code प्रोजेक्ट, Cloud Shell एडिटर के सबसे नीचे बाईं ओर (स्टेटस बार) में सेट हो. जैसा कि नीचे दी गई इमेज में हाइलाइट किया गया है. साथ ही, यह उस चालू Google Cloud प्रोजेक्ट पर सेट हो जिसके लिए बिलिंग की सुविधा चालू है. अगर कहा जाए, तो अनुमति दें पर क्लिक करें. अगर आपने पहले से ही पिछले निर्देश का पालन किया है, तो हो सकता है कि बटन, साइन इन बटन के बजाय सीधे तौर पर आपके चालू किए गए प्रोजेक्ट पर ले जाए
- इसके बाद, कोड सीखने की इस लैब के लिए, टेंप्लेट की वर्किंग डायरेक्ट्री को Github से क्लोन करें. इसके लिए, यहां दिया गया निर्देश चलाएं. इससे purchasing-concierge-a2a डायरेक्ट्री में वर्किंग डायरेक्ट्री बन जाएगी
git clone https://github.com/alphinside/purchasing-concierge-intro-a2a-codelab-starter.git purchasing-concierge-a2a
- इसके बाद, Cloud Shell Editor के सबसे ऊपर वाले सेक्शन पर जाएं और File->Open Folder पर क्लिक करें. इसके बाद, अपनी username डायरेक्ट्री ढूंढें और फिर purchasing-concierge-a2a डायरेक्ट्री ढूंढें. इसके बाद, OK बटन पर क्लिक करें. इससे चुनी गई डायरेक्ट्री, मुख्य वर्किंग डायरेक्ट्री बन जाएगी. इस उदाहरण में, उपयोगकर्ता का नाम alvinprayuda है. इसलिए, डायरेक्ट्री का पाथ यहां दिखाया गया है
अब आपका Cloud Shell Editor ऐसा दिखना चाहिए
एनवायरमेंट सेटअप करना
अगला चरण, डेवलपमेंट एनवायरमेंट तैयार करना है. आपका मौजूदा चालू टर्मिनल, purchasing-concierge-a2a वर्किंग डायरेक्ट्री में होना चाहिए. इस कोडलैब में, हम Python 3.12 का इस्तेमाल करेंगे. साथ ही, Python के वर्शन और वर्चुअल एनवायरमेंट को बनाने और मैनेज करने की ज़रूरत को आसान बनाने के लिए, हम uv python project manager का इस्तेमाल करेंगे
- अगर आपने अब तक टर्मिनल नहीं खोला है, तो टर्मिनल -> नया टर्मिनल पर क्लिक करके इसे खोलें. इसके अलावा, Ctrl + Shift + C का इस्तेमाल करें. इससे ब्राउज़र के सबसे नीचे एक टर्मिनल विंडो खुलेगी
- अब
uvका इस्तेमाल करके, खरीदारी में मदद करने वाले वर्चुअल एनवायरमेंट को शुरू करें. यह क्लाउड टर्मिनल पर पहले से इंस्टॉल होता है. इस कमांड को चलाएं
uv sync --frozen
इससे .venv डायरेक्ट्री बन जाएगी और डिपेंडेंसी इंस्टॉल हो जाएंगी. pyproject.toml की झलक से, आपको इस तरह दिखाई गई डिपेंडेंसी के बारे में जानकारी मिलेगी
dependencies = [
"a2a-sdk>=0.2.16",
"google-adk>=1.8.0",
"gradio>=5.38.2",
]
- वर्चुअल एनवायरमेंट की जांच करने के लिए, नई फ़ाइल main.py बनाएं और इस कोड को कॉपी करें
def main():
print("Hello from purchasing-concierge-a2a!")
if __name__ == "__main__":
main()
- इसके बाद, यह कमांड चलाएं
uv run main.py
आपको नीचे दिखाए गए जैसा आउटपुट मिलेगा
Using CPython 3.12 Creating virtual environment at: .venv Hello from purchasing-concierge-a2a!
इससे पता चलता है कि Python प्रोजेक्ट को सही तरीके से सेट अप किया जा रहा है.
अब हम अगले चरण पर जा सकते हैं. इसमें रिमोट सेलर एजेंट को कॉन्फ़िगर और डिप्लॉय करना है
3. A2A सर्वर से Cloud Run पर रिमोट सेलर एजेंट को डिप्लॉय करना
इस चरण में, हम लाल बॉक्स से मार्क किए गए इन दो रिमोट सेलर एजेंट को डिप्लॉय करेंगे. बर्गर एजेंट, CrewAI एजेंट फ़्रेमवर्क के साथ काम करेगा. वहीं, पिज़्ज़ा एजेंट, Langgraph एजेंट के साथ काम करेगा. ये दोनों ही Gemini Flash 2.0 मॉडल के साथ काम करते हैं
रिमोट बर्गर एजेंट को डिप्लॉय करना
बर्गर एजेंट का सोर्स कोड, remote_seller_agents/burger_agent डायरेक्ट्री में मौजूद है. agent.py स्क्रिप्ट पर, एजेंट के शुरू होने की प्रोसेस की जांच की जा सकती है. यहां एजेंट के शुरू किए गए कोड का स्निपेट दिया गया है
from crewai import Agent, Crew, LLM, Task, Process
from crewai.tools import tool
...
model = LLM(
model="vertex_ai/gemini-2.5-flash-lite", # Use base model name without provider prefix
)
burger_agent = Agent(
role="Burger Seller Agent",
goal=(
"Help user to understand what is available on burger menu and price also handle order creation."
),
backstory=("You are an expert and helpful burger seller agent."),
verbose=False,
allow_delegation=False,
tools=[create_burger_order],
llm=model,
)
agent_task = Task(
description=self.TaskInstruction,
agent=burger_agent,
expected_output="Response to the user in friendly and helpful manner",
)
crew = Crew(
tasks=[agent_task],
agents=[burger_agent],
verbose=False,
process=Process.sequential,
)
inputs = {"user_prompt": query, "session_id": sessionId}
response = crew.kickoff(inputs)
return response
...
remote_seller_agents/burger_agent डायरेक्ट्री में मौजूद सभी फ़ाइलें, Cloud Run पर हमारे एजेंट को डिप्लॉय करने के लिए पहले से ही काफ़ी हैं. इससे एजेंट को सेवा के तौर पर ऐक्सेस किया जा सकेगा. हम इस बारे में बाद में बात करेंगे. इसे डिप्लॉय करने के लिए, यह कमांड चलाएं
gcloud run deploy burger-agent \
--source remote_seller_agents/burger_agent \
--port=8080 \
--allow-unauthenticated \
--min 1 \
--region us-central1 \
--update-env-vars GOOGLE_CLOUD_LOCATION=us-central1 \
--update-env-vars GOOGLE_CLOUD_PROJECT={your-project-id}
अगर आपको यह सूचना मिलती है कि सोर्स से डिप्लॉय करने के लिए, कंटेनर रिपॉज़िटरी बनाई जाएगी, तो Y पर क्लिक करें. ऐसा सिर्फ़ तब होता है, जब आपने पहले कभी सोर्स से Cloud Run पर डिप्लॉय न किया हो. सफल डिप्लॉयमेंट के बाद, यह इस तरह का लॉग दिखाएगा.
Service [burger-agent] revision [burger-agent-xxxxx-xxx] has been deployed and is serving 100 percent of traffic. Service URL: https://burger-agent-xxxxxxxxx.us-central1.run.app
सेवा को डिप्लॉय करने पर, यहां मौजूद xxxx हिस्सा एक यूनीक आइडेंटिफ़ायर होगा. अब, ब्राउज़र के ज़रिए बर्गर एजेंट की सेवाएं डिप्लॉय करने के https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json तरीके को आज़माएं. यह डिप्लॉय किए गए A2A सर्वर एजेंट कार्ड को ऐक्सेस करने का यूआरएल है.
अगर इसे सही तरीके से डिप्लॉय किया गया है, तो https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json को ऐक्सेस करते समय आपको अपने ब्राउज़र में इस तरह का जवाब दिखेगा :
यह बर्गर एजेंट कार्ड की जानकारी है, जिसे खोज के लिए ऐक्सेस किया जाना चाहिए. हम इस बारे में बाद में बात करेंगे. ध्यान दें कि यहां url की वैल्यू अब भी http://0.0.0.0:8080/ पर सेट है. यह url वैल्यू, A2A क्लाइंट के लिए मुख्य जानकारी होनी चाहिए, ताकि वह बाहरी दुनिया से मैसेज भेज सके. हालांकि, इसे सही तरीके से कॉन्फ़िगर नहीं किया गया है. इस डेमो के लिए, हमें इस वैल्यू को बर्गर एजेंट सेवा के यूआरएल पर अपडेट करना होगा. इसके लिए, हमें एक और एनवायरमेंट वैरिएबल HOST_OVERRIDE जोड़ना होगा.
एनवायरमेंट वैरिएबल के ज़रिए, एजेंट कार्ड पर बर्गर एजेंट के यूआरएल की वैल्यू अपडेट करना
बर्गर एजेंट सेवा में HOST_OVERRIDE जोड़ने के लिए, यह तरीका अपनाएं
- अपने Cloud Console के सबसे ऊपर मौजूद खोज बार में Cloud Run खोजें
- पहले डिप्लॉय की गई burger-agent Cloud Run सेवा पर क्लिक करें
- बर्गर-सर्विस यूआरएल कॉपी करें. इसके बाद, बदलाव करें और नया वर्शन डिप्लॉय करें पर क्लिक करें
- इसके बाद, वैरिएबल और सीक्रेट सेक्शन पर क्लिक करें
- इसके बाद, वैरिएबल जोड़ें पर क्लिक करें और
HOST_OVERRIDEकी वैल्यू को सेवा के यूआरएल (https://burger-agent-xxxxxxxxx.us-central1.run.appपैटर्न वाला यूआरएल ) पर सेट करें
- आखिर में, अपनी सेवा को फिर से डिप्लॉय करने के लिए, डिप्लॉय करें बटन पर क्लिक करें
अब जब ब्राउज़र https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json में burger-agent एजेंट कार्ड को फिर से ऐक्सेस किया जाएगा, तब url वैल्यू पहले से ही सही तरीके से कॉन्फ़िगर की गई होगी
रिमोट पिज़्ज़ा एजेंट को डिप्लॉय करना
इसी तरह, पिज़्ज़ा एजेंट का सोर्स कोड remote_seller_agents/pizza_agent डायरेक्ट्री में मौजूद है. agent.py स्क्रिप्ट पर, एजेंट के शुरू होने की प्रोसेस की जांच की जा सकती है. यहां एजेंट के शुरू किए गए कोड का स्निपेट दिया गया है
from langchain_google_vertexai import ChatVertexAI
from langgraph.prebuilt import create_react_agent
...
self.model = ChatVertexAI(
model="gemini-2.5-flash-lite",
location=os.getenv("GOOGLE_CLOUD_LOCATION"),
project=os.getenv("GOOGLE_CLOUD_PROJECT"),
)
self.tools = [create_pizza_order]
self.graph = create_react_agent(
self.model,
tools=self.tools,
checkpointer=memory,
prompt=self.SYSTEM_INSTRUCTION,
)
...
बर्गर-एजेंट को डिप्लॉय करने के पिछले चरण की तरह ही, remote_seller_agents/pizza_agent डायरेक्ट्री में मौजूद सभी फ़ाइलें, हमारे एजेंट को Cloud Run पर डिप्लॉय करने के लिए पहले से ही काफ़ी हैं. इससे, एजेंट को सेवा के तौर पर ऐक्सेस किया जा सकेगा. इसे डिप्लॉय करने के लिए, यह कमांड चलाएं
gcloud run deploy pizza-agent \
--source remote_seller_agents/pizza_agent \
--port=8080 \
--allow-unauthenticated \
--min 1 \
--region us-central1 \
--update-env-vars GOOGLE_CLOUD_LOCATION=us-central1 \
--update-env-vars GOOGLE_CLOUD_PROJECT={your-project-id}
सफल डिप्लॉयमेंट के बाद, यह इस तरह का लॉग दिखाएगा.
Service [pizza-agent] revision [pizza-agent-xxxxx-xxx] has been deployed and is serving 100 percent of traffic. Service URL: https://pizza-agent-xxxxxxxxx.us-central1.run.app
सेवा को डिप्लॉय करने पर, यहां मौजूद xxxx हिस्सा एक यूनीक आइडेंटिफ़ायर होगा. बर्गर एजेंट के साथ भी ऐसा ही होता है. जब ब्राउज़र के ज़रिए, पिज़्ज़ा एजेंट की उन सेवाओं के https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json रूट पर जाने की कोशिश की जाती है जिन्हें डिप्लॉय किया गया है, ताकि A2A सर्वर एजेंट कार्ड को ऐक्सेस किया जा सके, तब पिज़्ज़ा एजेंट के एजेंट कार्ड पर url वैल्यू को अब तक ठीक से कॉन्फ़िगर नहीं किया गया होता है. हमें HOST_OVERRIDE को इसके एनवायरमेंट वैरिएबल में भी जोड़ना होगा
एनवायरमेंट वैरिएबल के ज़रिए, एजेंट कार्ड पर पिज़्ज़ा एजेंट के यूआरएल की वैल्यू अपडेट करना
पिज़्ज़ा एजेंट सेवा में HOST_OVERRIDE जोड़ने के लिए, यह तरीका अपनाएं
- अपने Cloud Console के सबसे ऊपर मौजूद खोज बार में Cloud Run खोजें
- पहले डिप्लॉय की गई pizza-agent Cloud Run सेवा पर क्लिक करें
- बदलाव करें और नई रीविज़न डिप्लॉय करें पर क्लिक करें
- पिज़्ज़ा-सर्विस यूआरएल को कॉपी करें. इसके बाद, वैरिएबल और सीक्रेट सेक्शन पर क्लिक करें
- इसके बाद, वैरिएबल जोड़ें पर क्लिक करें और
HOST_OVERRIDEकी वैल्यू को सेवा के यूआरएल (https://pizza-agent-xxxxxxxxx.us-central1.run.appपैटर्न वाला यूआरएल ) पर सेट करें
- आखिर में, अपनी सेवा को फिर से डिप्लॉय करने के लिए, डिप्लॉय करें बटन पर क्लिक करें
अब जब ब्राउज़र https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json में pizza-agent एजेंट कार्ड को फिर से ऐक्सेस किया जाएगा, तब url वैल्यू पहले से ही सही तरीके से कॉन्फ़िगर की गई होगी
इस समय, हम Cloud Run पर बर्गर और पिज़्ज़ा, दोनों सेवाओं को पहले ही डिप्लॉय कर चुके हैं. अब A2A सर्वर के मुख्य कॉम्पोनेंट के बारे में बात करते हैं
4. A2A सर्वर के मुख्य कॉम्पोनेंट
अब A2A सर्वर के मुख्य सिद्धांत और कॉम्पोनेंट के बारे में बात करते हैं
एजेंट कार्ड
हर A2A सर्वर के पास एक एजेंट कार्ड होना चाहिए, जिसे /.well-known/agent.json रिसॉर्स पर ऐक्सेस किया जा सकता हो. ऐसा A2A क्लाइंट पर डिस्कवरी फ़ेज़ को सपोर्ट करने के लिए किया जाता है. इससे एजेंट को ऐक्सेस करने और उसकी सभी क्षमताओं के बारे में पूरी जानकारी और कॉन्टेक्स्ट मिलना चाहिए. यह Swagger या Postman का इस्तेमाल करके, एपीआई के दस्तावेज़ों को अच्छी तरह से समझने जैसा है.
यह हमारे डिप्लॉय किए गए बर्गर एजेंट के एजेंट कार्ड का कॉन्टेंट है
{
"capabilities": {
"streaming": true
},
"defaultInputModes": [
"text",
"text/plain"
],
"defaultOutputModes": [
"text",
"text/plain"
],
"description": "Helps with creating burger orders",
"name": "burger_seller_agent",
"protocolVersion": "0.2.6",
"skills": [
{
"description": "Helps with creating burger orders",
"examples": [
"I want to order 2 classic cheeseburgers"
],
"id": "create_burger_order",
"name": "Burger Order Creation Tool",
"tags": [
"burger order creation"
]
}
],
"url": "https://burger-agent-109790610330.us-central1.run.app",
"version": "1.0.0"
}
इन एजेंट कार्ड में कई ज़रूरी कॉम्पोनेंट हाइलाइट किए जाते हैं. जैसे, एजेंट की स्किल, स्ट्रीमिंग की सुविधाएं, काम करने के तरीके, प्रोटोकॉल वर्शन, और अन्य चीज़ें.
इस पूरी जानकारी का इस्तेमाल, कम्यूनिकेशन का सही तरीका डेवलप करने के लिए किया जा सकता है, ताकि A2A क्लाइंट सही तरीके से कम्यूनिकेट कर सके. सपोर्ट की गई मोडैलिटी और पुष्टि करने के तरीके से यह पक्का किया जाता है कि बातचीत सही तरीके से शुरू हो सके. साथ ही, एजेंट skills की जानकारी को A2A क्लाइंट सिस्टम प्रॉम्प्ट में शामिल किया जा सकता है, ताकि क्लाइंट के एजेंट को रिमोट एजेंट की उन क्षमताओं और कौशल के बारे में जानकारी मिल सके जिनका इस्तेमाल किया जाना है. इस एजेंट कार्ड के बारे में ज़्यादा जानकारी देने वाले फ़ील्ड, इस दस्तावेज़ में देखे जा सकते हैं.
हमारे कोड में, A2A Python SDK का इस्तेमाल करके एजेंट कार्ड को लागू किया गया है. इसे लागू करने के लिए, यहां दिया गया remote_seller_agents/burger_agent/main.py स्निपेट देखें
...
capabilities = AgentCapabilities(streaming=True)
skill = AgentSkill(
id="create_burger_order",
name="Burger Order Creation Tool",
description="Helps with creating burger orders",
tags=["burger order creation"],
examples=["I want to order 2 classic cheeseburgers"],
)
agent_host_url = (
os.getenv("HOST_OVERRIDE")
if os.getenv("HOST_OVERRIDE")
else f"http://{host}:{port}/"
)
agent_card = AgentCard(
name="burger_seller_agent",
description="Helps with creating burger orders",
url=agent_host_url,
version="1.0.0",
defaultInputModes=BurgerSellerAgent.SUPPORTED_CONTENT_TYPES,
defaultOutputModes=BurgerSellerAgent.SUPPORTED_CONTENT_TYPES,
capabilities=capabilities,
skills=[skill],
)
...
हमें वहां कई फ़ील्ड दिखते हैं. जैसे:
AgentCapabilities: एजेंट सेवा के साथ काम करने वाले अतिरिक्त वैकल्पिक फ़ंक्शन का एलान. जैसे,स्ट्रीमिंग की सुविधा और/या पुश नोटिफ़िकेशन की सुविधाAgentSkill: ऐसे टूल या फ़ंक्शन जिनकी मदद से एजेंट सहायता कर सकता हैInput/OutputModes: इनपुट/आउटपुट टाइप की वह सुविधा जो काम करती हैUrl: एजेंट से संपर्क करने का पता
इस कॉन्फ़िगरेशन में, हम डाइनैमिक एजेंट होस्ट यूआरएल बनाने की सुविधा देते हैं, ताकि लोकल टेस्टिंग और क्लाउड डिप्लॉयमेंट के बीच आसानी से स्विच किया जा सके. इसलिए, हमें पिछले चरण में HOST_OVERRIDE वैरिएबल जोड़ने की ज़रूरत है.
टास्क क्यू और एजेंट एक्ज़ीक्यूटर
A2A सर्वर, अलग-अलग एजेंट या उपयोगकर्ताओं से मिले अनुरोधों को मैनेज कर सकता है. साथ ही, हर टास्क को अलग-अलग तरीके से पूरा कर सकता है. इनके कॉन्टेक्स्ट को बेहतर तरीके से समझने के लिए, यहां दी गई इमेज देखें
इसलिए, हर A2A सर्वर को आने वाले टास्क को ट्रैक करने और उनके बारे में सही जानकारी सेव करने में सक्षम होना चाहिए. A2A एसडीके, A2A सर्वर में इस समस्या को हल करने के लिए मॉड्यूल उपलब्ध कराता है. सबसे पहले, हम यह तय कर सकते हैं कि आने वाले अनुरोध को कैसे हैंडल करना है. AgentExecutor abstract क्लास को इनहेरिट करके, हम यह कंट्रोल कर सकते हैं कि हमें टास्क को कैसे मैनेज करना है और उसे कैसे रद्द करना है. लागू करने के इस उदाहरण की जांच remote_seller_agents/burger_agent/agent_executor.py मॉड्यूल में की जा सकती है. पिज़्ज़ा बेचने वाले के मामले में भी इसी तरह का पाथ होता है
...
class BurgerSellerAgentExecutor(AgentExecutor):
"""Burger Seller AgentExecutor."""
def __init__(self):
self.agent = BurgerSellerAgent()
async def execute(
self,
context: RequestContext,
event_queue: EventQueue,
) -> None:
query = context.get_user_input()
try:
result = self.agent.invoke(query, context.context_id)
print(f"Final Result ===> {result}")
parts = [Part(root=TextPart(text=str(result)))]
await event_queue.enqueue_event(
completed_task(
context.task_id,
context.context_id,
[new_artifact(parts, f"burger_{context.task_id}")],
[context.message],
)
)
except Exception as e:
print("Error invoking agent: %s", e)
raise ServerError(error=ValueError(f"Error invoking agent: {e}")) from e
async def cancel(
self, request: RequestContext, event_queue: EventQueue
) -> Task | None:
raise ServerError(error=UnsupportedOperationError())
...
ऊपर दिए गए कोड में, हमने बुनियादी प्रोसेसिंग स्कीम लागू की है. इसमें अनुरोध मिलने पर एजेंट को सीधे तौर पर शुरू किया जाएगा. साथ ही, शुरू करने की प्रोसेस पूरी होने के बाद, पूरे किए गए टास्क के इवेंट भेजे जाएंगे. हालांकि, हमने यहां सदस्यता रद्द करने का तरीका लागू नहीं किया है, क्योंकि इसे कम समय के लिए चलने वाली प्रोसेस माना गया था.
एक्ज़ीक्यूटर बनाने के बाद, हम एचटीटीपी सर्वर को स्पिन अप करने के लिए, सीधे तौर पर DefaultRequestHandler, InMemoryTaskStore, और A2AStarletteApplication का इस्तेमाल कर सकते हैं. इस सुविधा को remote_seller_agents/burger_agent/__main__.py में देखा जा सकता है
...
request_handler = DefaultRequestHandler(
agent_executor=BurgerSellerAgentExecutor(),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
agent_card=agent_card, http_handler=request_handler
)
uvicorn.run(server.build(), host=host, port=port)
...
इस मॉड्यूल से, आपको एजेंट कार्ड को ऐक्सेस करने के लिए /.well-known/agent.json रूट को लागू करने में मदद मिलेगी. साथ ही, A2A प्रोटोकॉल के लिए POST एंडपॉइंट को लागू करने में भी मदद मिलेगी
खास जानकारी
संक्षेप में, अब तक हमने Python SDK का इस्तेमाल करके A2A सर्वर को डिप्लॉय किया है. यह सर्वर, यहां दी गई दो सुविधाओं के साथ काम कर सकता है:
/.well-known/agent.jsonरूट पर एजेंट कार्ड पब्लिश करना- मेमोरी में मौजूद टास्क की कतार का इस्तेमाल करके, JSON-RPC अनुरोध को मैनेज करना
इन सुविधाओं को शुरू करने के एंट्री पॉइंट की जांच, __main__.py स्क्रिप्ट ( remote_seller_agents/burger_agent या remote_seller_agents/pizza_agent पर ) पर की जा सकती है.
5. Purchasing Concierge - A2A Client to Agent Engine को डिप्लॉय करना
इस चरण में, हम खरीदारी के लिए कंसीयर्ज़ एजेंट को डिप्लॉय करेंगे. हम इस एजेंट के साथ इंटरैक्ट करेंगे.
खरीदारी में मदद करने वाले कंसीयज एजेंट का सोर्स कोड, purchasing_concierge डायरेक्ट्री में मौजूद है. एजेंट के शुरू होने की प्रोसेस की जांच, purchasing_agent.py स्क्रिप्ट पर की जा सकती है. यहां एजेंट के शुरू किए गए कोड का स्निपेट दिया गया है.
from google.adk import Agent
...
def create_agent(self) -> Agent:
return Agent(
model="gemini-2.5-flash-lite",
name="purchasing_agent",
instruction=self.root_instruction,
before_model_callback=self.before_model_callback,
before_agent_callback=self.before_agent_callback,
description=(
"This purchasing agent orchestrates the decomposition of the user purchase request into"
" tasks that can be performed by the seller agents."
),
tools=[
self.send_task,
],
)
...
हम इस एजेंट को एजेंट इंजन में डिप्लॉय करेंगे. Vertex AI Agent Engine, सेवाओं का एक सेट है. इससे डेवलपर, प्रोडक्शन में एआई एजेंट को डिप्लॉय, मैनेज, और बड़े पैमाने पर उपलब्ध करा सकते हैं. यह प्रोडक्शन में एजेंटों को स्केल करने के लिए इन्फ़्रास्ट्रक्चर को मैनेज करता है, ताकि हम ऐप्लिकेशन बनाने पर ध्यान दे सकें. इस बारे में ज़्यादा जानने के लिए, यह दस्तावेज़ पढ़ें. अगर हमें एजेंट सेवा को डिप्लॉय करने के लिए ज़रूरी फ़ाइलें (जैसे कि main सर्वर स्क्रिप्ट और Dockerfile) तैयार करनी होती हैं, तो इस मामले में हम ADK और Agent Engine का इस्तेमाल करके, अपनी बैकएंड सेवा को डेवलप किए बिना, एजेंट को सीधे तौर पर Python स्क्रिप्ट से डिप्लॉय कर सकते हैं. इसे डिप्लॉय करने के लिए, यह तरीका अपनाएं :
- सबसे पहले, हमें Cloud Storage में अपना स्टेजिंग स्टोरेज बनाना होगा
gcloud storage buckets create gs://purchasing-concierge-{your-project-id} --location=us-central1
- अब हमें सबसे पहले .env वैरिएबल तैयार करना होगा. आइए, .env.example को .env फ़ाइल में कॉपी करें
cp .env.example .env
- अब .env फ़ाइल खोलें. आपको यह कॉन्टेंट दिखेगा
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
STAGING_BUCKET=gs://purchasing-concierge-{your-project-id}
PIZZA_SELLER_AGENT_URL={your-pizza-agent-url}
BURGER_SELLER_AGENT_URL={your-burger-agent-url}
AGENT_ENGINE_RESOURCE_NAME={your-agent-engine-resource-name}
यह एजेंट, बर्गर और पिज़्ज़ा एजेंट से बातचीत करेगा. इसलिए, हमें इन दोनों के लिए सही क्रेडेंशियल देने होंगे. हमें पिछले चरणों में मिले Cloud Run यूआरएल का इस्तेमाल करके, PIZZA_SELLER_AGENT_URL और BURGER_SELLER_AGENT_URL को अपडेट करना होगा.
अगर आपको यह याद नहीं है, तो Cloud Run कंसोल पर जाएं. अपने कंसोल में सबसे ऊपर मौजूद सर्च बार में "Cloud Run" टाइप करें. इसके बाद, Cloud Run आइकॉन पर राइट क्लिक करके, इसे नए टैब में खोलें
आपको रिमोट सेलर एजेंट की हमारी पिछली सेवाओं को नीचे दिए गए तरीके से देखना चाहिए
अब उन सेवाओं का सार्वजनिक यूआरएल देखने के लिए, किसी एक सेवा पर क्लिक करें. इसके बाद, आपको सेवा की जानकारी वाले पेज पर रीडायरेक्ट कर दिया जाएगा. आपको यूआरएल, सबसे ऊपर दाईं ओर क्षेत्र की जानकारी के ठीक बगल में दिखेगा
फ़ाइनल एनवायरमेंट वैरिएबल कुछ ऐसा दिखना चाहिए
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
STAGING_BUCKET=gs://purchasing-concierge-{your-project-id}
PIZZA_SELLER_AGENT_URL=https://pizza-agent-xxxxx.us-central1.run.app
BURGER_SELLER_AGENT_URL=https://burger-agent-xxxxx.us-central1.run.app
AGENT_ENGINE_RESOURCE_NAME={your-agent-engine-resource-name}
- अब हम खरीदारी में मदद करने वाले एजेंट को डिप्लॉय करने के लिए तैयार हैं. इस डेमो में, हम स्क्रिप्ट
deploy_to_agent_engine.pyका इस्तेमाल करके डिप्लॉय करेंगे. इसका कॉन्टेंट यहां दिखाया गया है
"""
Copyright 2025 Google LLC
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import vertexai
from vertexai.preview import reasoning_engines
from vertexai import agent_engines
from dotenv import load_dotenv
import os
from purchasing_concierge.agent import root_agent
load_dotenv()
PROJECT_ID = os.getenv("GOOGLE_CLOUD_PROJECT")
LOCATION = os.getenv("GOOGLE_CLOUD_LOCATION")
STAGING_BUCKET = os.getenv("STAGING_BUCKET")
vertexai.init(
project=PROJECT_ID,
location=LOCATION,
staging_bucket=STAGING_BUCKET,
)
adk_app = reasoning_engines.AdkApp(
agent=root_agent,
)
remote_app = agent_engines.create(
agent_engine=adk_app,
display_name="purchasing-concierge",
requirements=[
"google-cloud-aiplatform[adk,agent_engines]",
"a2a-sdk==0.2.16",
],
extra_packages=[
"./purchasing_concierge",
],
env_vars={
"GOOGLE_GENAI_USE_VERTEXAI": os.environ["GOOGLE_GENAI_USE_VERTEXAI"],
"PIZZA_SELLER_AGENT_URL": os.environ["PIZZA_SELLER_AGENT_URL"],
"BURGER_SELLER_AGENT_URL": os.environ["BURGER_SELLER_AGENT_URL"],
},
)
print(f"Deployed remote app resource: {remote_app.resource_name}")
एजेंट इंजन में हमारे एडीके एजेंट को डिप्लॉय करने के लिए, यह तरीका अपनाएं. सबसे पहले, हमें अपने एडीके root_agent से एक AdkApp ऑब्जेक्ट बनाना होगा. इसके बाद, हम adk_app ऑब्जेक्ट देकर agent_engines.create तरीके को चला सकते हैं. इसके लिए, requirements फ़ील्ड में ज़रूरी शर्तें तय करें, extra_packages में एजेंट डायरेक्ट्री का पाथ तय करें, और ज़रूरी env वैरिएबल दें.
हम स्क्रिप्ट चलाकर इसे डिप्लॉय कर सकते हैं:
uv run deploy_to_agent_engine.py
सफल डिप्लॉयमेंट के बाद, यह इस तरह का लॉग दिखाएगा. ध्यान दें कि xxxx आपका प्रोजेक्ट आईडी है और yyyy आपके एजेंट इंजन का संसाधन आईडी है
AgentEngine created. Resource name: projects/xxxx/locations/us-central1/reasoningEngines/yyyy
To use this AgentEngine in another session:
agent_engine = vertexai.agent_engines.get('projects/xxxx/locations/us-central1/reasoningEngines/yyyy)
Deployed remote app resource: projects/xxxx/locations/us-central1/reasoningEngines/xxxx
जब हम इसे एजेंट इंजन डैशबोर्ड में देखते हैं, तो (सर्च बार पर "एजेंट इंजन" खोजें) यह हमारा पिछला डिप्लॉयमेंट दिखाएगा
एजेंट इंजन पर डिप्लॉय किए गए एजेंट की जांच करना
एजेंट इंजन के साथ इंटरैक्ट करने के लिए, curl कमांड और एसडीके का इस्तेमाल किया जा सकता है. उदाहरण के लिए, डिप्लॉय किए गए एजेंट के साथ इंटरैक्ट करने के लिए, यह कमांड चलाएं.
यह क्वेरी भेजकर देखें कि एजेंट को सही तरीके से डिप्लॉय किया गया है या नहीं
curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://us-central1-aiplatform.googleapis.com/v1/projects/{YOUR_PROJECT_ID}/locations/us-central1/reasoningEngines/{YOUR_AGENT_ENGINE_RESOURCE_ID}:streamQuery?alt=sse -d '{
"class_method": "stream_query",
"input": {
"user_id": "user_123",
"message": "List available burger menu please",
}
}'
अगर कार्रवाई पूरी होती है, तो आपकी कंसोल पर इस तरह से स्ट्रीम किए गए कई जवाब इवेंट दिखेंगे
{
"content": {
"parts": [
{
"text": "Here is our burger menu:\n- Classic Cheeseburger: IDR 85K\n- Double Cheeseburger: IDR 110K\n- Spicy Chicken Burger: IDR 80K\n- Spicy Cajun Burger: IDR 85K"
}
],
"role": "model"
},
"usage_metadata": {
"candidates_token_count": 51,
"candidates_tokens_details": [
{
"modality": "TEXT",
"token_count": 51
}
],
"prompt_token_count": 907,
"prompt_tokens_details": [
{
"modality": "TEXT",
"token_count": 907
}
],
"total_token_count": 958,
"traffic_type": "ON_DEMAND"
},
"invocation_id": "e-14679918-af68-45f1-b942-cf014368a733",
"author": "purchasing_agent",
"actions": {
"state_delta": {},
"artifact_delta": {},
"requested_auth_configs": {}
},
"id": "dbe7fc43-b82a-4f3e-82aa-dd97afa8f15b",
"timestamp": 1754287348.941454
}
हम अगले चरण में यूज़र इंटरफ़ेस (यूआई) का इस्तेमाल करेंगे. हालांकि, पहले हम A2A क्लाइंट के मुख्य कॉम्पोनेंट और सामान्य फ़्लो के बारे में बात करते हैं
6. A2A क्लाइंट के मुख्य कॉम्पोनेंट
ऊपर दिखाई गई इमेज में, A2A इंटरैक्शन का सामान्य फ़्लो दिखाया गया है:
- क्लाइंट, दिए गए रिमोट एजेंट यूआरएल में पब्लिश किए गए किसी एजेंट कार्ड को ढूंढने की कोशिश करेगा
/.well-known/agent.json - इसके बाद, ज़रूरत पड़ने पर यह एजेंट को मैसेज और ज़रूरी मेटाडेटा पैरामीटर ( जैसे, सेशन आईडी, पुराना कॉन्टेक्स्ट वगैरह) भेजेगा. सर्वर इस मैसेज को पूरा किए जाने वाले टास्क के तौर पर देखेगा
- A2A सर्वर, अनुरोध को प्रोसेस करता है. अगर सर्वर पर पुश नोटिफ़िकेशन की सुविधा काम करती है, तो वह टास्क प्रोसेस करने के दौरान कुछ सूचनाएं भी पब्लिश कर पाएगा. ( यह सुविधा, इस कोडलैब के दायरे से बाहर है )
- प्रोसेस पूरी होने के बाद, A2A सर्वर, जवाब देने वाला आर्टफ़ैक्ट वापस क्लाइंट को भेज देगा
ऊपर दिए गए इंटरैक्शन के लिए, कुछ मुख्य ऑब्जेक्ट ये आइटम हैं. ज़्यादा जानकारी के लिए, यहां पढ़ें:
- मैसेज: क्लाइंट और रिमोट एजेंट के बीच बातचीत का एक राउंड
- टास्क: यह A2A की मैनेज की गई बुनियादी यूनिट है. इसकी पहचान एक यूनीक आईडी से की जाती है
- आर्टफ़ैक्ट: एजेंट की ओर से किसी टास्क के नतीजे के तौर पर जनरेट किया गया आउटपुट.जैसे, कोई दस्तावेज़, इमेज, स्ट्रक्चर्ड डेटा. इसमें कई पार्ट शामिल होते हैं
- पार्ट: मैसेज या आर्टफ़ैक्ट में मौजूद कॉन्टेंट की सबसे छोटी यूनिट. यह हिस्सा टेक्स्ट, इमेज, वीडियो, फ़ाइल वगैरह हो सकता है.
कार्ड डिस्कवरी
जब A2A क्लाइंट सेवा शुरू की जाती है, तो सामान्य प्रक्रिया यह होती है कि एजेंट कार्ड की जानकारी हासिल की जाए और उसे सेव किया जाए, ताकि ज़रूरत पड़ने पर उसे आसानी से ऐक्सेस किया जा सके. इस कोडलैब में, हमने इसे before_agent_callback पर लागू किया है. इसे purchasing_concierge/purchasing_agent.py पर लागू किया जा सकता है. यहां दिया गया कोड स्निपेट देखें
...
async def before_agent_callback(self, callback_context: CallbackContext):
if not self.a2a_client_init_status:
httpx_client = httpx.AsyncClient(timeout=httpx.Timeout(timeout=30))
for address in self.remote_agent_addresses:
card_resolver = A2ACardResolver(
base_url=address, httpx_client=httpx_client
)
try:
card = await card_resolver.get_agent_card()
remote_connection = RemoteAgentConnections(
agent_card=card, agent_url=card.url
)
self.remote_agent_connections[card.name] = remote_connection
self.cards[card.name] = card
except httpx.ConnectError:
print(f"ERROR: Failed to get agent card from : {address}")
agent_info = []
for ra in self.list_remote_agents():
agent_info.append(json.dumps(ra))
self.agents = "\n".join(agent_info)
...
यहां, हम बिल्ट-इन A2A क्लाइंट A2ACardResolver मॉड्यूल का इस्तेमाल करके, उपलब्ध सभी एजेंट कार्ड ऐक्सेस करने की कोशिश करते हैं. इसके बाद, हम एजेंट को मैसेज भेजने के लिए ज़रूरी कनेक्शन इकट्ठा करते हैं. इसके बाद, हमें प्रॉम्प्ट में उपलब्ध सभी एजेंट और उनकी खास बातों को भी शामिल करना होता है, ताकि हमारे एजेंट को पता चल सके कि वह इन एजेंट से कम्यूनिकेट कर सकता है
प्रॉम्प्ट और टास्क भेजने वाला टूल
यह वह प्रॉम्प्ट और टूल है जो हम यहां अपने एडीके एजेंट को उपलब्ध कराते हैं
...
def root_instruction(self, context: ReadonlyContext) -> str:
current_agent = self.check_active_agent(context)
return f"""You are an expert purchasing delegator that can delegate the user product inquiry and purchase request to the
appropriate seller remote agents.
Execution:
- For actionable tasks, you can use `send_task` to assign tasks to remote agents to perform.
- When the remote agent is repeatedly asking for user confirmation, assume that the remote agent doesn't have access to user's conversation context.
So improve the task description to include all the necessary information related to that agent
- Never ask user permission when you want to connect with remote agents. If you need to make connection with multiple remote agents, directly
connect with them without asking user permission or asking user preference
- Always show the detailed response information from the seller agent and propagate it properly to the user.
- If the remote seller is asking for confirmation, rely the confirmation question to the user if the user haven't do so.
- If the user already confirmed the related order in the past conversation history, you can confirm on behalf of the user
- Do not give irrelevant context to remote seller agent. For example, ordered pizza item is not relevant for the burger seller agent
- Never ask order confirmation to the remote seller agent
Please rely on tools to address the request, and don't make up the response. If you are not sure, please ask the user for more details.
Focus on the most recent parts of the conversation primarily.
If there is an active agent, send the request to that agent with the update task tool.
Agents:
{self.agents}
Current active seller agent: {current_agent["active_agent"]}
"""
...
async def send_task(self, agent_name: str, task: str, tool_context: ToolContext):
"""Sends a task to remote seller agent
This will send a message to the remote agent named agent_name.
Args:
agent_name: The name of the agent to send the task to.
task: The comprehensive conversation context summary
and goal to be achieved regarding user inquiry and purchase request.
tool_context: The tool context this method runs in.
Yields:
A dictionary of JSON data.
"""
if agent_name not in self.remote_agent_connections:
raise ValueError(f"Agent {agent_name} not found")
state = tool_context.state
state["active_agent"] = agent_name
client = self.remote_agent_connections[agent_name]
if not client:
raise ValueError(f"Client not available for {agent_name}")
session_id = state["session_id"]
task: Task
message_id = ""
metadata = {}
if "input_message_metadata" in state:
metadata.update(**state["input_message_metadata"])
if "message_id" in state["input_message_metadata"]:
message_id = state["input_message_metadata"]["message_id"]
if not message_id:
message_id = str(uuid.uuid4())
payload = {
"message": {
"role": "user",
"parts": [
{"type": "text", "text": task}
], # Use the 'task' argument here
"messageId": message_id,
"contextId": session_id,
},
}
message_request = SendMessageRequest(
id=message_id, params=MessageSendParams.model_validate(payload)
)
send_response: SendMessageResponse = await client.send_message(
message_request=message_request
)
print(
"send_response",
send_response.model_dump_json(exclude_none=True, indent=2),
)
if not isinstance(send_response.root, SendMessageSuccessResponse):
print("received non-success response. Aborting get task ")
return None
if not isinstance(send_response.root.result, Task):
print("received non-task response. Aborting get task ")
return None
return send_response.root.result
...
प्रॉम्प्ट में, हम खरीदारी में मदद करने वाले अपने एजेंट को, रिमोट एजेंट के तौर पर उपलब्ध सभी एजेंट के नाम और जानकारी देते हैं. साथ ही, टूल self.send_task में, हम एजेंट से कनेक्ट करने के लिए सही क्लाइंट को वापस पाने का तरीका उपलब्ध कराते हैं. साथ ही, SendMessageRequest ऑब्जेक्ट का इस्तेमाल करके ज़रूरी मेटाडेटा भेजते हैं.
कम्यूनिकेशन प्रोटोकॉल
Task की परिभाषा, A2A सर्वर के मालिकाना हक वाला डोमेन है. हालांकि, A2A क्लाइंट के हिसाब से, यह सर्वर को भेजा गया एक मैसेज होता है. सर्वर यह तय करता है कि क्लाइंट से मिले मैसेज को किस टास्क के तौर पर तय किया जाए. साथ ही, यह भी तय करता है कि टास्क पूरा करने के लिए क्लाइंट के साथ इंटरैक्ट करना ज़रूरी है या नहीं. टास्क के लाइफ़साइकल के बारे में ज़्यादा जानकारी पाने के लिए, यह दस्तावेज़ पढ़ें. इसकी ज़्यादा जानकारी यहां दी गई है:
मैसेज -> टास्क के इस इंटरचेंज को, JSON-RPC स्टैंडर्ड के ऊपर पेलोड फ़ॉर्मैट का इस्तेमाल करके लागू किया जाता है. जैसे, message/send प्रोटोकॉल के इस उदाहरण में दिखाया गया है :
{
# identifier for this request
"id": "abc123",
# version of JSON-RPC protocol
"jsonrpc": "2.0",
# method name
"method": "message/send",
# parameters/arguments of the method
"params": {
"message": "hi, what can you help me with?"
}
}
इसके लिए, कई तरीके उपलब्ध हैं. उदाहरण के लिए, अलग-अलग तरह के कम्यूनिकेशन (जैसे, सिंक, स्ट्रीमिंग, एसिंक) के लिए या टास्क की स्थिति के बारे में सूचनाएं कॉन्फ़िगर करने के लिए. इन टास्क डेफ़िनिशन स्टैंडर्ड को मैनेज करने के लिए, A2A सर्वर को आसानी से कॉन्फ़िगर किया जा सकता है. इन तरीकों के बारे में ज़्यादा जानकारी के लिए, यह दस्तावेज़ पढ़ें.
7. इंटिग्रेशन टेस्टिंग और पेलोड की जांच
अब वेब यूज़र इंटरफ़ेस (यूआई) का इस्तेमाल करके, रिमोट एजेंट के साथ बातचीत करके खरीदारी से जुड़ी सलाह देने वाले हमारे एआई की जांच करते हैं.
सबसे पहले, हमें . में AGENT_ENGINE_RESOURCE_NAME को अपडेट करना होगा.env फ़ाइल. पक्का करें कि आपने एजेंट इंजन का सही संसाधन नाम दिया हो. आपकी .env फ़ाइल ऐसी दिखनी चाहिए:
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
STAGING_BUCKET=gs://purchasing-concierge-{your-project-id}
PIZZA_SELLER_AGENT_URL=https://pizza-agent-xxxxx.us-central1.run.app
BURGER_SELLER_AGENT_URL=https://burger-agent-xxxxx.us-central1.run.app
AGENT_ENGINE_RESOURCE_NAME=projects/xxxx/locations/us-central1/reasoningEngines/yyyy
इसके बाद, Gradio ऐप्लिकेशन को डिप्लॉय करने के लिए, यह कमांड चलाएं
uv run purchasing_concierge_ui.py
अगर यह प्रोसेस पूरी हो जाती है, तो आपको यह आउटपुट दिखेगा
* Running on local URL: http://0.0.0.0:8080 * To create a public link, set `share=True` in `launch()`.
इसके बाद, टर्मिनल पर http://0.0.0.0:8080 यूआरएल पर Ctrl + क्लिक करें या वेब यूज़र इंटरफ़ेस (यूआई) खोलने के लिए, वेब की झलक दिखाने वाले बटन पर क्लिक करें
इस तरह से बातचीत करने की कोशिश करें :
- मुझे बर्गर और पिज़्ज़ा का मेन्यू दिखाओ
- मुझे एक बारबेक्यू चिकन पिज़्ज़ा और एक स्पाइसी काजुन बर्गर ऑर्डर करना है
इसके बाद, ऑर्डर पूरा होने तक बातचीत जारी रखें. देखें कि इंटरैक्शन कैसा चल रहा है. साथ ही, टूल कॉल और जवाब क्या है? नीचे दी गई इमेज, इंटरैक्शन के नतीजे का एक उदाहरण है.
हम देख सकते हैं कि दो अलग-अलग एजेंटों से बातचीत करने पर, दो अलग-अलग तरह के नतीजे मिलते हैं. A2A इस समस्या को आसानी से हल कर सकता है. पिज़्ज़ा बेचने वाला एजेंट, खरीदारी करने वाले हमारे एजेंट के अनुरोध को सीधे तौर पर स्वीकार कर लेता है. वहीं, बर्गर बेचने वाले एजेंट को हमारे अनुरोध को पूरा करने से पहले, हमारी पुष्टि की ज़रूरत होती है. पुष्टि हो जाने के बाद, एजेंट बर्गर बेचने वाले एजेंट को इसकी जानकारी दे सकता है
अब हमने A2A के बुनियादी सिद्धांतों के बारे में जान लिया है. आइए, देखते हैं कि इसे क्लाइंट और सर्वर आर्किटेक्चर के तौर पर कैसे लागू किया जाता है
8. चुनौती
क्या अब ज़रूरी फ़ाइल तैयार की जा सकती है और Gradio ऐप्लिकेशन को Cloud Run में खुद डिप्लॉय किया जा सकता है? चुनौती स्वीकार करने का समय आ गया है!
9. व्यवस्थित करें
इस कोडलैब में इस्तेमाल किए गए संसाधनों के लिए, अपने Google Cloud खाते से शुल्क न लिए जाने के लिए, यह तरीका अपनाएं:
- Google Cloud Console में, संसाधन मैनेज करें पेज पर जाएं.
- प्रोजेक्ट की सूची में, वह प्रोजेक्ट चुनें जिसे आपको मिटाना है. इसके बाद, मिटाएं पर क्लिक करें.
- डायलॉग बॉक्स में, प्रोजेक्ट आईडी डालें. इसके बाद, प्रोजेक्ट मिटाने के लिए बंद करें पर क्लिक करें.
- इसके अलावा, कंसोल पर Cloud Run पर जाकर, अभी-अभी डिप्लॉय की गई सेवा को चुनें और मिटाएं.