1. The Late Night Code Review
अभी रात के 2 बजे हैं
आपने कई घंटों से डीबग किया है. फ़ंक्शन सही दिख रहा है, लेकिन कोई गड़बड़ी हुई है. आपको पता है कि कोड काम करना चाहिए, लेकिन वह काम नहीं करता. साथ ही, आपको यह भी नहीं पता चलता कि ऐसा क्यों हो रहा है, क्योंकि आपने उसे बहुत देर तक देखा है.
def dfs_search_v1(graph, start, target):
"""Find if target is reachable from start."""
visited = set()
stack = start # Looks innocent enough...
while stack:
current = stack.pop()
if current == target:
return True
if current not in visited:
visited.add(current)
for neighbor in graph[current]:
if neighbor not in visited:
stack.append(neighbor)
return False
एआई डेवलपर का सफ़र
अगर आप इसे पढ़ रहे हैं, तो इसका मतलब है कि आपने कोडिंग में एआई के इस्तेमाल से होने वाले बदलावों को अनुभव किया है. Gemini Code Assist, Claude Code, और Cursor जैसे टूल ने, कोड लिखने के तरीके को बदल दिया है. ये टूल, बॉयलरप्लेट जनरेट करने, सुझाव देने, और डेवलपमेंट की प्रोसेस को तेज़ करने के लिए बेहतरीन हैं.
हालांकि, आप यहां इसलिए हैं, क्योंकि आपको इस बारे में ज़्यादा जानकारी चाहिए. आपको इन एआई सिस्टम को सिर्फ़ इस्तेमाल करने का तरीका नहीं, बल्कि बनाने का तरीका भी जानना है. आपको ऐसा कॉन्टेंट बनाना है जो:
- जिसकी परफ़ॉर्मेंस का अनुमान लगाया जा सकता है और जिसे ट्रैक किया जा सकता है
- इसे प्रोडक्शन में आसानी से डिप्लॉय किया जा सकता है
- आपको भरोसेमंद और एक जैसे नतीजे मिलते हैं
- इससे आपको यह पता चलता है कि Gemini ने फ़ैसले कैसे लिए
उपयोगकर्ता से क्रिएटर बनने तक का सफ़र
आज, आपको एआई टूल इस्तेमाल करने के बजाय, उन्हें बनाने का मौका मिलेगा. आपको एक मल्टी-एजेंट सिस्टम बनाना होगा, जो ये काम करता हो:
- यह कोड स्ट्रक्चर का विश्लेषण करता है
- व्यवहार की पुष्टि करने के लिए, असल टेस्ट करता है
- पुष्टि करता है कि स्टाइल, असली लिंटर के साथ काम करती है
- इकट्ठा किए गए डेटा से, कार्रवाई में मदद करने वाली जानकारी निकालता है
- डिप्लॉयमेंट को Google Cloud पर पूरी तरह से मॉनिटर किया जा सकता है
2. आपके पहले एजेंट को डिप्लॉय करने की प्रोसेस
डेवलपर का सवाल
"मुझे एलएलएम के बारे में पता है और मैंने एपीआई का इस्तेमाल किया है. हालांकि, मुझे यह नहीं पता कि Python स्क्रिप्ट से, प्रोडक्शन एआई एजेंट तक कैसे पहुंचा जाए, ताकि उसे स्केल किया जा सके?"
आइए, इस सवाल का जवाब देने के लिए, अपने एनवायरमेंट को सही तरीके से सेट अप करें. इसके बाद, प्रोडक्शन पैटर्न के बारे में जानने से पहले, एक सामान्य एजेंट बनाएं, ताकि हम बुनियादी बातें समझ सकें.
सबसे पहले ज़रूरी सेटअप करें
इससे पहले कि हम कोई एजेंट बनाएं, आइए पक्का करें कि आपका Google Cloud एनवायरमेंट तैयार हो.
क्या आपको Google Cloud क्रेडिट चाहिए?
Google Cloud Console में सबसे ऊपर मौजूद, Cloud Shell चालू करें पर क्लिक करें. यह Cloud Shell पैनल में सबसे ऊपर मौजूद, टर्मिनल के आकार का आइकॉन है,
अपना Google Cloud प्रोजेक्ट आईडी ढूंढें:
- Google Cloud Console खोलें: https://console.cloud.google.com
- पेज पर सबसे ऊपर मौजूद प्रोजेक्ट ड्रॉपडाउन से, वह प्रोजेक्ट चुनें जिसका इस्तेमाल आपको इस वर्कशॉप के लिए करना है.
- आपका प्रोजेक्ट आईडी, डैशबोर्ड पर मौजूद प्रोजेक्ट की जानकारी देने वाले कार्ड में दिखता है
पहला चरण: अपना प्रोजेक्ट आईडी सेट करना
Cloud Shell में, gcloud कमांड-लाइन टूल पहले से ही कॉन्फ़िगर होता है. चालू प्रोजेक्ट सेट करने के लिए, यह कमांड चलाएं. यह $GOOGLE_CLOUD_PROJECT एनवायरमेंट वैरिएबल का इस्तेमाल करता है. यह आपके Cloud Shell सेशन में अपने-आप सेट हो जाता है.
gcloud config set project $GOOGLE_CLOUD_PROJECT
दूसरा चरण: सेटअप की पुष्टि करना
इसके बाद, यहां दिए गए कमांड चलाकर पुष्टि करें कि आपका प्रोजेक्ट सही तरीके से सेट किया गया है और आपने पुष्टि कर ली है.
# Confirm project is set
echo "Current project: $(gcloud config get-value project)"
# Check authentication status
gcloud auth list
आपको अपना प्रोजेक्ट आईडी दिखेगा. साथ ही, आपके उपयोगकर्ता खाते के बगल में (ACTIVE) लिखा होगा.
अगर आपका खाता चालू के तौर पर नहीं दिखता है या आपको पुष्टि करने से जुड़ी कोई गड़बड़ी मिलती है, तो लॉग इन करने के लिए यह कमांड चलाएं:
gcloud auth application-default login
तीसरा चरण: ज़रूरी एपीआई चालू करना
बेसिक एजेंट के लिए, हमें कम से कम इन एपीआई की ज़रूरत होती है:
gcloud services enable \
aiplatform.googleapis.com \
compute.googleapis.com
इसमें कुछ समय लग सकता है . आपको ये चीज़ें दिखेंगी:
Operation "operations/..." finished successfully.
चौथा चरण: ADK इंस्टॉल करना
# Install the ADK CLI
pip install google-adk --upgrade
# Verify installation
adk --version
आपको 1.15.0 या इससे ज़्यादा का वर्शन नंबर दिखना चाहिए.
अब अपना बेसिक एजेंट बनाएं
एनवायरमेंट तैयार होने के बाद, चलिए अब हम एक सामान्य एजेंट बनाते हैं.
पांचवां चरण: ADK Create का इस्तेमाल करना
adk create my_first_agent
इंटरैक्टिव निर्देशों का पालन करें:
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
Enter Google Cloud project ID [auto-detected-from-gcloud]:
Enter Google Cloud region [us-central1]:
छठा चरण: बनाई गई चीज़ों की जांच करना
cd my_first_agent
ls -la
आपको तीन फ़ाइलें दिखेंगी:
.env # Configuration (auto-populated with your project)
__init__.py # Package marker
agent.py # Your agent definition
सातवां चरण: कॉन्फ़िगरेशन की तुरंत जांच करना
# Verify the .env was created correctly
cat .env
# Should show something like:
# GOOGLE_CLOUD_PROJECT=your-project-id
# GOOGLE_CLOUD_LOCATION=us-central1
# GOOGLE_GENAI_USE_VERTEXAI=1
अगर प्रोजेक्ट आईडी मौजूद नहीं है या गलत है, तो .env फ़ाइल में बदलाव करें:
nano .env # or use your preferred editor
आठवां चरण: एजेंट कोड देखें
cat 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',
)
सिंपल, साफ़-सुथरा, और कम से कम शब्दों में. यह एजेंटों के लिए आपका "नमस्ते दुनिया" है.
अपने बुनियादी एजेंट को आज़माना
नौवां चरण: एजेंट को चलाना
cd ..
adk run my_first_agent
आपको इस तरह की विंडो दिखेगी:
Log setup complete: /tmp/agents_log/agent.20250930_162430.log
To access latest log: tail -F /tmp/agents_log/agent.latest.log
[user]:
दसवां चरण: कुछ क्वेरी डालकर देखें
adk run जिस टर्मिनल पर चल रहा है वहां आपको एक प्रॉम्प्ट दिखेगा. अपनी क्वेरी टाइप करें:
[user]: Hey, what can you do?
[root_agent]: I can help answer questions, generate text, explain concepts, assist with problem-solving, and provide information on a wide range of topics. What would you like to know or discuss?
[user]: What's the current weather in Mountain View?
[root_agent]: I'm sorry, but I cannot provide real-time weather information as I don't have access to current data or weather services. To get the current weather in Mountain View, I recommend checking a weather website or app like weather.com or using a search engine.
इसकी सीमा के बारे में ध्यान दें - यह मौजूदा डेटा को ऐक्सेस नहीं कर सकता. आइए, अब इसके बारे में ज़्यादा जानते हैं:
[user]: Review this Python code: def add(a, b): return a + b
[root_agent]: This is a simple Python function that adds two numbers. Here's my review:
**Strengths:**
- Clean and concise implementation
- Follows Python conventions for simple operations
- Clear function name that describes its purpose
**Suggestions for improvement:**
- Add a docstring to document the function
- Consider adding type hints for better code clarity
- You might want to handle edge cases
Here's an enhanced version:
def add(a: float, b: float) -> float:
"""Add two numbers and return their sum."""
return a + b
एजेंट कोड के बारे में चर्चा कर सकता है, लेकिन क्या वह:
- क्या स्ट्रक्चर को समझने के लिए, एएसटी को पार्स करना है?
- क्या इसकी जांच करने के लिए टेस्ट चलाएं?
- क्या स्टाइल से जुड़ी शर्तों का पालन किया गया है?
- क्या आपको अपनी पिछली समीक्षाएं याद हैं?
नहीं. यहां हमें आर्किटेक्चर की ज़रूरत है.
🏃🚪
Ctrl+C
जब आपको और एक्सप्लोर नहीं करना हो.
3. प्रोडक्शन वर्कस्पेस तैयार करना
समाधान: प्रोडक्शन के लिए तैयार आर्किटेक्चर
इस आसान एजेंट ने शुरुआती पॉइंट के बारे में बताया. हालांकि, प्रोडक्शन सिस्टम के लिए मज़बूत स्ट्रक्चर की ज़रूरत होती है. अब हम एक ऐसा पूरा प्रोजेक्ट सेट अप करेंगे जिसमें प्रोडक्शन के सिद्धांत शामिल हों.
बुनियादी सेटअप करना
आपने बुनियादी एजेंट के लिए, Google Cloud प्रोजेक्ट को पहले ही कॉन्फ़िगर कर लिया हो. अब हम पूरे प्रोडक्शन वर्कस्पेस को तैयार करेंगे. इसमें वे सभी टूल, पैटर्न, और बुनियादी ढांचा शामिल होगा जिनकी ज़रूरत किसी असली सिस्टम के लिए होती है.
पहला चरण: स्ट्रक्चर्ड प्रोजेक्ट पाना
सबसे पहले, Ctrl+C की मदद से, चल रही किसी भी adk run से बाहर निकलें और साफ़ करें:
# Clean up the basic agent
cd ~ # Make sure you're not inside my_first_agent
rm -rf my_first_agent
# Get the production scaffold
git clone https://github.com/ayoisio/adk-code-review-assistant.git
cd adk-code-review-assistant
git checkout codelab
दूसरा चरण: वर्चुअल एनवायरमेंट बनाना और उसे चालू करना
# Create the virtual environment
python -m venv .venv
# Activate it
# On macOS/Linux:
source .venv/bin/activate
# On Windows:
# .venv\Scripts\activate
पुष्टि: अब आपके प्रॉम्प्ट की शुरुआत में (.venv) दिखना चाहिए.
तीसरा चरण: डिपेंडेंसी इंस्टॉल करना
pip install -r code_review_assistant/requirements.txt
# Install the package in editable mode (enables imports)
pip install -e .
इससे ये इंस्टॉल होते हैं:
google-adk- ADK फ़्रेमवर्कpycodestyle- PEP 8 की जांच करने के लिएvertexai- क्लाउड डिप्लॉयमेंट के लिए- प्रॉडक्ट से जुड़ी अन्य डिपेंडेंसी
-e फ़्लैग की मदद से, code_review_assistant मॉड्यूल को कहीं से भी इंपोर्ट किया जा सकता है.
चौथा चरण: अपना एनवायरमेंट कॉन्फ़िगर करना
# Copy the example environment file
cp .env.example .env
# Edit .env and replace the placeholders:
# - GOOGLE_CLOUD_PROJECT=your-project-id → your actual project ID
# - Keep other defaults as-is
पुष्टि करना: अपने कॉन्फ़िगरेशन की जांच करें:
cat .env
इसमें यह जानकारी दिखनी चाहिए:
GOOGLE_CLOUD_PROJECT=your-actual-project-id
GOOGLE_CLOUD_LOCATION=us-central1
GOOGLE_GENAI_USE_VERTEXAI=TRUE
पांचवां चरण: पुष्टि करना
आपने पहले ही gcloud auth को चला लिया है. इसलिए, आइए सिर्फ़ इसकी पुष्टि करें:
# Check current authentication
gcloud auth list
# Should show your account with (ACTIVE)
# If not, run:
gcloud auth application-default login
छठा चरण: प्रोडक्शन के लिए उपलब्ध अन्य एपीआई चालू करना
हमने बुनियादी एपीआई पहले ही चालू कर दिए हैं. अब प्रोडक्शन वाले प्रॉडक्ट जोड़ें:
gcloud services enable \
sqladmin.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
artifactregistry.googleapis.com \
storage.googleapis.com \
cloudtrace.googleapis.com
इससे ये काम किए जा सकते हैं:
- SQL एडमिन: Cloud Run का इस्तेमाल करने पर Cloud SQL के लिए
- Cloud Run: सर्वरलेस डिप्लॉयमेंट के लिए
- Cloud Build: ऑटोमेटेड डिप्लॉयमेंट के लिए
- Artifact Registry: कंटेनर इमेज के लिए
- Cloud Storage: आर्टफ़ैक्ट और स्टेजिंग के लिए
- Cloud Trace: निगरानी के लिए
सातवां चरण: Artifact Registry रिपॉज़िटरी बनाना
हमारा डिप्लॉयमेंट, कंटेनर इमेज बनाएगा. इसके लिए, एक होम डायरेक्ट्री की ज़रूरत होगी:
gcloud artifacts repositories create code-review-assistant-repo \
--repository-format=docker \
--location=us-central1 \
--description="Docker repository for Code Review Assistant"
आपको यह दिखना चाहिए:
Created repository [code-review-assistant-repo].
अगर यह फ़ाइल पहले से मौजूद है (शायद पिछली कोशिश से), तो कोई बात नहीं. आपको गड़बड़ी का एक मैसेज दिखेगा, जिसे अनदेखा किया जा सकता है.
आठवां चरण: IAM अनुमतियां देना
# Get your project number
PROJECT_NUMBER=$(gcloud projects describe $GOOGLE_CLOUD_PROJECT \
--format="value(projectNumber)")
# Define the service account
SERVICE_ACCOUNT="${PROJECT_NUMBER}@cloudbuild.gserviceaccount.com"
# Grant necessary roles
gcloud projects add-iam-policy-binding $GOOGLE_CLOUD_PROJECT \
--member="serviceAccount:${SERVICE_ACCOUNT}" \
--role="roles/run.admin"
gcloud projects add-iam-policy-binding $GOOGLE_CLOUD_PROJECT \
--member="serviceAccount:${SERVICE_ACCOUNT}" \
--role="roles/iam.serviceAccountUser"
gcloud projects add-iam-policy-binding $GOOGLE_CLOUD_PROJECT \
--member="serviceAccount:${SERVICE_ACCOUNT}" \
--role="roles/cloudsql.admin"
gcloud projects add-iam-policy-binding $GOOGLE_CLOUD_PROJECT \
--member="serviceAccount:${SERVICE_ACCOUNT}" \
--role="roles/storage.admin"
हर कमांड का आउटपुट यह होगा:
Updated IAM policy for project [your-project-id].
आपने क्या-क्या हासिल किया
आपका प्रोडक्शन वर्कस्पेस अब पूरी तरह से तैयार है:
✅ Google Cloud प्रोजेक्ट कॉन्फ़िगर और पुष्टि किया गया हो
✅ बुनियादी एजेंट की जांच की गई हो, ताकि सीमाओं के बारे में पता चल सके
✅ रणनीतिक प्लेसहोल्डर वाला प्रोजेक्ट कोड तैयार हो
✅ वर्चुअल एनवायरमेंट में डिपेंडेंसी अलग की गई हों
✅ सभी ज़रूरी एपीआई चालू हों
✅ कंटेनर रजिस्ट्री, डिप्लॉयमेंट के लिए तैयार हो
✅ IAM अनुमतियां सही तरीके से कॉन्फ़िगर की गई हों
✅ एनवायरमेंट वैरिएबल सही तरीके से सेट किए गए हों
अब आपके पास, डिटरमिनिस्टिक टूल, स्टेट मैनेजमेंट, और सही आर्किटेक्चर की मदद से, एक असली एआई सिस्टम बनाने का विकल्प है.
4. अपना पहला एजेंट बनाना
टूल, एलएलएम से कैसे अलग होते हैं
जब किसी एलएलएम से पूछा जाता है कि "इस कोड में कितने फ़ंक्शन हैं?", तो वह पैटर्न मैचिंग और अनुमान लगाने की तकनीक का इस्तेमाल करता है. Python के ast.parse() को कॉल करने वाले टूल का इस्तेमाल करने पर, यह टूल असल सिंटैक्स ट्री को पार्स करता है. इसमें अनुमान लगाने की ज़रूरत नहीं होती और हर बार एक जैसा नतीजा मिलता है.
इस सेक्शन में एक ऐसा टूल बनाया गया है जो कोड स्ट्रक्चर का विश्लेषण करता है. इसके बाद, इसे ऐसे एजेंट से कनेक्ट किया जाता है जिसे यह पता होता है कि इसे कब लागू करना है.
पहला चरण: स्कैफ़ोल्ड को समझना
आइए, उस स्ट्रक्चर की जांच करें जिसे आपको भरना है.
👉 खोलें
code_review_assistant/tools.py
आपको analyze_code_structure फ़ंक्शन दिखेगा. इसमें प्लेसहोल्डर के तौर पर टिप्पणियां होंगी. इनसे पता चलेगा कि आपको कोड कहां जोड़ना है. फ़ंक्शन का बुनियादी स्ट्रक्चर पहले से मौजूद है. आपको इसे चरण दर चरण बेहतर बनाना होगा.
दूसरा चरण: स्टेट स्टोरेज जोड़ना
स्टेट स्टोरेज की मदद से, पाइपलाइन में मौजूद अन्य एजेंट आपके टूल के नतीजों को ऐक्सेस कर सकते हैं. इसके लिए, उन्हें विश्लेषण को फिर से चलाने की ज़रूरत नहीं होती.
👉 ढूंढें:
# MODULE_4_STEP_2_ADD_STATE_STORAGE
👉 उस एक लाइन को इससे बदलें:
# Store code and analysis for other agents to access
tool_context.state[StateKeys.CODE_TO_REVIEW] = code
tool_context.state[StateKeys.CODE_ANALYSIS] = analysis
tool_context.state[StateKeys.CODE_LINE_COUNT] = len(code.splitlines())
तीसरा चरण: थ्रेड पूल के साथ एसिंक पार्सिंग जोड़ना
हमारे टूल को अन्य कार्रवाइयों को ब्लॉक किए बिना, एएसटी को पार्स करने की ज़रूरत होती है. आइए, थ्रेड पूल के साथ एसिंक एक्ज़ीक्यूशन जोड़ते हैं.
👉 ढूंढें:
# MODULE_4_STEP_3_ADD_ASYNC
👉 उस एक लाइन को इससे बदलें:
# Parse in thread pool to avoid blocking the event loop
loop = asyncio.get_event_loop()
with ThreadPoolExecutor() as executor:
tree = await loop.run_in_executor(executor, ast.parse, code)
चौथा चरण: पूरी जानकारी निकालना
अब हम क्लास, इंपोर्ट, और मेट्रिक की जानकारी निकालेंगे. यह जानकारी, कोड की पूरी समीक्षा करने के लिए ज़रूरी है.
👉 ढूंढें:
# MODULE_4_STEP_4_EXTRACT_DETAILS
👉 उस एक लाइन को इससे बदलें:
# Extract comprehensive structural information
analysis = await loop.run_in_executor(
executor, _extract_code_structure, tree, code
)
👉 पुष्टि करें: फ़ंक्शन
analyze_code_structure
में
tools.py
में एक सेंट्रल बॉडी होती है, जो इस तरह दिखती है:
# Parse in thread pool to avoid blocking the event loop
loop = asyncio.get_event_loop()
with ThreadPoolExecutor() as executor:
tree = await loop.run_in_executor(executor, ast.parse, code)
# Extract comprehensive structural information
analysis = await loop.run_in_executor(
executor, _extract_code_structure, tree, code
)
# Store code and analysis for other agents to access
tool_context.state[StateKeys.CODE_TO_REVIEW] = code
tool_context.state[StateKeys.CODE_ANALYSIS] = analysis
tool_context.state[StateKeys.CODE_LINE_COUNT] = len(code.splitlines())
👉 अब सबसे नीचे तक स्क्रोल करें
tools.py
और ढूंढें:
# MODULE_4_STEP_4_HELPER_FUNCTION
👉 उस एक लाइन को पूरे हेल्पर फ़ंक्शन से बदलें:
def _extract_code_structure(tree: ast.AST, code: str) -> Dict[str, Any]:
"""
Helper function to extract structural information from AST.
Runs in thread pool for CPU-bound work.
"""
functions = []
classes = []
imports = []
docstrings = []
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef):
func_info = {
'name': node.name,
'args': [arg.arg for arg in node.args.args],
'lineno': node.lineno,
'has_docstring': ast.get_docstring(node) is not None,
'is_async': isinstance(node, ast.AsyncFunctionDef),
'decorators': [d.id for d in node.decorator_list
if isinstance(d, ast.Name)]
}
functions.append(func_info)
if func_info['has_docstring']:
docstrings.append(f"{node.name}: {ast.get_docstring(node)[:50]}...")
elif isinstance(node, ast.ClassDef):
methods = []
for item in node.body:
if isinstance(item, ast.FunctionDef):
methods.append(item.name)
class_info = {
'name': node.name,
'lineno': node.lineno,
'methods': methods,
'has_docstring': ast.get_docstring(node) is not None,
'base_classes': [base.id for base in node.bases
if isinstance(base, ast.Name)]
}
classes.append(class_info)
elif isinstance(node, ast.Import):
for alias in node.names:
imports.append({
'module': alias.name,
'alias': alias.asname,
'type': 'import'
})
elif isinstance(node, ast.ImportFrom):
imports.append({
'module': node.module or '',
'names': [alias.name for alias in node.names],
'type': 'from_import',
'level': node.level
})
return {
'functions': functions,
'classes': classes,
'imports': imports,
'docstrings': docstrings,
'metrics': {
'line_count': len(code.splitlines()),
'function_count': len(functions),
'class_count': len(classes),
'import_count': len(imports),
'has_main': any(f['name'] == 'main' for f in functions),
'has_if_main': '__main__' in code,
'avg_function_length': _calculate_avg_function_length(tree)
}
}
def _calculate_avg_function_length(tree: ast.AST) -> float:
"""Calculate average function length in lines."""
function_lengths = []
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef):
if hasattr(node, 'end_lineno') and hasattr(node, 'lineno'):
length = node.end_lineno - node.lineno + 1
function_lengths.append(length)
if function_lengths:
return sum(function_lengths) / len(function_lengths)
return 0.0
पाँचवाँ चरण: एजेंट से कनेक्ट करना
अब हम टूल को ऐसे एजेंट से जोड़ते हैं जिसे पता है कि इसका इस्तेमाल कब करना है और इसके नतीजों को कैसे समझना है.
👉 खोलें
code_review_assistant/sub_agents/review_pipeline/code_analyzer.py
👉 ढूंढें:
# MODULE_4_STEP_5_CREATE_AGENT
👉 उस एक लाइन को पूरे प्रोडक्शन एजेंट से बदलें:
code_analyzer_agent = Agent(
name="CodeAnalyzer",
model=config.worker_model,
description="Analyzes Python code structure and identifies components",
instruction="""You are a code analysis specialist responsible for understanding code structure.
Your task:
1. Take the code submitted by the user (it will be provided in the user message)
2. Use the analyze_code_structure tool to parse and analyze it
3. Pass the EXACT code to your tool - do not modify, fix, or "improve" it
4. Identify all functions, classes, imports, and structural patterns
5. Note any syntax errors or structural issues
6. Store the analysis in state for other agents to use
CRITICAL:
- Pass the code EXACTLY as provided to the analyze_code_structure tool
- Do not fix syntax errors, even if obvious
- Do not add missing imports or fix indentation
- The goal is to analyze what IS there, not what SHOULD be there
When calling the tool, pass the code as a string to the 'code' parameter.
If the analysis fails due to syntax errors, clearly report the error location and type.
Provide a clear summary including:
- Number of functions and classes found
- Key structural observations
- Any syntax errors or issues detected
- Overall code organization assessment""",
tools=[FunctionTool(func=analyze_code_structure)],
output_key="structure_analysis_summary"
)
कोड की जांच करने वाले टूल को आज़माना
अब पुष्टि करें कि आपका विश्लेषण करने वाला टूल सही तरीके से काम कर रहा है.
👉 टेस्ट स्क्रिप्ट चलाएं:
python tests/test_code_analyzer.py
टेस्ट स्क्रिप्ट, python-dotenv का इस्तेमाल करके आपकी .env फ़ाइल से कॉन्फ़िगरेशन अपने-आप लोड करती है. इसलिए, एनवायरमेंट वैरिएबल को मैन्युअल तरीके से सेट अप करने की ज़रूरत नहीं होती.
अनुमानित आउटपुट:
INFO:code_review_assistant.config:Code Review Assistant Configuration Loaded:
INFO:code_review_assistant.config: - GCP Project: your-project-id
INFO:code_review_assistant.config: - Artifact Bucket: gs://your-project-artifacts
INFO:code_review_assistant.config: - Models: worker=gemini-2.5-flash, critic=gemini-2.5-pro
Testing code analyzer...
INFO:code_review_assistant.tools:Tool: Analysis complete - 2 functions, 1 classes
=== Analyzer Response ===
The analysis of the provided code shows the following:
* **Functions Found:** 2
* `add(a, b)`: A global function at line 2.
* `multiply(self, x, y)`: A method within the `Calculator` class.
* **Classes Found:** 1
* `Calculator`: A class defined at line 5. Contains one method, `multiply`.
* **Imports:** 0
* **Structural Patterns:** The code defines one global function and one class
with a single method. Both are simple, each with a single return statement.
* **Syntax Errors/Issues:** No syntax errors detected.
* **Overall Code Organization:** The code is well-organized for its small size,
clearly defining a function and a class with a method.
अभी क्या हुआ:
- टेस्ट स्क्रिप्ट ने आपके
.envकॉन्फ़िगरेशन को अपने-आप लोड कर दिया है analyze_code_structure()टूल ने Python के AST का इस्तेमाल करके कोड को पार्स किया_extract_code_structure()हेल्पर ने फ़ंक्शन, क्लास, और मेट्रिक निकाली हैंStateKeysकॉन्स्टेंट का इस्तेमाल करके, नतीजों को सेशन की मौजूदा स्थिति में सेव किया गया था- कोड की जांच करने वाले एजेंट ने नतीजों को समझा और उनकी खास जानकारी दी
समस्या हल करना:
- "No module named ‘code_review_assistant'": प्रोजेक्ट रूट से
pip install -e .चलाएं - "Missing key inputs argument": पुष्टि करें कि आपके
.envमेंGOOGLE_CLOUD_PROJECT,GOOGLE_CLOUD_LOCATION, औरGOOGLE_GENAI_USE_VERTEXAI=trueमौजूद हैं
आपने क्या बनाया है
अब आपके पास प्रोडक्शन के लिए तैयार कोड का विश्लेषण करने वाला टूल है. यह टूल:
✅ असल Python AST को पार्स करता है - यह पैटर्न मैचिंग नहीं है, बल्कि डिटरमिनिस्टिक है
✅ नतीजों को स्टेट में सेव करता है - अन्य एजेंट विश्लेषण को ऐक्सेस कर सकते हैं
✅ एसिंक्रोनस तरीके से काम करता है - यह अन्य टूल को ब्लॉक नहीं करता है
✅ पूरी जानकारी निकालता है - फ़ंक्शन, क्लास, इंपोर्ट, मेट्रिक
✅ गड़बड़ियों को आसानी से ठीक करता है - लाइन नंबर के साथ सिंटैक्स से जुड़ी गड़बड़ियों की जानकारी देता है
✅ किसी एजेंट से कनेक्ट होता है - एलएलएम को पता होता है कि इसका इस्तेमाल कब और कैसे करना है
मास्टर किए गए मुख्य कॉन्सेप्ट
टूल बनाम एजेंट:
- टूल, तय किए गए तरीके से काम करते हैं (एएसटी पार्सिंग)
- एजेंट यह तय करते हैं कि टूल का इस्तेमाल कब करना है और नतीजों की व्याख्या कैसे करनी है
वैल्यू बनाम स्थिति:
- जवाब: एलएलएम को तुरंत क्या दिखता है
- स्टेट: अन्य एजेंट के लिए क्या बना रहता है
State Keys Constants:
- मल्टी-एजेंट सिस्टम में टाइपिंग की गलतियों को रोकना
- एजेंट के बीच कानूनी समझौते के तौर पर काम करना
- एजेंट के डेटा शेयर करने पर यह ज़रूरी है
एसिंक + थ्रेड पूल:
async defटूल को एक्ज़ीक्यूशन रोकने की अनुमति देता है- थ्रेड पूल, सीपीयू से जुड़े काम को बैकग्राउंड में करते हैं
- ये दोनों मिलकर, इवेंट लूप को रिस्पॉन्सिव बनाए रखते हैं
हेल्पर फ़ंक्शन:
- सिंक हेल्पर को एसिंक टूल से अलग करना
- इससे कोड की जांच करना और उसे फिर से इस्तेमाल करना आसान हो जाता है
Agent Instructions:
- ज़्यादा जानकारी देने वाले निर्देशों से, एलएलएम से होने वाली सामान्य गड़बड़ियों को रोका जा सकता है
- साफ़ तौर पर बताया गया है कि क्या नहीं करना है (कोड ठीक न करें)
- वर्कफ़्लो के चरणों को एक जैसा रखने के लिए, उन्हें साफ़ तौर पर बताएं
आगे क्या करना है
पांचवें मॉड्यूल में, ये चीज़ें जोड़ी जाएंगी:
- स्टाइल चेक करने वाला टूल, जो स्थिति से कोड को पढ़ता है
- टेस्ट रनर, जो असल में टेस्ट को एक्ज़ीक्यूट करता है
- सुझाव, शिकायत या राय को एक साथ रखने वाला सिंथेसाइज़र, जो सभी विश्लेषणों को एक साथ रखता है
आपको यह दिखेगा कि स्टेट, सीक्वेंशियल पाइपलाइन से कैसे फ़्लो होती है. साथ ही, यह भी दिखेगा कि जब कई एजेंट एक ही डेटा को पढ़ते और लिखते हैं, तो कॉन्स्टेंट पैटर्न क्यों ज़रूरी होता है.
5. पाइपलाइन बनाना: एक साथ काम करने वाले कई एजेंट
परिचय
मॉड्यूल 4 में, आपने एक ऐसा एजेंट बनाया था जो कोड के स्ट्रक्चर का विश्लेषण करता है. हालांकि, कोड की पूरी तरह से समीक्षा करने के लिए, सिर्फ़ पार्सिंग से ज़्यादा चीज़ों की ज़रूरत होती है. जैसे, स्टाइल की जांच करना, टेस्ट लागू करना, और बेहतर तरीके से सुझाव देना.
यह मॉड्यूल, चार एजेंट की पाइपलाइन बनाता है. ये एजेंट एक के बाद एक काम करते हैं. हर एजेंट, खास विश्लेषण में योगदान देता है:
- कोड विश्लेषक (मॉड्यूल 4 से) - स्ट्रक्चर पार्स करता है
- स्टाइल की जांच करने वाला टूल - स्टाइल से जुड़े उल्लंघन की पहचान करता है
- टेस्ट रनर - टेस्ट को लागू करता है और उनकी पुष्टि करता है
- सुझाव, शिकायत या राय को व्यवस्थित करने वाला टूल - यह टूल, सभी सुझावों, शिकायतों या राय को एक साथ मिलाकर, कार्रवाई करने लायक सुझाव बनाता है
मुख्य कॉन्सेप्ट: स्टेट को कम्यूनिकेशन चैनल के तौर पर इस्तेमाल करना. हर एजेंट, पिछले एजेंट के लिखे गए जवाब को पढ़ता है. इसके बाद, वह अपने हिसाब से विश्लेषण करके जवाब में जानकारी जोड़ता है और उसे अगले एजेंट को भेज देता है. जब कई एजेंट डेटा शेयर करते हैं, तब मॉड्यूल 4 से कॉन्स्टेंट पैटर्न का इस्तेमाल करना ज़रूरी हो जाता है.
आपको क्या बनाना है, इसकी झलक: गड़बड़ कोड सबमिट करें → चार एजेंट के ज़रिए स्टेट फ़्लो देखें → पिछले पैटर्न के आधार पर, अपनी पसंद के मुताबिक सुझावों वाली पूरी रिपोर्ट पाएं.
पहला चरण: स्टाइल चेक करने वाले टूल और एजेंट को जोड़ना
स्टाइल चेक करने वाला टूल, pycodestyle का इस्तेमाल करके PEP 8 के उल्लंघन की पहचान करता है. यह एक डिटरमिनिस्टिक लिंटर है, न कि एलएलएम पर आधारित इंटरप्रिटेशन.
स्टाइल की जांच करने वाला टूल जोड़ना
👉 खोलें
code_review_assistant/tools.py
👉 ढूंढें:
# MODULE_5_STEP_1_STYLE_CHECKER_TOOL
👉 उस एक लाइन को इससे बदलें:
async def check_code_style(code: str, tool_context: ToolContext) -> Dict[str, Any]:
"""
Checks code style compliance using pycodestyle (PEP 8).
Args:
code: Python source code to check (or will retrieve from state)
tool_context: ADK tool context
Returns:
Dictionary containing style score and issues
"""
logger.info("Tool: Checking code style...")
try:
# Retrieve code from state if not provided
if not code:
code = tool_context.state.get(StateKeys.CODE_TO_REVIEW, '')
if not code:
return {
"status": "error",
"message": "No code provided or found in state"
}
# Run style check in thread pool
loop = asyncio.get_event_loop()
with ThreadPoolExecutor() as executor:
result = await loop.run_in_executor(
executor, _perform_style_check, code
)
# Store results in state
tool_context.state[StateKeys.STYLE_SCORE] = result['score']
tool_context.state[StateKeys.STYLE_ISSUES] = result['issues']
tool_context.state[StateKeys.STYLE_ISSUE_COUNT] = result['issue_count']
logger.info(f"Tool: Style check complete - Score: {result['score']}/100, "
f"Issues: {result['issue_count']}")
return result
except Exception as e:
error_msg = f"Style check failed: {str(e)}"
logger.error(f"Tool: {error_msg}", exc_info=True)
# Set default values on error
tool_context.state[StateKeys.STYLE_SCORE] = 0
tool_context.state[StateKeys.STYLE_ISSUES] = []
return {
"status": "error",
"message": error_msg,
"score": 0
}
👉 अब फ़ाइल के आखिर तक स्क्रोल करें और यह ढूंढें:
# MODULE_5_STEP_1_STYLE_HELPERS
👉 उस एक लाइन को हेल्पर फ़ंक्शन से बदलें:
def _perform_style_check(code: str) -> Dict[str, Any]:
"""Helper to perform style check in thread pool."""
import io
import sys
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as tmp:
tmp.write(code)
tmp_path = tmp.name
try:
# Capture stdout to get pycodestyle output
old_stdout = sys.stdout
sys.stdout = captured_output = io.StringIO()
style_guide = pycodestyle.StyleGuide(
quiet=False, # We want output
max_line_length=100,
ignore=['E501', 'W503']
)
result = style_guide.check_files([tmp_path])
# Restore stdout
sys.stdout = old_stdout
# Parse captured output
output = captured_output.getvalue()
issues = []
for line in output.strip().split('\n'):
if line and ':' in line:
parts = line.split(':', 4)
if len(parts) >= 4:
try:
issues.append({
'line': int(parts[1]),
'column': int(parts[2]),
'code': parts[3].split()[0] if len(parts) > 3 else 'E000',
'message': parts[3].strip() if len(parts) > 3 else 'Unknown error'
})
except (ValueError, IndexError):
pass
# Add naming convention checks
try:
tree = ast.parse(code)
naming_issues = _check_naming_conventions(tree)
issues.extend(naming_issues)
except SyntaxError:
pass # Syntax errors will be caught elsewhere
# Calculate weighted score
score = _calculate_style_score(issues)
return {
"status": "success",
"score": score,
"issue_count": len(issues),
"issues": issues[:10], # First 10 issues
"summary": f"Style score: {score}/100 with {len(issues)} violations"
}
finally:
if os.path.exists(tmp_path):
os.unlink(tmp_path)
def _check_naming_conventions(tree: ast.AST) -> List[Dict[str, Any]]:
"""Check PEP 8 naming conventions."""
naming_issues = []
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef):
# Skip private/protected methods and __main__
if not node.name.startswith('_') and node.name != node.name.lower():
naming_issues.append({
'line': node.lineno,
'column': node.col_offset,
'code': 'N802',
'message': f"N802 function name '{node.name}' should be lowercase"
})
elif isinstance(node, ast.ClassDef):
# Check if class name follows CapWords convention
if not node.name[0].isupper() or '_' in node.name:
naming_issues.append({
'line': node.lineno,
'column': node.col_offset,
'code': 'N801',
'message': f"N801 class name '{node.name}' should use CapWords convention"
})
return naming_issues
def _calculate_style_score(issues: List[Dict[str, Any]]) -> int:
"""Calculate weighted style score based on violation severity."""
if not issues:
return 100
# Define weights by error type
weights = {
'E1': 10, # Indentation errors
'E2': 3, # Whitespace errors
'E3': 5, # Blank line errors
'E4': 8, # Import errors
'E5': 5, # Line length
'E7': 7, # Statement errors
'E9': 10, # Syntax errors
'W2': 2, # Whitespace warnings
'W3': 2, # Blank line warnings
'W5': 3, # Line break warnings
'N8': 7, # Naming conventions
}
total_deduction = 0
for issue in issues:
code_prefix = issue['code'][:2] if len(issue['code']) >= 2 else 'E2'
weight = weights.get(code_prefix, 3)
total_deduction += weight
# Cap at 100 points deduction
return max(0, 100 - min(total_deduction, 100))
स्टाइल की जांच करने वाले एजेंट को जोड़ना
👉 खोलें
code_review_assistant/sub_agents/review_pipeline/style_checker.py
👉 ढूंढें:
# MODULE_5_STEP_1_INSTRUCTION_PROVIDER
👉 उस एक लाइन को इससे बदलें:
async def style_checker_instruction_provider(context: ReadonlyContext) -> str:
"""Dynamic instruction provider that injects state variables."""
template = """You are a code style expert focused on PEP 8 compliance.
Your task:
1. Use the check_code_style tool to validate PEP 8 compliance
2. The tool will retrieve the ORIGINAL code from state automatically
3. Report violations exactly as found
4. Present the results clearly and confidently
CRITICAL:
- The tool checks the code EXACTLY as provided by the user
- Do not suggest the code was modified or fixed
- Report actual violations found in the original code
- If there are style issues, they should be reported honestly
Call the check_code_style tool with an empty string for the code parameter,
as the tool will retrieve the code from state automatically.
When presenting results based on what the tool returns:
- State the exact score from the tool results
- If score >= 90: "Excellent style compliance!"
- If score 70-89: "Good style with minor improvements needed"
- If score 50-69: "Style needs attention"
- If score < 50: "Significant style improvements needed"
List the specific violations found (the tool will provide these):
- Show line numbers, error codes, and messages
- Focus on the top 10 most important issues
Previous analysis: {structure_analysis_summary}
Format your response as:
## Style Analysis Results
- Style Score: [exact score]/100
- Total Issues: [count]
- Assessment: [your assessment based on score]
## Top Style Issues
[List issues with line numbers and descriptions]
## Recommendations
[Specific fixes for the most critical issues]"""
return await instructions_utils.inject_session_state(template, context)
👉 ढूंढें:
# MODULE_5_STEP_1_STYLE_CHECKER_AGENT
👉 उस एक लाइन को इससे बदलें:
style_checker_agent = Agent(
name="StyleChecker",
model=config.worker_model,
description="Checks Python code style against PEP 8 guidelines",
instruction=style_checker_instruction_provider,
tools=[FunctionTool(func=check_code_style)],
output_key="style_check_summary"
)
दूसरा चरण: Test Runner Agent जोड़ना
टेस्ट रनर, पूरी तरह से टेस्ट जनरेट करता है और उन्हें पहले से मौजूद कोड एक्ज़ीक्यूटर का इस्तेमाल करके लागू करता है.
👉 खोलें
code_review_assistant/sub_agents/review_pipeline/test_runner.py
👉 ढूंढें:
# MODULE_5_STEP_2_INSTRUCTION_PROVIDER
👉 उस एक लाइन को इससे बदलें:
async def test_runner_instruction_provider(context: ReadonlyContext) -> str:
"""Dynamic instruction provider that injects the code_to_review directly."""
template = """You are a testing specialist who creates and runs tests for Python code.
THE CODE TO TEST IS:
{code_to_review}
YOUR TASK:
1. Understand what the function appears to do based on its name and structure
2. Generate comprehensive tests (15-20 test cases)
3. Execute the tests using your code executor
4. Analyze results to identify bugs vs expected behavior
5. Output a detailed JSON analysis
TESTING METHODOLOGY:
- Test with the most natural interpretation first
- When something fails, determine if it's a bug or unusual design
- Test edge cases, boundaries, and error scenarios
- Document any surprising behavior
Execute your tests and output ONLY valid JSON with this structure:
- "test_summary": object with "total_tests_run", "tests_passed", "tests_failed", "tests_with_errors", "critical_issues_found"
- "critical_issues": array of objects, each with "type", "description", "example_input", "expected_behavior", "actual_behavior", "severity"
- "test_categories": object with "basic_functionality", "edge_cases", "error_handling" (each containing "passed", "failed", "errors" counts)
- "function_behavior": object with "apparent_purpose", "actual_interface", "unexpected_requirements"
- "verdict": object with "status" (WORKING/BUGGY/BROKEN), "confidence" (high/medium/low), "recommendation"
Do NOT output the test code itself, only the JSON analysis."""
return await instructions_utils.inject_session_state(template, context)
👉 ढूंढें:
# MODULE_5_STEP_2_TEST_RUNNER_AGENT
👉 उस एक लाइन को इससे बदलें:
test_runner_agent = Agent(
name="TestRunner",
model=config.critic_model,
description="Generates and runs tests for Python code using safe code execution",
instruction=test_runner_instruction_provider,
code_executor=BuiltInCodeExecutor(),
output_key="test_execution_summary"
)
तीसरा चरण: अलग-अलग सेशन के लिए, मेमोरी को समझना
फ़ीडबैक सिंथेसाइज़र बनाने से पहले, आपको स्टेट और मेमोरी के बीच का अंतर समझना होगा. ये दोनों, अलग-अलग मकसद के लिए स्टोरेज की दो अलग-अलग सुविधाएं हैं.
स्टेट बनाम मेमोरी: मुख्य अंतर
आइए, कोड की समीक्षा के एक उदाहरण से इसे समझते हैं:
राज्य (सिर्फ़ मौजूदा सेशन के लिए):
# Data from THIS review session
tool_context.state[StateKeys.STYLE_ISSUES] = [
{"line": 5, "code": "E231", "message": "missing whitespace"},
{"line": 12, "code": "E701", "message": "multiple statements"}
]
- स्कोप: सिर्फ़ यह बातचीत
- मकसद: मौजूदा पाइपलाइन में एजेंट के बीच डेटा ट्रांसफ़र करना
- Lives in:
Sessionऑब्जेक्ट - लाइफ़टाइम: सेशन खत्म होने पर हटा दी जाती है
मेमोरी (पिछले सभी सेशन):
# Learned from 50 previous reviews
"User frequently forgets docstrings on helper functions"
"User tends to write long functions (avg 45 lines)"
"User improved error handling after feedback in session #23"
- स्कोप: इस उपयोगकर्ता के सभी पिछले सेशन
- मकसद: पैटर्न के बारे में जानना, लोगों की दिलचस्पी के हिसाब से सुझाव देना
- यहां रहता/रहती है:
MemoryService - लाइफ़टाइम: यह कुकी सभी सेशन में बनी रहती है और इसे खोजा जा सकता है
सुझाव/राय देने या शिकायत करने के लिए, दोनों की ज़रूरत क्यों होती है:
मान लें कि सिंथेसाइज़र फ़ीडबैक जनरेट कर रहा है:
सिर्फ़ राज्य के हिसाब से समीक्षा करने पर (मौजूदा समीक्षा):
"Function `calculate_total` has no docstring."
सामान्य, मशीन से जुड़ा फ़ीडबैक.
स्टेट + मेमोरी (मौजूदा + पिछले पैटर्न) का इस्तेमाल करके:
"Function `calculate_total` has no docstring. This is the 4th review
where helper functions lacked documentation. Consider adding docstrings
as you write functions, not afterwards - you mentioned in our last
session that you find it easier that way."
समय के साथ, जवाब को ज़्यादा निजी, संदर्भ के हिसाब से, और बेहतर बनाना.
प्रोडक्शन डिप्लॉयमेंट के लिए, आपके पास ये विकल्प हैं:
पहला विकल्प: VertexAiMemoryBankService (ऐडवांस)
- यह सुविधा क्या करती है: एलएलएम की मदद से, बातचीत से काम के तथ्यों को निकालना
- खोज: सिमैंटिक सर्च (सिर्फ़ कीवर्ड नहीं, मतलब भी समझता है)
- मेमोरी मैनेजमेंट: समय के साथ यादों को अपने-आप अपडेट और व्यवस्थित करता है
- ज़रूरी शर्तें: Google Cloud Project + Agent Engine का सेटअप
- इसका इस्तेमाल तब करें, जब: आपको बेहतर, नई, और आपकी पसंद के हिसाब से यादें चाहिए
- उदाहरण: "उपयोगकर्ता को फ़ंक्शनल प्रोग्रामिंग पसंद है" (कोड स्टाइल के बारे में हुई 10 बातचीत से निकाला गया)
दूसरा विकल्प: InMemoryMemoryService + Persistent Sessions का इस्तेमाल जारी रखें
- यह क्या करता है: यह कुकी, कीवर्ड खोज के लिए बातचीत का पूरा इतिहास सेव करती है
- खोज: पिछले सेशन में कीवर्ड मैच करने की बुनियादी सुविधा
- मेमोरी मैनेजमेंट: आपके पास यह कंट्रोल होता है कि क्या सेव किया जाए (
add_session_to_memoryके ज़रिए) - ज़रूरी है: सिर्फ़ एक परसिस्टेंट
SessionService(जैसे किVertexAiSessionServiceयाDatabaseSessionService) - इसका इस्तेमाल तब करें, जब: आपको एलएलएम प्रोसेसिंग के बिना, पिछली बातचीत में आसान तरीके से खोजना हो
- उदाहरण: "docstring" खोजने पर, उस शब्द का ज़िक्र करने वाले सभी सेशन दिखते हैं
'यादें' सेक्शन में फ़ोटो कैसे दिखती हैं
कोड की हर समीक्षा पूरी होने के बाद:
# At the end of a session (typically in your application code)
await memory_service.add_session_to_memory(session)
क्या होता है:
- InMemoryMemoryService: यह कुकी, कीवर्ड खोजने के लिए पूरे सेशन के इवेंट सेव करती है
- VertexAiMemoryBankService: एलएलएम, मुख्य तथ्यों को निकालता है और उन्हें मौजूदा यादों के साथ जोड़ता है
इसके बाद, आने वाले सेशन में ये क्वेरी की जा सकती हैं:
# In a tool, search for relevant past feedback
results = tool_context.search_memory("feedback about docstrings")
चौथा चरण: सुझाव/राय देने या शिकायत करने की सुविधा देने वाले टूल और एजेंट जोड़ना
फ़ीडबैक सिंथेसाइज़र, पाइपलाइन में सबसे बेहतर एजेंट है. यह तीन टूल को मैनेज करता है, डाइनैमिक निर्देशों का इस्तेमाल करता है, और स्थिति, मेमोरी, और आर्टफ़ैक्ट को जोड़ता है.
सिंथेसाइज़र के तीन टूल जोड़ना
👉 खोलें
code_review_assistant/tools.py
👉 ढूंढें:
# MODULE_5_STEP_4_SEARCH_PAST_FEEDBACK
👉 Replace with Tool 1 - Memory Search (production version):
async def search_past_feedback(developer_id: str, tool_context: ToolContext) -> Dict[str, Any]:
"""
Search for past feedback in memory service.
Args:
developer_id: ID of the developer (defaults to "default_user")
tool_context: ADK tool context with potential memory service access
Returns:
Dictionary containing feedback search results
"""
logger.info(f"Tool: Searching for past feedback for developer {developer_id}...")
try:
# Default developer ID if not provided
if not developer_id:
developer_id = tool_context.state.get(StateKeys.USER_ID, 'default_user')
# Check if memory service is available
if hasattr(tool_context, 'search_memory'):
try:
# Perform structured searches
queries = [
f"developer:{developer_id} code review feedback",
f"developer:{developer_id} common issues",
f"developer:{developer_id} improvements"
]
all_feedback = []
patterns = {
'common_issues': [],
'improvements': [],
'strengths': []
}
for query in queries:
search_result = await tool_context.search_memory(query)
if search_result and hasattr(search_result, 'memories'):
for memory in search_result.memories[:5]:
memory_text = memory.text if hasattr(memory, 'text') else str(memory)
all_feedback.append(memory_text)
# Extract patterns
if 'style' in memory_text.lower():
patterns['common_issues'].append('style compliance')
if 'improved' in memory_text.lower():
patterns['improvements'].append('showing improvement')
if 'excellent' in memory_text.lower():
patterns['strengths'].append('consistent quality')
# Store in state
tool_context.state[StateKeys.PAST_FEEDBACK] = all_feedback
tool_context.state[StateKeys.FEEDBACK_PATTERNS] = patterns
logger.info(f"Tool: Found {len(all_feedback)} past feedback items")
return {
"status": "success",
"feedback_found": True,
"count": len(all_feedback),
"summary": " | ".join(all_feedback[:3]) if all_feedback else "No feedback",
"patterns": patterns
}
except Exception as e:
logger.warning(f"Tool: Memory search error: {e}")
# Fallback: Check state for cached feedback
cached_feedback = tool_context.state.get(StateKeys.USER_PAST_FEEDBACK_CACHE, [])
if cached_feedback:
tool_context.state[StateKeys.PAST_FEEDBACK] = cached_feedback
return {
"status": "success",
"feedback_found": True,
"count": len(cached_feedback),
"summary": "Using cached feedback",
"patterns": {}
}
# No feedback found
tool_context.state[StateKeys.PAST_FEEDBACK] = []
logger.info("Tool: No past feedback found")
return {
"status": "success",
"feedback_found": False,
"message": "No past feedback available - this appears to be a first submission",
"patterns": {}
}
except Exception as e:
error_msg = f"Feedback search error: {str(e)}"
logger.error(f"Tool: {error_msg}", exc_info=True)
tool_context.state[StateKeys.PAST_FEEDBACK] = []
return {
"status": "error",
"message": error_msg,
"feedback_found": False
}
👉 ढूंढें:
# MODULE_5_STEP_4_UPDATE_GRADING_PROGRESS
👉 टूल 2 - ग्रेडिंग ट्रैकर (प्रोडक्शन वर्शन) से बदलें:
async def update_grading_progress(tool_context: ToolContext) -> Dict[str, Any]:
"""
Updates grading progress counters and metrics in state.
"""
logger.info("Tool: Updating grading progress...")
try:
current_time = datetime.now().isoformat()
# Build all state changes
state_updates = {}
# Temporary (invocation-level) state
state_updates[StateKeys.TEMP_PROCESSING_TIMESTAMP] = current_time
# Session-level state
attempts = tool_context.state.get(StateKeys.GRADING_ATTEMPTS, 0) + 1
state_updates[StateKeys.GRADING_ATTEMPTS] = attempts
state_updates[StateKeys.LAST_GRADING_TIME] = current_time
# User-level persistent state
lifetime_submissions = tool_context.state.get(StateKeys.USER_TOTAL_SUBMISSIONS, 0) + 1
state_updates[StateKeys.USER_TOTAL_SUBMISSIONS] = lifetime_submissions
state_updates[StateKeys.USER_LAST_SUBMISSION_TIME] = current_time
# Calculate improvement metrics
current_style_score = tool_context.state.get(StateKeys.STYLE_SCORE, 0)
last_style_score = tool_context.state.get(StateKeys.USER_LAST_STYLE_SCORE, 0)
score_improvement = current_style_score - last_style_score
state_updates[StateKeys.USER_LAST_STYLE_SCORE] = current_style_score
state_updates[StateKeys.SCORE_IMPROVEMENT] = score_improvement
# Track test results if available
test_results = tool_context.state.get(StateKeys.TEST_EXECUTION_SUMMARY, {})
# Parse if it's a string
if isinstance(test_results, str):
try:
test_results = json.loads(test_results)
except:
test_results = {}
if test_results and test_results.get('test_summary', {}).get('total_tests_run', 0) > 0:
summary = test_results['test_summary']
total = summary.get('total_tests_run', 0)
passed = summary.get('tests_passed', 0)
if total > 0:
pass_rate = (passed / total) * 100
state_updates[StateKeys.USER_LAST_TEST_PASS_RATE] = pass_rate
# Apply all updates atomically
for key, value in state_updates.items():
tool_context.state[key] = value
logger.info(f"Tool: Progress updated - Attempt #{attempts}, "
f"Lifetime: {lifetime_submissions}")
return {
"status": "success",
"session_attempts": attempts,
"lifetime_submissions": lifetime_submissions,
"timestamp": current_time,
"improvement": {
"style_score_change": score_improvement,
"direction": "improved" if score_improvement > 0 else "declined"
},
"summary": f"Attempt #{attempts} recorded, {lifetime_submissions} total submissions"
}
except Exception as e:
error_msg = f"Progress update error: {str(e)}"
logger.error(f"Tool: {error_msg}", exc_info=True)
return {
"status": "error",
"message": error_msg
}
👉 ढूंढें:
# MODULE_5_STEP_4_SAVE_GRADING_REPORT
👉 Tool 3 - Artifact Saver (production version) से बदलें:
async def save_grading_report(feedback_text: str, tool_context: ToolContext) -> Dict[str, Any]:
"""
Saves a detailed grading report as an artifact.
Args:
feedback_text: The feedback text to include in the report
tool_context: ADK tool context for state management
Returns:
Dictionary containing save status and details
"""
logger.info("Tool: Saving grading report...")
try:
# Gather all relevant data from state
code = tool_context.state.get(StateKeys.CODE_TO_REVIEW, '')
analysis = tool_context.state.get(StateKeys.CODE_ANALYSIS, {})
style_score = tool_context.state.get(StateKeys.STYLE_SCORE, 0)
style_issues = tool_context.state.get(StateKeys.STYLE_ISSUES, [])
# Get test results
test_results = tool_context.state.get(StateKeys.TEST_EXECUTION_SUMMARY, {})
# Parse if it's a string
if isinstance(test_results, str):
try:
test_results = json.loads(test_results)
except:
test_results = {}
timestamp = datetime.now().isoformat()
# Create comprehensive report dictionary
report = {
'timestamp': timestamp,
'grading_attempt': tool_context.state.get(StateKeys.GRADING_ATTEMPTS, 1),
'code': {
'content': code,
'line_count': len(code.splitlines()),
'hash': hashlib.md5(code.encode()).hexdigest()
},
'analysis': analysis,
'style': {
'score': style_score,
'issues': style_issues[:5] # First 5 issues
},
'tests': test_results,
'feedback': feedback_text,
'improvements': {
'score_change': tool_context.state.get(StateKeys.SCORE_IMPROVEMENT, 0),
'from_last_score': tool_context.state.get(StateKeys.USER_LAST_STYLE_SCORE, 0)
}
}
# Convert report to JSON string
report_json = json.dumps(report, indent=2)
report_part = types.Part.from_text(text=report_json)
# Try to save as artifact if the service is available
if hasattr(tool_context, 'save_artifact'):
try:
# Generate filename with timestamp (replace colons for filesystem compatibility)
filename = f"grading_report_{timestamp.replace(':', '-')}.json"
# Save the main report
version = await tool_context.save_artifact(filename, report_part)
# Also save a "latest" version for easy access
await tool_context.save_artifact("latest_grading_report.json", report_part)
logger.info(f"Tool: Report saved as {filename} (version {version})")
# Store report in state as well for redundancy
tool_context.state[StateKeys.USER_LAST_GRADING_REPORT] = report
return {
"status": "success",
"artifact_saved": True,
"filename": filename,
"version": str(version),
"size": len(report_json),
"summary": f"Report saved as {filename}"
}
except Exception as artifact_error:
logger.warning(f"Artifact service error: {artifact_error}, falling back to state storage")
# Continue to fallback below
# Fallback: Store in state if artifact service is not available or failed
tool_context.state[StateKeys.USER_LAST_GRADING_REPORT] = report
logger.info("Tool: Report saved to state (artifact service not available)")
return {
"status": "success",
"artifact_saved": False,
"message": "Report saved to state only",
"size": len(report_json),
"summary": "Report saved to session state"
}
except Exception as e:
error_msg = f"Report save error: {str(e)}"
logger.error(f"Tool: {error_msg}", exc_info=True)
# Still try to save minimal data to state
try:
tool_context.state[StateKeys.USER_LAST_GRADING_REPORT] = {
'error': error_msg,
'feedback': feedback_text,
'timestamp': datetime.now().isoformat()
}
except:
pass
return {
"status": "error",
"message": error_msg,
"artifact_saved": False,
"summary": f"Failed to save report: {error_msg}"
}
सिंथसाइज़र एजेंट बनाना
👉 खोलें
code_review_assistant/sub_agents/review_pipeline/feedback_synthesizer.py
👉 ढूंढें:
# MODULE_5_STEP_4_INSTRUCTION_PROVIDER
👉 प्रोडक्शन के निर्देश देने वाली कंपनी का नाम डालें:
async def feedback_instruction_provider(context: ReadonlyContext) -> str:
"""Dynamic instruction provider that injects state variables."""
template = """You are an expert code reviewer and mentor providing constructive, educational feedback.
CONTEXT FROM PREVIOUS AGENTS:
- Structure analysis summary: {structure_analysis_summary}
- Style check summary: {style_check_summary}
- Test execution summary: {test_execution_summary}
YOUR TASK requires these steps IN ORDER:
1. Call search_past_feedback tool with developer_id="default_user"
2. Call update_grading_progress tool with no parameters
3. Carefully analyze the test results to understand what really happened
4. Generate comprehensive feedback following the structure below
5. Call save_grading_report tool with the feedback_text parameter
6. Return the feedback as your final output
CRITICAL - Understanding Test Results:
The test_execution_summary contains structured JSON. Parse it carefully:
- tests_passed = Code worked correctly
- tests_failed = Code produced wrong output
- tests_with_errors = Code crashed
- critical_issues = Fundamental problems with the code
If critical_issues array contains items, these are serious bugs that need fixing.
Do NOT count discovering bugs as test successes.
FEEDBACK STRUCTURE TO FOLLOW:
## 📊 Summary
Provide an honest assessment. Be encouraging but truthful about problems found.
## ✅ Strengths
List 2-3 things done well, referencing specific code elements.
## 📈 Code Quality Analysis
### Structure & Organization
Comment on code organization, readability, and documentation.
### Style Compliance
Report the actual style score and any specific issues.
### Test Results
Report the actual test results accurately:
- If critical_issues exist, report them as bugs to fix
- Be clear: "X tests passed, Y critical issues were found"
- List each critical issue
- Don't hide or minimize problems
## 💡 Recommendations for Improvement
Based on the analysis, provide specific actionable fixes.
If critical issues exist, fixing them is top priority.
## 🎯 Next Steps
Prioritized action list based on severity of issues.
## 💬 Encouragement
End with encouragement while being honest about what needs fixing.
Remember: Complete ALL steps including calling save_grading_report."""
return await instructions_utils.inject_session_state(template, context)
👉 ढूंढें:
# MODULE_5_STEP_4_SYNTHESIZER_AGENT
👉 इससे बदलें:
feedback_synthesizer_agent = Agent(
name="FeedbackSynthesizer",
model=config.critic_model,
description="Synthesizes all analysis into constructive, personalized feedback",
instruction=feedback_instruction_provider,
tools=[
FunctionTool(func=search_past_feedback),
FunctionTool(func=update_grading_progress),
FunctionTool(func=save_grading_report)
],
output_key="final_feedback"
)
पांचवां चरण: पाइपलाइन को वायर करना
अब इन चारों एजेंट को एक क्रमवार पाइपलाइन में कनेक्ट करें और रूट एजेंट बनाएं.
👉 खोलें
code_review_assistant/agent.py
👉 फ़ाइल में सबसे ऊपर (मौजूदा इंपोर्ट के बाद), ज़रूरी इंपोर्ट जोड़ें:
from google.adk.agents import Agent, SequentialAgent
from code_review_assistant.sub_agents.review_pipeline.code_analyzer import code_analyzer_agent
from code_review_assistant.sub_agents.review_pipeline.style_checker import style_checker_agent
from code_review_assistant.sub_agents.review_pipeline.test_runner import test_runner_agent
from code_review_assistant.sub_agents.review_pipeline.feedback_synthesizer import feedback_synthesizer_agent
अब आपकी फ़ाइल ऐसी दिखनी चाहिए:
"""
Main agent orchestration for the Code Review Assistant.
"""
from google.adk.agents import Agent, SequentialAgent
from .config import config
from code_review_assistant.sub_agents.review_pipeline.code_analyzer import code_analyzer_agent
from code_review_assistant.sub_agents.review_pipeline.style_checker import style_checker_agent
from code_review_assistant.sub_agents.review_pipeline.test_runner import test_runner_agent
from code_review_assistant.sub_agents.review_pipeline.feedback_synthesizer import feedback_synthesizer_agent
# MODULE_5_STEP_5_CREATE_PIPELINE
# MODULE_6_STEP_5_CREATE_FIX_LOOP
# MODULE_6_STEP_5_UPDATE_ROOT_AGENT
👉 ढूंढें:
# MODULE_5_STEP_5_CREATE_PIPELINE
👉 उस लाइन को इससे बदलें:
# Create sequential pipeline
code_review_pipeline = SequentialAgent(
name="CodeReviewPipeline",
description="Complete code review pipeline with analysis, testing, and feedback",
sub_agents=[
code_analyzer_agent,
style_checker_agent,
test_runner_agent,
feedback_synthesizer_agent
]
)
# Root agent - coordinates the review pipeline
root_agent = Agent(
name="CodeReviewAssistant",
model=config.worker_model,
description="An intelligent code review assistant that analyzes Python code and provides educational feedback",
instruction="""You are a specialized Python code review assistant focused on helping developers improve their code quality.
When a user provides Python code for review:
1. Immediately delegate to CodeReviewPipeline and pass the code EXACTLY as it was provided by the user.
2. The pipeline will handle all analysis and feedback
3. Return ONLY the final feedback from the pipeline - do not add any commentary
When a user asks what you can do or asks general questions:
- Explain your capabilities for code review
- Do NOT trigger the pipeline for non-code messages
The pipeline handles everything for code review - just pass through its final output.""",
sub_agents=[code_review_pipeline],
output_key="assistant_response"
)
छठा चरण: पूरी पाइपलाइन की जांच करना
अब चारों एजेंट को एक साथ काम करते हुए देखने का समय है.
👉 सिस्टम शुरू करें:
adk web code_review_assistant
adk web कमांड चलाने के बाद, आपको अपने टर्मिनल में आउटपुट दिखेगा. इससे पता चलेगा कि ADK वेब सर्वर शुरू हो गया है. यह आउटपुट कुछ ऐसा दिखेगा:
+-----------------------------------------------------------------------------+
| ADK Web Server started |
| |
| For local testing, access at http://localhost:8000. |
+-----------------------------------------------------------------------------+
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
👉 इसके बाद, अपने ब्राउज़र से ADK Dev UI को ऐक्सेस करने के लिए:
Cloud Shell टूलबार (आम तौर पर सबसे ऊपर दाईं ओर) में मौजूद, वेब की झलक दिखाने वाले आइकॉन (अक्सर यह आंख या ऐरो वाला स्क्वेयर दिखता है) से, पोर्ट बदलें को चुनें. पॉप-अप विंडो में, पोर्ट को 8000 पर सेट करें. इसके बाद, "बदलें और झलक देखें" पर क्लिक करें. इसके बाद, Cloud Shell एक नया ब्राउज़र टैब या विंडो खोलेगा. इसमें ADK Dev UI दिखेगा.
👉 एजेंट अब काम कर रहा है. आपके ब्राउज़र में मौजूद ADK Dev UI, एजेंट से सीधे तौर पर कनेक्ट होने का तरीका है.
- अपना टारगेट चुनें: यूज़र इंटरफ़ेस (यूआई) में सबसे ऊपर मौजूद ड्रॉपडाउन मेन्यू में,
code_review_assistantएजेंट चुनें.
👉 टेस्ट प्रॉम्प्ट:
Please analyze the following:
def dfs_search_v1(graph, start, target):
"""Find if target is reachable from start."""
visited = set()
stack = start
while stack:
current = stack.pop()
if current == target:
return True
if current not in visited:
visited.add(current)
for neighbor in graph[current]:
if neighbor not in visited:
stack.append(neighbor)
return False
👉 कोड की समीक्षा करने वाली पाइपलाइन को काम करते हुए देखें:
बग वाले dfs_search_v1 फ़ंक्शन को सबमिट करने पर, आपको सिर्फ़ एक जवाब नहीं मिलता. आपको मल्टी-एजेंट पाइपलाइन काम करती हुई दिख रही है. आपको जो स्ट्रीमिंग आउटपुट दिखता है वह चार खास एजेंटों के क्रम से काम करने का नतीजा है. हर एजेंट, पिछले एजेंट के काम को आगे बढ़ाता है.
यहां बताया गया है कि हर एजेंट, फ़ाइनल और पूरी जानकारी वाली समीक्षा में क्या योगदान देता है. इससे रॉ डेटा को कार्रवाई लायक जानकारी में बदला जा सकता है.
1. कोड विश्लेषण करने वाले टूल की स्ट्रक्चरल रिपोर्ट
सबसे पहले, CodeAnalyzer एजेंट को रॉ कोड मिलता है. यह कोड के काम करने के तरीके का अंदाज़ा नहीं लगाता. यह analyze_code_structure टूल का इस्तेमाल करके, डिटरमिनिस्टिक ऐब्स्ट्रैक्ट सिंटैक्स ट्री (एएसटी) पार्स करता है.
इसका आउटपुट, कोड के स्ट्रक्चर के बारे में सटीक जानकारी देता है:
The analysis of the provided code reveals the following:
Summary:
- Functions Found: 1
- Classes Found: 0
Key Structural Observations:
- A single function, dfs_search_v1, is defined.
- It includes a docstring: "Find if target is reachable from start."
- No syntax errors were detected.
Overall Code Organization Assessment:
- The code snippet is a well-defined, self-contained function.
⭐ वैल्यू: इस शुरुआती चरण से, अन्य एजेंट के लिए साफ़-सुथरा और भरोसेमंद आधार मिलता है. इससे पुष्टि होती है कि कोड, मान्य Python है. साथ ही, इससे उन कॉम्पोनेंट की पहचान होती है जिनकी समीक्षा की जानी है.
2. स्टाइल चेकर का पीईपी 8 ऑडिट
इसके बाद, StyleChecker एजेंट की बारी आती है. यह शेयर की गई स्थिति से कोड को पढ़ता है और check_code_style टूल का इस्तेमाल करता है. यह टूल, pycodestyle लिंटर का फ़ायदा उठाता है.
इसका आउटपुट, क्वालिटी स्कोर और खास उल्लंघनों के बारे में जानकारी देता है:
Style Analysis Results
- Style Score: 88/100
- Total Issues: 6
- Assessment: Good style with minor improvements needed
Top Style Issues
- Line 5, W293: blank line contains whitespace
- Line 19, W292: no newline at end of file
⭐ वैल्यू: यह एजेंट, कम्यूनिटी के तय किए गए स्टैंडर्ड (पीईपी 8) के आधार पर, निष्पक्ष और बिना किसी समझौते के सुझाव देता है. वेटेड स्कोरिंग सिस्टम से, उपयोगकर्ता को समस्याओं की गंभीरता के बारे में तुरंत पता चल जाता है.
3. टेस्ट रनर को मिली गंभीर गड़बड़ी
यहां सिस्टम, सामान्य विश्लेषण से आगे बढ़कर काम करता है. TestRunner एजेंट, कोड के व्यवहार की पुष्टि करने के लिए, टेस्ट का एक पूरा सेट जनरेट और लागू करता है.
इसका आउटपुट, स्ट्रक्चर्ड JSON ऑब्जेक्ट होता है. इसमें एक गंभीर फ़ैसला शामिल होता है:
{
"critical_issues": [
{
"type": "Critical Bug",
"description": "The function's initialization `stack = start` is incorrect... When a common input like a string... is provided... the function crashes with an AttributeError.",
"severity": "Critical"
}
],
"verdict": {
"status": "BROKEN",
"confidence": "high",
"recommendation": "The function is fundamentally broken... the stack initialization line `stack = start` must be changed to `stack = [start]`."
}
}
⭐ वैल्यू: यह सबसे अहम जानकारी होती है. एजेंट ने सिर्फ़ अनुमान नहीं लगाया, बल्कि कोड को चलाकर यह साबित किया कि कोड में गड़बड़ी है. इसने रनटाइम से जुड़ी एक ऐसी गड़बड़ी का पता लगाया जो मामूली होने के साथ-साथ गंभीर भी थी. इसकी समीक्षा करने वाला व्यक्ति शायद इस गड़बड़ी का पता न लगा पाता. इसने गड़बड़ी की सटीक वजह और उसे ठीक करने के तरीके का भी पता लगाया.
4. सुझावों को व्यवस्थित करने वाले टूल की फ़ाइनल रिपोर्ट
आखिर में, FeedbackSynthesizer एजेंट कंडक्टर के तौर पर काम करता है. यह पिछले तीन एजेंट से स्ट्रक्चर्ड डेटा लेता है और एक ऐसी रिपोर्ट बनाता है जो इस्तेमाल करने में आसान होती है. साथ ही, इसमें विश्लेषण भी शामिल होता है और यह उत्साह बढ़ाने वाली होती है.
इसका आउटपुट, आपको दिखने वाली फ़ाइनल और बेहतर समीक्षा होती है:
📊 Summary
Great effort on implementing the Depth-First Search algorithm! ... However, a critical bug in the initialization of the stack prevents the function from working correctly...
✅ Strengths
- Good Algorithm Structure
- Correct Use of `visited` Set
📈 Code Quality Analysis
...
### Style Compliance
The style analysis returned a good score of 88/100.
...
### Test Results
The automated testing revealed a critical issue... The line `stack = start` directly assigns the input... which results in an `AttributeError`.
💡 Recommendations for Improvement
**Fix the Critical Stack Initialization Bug:**
- Incorrect Code: `stack = start`
- Correct Code: `stack = [start]`
💬 Encouragement
You are very close to a perfect implementation! The core logic of your DFS algorithm is sound, which is the hardest part.
⭐ वैल्यू: यह एजेंट, तकनीकी डेटा को मददगार और शिक्षा से जुड़े अनुभव में बदलता है. इसमें सबसे अहम समस्या (बग) को प्राथमिकता दी गई है. साथ ही, इसे साफ़ तौर पर बताया गया है और इसका सटीक समाधान दिया गया है. इसके अलावा, इसमें सकारात्मक लहज़े का इस्तेमाल किया गया है. इसमें पिछले सभी चरणों के नतीजों को एक साथ और अहम जानकारी के तौर पर शामिल किया जाता है.
कई चरणों वाली इस प्रोसेस से, एजेंटिक पाइपलाइन की क्षमता का पता चलता है. आपको एक ही जवाब नहीं मिलता. इसके बजाय, आपको लेयर के हिसाब से विश्लेषण मिलता है. इसमें हर एजेंट, पुष्टि किया जा सकने वाला खास टास्क पूरा करता है. इससे ऐसी समीक्षा मिलती है जो न सिर्फ़ जानकारी देने वाली होती है, बल्कि भरोसेमंद, शिक्षा देने वाली, और सटीक भी होती है.
👉💻 टेस्टिंग पूरी होने के बाद, Cloud Shell Editor टर्मिनल पर वापस जाएं और ADK Dev UI को बंद करने के लिए, Ctrl+C दबाएं.
आपने क्या बनाया है
अब आपके पास कोड की समीक्षा करने के लिए पूरी पाइपलाइन है. इससे ये काम किए जा सकते हैं:
✅ कोड स्ट्रक्चर को पार्स करता है - हेल्पर फ़ंक्शन
के साथ एएसटी का विश्लेषण ✅ स्टाइल की जांच करता है - नाम रखने के नियमों
के साथ वज़न के हिसाब से स्कोरिंग ✅ टेस्ट चलाता है - स्ट्रक्चर्ड JSON आउटपुट
के साथ टेस्ट जनरेट करता है ✅ सुझाव देता है - स्थिति, मेमोरी, और आर्टफ़ैक्ट को इंटिग्रेट करता है
✅ प्रोग्रेस को ट्रैक करता है - इनवोकेशन/सेशन/उपयोगकर्ताओं के लिए मल्टी-टियर स्थिति
✅ समय के साथ सीखता है - क्रॉस-सेशन पैटर्न के लिए मेमोरी सेवा
✅ आर्टफ़ैक्ट उपलब्ध कराता है - पूरी ऑडिट ट्रेल के साथ डाउनलोड की जा सकने वाली JSON रिपोर्ट
मास्टर किए गए मुख्य कॉन्सेप्ट
सीक्वेंशियल पाइपलाइन:
- चार एजेंटों को एक के बाद एक काम करते हुए दिखाया गया है
- हर एक, अगले के लिए स्टेट को बेहतर बनाता है
- डिपेंडेंसी से, एक्ज़ीक्यूशन का क्रम तय होता है
प्रोडक्शन पैटर्न:
- हेल्पर फ़ंक्शन को अलग करना (थ्रेड पूल में सिंक करना)
- ग्रेसफ़ुल डिग्रेडेशन (फ़ॉलबैक रणनीतियां)
- मल्टी-टियर स्टेट मैनेजमेंट (टेंप/सेशन/उपयोगकर्ता)
- कॉन्टेक्स्ट के हिसाब से डाइनैमिक निर्देश देने वाले टूल
- डुअल स्टोरेज (आर्टफ़ैक्ट + स्टेट रिडंडेंसी)
स्टेटस को 'कम्यूनिकेशन' के तौर पर सेट करें:
- कॉन्स्टेंट की मदद से, सभी एजेंट के लिए टाइपिंग की गलतियों को रोका जा सकता है
output_keyएजेंट की खास जानकारी लिखता है- बाद में एजेंट, StateKeys के ज़रिए पढ़ते हैं
- पाइपलाइन में स्टेट का फ़्लो लीनियर होता है
मेमोरी बनाम स्टेट:
- स्टेट: मौजूदा सेशन का डेटा
- मेमोरी: अलग-अलग सेशन में पैटर्न
- अलग-अलग मकसद, अलग-अलग लाइफ़टाइम
टूल ऑर्केस्ट्रेशन:
- सिंगल-टूल एजेंट (विश्लेषण करने वाला, स्टाइल की जांच करने वाला)
- पहले से मौजूद एक्ज़ीक्यूटर (test_runner)
- एक से ज़्यादा टूल का इस्तेमाल करके काम करना (सिंथेसाइज़र)
मॉडल चुनने की रणनीति:
- वर्कर मॉडल: मैकेनिकल टास्क (पार्सिंग, लिंटिंग, राउटिंग)
- आलोचक मॉडल: तर्क से जुड़े टास्क (जांच, सिंथेसिस)
- सही विकल्प चुनकर लागत को ऑप्टिमाइज़ करना
आगे क्या करना है
मॉड्यूल 6 में, फ़िक्स पाइपलाइन बनाई जाएगी:
- बार-बार ठीक करने के लिए, LoopAgent आर्किटेक्चर
- सूचना देने से जुड़ी नीति के तहत, बाहर निकलने की शर्तें
- सभी इटरेशन में स्टेट का इकट्ठा होना
- पुष्टि करने और फिर से कोशिश करने का लॉजिक
- समस्याओं को ठीक करने के लिए, समीक्षा पाइपलाइन के साथ इंटिग्रेशन
आपको यह पता चलेगा कि एक ही तरह के पैटर्न को जटिल और बार-बार किए जाने वाले वर्कफ़्लो में कैसे बढ़ाया जाता है. इन वर्कफ़्लो में एजेंट, काम पूरा होने तक कई बार कोशिश करते हैं. साथ ही, आपको यह भी पता चलेगा कि एक ही ऐप्लिकेशन में कई पाइपलाइन को कैसे मैनेज किया जाता है.
6. फ़िक्स पाइपलाइन: लूप आर्किटेक्चर जोड़ना
परिचय
मॉड्यूल 5 में, आपने समीक्षा करने वाली एक ऐसी पाइपलाइन बनाई थी जो कोड का विश्लेषण करती है और फ़ीडबैक देती है. हालांकि, समस्याओं का पता लगाना सिर्फ़ आधा समाधान है. डेवलपर को उन्हें ठीक करने में मदद की ज़रूरत होती है.
यह मॉड्यूल, अपने-आप होने वाले सुधार की पाइपलाइन बनाता है. यह पाइपलाइन:
- समीक्षा के नतीजों के आधार पर सुधारों को जनरेट करता है
- समस्या ठीक होने की पुष्टि करता है. इसके लिए, वह कई तरह के टेस्ट करता है
- समस्या ठीक न होने पर, अपने-आप फिर से कोशिश करता है (तीन बार तक)
- रिपोर्ट के नतीजे, जिनमें पहले और बाद की तुलनाएं शामिल हैं
मुख्य कॉन्सेप्ट: LoopAgent की मदद से, अनुरोध को अपने-आप फिर से भेजा जा सकता है. सीक्वेंशियल एजेंट एक बार चलते हैं. वहीं, LoopAgent अपने सब-एजेंट को तब तक दोहराता है, जब तक कि बाहर निकलने की शर्त पूरी नहीं हो जाती या ज़्यादा से ज़्यादा इटरेशन पूरे नहीं हो जाते. टूल, tool_context.actions.escalate = True सेट करके सिग्नल की स्थिति के बारे में बताता है.
आपको क्या बनाना है, इसकी झलक: गड़बड़ी वाला कोड सबमिट करें → समीक्षा में समस्याएं पहचानें → ठीक करने के लिए लूप में सुधार करें → जांच में पुष्टि करें → ज़रूरत पड़ने पर फिर से कोशिश करें → पूरी जानकारी वाली फ़ाइनल रिपोर्ट.
मुख्य सिद्धांत: LoopAgent बनाम Sequential
सीक्वेंशियल पाइपलाइन (मॉड्यूल 5):
SequentialAgent(agents=[A, B, C])
# Executes: A → B → C → Done
- एकतरफ़ा फ़्लो
- हर एजेंट को सिर्फ़ एक बार चलाया जाता है
- दोबारा कोशिश करने का लॉजिक नहीं
लूप पाइपलाइन (मॉड्यूल 6):
LoopAgent(agents=[A, B, C], max_iterations=3)
# Executes: A → B → C → (check exit) → A → B → C → (check exit) → ...
- साइक्लिक फ़्लो
- एजेंट को कई बार चलाया जा सकता है
- यह कुकी तब हट जाती है, जब:
- कोई टूल
tool_context.actions.escalate = True(सफल) सेट करता है max_iterationsतक पहुंच गया (सुरक्षा सीमा)- अनहैंडल किया गया अपवाद (गड़बड़ी)
- कोई टूल
कोड ठीक करने के लिए लूप का इस्तेमाल क्यों किया जाता है:
कोड ठीक करने के लिए, अक्सर कई बार कोशिश करनी पड़ती है:
- पहली कोशिश: साफ़ तौर पर दिख रही गड़बड़ियों को ठीक करना (जैसे, गलत वैरिएबल टाइप)
- दूसरा चरण: जांच के दौरान मिली अन्य समस्याओं को ठीक करना (खास मामलों में)
- तीसरा चरण: सभी टेस्ट पास होने चाहिए. इसके लिए, ज़रूरी बदलाव करें और पुष्टि करें
लूप के बिना, आपको एजेंट के निर्देशों में जटिल शर्तों वाला लॉजिक इस्तेमाल करना होगा. LoopAgent के साथ, फिर से कोशिश करने की सुविधा अपने-आप काम करती है.
आर्किटेक्चर की तुलना:
Sequential (Module 5):
User → Review Pipeline → Feedback → Done
Loop (Module 6):
User → Review Pipeline → Feedback → Fix Pipeline
↓
┌──────────────┴──────────────┐
│ Fix Attempt Loop (1-3x) │
│ ┌─────────────────────┐ │
│ │ 1. Generate Fixes │ │
│ │ 2. Test Fixes │ │
│ │ 3. Validate & Exit? │────┼─→ If escalate=True
│ └─────────────────────┘ │ exit loop
│ ↓ If not │
│ Try Again (max 3) │
└─────────────────────────────┘
↓
4. Synthesize Final Report → Done
पहला चरण: Code Fixer एजेंट जोड़ना
कोड ठीक करने वाला टूल, समीक्षा के नतीजों के आधार पर सही किया गया Python कोड जनरेट करता है.
👉 खोलें
code_review_assistant/sub_agents/fix_pipeline/code_fixer.py
👉 ढूंढें:
# MODULE_6_STEP_1_CODE_FIXER_INSTRUCTION_PROVIDER
👉 उस एक लाइन को इससे बदलें:
async def code_fixer_instruction_provider(context: ReadonlyContext) -> str:
"""Dynamic instruction provider that injects state variables."""
template = """You are an expert code fixing specialist.
Original Code:
{code_to_review}
Analysis Results:
- Style Score: {style_score}/100
- Style Issues: {style_issues}
- Test Results: {test_execution_summary}
Based on the test results, identify and fix ALL issues including:
- Interface bugs (e.g., if start parameter expects wrong type)
- Logic errors (e.g., KeyError when accessing graph nodes)
- Style violations
- Missing documentation
YOUR TASK:
Generate the complete fixed Python code that addresses all identified issues.
CRITICAL INSTRUCTIONS:
- Output ONLY the corrected Python code
- Do NOT include markdown code blocks (```python)
- Do NOT include any explanations or commentary
- The output should be valid, executable Python code and nothing else
Common fixes to apply based on test results:
- If tests show AttributeError with 'pop', fix: stack = [start] instead of stack = start
- If tests show KeyError accessing graph, fix: use graph.get(current, [])
- Add docstrings if missing
- Fix any style violations identified
Output the complete fixed code now:"""
return await instructions_utils.inject_session_state(template, context)
👉 ढूंढें:
# MODULE_6_STEP_1_CODE_FIXER_AGENT
👉 उस एक लाइन को इससे बदलें:
code_fixer_agent = Agent(
name="CodeFixer",
model=config.worker_model,
description="Generates comprehensive fixes for all identified code issues",
instruction=code_fixer_instruction_provider,
code_executor=BuiltInCodeExecutor(),
output_key="code_fixes"
)
दूसरा चरण: Fix Test Runner Agent जोड़ना
फ़िक्स टेस्ट रनर, ठीक किए गए कोड पर पूरी तरह से जांच करके, सुधारों की पुष्टि करता है.
👉 खोलें
code_review_assistant/sub_agents/fix_pipeline/fix_test_runner.py
👉 ढूंढें:
# MODULE_6_STEP_2_FIX_TEST_RUNNER_INSTRUCTION_PROVIDER
👉 उस एक लाइन को इससे बदलें:
async def fix_test_runner_instruction_provider(context: ReadonlyContext) -> str:
"""Dynamic instruction provider that uses the clean code from the previous step."""
template = """You are responsible for validating the fixed code by running tests.
THE FIXED CODE TO TEST:
{code_fixes}
ORIGINAL TEST RESULTS: {test_execution_summary}
YOUR TASK:
1. Understand the fixes that were applied
2. Generate the same comprehensive tests (15-20 test cases)
3. Execute the tests on the FIXED code using your code executor
4. Compare results with original test results
5. Output a detailed JSON analysis
TESTING METHODOLOGY:
- Run the same tests that revealed issues in the original code
- Verify that previously failing tests now pass
- Ensure no regressions were introduced
- Document the improvement
Execute your tests and output ONLY valid JSON with this structure:
- "passed": number of tests that passed
- "failed": number of tests that failed
- "total": total number of tests
- "pass_rate": percentage as a number
- "comparison": object with "original_pass_rate", "new_pass_rate", "improvement"
- "newly_passing_tests": array of test names that now pass
- "still_failing_tests": array of test names still failing
Do NOT output the test code itself, only the JSON analysis."""
return await instructions_utils.inject_session_state(template, context)
👉 ढूंढें:
# MODULE_6_STEP_2_FIX_TEST_RUNNER_AGENT
👉 उस एक लाइन को इससे बदलें:
fix_test_runner_agent = Agent(
name="FixTestRunner",
model=config.critic_model,
description="Runs comprehensive tests on fixed code to verify all issues are resolved",
instruction=fix_test_runner_instruction_provider,
code_executor=BuiltInCodeExecutor(),
output_key="fix_test_execution_summary"
)
तीसरा चरण: Fix Validator Agent जोड़ना
पुष्टि करने वाला टूल यह जांच करता है कि गड़बड़ियां ठीक हुई हैं या नहीं. साथ ही, यह तय करता है कि लूप से बाहर निकलना है या नहीं.
टूल के बारे में जानकारी
सबसे पहले, पुष्टि करने वाले प्रोग्राम के लिए ज़रूरी तीन टूल जोड़ें.
👉 खोलें
code_review_assistant/tools.py
👉 ढूंढें:
# MODULE_6_STEP_3_VALIDATE_FIXED_STYLE
👉 स्टाइल की पुष्टि करने वाले टूल 1 से बदलें:
async def validate_fixed_style(tool_context: ToolContext) -> Dict[str, Any]:
"""
Validates style compliance of the fixed code.
Args:
tool_context: ADK tool context containing fixed code in state
Returns:
Dictionary with style validation results
"""
logger.info("Tool: Validating style of fixed code...")
try:
# Get the fixed code from state
code_fixes = tool_context.state.get(StateKeys.CODE_FIXES, '')
# Try to extract from markdown if present
if '```python' in code_fixes:
start = code_fixes.rfind('```python') + 9
end = code_fixes.rfind('```')
if start < end:
code_fixes = code_fixes[start:end].strip()
if not code_fixes:
return {
"status": "error",
"message": "No fixed code found in state"
}
# Store the extracted fixed code
tool_context.state[StateKeys.CODE_FIXES] = code_fixes
# Run style check on fixed code
loop = asyncio.get_event_loop()
with ThreadPoolExecutor() as executor:
style_result = await loop.run_in_executor(
executor, _perform_style_check, code_fixes
)
# Compare with original
original_score = tool_context.state.get(StateKeys.STYLE_SCORE, 0)
improvement = style_result['score'] - original_score
# Store results
tool_context.state[StateKeys.FIXED_STYLE_SCORE] = style_result['score']
tool_context.state[StateKeys.FIXED_STYLE_ISSUES] = style_result['issues']
logger.info(f"Tool: Fixed code style score: {style_result['score']}/100 "
f"(improvement: +{improvement})")
return {
"status": "success",
"fixed_style_score": style_result['score'],
"original_style_score": original_score,
"improvement": improvement,
"remaining_issues": style_result['issues'],
"perfect_style": style_result['score'] == 100
}
except Exception as e:
logger.error(f"Tool: Style validation failed: {e}", exc_info=True)
return {
"status": "error",
"message": str(e)
}
👉 ढूंढें:
# MODULE_6_STEP_3_COMPILE_FIX_REPORT
👉 दूसरे टूल, रिपोर्ट कंपाइलर का इस्तेमाल करें:
async def compile_fix_report(tool_context: ToolContext) -> Dict[str, Any]:
"""
Compiles comprehensive report of the fix process.
Args:
tool_context: ADK tool context with all fix pipeline data
Returns:
Comprehensive fix report
"""
logger.info("Tool: Compiling comprehensive fix report...")
try:
# Gather all data
original_code = tool_context.state.get(StateKeys.CODE_TO_REVIEW, '')
code_fixes = tool_context.state.get(StateKeys.CODE_FIXES, '')
# Test results
original_tests = tool_context.state.get(StateKeys.TEST_EXECUTION_SUMMARY, {})
fixed_tests = tool_context.state.get(StateKeys.FIX_TEST_EXECUTION_SUMMARY, {})
# Parse if strings
if isinstance(original_tests, str):
try:
original_tests = json.loads(original_tests)
except:
original_tests = {}
if isinstance(fixed_tests, str):
try:
fixed_tests = json.loads(fixed_tests)
except:
fixed_tests = {}
# Extract pass rates
original_pass_rate = 0
if original_tests:
if 'pass_rate' in original_tests:
original_pass_rate = original_tests['pass_rate']
elif 'test_summary' in original_tests:
# Handle test_runner_agent's JSON structure
summary = original_tests['test_summary']
total = summary.get('total_tests_run', 0)
passed = summary.get('tests_passed', 0)
if total > 0:
original_pass_rate = (passed / total) * 100
elif 'passed' in original_tests and 'total' in original_tests:
if original_tests['total'] > 0:
original_pass_rate = (original_tests['passed'] / original_tests['total']) * 100
fixed_pass_rate = 0
all_tests_pass = False
if fixed_tests:
if 'pass_rate' in fixed_tests:
fixed_pass_rate = fixed_tests['pass_rate']
all_tests_pass = fixed_tests.get('failed', 1) == 0
elif 'passed' in fixed_tests and 'total' in fixed_tests:
if fixed_tests['total'] > 0:
fixed_pass_rate = (fixed_tests['passed'] / fixed_tests['total']) * 100
all_tests_pass = fixed_tests.get('failed', 0) == 0
# Style scores
original_style = tool_context.state.get(StateKeys.STYLE_SCORE, 0)
fixed_style = tool_context.state.get(StateKeys.FIXED_STYLE_SCORE, 0)
# Calculate improvements
test_improvement = {
'original_pass_rate': original_pass_rate,
'fixed_pass_rate': fixed_pass_rate,
'improvement': fixed_pass_rate - original_pass_rate,
'all_tests_pass': all_tests_pass
}
style_improvement = {
'original_score': original_style,
'fixed_score': fixed_style,
'improvement': fixed_style - original_style,
'perfect_style': fixed_style == 100
}
# Determine overall status
if all_tests_pass and style_improvement['perfect_style']:
fix_status = 'SUCCESSFUL'
status_emoji = '✅'
elif test_improvement['improvement'] > 0 or style_improvement['improvement'] > 0:
fix_status = 'PARTIAL'
status_emoji = '⚠️'
else:
fix_status = 'FAILED'
status_emoji = '❌'
# Build comprehensive report
report = {
'status': fix_status,
'status_emoji': status_emoji,
'timestamp': datetime.now().isoformat(),
'original_code': original_code,
'code_fixes': code_fixes,
'improvements': {
'tests': test_improvement,
'style': style_improvement
},
'summary': f"{status_emoji} Fix Status: {fix_status}\n"
f"Tests: {original_pass_rate:.1f}% → {fixed_pass_rate:.1f}%\n"
f"Style: {original_style}/100 → {fixed_style}/100"
}
# Store report in state
tool_context.state[StateKeys.FIX_REPORT] = report
tool_context.state[StateKeys.FIX_STATUS] = fix_status
logger.info(f"Tool: Fix report compiled - Status: {fix_status}")
logger.info(f"Tool: Test improvement: {original_pass_rate:.1f}% → {fixed_pass_rate:.1f}%")
logger.info(f"Tool: Style improvement: {original_style} → {fixed_style}")
return {
"status": "success",
"fix_status": fix_status,
"report": report
}
except Exception as e:
logger.error(f"Tool: Failed to compile fix report: {e}", exc_info=True)
return {
"status": "error",
"message": str(e)
}
👉 ढूंढें:
# MODULE_6_STEP_3_EXIT_FIX_LOOP
👉 तीसरे टूल - लूप से बाहर निकलने के सिग्नल से बदलें:
def exit_fix_loop(tool_context: ToolContext) -> Dict[str, Any]:
"""
Signal that fixing is complete and should exit the loop.
Args:
tool_context: ADK tool context
Returns:
Confirmation message
"""
logger.info("Tool: Setting escalate flag to exit fix loop")
# This is the critical line that exits the LoopAgent
tool_context.actions.escalate = True
return {
"status": "success",
"message": "Fix complete, exiting loop"
}
पुष्टि करने वाला एजेंट बनाना
👉 खोलें
code_review_assistant/sub_agents/fix_pipeline/fix_validator.py
👉 ढूंढें:
# MODULE_6_STEP_3_FIX_VALIDATOR_INSTRUCTION_PROVIDER
👉 उस एक लाइन को इससे बदलें:
async def fix_validator_instruction_provider(context: ReadonlyContext) -> str:
"""Dynamic instruction provider that injects state variables."""
template = """You are the final validation specialist for code fixes.
You have access to:
- Original issues from initial review
- Applied fixes: {code_fixes}
- Test results after fix: {fix_test_execution_summary}
- All state data from the fix process
Your responsibilities:
1. Use validate_fixed_style tool to check style compliance of fixed code
- Pass no arguments, it will retrieve fixed code from state
2. Use compile_fix_report tool to generate comprehensive report
- Pass no arguments, it will gather all data from state
3. Based on the report, determine overall fix status:
- ✅ SUCCESSFUL: All tests pass, style score 100
- ⚠️ PARTIAL: Improvements made but issues remain
- ❌ FAILED: Fix didn't work or made things worse
4. CRITICAL: If status is SUCCESSFUL, call the exit_fix_loop tool to stop iterations
- This prevents unnecessary additional fix attempts
- If not successful, the loop will continue for another attempt
5. Provide clear summary of:
- What was fixed
- What improvements were achieved
- Any remaining issues requiring manual attention
Be precise and quantitative in your assessment.
"""
return await instructions_utils.inject_session_state(template, context)
👉 ढूंढें:
# MODULE_6_STEP_3_FIX_VALIDATOR_AGENT
👉 उस एक लाइन को इससे बदलें:
fix_validator_agent = Agent(
name="FixValidator",
model=config.worker_model,
description="Validates fixes and generates final fix report",
instruction=fix_validator_instruction_provider,
tools=[
FunctionTool(func=validate_fixed_style),
FunctionTool(func=compile_fix_report),
FunctionTool(func=exit_fix_loop)
],
output_key="final_fix_report"
)
चौथा चरण: LoopAgent से बाहर निकलने की शर्तों को समझना
LoopAgent से बाहर निकलने के तीन तरीके हैं:
1. सफलतापूर्वक बाहर निकलें (बढ़ाएं)
# Inside any tool in the loop:
tool_context.actions.escalate = True
# Effect: Loop completes current iteration, then exits
# Use when: Fix is successful and no more attempts needed
उदाहरण के लिए फ़्लो:
Iteration 1:
CodeFixer → generates fixes
FixTestRunner → tests show 90% pass rate
FixValidator → compiles report, sees PARTIAL status
→ Does NOT set escalate
→ Loop continues
Iteration 2:
CodeFixer → refines fixes based on failures
FixTestRunner → tests show 100% pass rate
FixValidator → compiles report, sees SUCCESSFUL status
→ Calls exit_fix_loop() which sets escalate = True
→ Loop exits after this iteration
2. Max Iterations Exit
LoopAgent(
name="FixAttemptLoop",
sub_agents=[...],
max_iterations=3 # Safety limit
)
# Effect: After 3 complete iterations, loop exits regardless of escalate
# Use when: Prevent infinite loops if fixes never succeed
उदाहरण के लिए फ़्लो:
Iteration 1: PARTIAL (continue)
Iteration 2: PARTIAL (continue)
Iteration 3: PARTIAL (but max reached)
→ Loop exits, synthesizer presents best attempt
3. गड़बड़ी की वजह से बंद हुआ
# If any agent throws unhandled exception:
raise Exception("Unexpected error")
# Effect: Loop exits immediately with error state
# Use when: Critical failure that can't be recovered
अलग-अलग वर्शन में स्थिति में बदलाव:
हर बार अपडेट करने पर, पिछली कोशिश से अपडेट की गई स्थिति दिखती है:
# Before Iteration 1:
state = {
"code_to_review": "def add(a,b):return a+b", # Original
"style_score": 40,
"test_execution_summary": {...}
}
# After Iteration 1:
state = {
"code_to_review": "def add(a,b):return a+b", # Unchanged
"code_fixes": "def add(a, b):\n return a + b", # NEW
"style_score": 40, # Unchanged
"fixed_style_score": 100, # NEW
"test_execution_summary": {...}, # Unchanged
"fix_test_execution_summary": {...} # NEW
}
# Iteration 2 starts with all this state
# If fixes still not perfect, code_fixes gets overwritten
क्यों
escalate
रिटर्न वैल्यू के बजाय:
# Bad: Using return value to signal exit
def validator_agent():
report = compile_report()
if report['status'] == 'SUCCESSFUL':
return {"exit": True} # How does loop know?
# Good: Using escalate
def validator_tool(tool_context):
report = compile_report()
if report['status'] == 'SUCCESSFUL':
tool_context.actions.escalate = True # Loop knows immediately
return {"report": report}
फ़ायदे:
- यह सुविधा, किसी भी टूल से काम करती है. सिर्फ़ आखिरी टूल से नहीं
- इससे, डेटा वापस पाने में कोई समस्या नहीं आती
- सिमेंटिक अर्थ साफ़ तौर पर समझ में आना चाहिए
- फ़्रेमवर्क, बाहर निकलने के लॉजिक को हैंडल करता है
पांचवां चरण: फ़िक्स पाइपलाइन को वायर करना
👉 खोलें
code_review_assistant/agent.py
👉 मौजूदा इंपोर्ट के बाद, पाइपलाइन इंपोर्ट से जुड़ी समस्या को ठीक करने के लिए यह कोड जोड़ें:
from google.adk.agents import LoopAgent # Add this to the existing Agent, SequentialAgent line
from code_review_assistant.sub_agents.fix_pipeline.code_fixer import code_fixer_agent
from code_review_assistant.sub_agents.fix_pipeline.fix_test_runner import fix_test_runner_agent
from code_review_assistant.sub_agents.fix_pipeline.fix_validator import fix_validator_agent
from code_review_assistant.sub_agents.fix_pipeline.fix_synthesizer import fix_synthesizer_agent
अब आपके इंपोर्ट किए गए डेटा में ये चीज़ें होनी चाहिए:
from google.adk.agents import Agent, SequentialAgent, LoopAgent
from .config import config
# Review pipeline imports (from Module 5)
from code_review_assistant.sub_agents.review_pipeline.code_analyzer import code_analyzer_agent
from code_review_assistant.sub_agents.review_pipeline.style_checker import style_checker_agent
from code_review_assistant.sub_agents.review_pipeline.test_runner import test_runner_agent
from code_review_assistant.sub_agents.review_pipeline.feedback_synthesizer import feedback_synthesizer_agent
# Fix pipeline imports (NEW)
from code_review_assistant.sub_agents.fix_pipeline.code_fixer import code_fixer_agent
from code_review_assistant.sub_agents.fix_pipeline.fix_test_runner import fix_test_runner_agent
from code_review_assistant.sub_agents.fix_pipeline.fix_validator import fix_validator_agent
from code_review_assistant.sub_agents.fix_pipeline.fix_synthesizer import fix_synthesizer_agent
👉 ढूंढें:
# MODULE_6_STEP_5_CREATE_FIX_LOOP
👉 उस एक लाइन को इससे बदलें:
# Create the fix attempt loop (retries up to 3 times)
fix_attempt_loop = LoopAgent(
name="FixAttemptLoop",
sub_agents=[
code_fixer_agent, # Step 1: Generate fixes
fix_test_runner_agent, # Step 2: Validate with tests
fix_validator_agent # Step 3: Check success & possibly exit
],
max_iterations=3 # Try up to 3 times
)
# Wrap loop with synthesizer for final report
code_fix_pipeline = SequentialAgent(
name="CodeFixPipeline",
description="Automated code fixing pipeline with iterative validation",
sub_agents=[
fix_attempt_loop, # Try to fix (1-3 times)
fix_synthesizer_agent # Present final results (always runs once)
]
)
👉 मौजूदा
root_agent
definition:
root_agent = Agent(...)
👉 ढूंढें:
# MODULE_6_STEP_5_UPDATE_ROOT_AGENT
👉 उस एक लाइन को इससे बदलें:
# Update root agent to include both pipelines
root_agent = Agent(
name="CodeReviewAssistant",
model=config.worker_model,
description="An intelligent code review assistant that analyzes Python code and provides educational feedback",
instruction="""You are a specialized Python code review assistant focused on helping developers improve their code quality.
When a user provides Python code for review:
1. Immediately delegate to CodeReviewPipeline and pass the code EXACTLY as it was provided by the user.
2. The pipeline will handle all analysis and feedback
3. Return ONLY the final feedback from the pipeline - do not add any commentary
After completing a review, if significant issues were identified:
- If style score < 100 OR tests are failing OR critical issues exist:
* Add at the end: "\n\n💡 I can fix these issues for you. Would you like me to do that?"
- If the user responds yes or requests fixes:
* Delegate to CodeFixPipeline
* Return the fix pipeline's complete output AS-IS
When a user asks what you can do or general questions:
- Explain your capabilities for code review and fixing
- Do NOT trigger the pipeline for non-code messages
The pipelines handle everything for code review and fixing - just pass through their final output.""",
sub_agents=[code_review_pipeline, code_fix_pipeline],
output_key="assistant_response"
)
छठा चरण: Fix Synthesizer Agent जोड़ना
लूप पूरा होने के बाद, सिंथेसाइज़र गड़बड़ी ठीक करने के नतीजों को उपयोगकर्ता के लिए आसान तरीके से दिखाता है.
👉 खोलें
code_review_assistant/sub_agents/fix_pipeline/fix_synthesizer.py
👉 ढूंढें:
# MODULE_6_STEP_6_FIX_SYNTHESIZER_INSTRUCTION_PROVIDER
👉 उस एक लाइन को इससे बदलें:
async def fix_synthesizer_instruction_provider(context: ReadonlyContext) -> str:
"""Dynamic instruction provider that injects state variables."""
template = """You are responsible for presenting the fix results to the user.
Based on the validation report: {final_fix_report}
Fixed code from state: {code_fixes}
Fix status: {fix_status}
Create a comprehensive yet friendly response that includes:
## 🔧 Fix Summary
[Overall status and key improvements - be specific about what was achieved]
## 📊 Metrics
- Test Results: [original pass rate]% → [new pass rate]%
- Style Score: [original]/100 → [new]/100
- Issues Fixed: X of Y
## ✅ What Was Fixed
[List each fixed issue with brief explanation of the correction made]
## 📝 Complete Fixed Code
[Include the complete, corrected code from state - this is critical]
## 💡 Explanation of Key Changes
[Brief explanation of the most important changes made and why]
[If any issues remain]
## ⚠️ Remaining Issues
[List what still needs manual attention]
## 🎯 Next Steps
[Guidance on what to do next - either use the fixed code or address remaining issues]
Save the fix report using save_fix_report tool before presenting.
Call it with no parameters - it will retrieve the report from state automatically.
Be encouraging about improvements while being honest about any remaining issues.
Focus on the educational aspect - help the user understand what was wrong and how it was fixed.
"""
return await instructions_utils.inject_session_state(template, context)
👉 ढूंढें:
# MODULE_6_STEP_6_FIX_SYNTHESIZER_AGENT
👉 उस एक लाइन को इससे बदलें:
fix_synthesizer_agent = Agent(
name="FixSynthesizer",
model=config.critic_model,
description="Creates comprehensive user-friendly fix report",
instruction=fix_synthesizer_instruction_provider,
tools=[FunctionTool(func=save_fix_report)],
output_key="fix_summary"
)
👉 जोड़ें
save_fix_report
टूल से
tools.py
:
👉 ढूंढें:
# MODULE_6_STEP_6_SAVE_FIX_REPORT
👉 इससे बदलें:
async def save_fix_report(tool_context: ToolContext) -> Dict[str, Any]:
"""
Saves the fix report as an artifact.
Args:
tool_context: ADK tool context
Returns:
Save status
"""
logger.info("Tool: Saving fix report...")
try:
# Get the report from state
fix_report = tool_context.state.get(StateKeys.FIX_REPORT, {})
if not fix_report:
return {
"status": "error",
"message": "No fix report found in state"
}
# Convert to JSON
report_json = json.dumps(fix_report, indent=2)
report_part = types.Part.from_text(text=report_json)
# Generate filename
timestamp = datetime.now().isoformat().replace(':', '-')
filename = f"fix_report_{timestamp}.json"
# Try to save as artifact
if hasattr(tool_context, 'save_artifact'):
try:
version = await tool_context.save_artifact(filename, report_part)
await tool_context.save_artifact("latest_fix_report.json", report_part)
logger.info(f"Tool: Fix report saved as {filename}")
return {
"status": "success",
"filename": filename,
"version": str(version),
"size": len(report_json)
}
except Exception as e:
logger.warning(f"Could not save as artifact: {e}")
# Fallback: store in state
tool_context.state[StateKeys.LAST_FIX_REPORT] = fix_report
return {
"status": "success",
"message": "Fix report saved to state",
"size": len(report_json)
}
except Exception as e:
logger.error(f"Tool: Failed to save fix report: {e}", exc_info=True)
return {
"status": "error",
"message": str(e)
}
सातवां चरण: 'समस्या हल हो गई' पाइपलाइन की जांच करना
अब पूरे लूप को ऐक्शन में देखने का समय है.
👉 सिस्टम शुरू करें:
adk web code_review_assistant
adk web कमांड चलाने के बाद, आपको अपने टर्मिनल में आउटपुट दिखेगा. इससे पता चलेगा कि ADK वेब सर्वर शुरू हो गया है. यह आउटपुट कुछ ऐसा दिखेगा:
+-----------------------------------------------------------------------------+
| ADK Web Server started |
| |
| For local testing, access at http://localhost:8000. |
+-----------------------------------------------------------------------------+
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
👉 टेस्ट प्रॉम्प्ट:
Please analyze the following:
def dfs_search_v1(graph, start, target):
"""Find if target is reachable from start."""
visited = set()
stack = start
while stack:
current = stack.pop()
if current == target:
return True
if current not in visited:
visited.add(current)
for neighbor in graph[current]:
if neighbor not in visited:
stack.append(neighbor)
return False
सबसे पहले, गड़बड़ी वाला कोड सबमिट करें, ताकि समीक्षा पाइपलाइन शुरू हो सके. गड़बड़ियों का पता लगाने के बाद, एजेंट से "कृपया कोड ठीक करें" कहा जाएगा. इससे, गड़बड़ी ठीक करने की प्रोसेस शुरू हो जाएगी.
1. शुरुआती समीक्षा (कमियां ढूंढना)
यह प्रोसेस का पहला हिस्सा है. चार एजेंट वाली समीक्षा पाइपलाइन, कोड का विश्लेषण करती है, उसकी स्टाइल की जांच करती है, और जनरेट की गई टेस्ट सुइट को चलाती है. यह AttributeError और अन्य समस्याओं का सही तरीके से पता लगाता है. साथ ही, यह बताता है कि कोड BROKEN है और टेस्ट पास करने की दर सिर्फ़ 84.21% है.
2. अपने-आप ठीक होने की सुविधा (लूप का इस्तेमाल)
यह सबसे दिलचस्प हिस्सा है. जब एजेंट से कोड ठीक करने के लिए कहा जाता है, तो वह सिर्फ़ एक बदलाव नहीं करता. यह गड़बड़ी ठीक करने और पुष्टि करने की प्रोसेस को शुरू करता है. यह प्रोसेस, एक मेहनती डेवलपर की तरह काम करती है: यह गड़बड़ी को ठीक करने की कोशिश करती है, उसकी अच्छी तरह से जांच करती है, और अगर गड़बड़ी पूरी तरह से ठीक नहीं होती है, तो यह प्रोसेस फिर से शुरू हो जाती है.
पहला वर्शन: पहला जवाब (कुछ हद तक सही)
- समस्या हल करने का तरीका:
CodeFixerएजेंट, शुरुआती रिपोर्ट को पढ़ता है और उसमें मौजूद सबसे सामान्य गड़बड़ियों को ठीक करता है. यहstack = startकोstack = [start]में बदलता है औरKeyErrorअपवादों को रोकने के लिएgraph.get()का इस्तेमाल करता है. - पुष्टि करना:
TestRunnerइस नए कोड के ख़िलाफ़, पूरे टेस्ट सुइट को तुरंत फिर से चलाता है. - नतीजा: पास होने की दर में काफ़ी सुधार हुआ है. अब यह 88.89% हो गई है! गंभीर बग ठीक कर दिए गए हैं. हालांकि, ये टेस्ट इतने ज़्यादा हैं कि इनसे दो नई, छोटी-मोटी गड़बड़ियां (रिग्रेशन) पता चलती हैं. ये गड़बड़ियां,
Noneको ग्राफ़ या सूची में शामिल न की गई आस-पास की वैल्यू के तौर पर हैंडल करने से जुड़ी हैं. सिस्टम, समस्या को आंशिक रूप से ठीक हुई के तौर पर मार्क करता है.
दूसरा वर्शन: फ़ाइनल पॉलिश (100% सफलता)
- समस्या हल करना: लूप के खत्म होने की शर्त (100% पास रेट) पूरी नहीं हुई है. इसलिए, यह फिर से चलेगा.
CodeFixerमें अब ज़्यादा जानकारी मौजूद है. इसमें रिग्रेशन से जुड़ी दो नई गड़बड़ियों के बारे में बताया गया है. यह कोड का फ़ाइनल और ज़्यादा बेहतर वर्शन जनरेट करता है. इसमें उन खास मामलों को साफ़ तौर पर हैंडल किया जाता है. - पुष्टि करना:
TestRunner, कोड के फ़ाइनल वर्शन के ख़िलाफ़ टेस्ट सुइट को एक बार फिर से एक्ज़ीक्यूट करता है. - नतीजा: 100% पास रेट. सभी ओरिजनल बग और सभी रिग्रेशन ठीक कर दिए गए हैं. सिस्टम, समस्या को ठीक होने के तौर पर SUCCESSFUL मार्क करता है और लूप बंद हो जाता है.
3. फ़ाइनल रिपोर्ट: परफ़ेक्ट स्कोर
समस्या ठीक होने की पुष्टि हो जाने के बाद, FixSynthesizer एजेंट फ़ाइनल रिपोर्ट दिखाता है. इसमें तकनीकी डेटा को साफ़ तौर पर जानकारी देने वाली खास जानकारी में बदल दिया जाता है.
मेट्रिक | पहले | बाद में | सुधार |
टेस्ट पास करने की दर | 84.21% | 100% | ▲ 15.79% |
स्टाइल स्कोर | 88 / 100 | 98 / 100 | ▲ 10 पॉइंट |
गड़बड़ियां ठीक की गईं | 3 में से 0 | 3 में से 3 | ✅ |
✅ पुष्टि किया गया आखिरी कोड
यहां पूरा और सही किया गया कोड दिया गया है. अब यह कोड सभी 19 टेस्ट पास कर लेता है. इससे पता चलता है कि समस्या को ठीक कर दिया गया है:
def dfs_search_v1(graph, start, target):
"""Find if target is reachable from start."""
# Handles 'None' graph input
if graph is None:
return False
visited = set()
# Fixes the critical AttributeError
stack = [start]
while stack:
current = stack.pop()
if current == target:
return True
if current not in visited:
visited.add(current)
# Safely gets neighbors to prevent KeyError
neighbors = graph.get(current)
if neighbors is None:
continue
# Validates that neighbors are iterable
if not isinstance(neighbors, (list, set, tuple)):
raise TypeError(
f"Graph value for node '{current}' is of type "
f"{type(neighbors).__name__}. Expected a list, set, or tuple."
)
for neighbor in neighbors:
if neighbor not in visited:
stack.append(neighbor)
return False
👉💻 टेस्टिंग पूरी होने के बाद, Cloud Shell Editor टर्मिनल पर वापस जाएं और ADK Dev UI को बंद करने के लिए, Ctrl+C दबाएं.
आपने क्या बनाया है
अब आपके पास, समस्या को अपने-आप ठीक करने की पूरी पाइपलाइन है. इससे ये काम किए जा सकते हैं:
✅ समस्या ठीक करने के तरीके सुझाता है - समीक्षा के विश्लेषण के आधार पर
✅ बार-बार पुष्टि करता है - समस्या ठीक करने की हर कोशिश के बाद जांच करता है
✅ अपने-आप फिर से कोशिश करता है - समस्या ठीक होने तक तीन बार कोशिश करता है
✅ स्मार्ट तरीके से बंद होता है - समस्या ठीक होने पर, एस्केलेट करके बंद होता है
✅ बेहतर परफ़ॉर्मेंस को ट्रैक करता है - समस्या ठीक होने से पहले और बाद के मेट्रिक की तुलना करता है
✅ आर्टफ़ैक्ट उपलब्ध कराता है - समस्या ठीक करने की रिपोर्ट डाउनलोड की जा सकती हैं
मास्टर किए गए मुख्य कॉन्सेप्ट
LoopAgent बनाम Sequential:
- सीक्वेंशियल: एजेंट के ज़रिए एक पास
- LoopAgent: यह तब तक दोहराता है, जब तक कि बाहर निकलने की शर्त पूरी नहीं हो जाती या ज़्यादा से ज़्यादा पुनरावृत्तियां नहीं हो जातीं
tool_context.actions.escalate = Trueसे होकर बाहर निकलें
अलग-अलग वर्शन में स्थिति में बदलाव:
CODE_FIXESको हर बार अपडेट किया जाता है- समय के साथ टेस्ट के नतीजों में सुधार दिखता है
- मान्य करने वाले व्यक्ति को सभी बदलाव दिखते हैं
मल्टी-पाइपलाइन आर्किटेक्चर:
- समीक्षा पाइपलाइन: सिर्फ़ पढ़ने के लिए विश्लेषण (मॉड्यूल 5)
- लूप ठीक करना: बार-बार सुधार करना (मॉड्यूल 6 का इनर लूप)
- फ़िक्स पाइपलाइन: लूप + सिंथेसाइज़र (मॉड्यूल 6 का बाहरी हिस्सा)
- रूट एजेंट: उपयोगकर्ता के मकसद के आधार पर काम करता है
फ़्लो को कंट्रोल करने वाले टूल:
exit_fix_loop()सेट एस्केलेट करता है- कोई भी टूल, लूप पूरा होने का सिग्नल दे सकता है
- इससे, एजेंट के निर्देशों से बाहर निकलने के लॉजिक को अलग किया जाता है
ज़्यादा से ज़्यादा बार दोहराने की सुविधा की सुरक्षा:
- यह कुकी, लगातार लूप को रोकने में मदद करती है
- इससे यह पक्का किया जाता है कि सिस्टम हमेशा जवाब दे
- जवाब में पूरी जानकारी न होने पर भी, सबसे सही जवाब देने की कोशिश करता है
आगे क्या करना है
आखिरी मॉड्यूल में, आपको अपने एजेंट को प्रोडक्शन में डिप्लॉय करने का तरीका बताया जाएगा:
- VertexAiSessionService की मदद से, परसिस्टेंट स्टोरेज सेट अप करना
- Google Cloud पर Agent Engine को डिप्लॉय करना
- प्रोडक्शन एजेंट को मॉनिटर और डीबग करना
- स्केलिंग और भरोसेमंद तरीके से काम करने के सबसे सही तरीके
आपने क्रमवार और लूप आर्किटेक्चर के साथ एक पूरा मल्टी-एजेंट सिस्टम बनाया है. आपने जो पैटर्न सीखे हैं - स्टेट मैनेजमेंट, डाइनैमिक निर्देश, टूल ऑर्केस्ट्रेशन, और बार-बार सुधार करना - ये प्रोडक्शन-रेडी तकनीकें हैं. इनका इस्तेमाल, असल एजेंटिक सिस्टम में किया जाता है.
7. प्रोडक्शन में डिप्लॉय करना
परिचय
अब आपके पास कोड की समीक्षा करने वाला असिस्टेंट उपलब्ध है. साथ ही, समीक्षा और गड़बड़ी ठीक करने वाली पाइपलाइन, स्थानीय तौर पर काम कर रही हैं. यह सिर्फ़ आपकी मशीन पर काम करता है. इस मॉड्यूल में, आपको अपने एजेंट को Google Cloud पर डिप्लॉय करने का तरीका बताया जाएगा. इससे आपकी टीम, एजेंट को लगातार सेशन और प्रोडक्शन-ग्रेड इंफ़्रास्ट्रक्चर के साथ ऐक्सेस कर पाएगी.
आपको यह जानकारी मिलेगी:
- डिप्लॉयमेंट के तीन तरीके: लोकल, Cloud Run, और Agent Engine
- अपने-आप इंफ़्रास्ट्रक्चर उपलब्ध कराने की सुविधा
- सेशन को बनाए रखने की रणनीतियां
- तैनात किए गए एजेंटों की टेस्टिंग करना
डिप्लॉयमेंट के विकल्पों के बारे में जानकारी
ADK, डिप्लॉयमेंट के कई टारगेट के साथ काम करता है. हर टारगेट के अपने फ़ायदे और नुकसान होते हैं:
डिप्लॉयमेंट पाथ
सिग्नल | स्थानीय ( | Cloud Run ( | एजेंट इंजन ( |
जटिलता | मिनिमल | मीडियम | कम |
सेशन बनाए रखना | सिर्फ़ मेमोरी में (रीस्टार्ट करने पर मिट जाता है) | Cloud SQL (PostgreSQL) | Vertex AI मैनेज करता है (अपने-आप) |
इन्फ़्रास्ट्रक्चर | कोई नहीं (सिर्फ़ डेवलपर मशीन) | कंटेनर + डेटाबेस | पूरी तरह से प्रबंधित |
कोल्ड स्टार्ट | लागू नहीं | 100-2000 मि॰से॰ | 100-500 मि॰से॰ |
स्केलिंग | सिंगल इंस्टेंस | अपने-आप (शून्य पर) | ऑटोमैटिक |
लागत मॉडल | मुफ़्त (लोकल कंप्यूट) | अनुरोध पर आधारित + फ़्री टियर | कंप्यूट के आधार पर |
यूज़र इंटरफ़ेस (यूआई) से जुड़ी सहायता | हां ( | हां ( | नहीं (सिर्फ़ एपीआई) |
इन स्थितियों में बेहतर है | डेवलपमेंट/टेस्टिंग | ट्रैफ़िक में बदलाव, लागत कंट्रोल करना | प्रोडक्शन एजेंट |
डिप्लॉयमेंट का अतिरिक्त विकल्प: Google Kubernetes Engine (GKE) उन ऐडवांस उपयोगकर्ताओं के लिए उपलब्ध है जिन्हें Kubernetes-लेवल कंट्रोल, कस्टम नेटवर्किंग या मल्टी-सर्विस ऑर्केस्ट्रेशन की ज़रूरत होती है. GKE डिप्लॉयमेंट के बारे में इस कोडलैब में नहीं बताया गया है. हालांकि, इसके बारे में ADK डिप्लॉयमेंट गाइड में बताया गया है.
क्या-क्या डिप्लॉय किया जाता है
Cloud Run या Agent Engine पर डिप्लॉय करते समय, यहां दी गई चीज़ें पैकेज की जाती हैं और डिप्लॉय की जाती हैं:
- आपका एजेंट कोड (
agent.py, सभी सब-एजेंट, टूल) - डिपेंडेंसी (
requirements.txt) - ADK API सर्वर (अपने-आप शामिल होता है)
- वेब यूज़र इंटरफ़ेस (सिर्फ़ Cloud Run के लिए, जब
--with_uiतय किया गया हो)
मुख्य अंतर:
- Cloud Run:
adk deploy cloud_runसीएलआई का इस्तेमाल करता है. इससे कंटेनर अपने-आप बन जाता है. इसके अलावा,gcloud run deployका इस्तेमाल भी किया जा सकता है. इसके लिए, कस्टम Dockerfile की ज़रूरत होती है - Agent Engine:
adk deploy agent_engineCLI का इस्तेमाल करता है. इसके लिए, कंटेनर बनाने की ज़रूरत नहीं होती. यह सीधे तौर पर Python कोड को पैकेज करता है
पहला चरण: अपना एनवायरमेंट कॉन्फ़िगर करना
अपनी .env फ़ाइल कॉन्फ़िगर करना
क्लाउड पर डिप्लॉय करने के लिए, आपकी .env फ़ाइल (मॉड्यूल 3 में बनाई गई) को अपडेट करने की ज़रूरत है. .env खोलें और इन सेटिंग की पुष्टि करें/इन्हें अपडेट करें:
सभी क्लाउड डिप्लॉयमेंट के लिए ज़रूरी है:
# Your actual GCP Project ID (REQUIRED)
GOOGLE_CLOUD_PROJECT=your-project-id
# GCP region for deployments (REQUIRED)
GOOGLE_CLOUD_LOCATION=us-central1
# Use Vertex AI (REQUIRED)
GOOGLE_GENAI_USE_VERTEXAI=true
# Model configuration (already set)
WORKER_MODEL=gemini-2.5-flash
CRITIC_MODEL=gemini-2.5-pro
बकेट के नाम सेट करें (deploy.sh चलाने से पहले यह ज़रूरी है):
डिपार्टमेंट स्क्रिप्ट, इन नामों के आधार पर बकेट बनाती है. इन्हें अभी सेट करें:
# Staging bucket for Agent Engine code uploads (REQUIRED for agent-engine)
STAGING_BUCKET=gs://your-project-id-staging
# Artifact storage for reports and fixed code (REQUIRED for both cloud-run and agent-engine)
ARTIFACT_BUCKET=gs://your-project-id-artifacts
दोनों बकेट के नामों में, your-project-id की जगह अपना असल प्रोजेक्ट आईडी डालें. अगर ये बकेट मौजूद नहीं हैं, तो स्क्रिप्ट इन्हें बना देगी.
ज़रूरी नहीं है कि ये वैरिएबल मौजूद हों (अगर ये वैरिएबल मौजूद नहीं हैं, तो इन्हें अपने-आप बना दिया जाता है):
# Agent Engine ID (populated after first deployment)
AGENT_ENGINE_ID=
# Cloud Run Database credentials (created automatically if blank)
CLOUD_SQL_INSTANCE_NAME=
DB_USER=
DB_PASSWORD=
DB_NAME=
पुष्टि की जांच
अगर आपको डिप्लॉयमेंट के दौरान पुष्टि करने से जुड़ी गड़बड़ियों का सामना करना पड़ता है, तो:
gcloud auth application-default login
gcloud config set project $GOOGLE_CLOUD_PROJECT
दूसरा चरण: डिप्लॉयमेंट स्क्रिप्ट को समझना
deploy.sh स्क्रिप्ट, सभी डिप्लॉयमेंट मोड के लिए एक जैसा इंटरफ़ेस उपलब्ध कराती है:
./deploy.sh {local|cloud-run|agent-engine}
स्क्रिप्ट की सुविधाएं
इन्फ़्रास्ट्रक्चर प्रोविज़निंग:
- एपीआई चालू करना (AI Platform, Storage, Cloud Build, Cloud Trace, Cloud SQL)
- आईएएम की अनुमति का कॉन्फ़िगरेशन (सेवा खाते, भूमिकाएं)
- संसाधन बनाना (बकेट, डेटाबेस, इंस्टेंस)
- सही फ़्लैग के साथ डिप्लॉयमेंट
- डिप्लॉयमेंट के बाद की पुष्टि
स्क्रिप्ट के मुख्य सेक्शन
- कॉन्फ़िगरेशन (लाइनें 1 से 35): प्रोजेक्ट, क्षेत्र, सेवा के नाम, डिफ़ॉल्ट
- हेल्पर फ़ंक्शन (लाइन 37 से 200): एपीआई चालू करना, बकेट बनाना, IAM सेटअप करना
- मुख्य लॉजिक (लाइनें 202-400): मोड के हिसाब से डिप्लॉयमेंट ऑर्केस्ट्रेशन
तीसरा चरण: एजेंट को एजेंट इंजन के लिए तैयार करना
Agent Engine पर डिप्लॉय करने से पहले, एक agent_engine_app.py फ़ाइल की ज़रूरत होती है. यह फ़ाइल, मैनेज किए गए रनटाइम के लिए आपके एजेंट को रैप करती है. यह आपके लिए पहले ही बना दिया गया है.
code_review_assistant/agent_engine_app.py देखें
👉 फ़ाइल खोलें:
"""
Agent Engine application wrapper.
This file prepares the agent for deployment to Vertex AI Agent Engine.
"""
from vertexai import agent_engines
from .agent import root_agent
# Wrap the agent in an AdkApp object for Agent Engine deployment
app = agent_engines.AdkApp(
agent=root_agent,
enable_tracing=True,
)
चौथा चरण: Agent Engine पर डिप्लॉय करना
ADK एजेंट के लिए, Agent Engine को प्रोडक्शन डिप्लॉयमेंट के लिए सुझाया गया है. इसकी वजह यह है कि यह:
- पूरी तरह से मैनेज किया गया इंफ़्रास्ट्रक्चर (कंटेनर बनाने की ज़रूरत नहीं होती)
VertexAiSessionServiceके ज़रिए सेशन को बनाए रखने की सुविधा- शून्य से अपने-आप स्केलिंग होना
- Cloud Trace इंटिग्रेशन की सुविधा डिफ़ॉल्ट रूप से चालू होती है
एजेंट इंजन, अन्य डिप्लॉयमेंट से किस तरह अलग है
इस सुविधा के काम करने के तरीके के बारे में जानें.
deploy.sh agent-engine
uses:
adk deploy agent_engine \
--project=$GOOGLE_CLOUD_PROJECT \
--region=$GOOGLE_CLOUD_LOCATION \
--staging_bucket=$STAGING_BUCKET \
--display_name="Code Review Assistant" \
--trace_to_cloud \
code_review_assistant
इस कमांड से:
- आपके Python कोड को सीधे तौर पर पैकेज करता है (Docker बिल्ड नहीं)
.envमें बताए गए स्टेजिंग बकेट में अपलोड किए गए वीडियो- यह मैनेज किया गया एजेंट इंजन इंस्टेंस बनाता है
- यह विकल्प, निगरानी के लिए Cloud Trace को चालू करता है
- यह कुकी, रनटाइम को कॉन्फ़िगर करने के लिए
agent_engine_app.pyका इस्तेमाल करती है
Cloud Run आपके कोड को कंटेनर में बदलता है. वहीं, Agent Engine आपके Python कोड को सीधे तौर पर मैनेज किए गए रनटाइम एनवायरमेंट में चलाता है. यह बिना सर्वर वाले फ़ंक्शन की तरह काम करता है.
डिप्लॉयमेंट शुरू करना
अपने प्रोजेक्ट के रूट से:
./deploy.sh agent-engine
डिप्लॉयमेंट के चरण
स्क्रिप्ट को इन फ़ेज़ में एक्ज़ीक्यूट होते हुए देखें:
Phase 1: API Enablement
✓ aiplatform.googleapis.com
✓ storage-api.googleapis.com
✓ cloudbuild.googleapis.com
✓ cloudtrace.googleapis.com
Phase 2: IAM Setup
✓ Getting project number
✓ Granting Storage Object Admin
✓ Granting AI Platform User
✓ Granting Cloud Trace Agent
Phase 3: Staging Bucket
✓ Creating gs://your-project-id-staging
✓ Setting permissions
Phase 4: Artifact Bucket
✓ Creating gs://your-project-id-artifacts
✓ Configuring access
Phase 5: Validation
✓ Checking agent.py exists
✓ Verifying root_agent defined
✓ Checking agent_engine_app.py exists
✓ Validating requirements.txt
Phase 6: Build & Deploy
✓ Packaging agent code
✓ Uploading to staging bucket
✓ Creating Agent Engine instance
✓ Configuring session persistence
✓ Setting up Cloud Trace integration
✓ Running health checks
इस प्रोसेस में 5 से 10 मिनट लगते हैं, क्योंकि इसमें एजेंट को पैकेज किया जाता है और Vertex AI के इन्फ़्रास्ट्रक्चर पर डिप्लॉय किया जाता है.
अपने एजेंट इंजन का आईडी सेव करना
डिप्लॉयमेंट पूरा होने पर:
✅ Deployment successful!
Agent Engine ID: 7917477678498709504
Resource Name: projects/123456789/locations/us-central1/reasoningEngines/7917477678498709504
Endpoint: https://us-central1-aiplatform.googleapis.com/v1/...
⚠️ IMPORTANT: Save this in your .env file:
AGENT_ENGINE_ID=7917477678498709504
अपना
.env
फ़ाइल तुरंत:
echo "AGENT_ENGINE_ID=7917477678498709504" >> .env
इस आईडी की ज़रूरत इन कामों के लिए होती है:
- डिप्लॉय किए गए एजेंट की जांच करना
- डिप्लॉयमेंट को बाद में अपडेट करना
- लॉग और ट्रेस ऐक्सेस करना
क्या डिप्लॉय किया गया था
अब आपके एजेंट इंजन के डिप्लॉयमेंट में ये शामिल हैं:
✅ समीक्षा की पूरी प्रोसेस (चार एजेंट)
✅ गड़बड़ी ठीक करने की पूरी प्रोसेस (लूप + सिंथेसाइज़र)
✅ सभी टूल (एएसटी विश्लेषण, स्टाइल की जांच, आर्टफ़ैक्ट जनरेशन)
✅ सेशन जारी रखने की सुविधा (VertexAiSessionService के ज़रिए अपने-आप चालू होती है)
✅ स्टेट मैनेजमेंट (सेशन/उपयोगकर्ता/लाइफ़टाइम टियर)
✅ ऑब्ज़र्वेबिलिटी (Cloud Trace चालू है)
✅ अपने-आप स्केल होने वाला इन्फ़्रास्ट्रक्चर
पांचवां चरण: डिप्लॉय किए गए एजेंट की जांच करना
अपनी .env फ़ाइल अपडेट करना
डेटा डिप्लॉय करने के बाद, पुष्टि करें कि आपके .env में ये शामिल हैं:
AGENT_ENGINE_ID=7917477678498709504 # From deployment output
GOOGLE_CLOUD_PROJECT=your-project-id
GOOGLE_CLOUD_LOCATION=us-central1
टेस्ट स्क्रिप्ट चलाना
इस प्रोजेक्ट में, tests/test_agent_engine.py को खास तौर पर एजेंट इंजन डिप्लॉयमेंट की टेस्टिंग के लिए शामिल किया गया है:
python tests/test_agent_engine.py
टेस्ट क्या करता है
- आपके Google Cloud प्रोजेक्ट की पुष्टि करता है
- यह कुकी, डिप्लॉय किए गए एजेंट के साथ एक सेशन बनाती है
- कोड की समीक्षा करने का अनुरोध भेजता है (डीएफ़एस बग का उदाहरण)
- सर्वर-सेंट इवेंट (एसएसई) के ज़रिए, जवाब को वापस स्ट्रीम करता है
- यह कुकी, सेशन के बने रहने और उसकी स्थिति को मैनेज करने की पुष्टि करती है
अनुमानित आउटपुट
Authenticated with project: your-project-id
Targeting Agent Engine: projects/.../reasoningEngines/7917477678498709504
Creating new session...
Created session: 4857885913439920384
Sending query to agent and streaming response:
data: {"content": {"parts": [{"text": "I'll analyze your code..."}]}}
data: {"content": {"parts": [{"text": "**Code Structure Analysis**\n..."}]}}
data: {"content": {"parts": [{"text": "**Style Check Results**\n..."}]}}
data: {"content": {"parts": [{"text": "**Test Results**\n..."}]}}
data: {"content": {"parts": [{"text": "**Final Feedback**\n..."}]}}
Stream finished.
पुष्टि करने के लिए चेकलिस्ट
- ✅ समीक्षा की पूरी प्रोसेस (चारों एजेंट)
- ✅ स्ट्रीमिंग रिस्पॉन्स में, प्रोग्रेसिव आउटपुट दिखता है
- ✅ सेशन की स्थिति, सभी अनुरोधों में बनी रहती है
- ✅ पुष्टि करने या कनेक्शन से जुड़ी कोई गड़बड़ी नहीं है
- ✅ टूल कॉल सही तरीके से काम कर रहे हैं (एएसटी विश्लेषण, स्टाइल की जांच)
- ✅ आर्टफ़ैक्ट सेव किए जाते हैं (ग्रेडिंग रिपोर्ट ऐक्सेस की जा सकती है)
वैकल्पिक तरीका: Cloud Run पर डिप्लॉय करना
हालांकि, प्रोडक्शन डिप्लॉयमेंट को स्ट्रीमलाइन करने के लिए, Agent Engine का इस्तेमाल करने का सुझाव दिया जाता है. वहीं, Cloud Run आपको ज़्यादा कंट्रोल देता है और ADK के वेब यूज़र इंटरफ़ेस (यूआई) के साथ काम करता है. इस सेक्शन में खास जानकारी दी गई है.
Cloud Run का इस्तेमाल कब करना चाहिए
अगर आपको ये सुविधाएं चाहिए, तो Cloud Run चुनें:
- उपयोगकर्ता के इंटरैक्शन के लिए ADK का वेब यूज़र इंटरफ़ेस (यूआई)
- कंटेनर एनवायरमेंट पर पूरा कंट्रोल
- डेटाबेस के कस्टम कॉन्फ़िगरेशन
- मौजूदा Cloud Run सेवाओं के साथ इंटिग्रेशन
Cloud Run डिप्लॉयमेंट कैसे काम करता है
इस सुविधा के काम करने के तरीके के बारे में जानें.
deploy.sh cloud-run
uses:
adk deploy cloud_run \
--project=$GOOGLE_CLOUD_PROJECT \
--region=$GOOGLE_CLOUD_LOCATION \
--service_name="code-review-assistant" \
--app_name="code_review_assistant" \
--port=8080 \
--with_ui \
--artifact_service_uri="gs://$ARTIFACT_BUCKET" \
--trace_to_cloud \
code_review_assistant
इस कमांड से:
- यह आपके एजेंट कोड के साथ एक Docker कंटेनर बनाता है
- Google Artifact Registry में पुश करता है
- Cloud Run सेवा के तौर पर डिप्लॉय करता है
- इसमें ADK का वेब यूज़र इंटरफ़ेस (यूआई) (
--with_ui) शामिल है - Cloud SQL कनेक्शन कॉन्फ़िगर करता है (इसे स्क्रिप्ट ने शुरुआती डिप्लॉयमेंट के बाद जोड़ा है)
Agent Engine और Cloud Run में मुख्य अंतर यह है कि Cloud Run आपके कोड को कंटेनर में रखता है और सेशन को बनाए रखने के लिए डेटाबेस की ज़रूरत होती है. वहीं, Agent Engine इन दोनों को अपने-आप मैनेज करता है.
Cloud Run डिप्लॉयमेंट कमांड
./deploy.sh cloud-run
क्या अलग है
इन्फ़्रास्ट्रक्चर:
- कंटेनर में डिप्लॉयमेंट (ADK अपने-आप Docker बनाता है)
- सेशन को बनाए रखने के लिए Cloud SQL (PostgreSQL)
- स्क्रिप्ट की मदद से अपने-आप बनने वाला डेटाबेस या मौजूदा इंस्टेंस का इस्तेमाल करने वाला डेटाबेस
सेशन मैनेजमेंट:
VertexAiSessionServiceके बजायDatabaseSessionServiceका इस्तेमाल करता है.envमें डेटाबेस क्रेडेंशियल (या अपने-आप जनरेट होने वाले क्रेडेंशियल) ज़रूरी हैं- PostgreSQL डेटाबेस में स्थिति बनी रहती है
यूज़र इंटरफ़ेस (यूआई) के साथ काम करता है:
--with_uiफ़्लैग के ज़रिए वेब यूज़र इंटरफ़ेस (स्क्रिप्ट से मैनेज किया जाता है)- ऐक्सेस करने की तारीख:
https://code-review-assistant-xyz.a.run.app
आपने क्या-क्या हासिल किया
आपके प्रोडक्शन डिप्लॉयमेंट में ये शामिल हैं:
✅ deploy.shस्क्रिप्ट
के ज़रिए अपने-आप चालू होने की सुविधा ✅ मैनेज किया गया इन्फ़्रास्ट्रक्चर (एजेंट इंजन, स्केलिंग, डेटा सेव करने, और मॉनिटरिंग को मैनेज करता है)
✅ सभी मेमोरी टियर (सेशन/उपयोगकर्ता/लाइफ़टाइम) में डेटा सेव करने की सुविधा ✅ क्रेडेंशियल को सुरक्षित तरीके से मैनेज करने की सुविधा (अपने-आप जनरेट होने और IAM सेटअप होने की सुविधा)
✅ स्केल करने लायक आर्किटेक्चर (एक साथ इस्तेमाल करने वाले उपयोगकर्ताओं की संख्या शून्य से लेकर हज़ारों तक हो सकती है)
✅ निगरानी करने की सुविधा पहले से मौजूद है (Cloud Trace इंटिग्रेशन चालू है)
✅ प्रोडक्शन-ग्रेड गड़बड़ी ठीक करने और डेटा वापस पाने की सुविधा
मास्टर किए गए मुख्य कॉन्सेप्ट
डिप्लॉयमेंट की तैयारी:
agent_engine_app.py: यह एजेंट कोAdkAppके साथ रैप करता है, ताकि एजेंट इंजन का इस्तेमाल किया जा सकेAdkApp,VertexAiSessionServiceको लगातार इस्तेमाल करने के लिए अपने-आप कॉन्फ़िगर करता हैenable_tracing=Trueके ज़रिए ट्रेसिंग की सुविधा चालू की गई
डिप्लॉयमेंट के निर्देश:
adk deploy agent_engine: इसमें Python कोड को पैकेज किया जाता है, लेकिन कंटेनर नहीं होतेadk deploy cloud_run: Docker कंटेनर अपने-आप बनाता हैgcloud run deploy: कस्टम Dockerfile वाला विकल्प
डिप्लॉयमेंट के विकल्प:
- एजेंट इंजन: पूरी तरह से मैनेज किया जाता है, प्रोडक्शन के लिए सबसे तेज़
- Cloud Run: ज़्यादा कंट्रोल, वेब यूज़र इंटरफ़ेस (यूआई) के साथ काम करता है
- GKE: Kubernetes को बेहतर तरीके से कंट्रोल करना (GKE डिप्लॉयमेंट गाइड देखें)
मैनेज की जा रही सेवाएं:
- Agent Engine, सेशन को बनाए रखने की सुविधा को अपने-आप मैनेज करता है
- Cloud Run के लिए, डेटाबेस सेटअप करना ज़रूरी है. हालांकि, इसे अपने-आप भी बनाया जा सकता है
- दोनों, GCS के ज़रिए आर्टफ़ैक्ट स्टोरेज की सुविधा देते हैं
सेशन मैनेजमेंट:
- एजेंट इंजन:
VertexAiSessionService(अपने-आप) - Cloud Run:
DatabaseSessionService(Cloud SQL) - लोकल:
InMemorySessionService(कुछ समय के लिए उपलब्ध)
आपका एजेंट लाइव है
कोड की समीक्षा करने वाला आपका असिस्टेंट अब यह है:
- एचटीटीपीएस एपीआई एंडपॉइंट के ज़रिए ऐक्सेस किया जा सकता है
- Persistent, जिसमें रीस्टार्ट होने के बाद भी स्थिति बनी रहती है
- टीम के बढ़ने पर, स्केल करने की सुविधा अपने-आप चालू हो जाती है
- अनुरोध के पूरे ट्रेस के साथ ऑब्ज़र्वेबल
- स्क्रिप्ट किए गए डिप्लॉयमेंट की मदद से बनाए रखने में आसान
आगे क्या करना है? आठवें मॉड्यूल में, आपको Cloud Trace का इस्तेमाल करना सिखाया जाएगा. इससे आपको अपने एजेंट की परफ़ॉर्मेंस को समझने, समीक्षा और ठीक करने वाली पाइपलाइन में आने वाली समस्याओं की पहचान करने, और एक्ज़ीक्यूशन के समय को ऑप्टिमाइज़ करने में मदद मिलेगी.
8. प्रोडक्शन की जांच करने की क्षमता
परिचय
कोड की समीक्षा करने वाला आपका असिस्टेंट अब Agent Engine पर डिप्लॉय हो गया है और प्रोडक्शन में काम कर रहा है. हालांकि, आपको कैसे पता चलेगा कि यह सुविधा ठीक से काम कर रही है? क्या आपके पास इन अहम सवालों के जवाब हैं:
- क्या एजेंट तुरंत जवाब दे रहा है?
- कौनसी कार्रवाइयां सबसे धीमी हैं?
- क्या फ़िक्स लूप सही तरीके से काम कर रहे हैं?
- परफ़ॉर्मेंस बॉटलनेक कहां हैं?
ऑब्ज़र्वेबिलिटी के बिना, आपको कुछ भी नहीं दिखेगा. आपने डिप्लॉयमेंट के दौरान जिस --trace-to-cloud फ़्लैग का इस्तेमाल किया था उसने Cloud Trace को अपने-आप चालू कर दिया. इससे आपको हर उस अनुरोध की पूरी जानकारी मिलती है जिसे आपका एजेंट प्रोसेस करता है.
इस मॉड्यूल में, आपको ट्रेस पढ़ने, एजेंट की परफ़ॉर्मेंस की विशेषताओं को समझने, और ऑप्टिमाइज़ेशन के लिए क्षेत्रों की पहचान करने का तरीका बताया जाएगा.
ट्रेस और स्पैन को समझना
ट्रेस क्या होता है?
ट्रेस, किसी एक अनुरोध को हैंडल करने के दौरान एजेंट की पूरी टाइमलाइन होती है. यह उपयोगकर्ता की क्वेरी भेजने से लेकर, फ़ाइनल जवाब मिलने तक की पूरी प्रोसेस को कैप्चर करता है. हर ट्रेस में यह जानकारी दिखती है:
- अनुरोध पूरा होने में लगने वाला कुल समय
- वे सभी कार्रवाइयां जो पूरी हो चुकी हैं
- ऑपरेशन एक-दूसरे से कैसे जुड़े होते हैं (पैरंट-चाइल्ड रिलेशनशिप)
- हर ऑपरेशन कब शुरू और खत्म हुआ
स्पैन क्या होता है?
स्पैन, ट्रेस में काम की एक यूनिट को दिखाता है. कोड की समीक्षा करने वाले असिस्टेंट में, स्पैन के ये टाइप आम तौर पर इस्तेमाल किए जाते हैं:
agent_run: किसी एजेंट (रूट एजेंट या सब-एजेंट) को लागू करनाcall_llm: किसी लैंग्वेज मॉडल से किया गया अनुरोधexecute_tool: टूल फ़ंक्शन को लागू करनाstate_read/state_write: स्टेट मैनेजमेंट के ऑपरेशनcode_executor: जांचों के साथ कोड चलाया जा रहा है
स्पैन में ये शामिल होते हैं:
- नाम: यह किस कार्रवाई को दिखाता है
- अवधि: इसमें कितना समय लगा
- एट्रिब्यूट: मेटाडेटा, जैसे कि मॉडल का नाम, टोकन की संख्या, इनपुट/आउटपुट
- स्टेटस: सफल या असफल
- पैरंट/चाइल्ड रिलेशनशिप: किस ऑपरेशन से कौनसी कार्रवाई ट्रिगर हुई
ऑटोमैटिक इंस्ट्रूमेंटेशन
--trace-to-cloud की मदद से डिप्लॉय करने पर, ADK अपने-आप इन मेट्रिक को ट्रैक करता है:
- हर एजेंट इनवोकेशन और सब-एजेंट कॉल
- टोकन की संख्या के साथ एलएलएम के सभी अनुरोध
- इनपुट/आउटपुट के साथ टूल के इस्तेमाल की जानकारी
- स्टेट से जुड़ी कार्रवाइयां (पढ़ें/लिखें)
- फ़िक्स पाइपलाइन में लूप के दोहराव
- गड़बड़ी की स्थितियां और फिर से कोशिश करना
कोड में कोई बदलाव करने की ज़रूरत नहीं है - ट्रेसिंग की सुविधा, ADK के रनटाइम में पहले से मौजूद होती है.
पहला चरण: Cloud Trace Explorer को ऐक्सेस करना
Google Cloud Console में Cloud Trace खोलें:
- Cloud Trace Explorer पर जाएं
- ड्रॉपडाउन से अपना प्रोजेक्ट चुनें (यह पहले से चुना हुआ होना चाहिए)
- आपको मॉड्यूल 7 में, अपने टेस्ट के ट्रेस दिखने चाहिए
अगर आपको अब तक ट्रेस नहीं दिख रहे हैं, तो:
मॉड्यूल 7 में किए गए टेस्ट से ट्रेस जनरेट होने चाहिए. अगर सूची खाली है, तो कुछ ट्रेस डेटा जनरेट करें:
python tests/test_agent_engine.py
कंसोल में ट्रेस दिखने में एक से दो मिनट लगते हैं.
आपको क्या दिख रहा है
ट्रेस एक्सप्लोरर में यह जानकारी दिखती है:
- ट्रेस की सूची: हर पंक्ति एक पूरे अनुरोध को दिखाती है
- टाइमलाइन: अनुरोध कब किए गए
- कुल समय: हर अनुरोध को पूरा होने में कितना समय लगा
- अनुरोध की जानकारी: टाइमस्टैंप, इंतज़ार का समय, स्पैन की संख्या
यह आपके प्रोडक्शन ट्रैफ़िक का लॉग है. आपके एजेंट के साथ होने वाला हर इंटरैक्शन, एक ट्रेस बनाता है.
दूसरा चरण: समीक्षा पाइपलाइन के ट्रेस की जांच करना
वॉटरफ़ॉल व्यू खोलने के लिए, सूची में मौजूद किसी भी ट्रेस पर क्लिक करें
आपको एक गैंट चार्ट दिखेगा, जिसमें कैंपेन को लागू करने की पूरी टाइमलाइन दिखाई जाएगी. समीक्षा पाइपलाइन का सामान्य ट्रेस ऐसा दिखता है:
invocation (2.3s) ────────────────────────────────────────────►
├── agent_run: CodeReviewAssistant (2.2s) ──────────────────►
│ ├── state_read: CODE_TO_REVIEW (0.01s) ►
│ ├── agent_run: CodeReviewPipeline (2.1s) ─────────────►
│ │ ├── agent_run: CodeAnalyzer (0.3s) ──────►
│ │ │ ├── execute_tool: analyze_code_structure (0.1s) ──►
│ │ │ └── call_llm: gemini-2.5-flash (0.15s) ────►
│ │ ├── agent_run: StyleChecker (0.2s) ──────►
│ │ │ ├── execute_tool: check_code_style (0.1s) ──►
│ │ │ └── call_llm: gemini-2.5-flash (0.08s) ──►
│ │ ├── agent_run: TestRunner (1.2s) ─────────────►
│ │ │ └── code_executor: BuiltInCodeExecutor (0.9s) ────►
│ │ └── agent_run: FeedbackSynthesizer (0.4s) ────────►
│ │ └── call_llm: gemini-2.5-flash (0.28s) ────►
वॉटरफ़ॉल चार्ट को पढ़ना
हर बार एक स्पैन दिखाता है. हॉरिजॉन्टल पोज़िशन से पता चलता है कि यह कब शुरू हुआ था. वहीं, इसकी लंबाई से पता चलता है कि इसमें कितना समय लगा.
इस ट्रेस से मिली अहम जानकारी:
- कुल इंतज़ार का समय: अनुरोध से लेकर जवाब मिलने तक 2.3 सेकंड
- क्रिटिकल पाथ: TestRunner को 1.2 सेकंड (कुल समय का 52%) लगता है
- बॉटलनैक: TestRunner में कोड को एक्ज़ीक्यूट होने में 0.9 सेकंड लगते हैं. यह TestRunner के समय का 75% है
- स्टेट ऑपरेशन: बहुत तेज़ (हर ऑपरेशन में 10 मि॰से॰) - कोई समस्या नहीं है
- पाइपलाइन का स्ट्रक्चर: क्रम से लागू करना - CodeAnalyzer → StyleChecker → TestRunner → FeedbackSynthesizer
स्पैन की जानकारी देखना
पर क्लिक करें
call_llm: gemini-2.5-flash
span under FeedbackSynthesizer
आपको इस एलएलएम कॉल के लिए एट्रिब्यूट की पूरी जानकारी दिखेगी:
{
"name": "call_llm",
"span_kind": "LLM",
"duration": "280ms",
"attributes": {
"llm.model": "models/gemini-2.5-flash",
"llm.request_type": "GenerateContent",
"llm.usage.prompt_tokens": 845,
"llm.usage.completion_tokens": 234,
"llm.usage.total_tokens": 1079,
"llm.response.finish_reason": "STOP",
"status_code": "OK"
}
}
इससे यह पता चलता है कि:
- किस मॉडल का इस्तेमाल किया गया
- कितने टोकन इस्तेमाल किए गए (इनपुट + आउटपुट)
- अनुरोध की अवधि
- सफल/असफल स्थिति
- पूरा प्रॉम्प्ट, एट्रिब्यूट में भी दिखता है. इसे देखने के लिए स्क्रोल करें
पाइपलाइन फ़्लो को समझना
ध्यान दें कि ट्रेस से आपके आर्किटेक्चर के बारे में कैसे पता चलता है:
- रूट एजेंट (CodeReviewAssistant) को अनुरोध मिलता है
- State read, समीक्षा करने के लिए कोड वापस लाता है
- समीक्षा पाइपलाइन, चार सब-एजेंट को क्रम से व्यवस्थित करती है
- हर सब-एजेंट, अपना काम पूरा करने के लिए टूल और एलएलएम कॉल का इस्तेमाल करता है
- आखिरी जवाब, वापस ऊपर की ओर जाता है
इस जानकारी से, आपको यह समझने में मदद मिलती है कि हर अनुरोध के दौरान क्या होता है.
तीसरा चरण: फ़िक्स पाइपलाइन के ट्रेस का विश्लेषण करना
फ़िक्स पाइपलाइन ज़्यादा जटिल होती है, क्योंकि इसमें लूप शामिल होते हैं. आइए, देखते हैं कि ट्रेस, बार-बार होने वाली गतिविधि को कैसे कैप्चर करते हैं.
ऐसा ट्रेस ढूंढें जिसके स्पैन के नाम में "CodeFixPipeline" शामिल हो
आपको अपने ट्रेस में स्क्रोल करना पड़ सकता है या ऐसा अनुरोध भेजना पड़ सकता है जिससे समस्या ठीक करने की प्रोसेस शुरू हो जाए. अगर आपके पास ऐसा कोई कोड नहीं है, तो इसे जनरेट किया जा सकता है:
# In your test script, respond "yes" when asked to fix issues
python tests/test_agent_engine.py
लूप स्ट्रक्चर की जांच करना
दो इटरेशन वाली फ़िक्स पाइपलाइन का ट्रेस ऐसा दिखता है:
agent_run: CodeFixPipeline (8.5s) ───────────────────────►
├── agent_run: FixAttemptLoop (7.8s) ───────────────────►
│ ├── loop_iteration: 1 (3.2s) ──────────►
│ │ ├── agent_run: CodeFixer (0.8s) ────►
│ │ │ └── call_llm: gemini-2.5-flash (0.7s) ───►
│ │ ├── agent_run: FixTestRunner (1.8s) ─────────►
│ │ │ └── code_executor: BuiltInCodeExecutor (1.5s) ─────►
│ │ └── agent_run: FixValidator (0.6s) ────►
│ │ ├── execute_tool: validate_fixed_style (0.2s) ──►
│ │ └── state_write: FIX_STATUS = "PARTIAL" ►
│ │
│ ├── loop_iteration: 2 (4.5s) ─────────────────►
│ │ ├── agent_run: CodeFixer (1.0s) ──────►
│ │ │ └── call_llm: gemini-2.5-flash (0.9s) ───►
│ │ ├── agent_run: FixTestRunner (2.0s) ────────►
│ │ │ └── code_executor: BuiltInCodeExecutor (1.7s) ─────►
│ │ └── agent_run: FixValidator (1.5s) ──────►
│ │ ├── execute_tool: compile_fix_report (0.3s) ──►
│ │ └── state_write: FIX_STATUS = "SUCCESSFUL" ►
│ │
│ └── loop_exit: escalation_triggered ►
│
└── agent_run: FixSynthesizer (0.7s) ────►
├── execute_tool: save_fix_report (0.2s) ──►
└── call_llm: gemini-2.5 (0.4s) ────►
लूप के बारे में मुख्य बातें
इटरेशन पैटर्न:
- दो बार कोशिश की गई: पहली कोशिश में कुछ हद तक सफलता मिली, दूसरी कोशिश में पूरी तरह से सफलता मिली
- बढ़ी हुई लागत: दूसरे इटरेशन में ज़्यादा समय लगता है (4.5 सेकंड बनाम 3.2 सेकंड)
- स्टेट ट्रैकिंग: हर इटरेशन, FIX_STATUS को स्टेट में लिखता है
- लूप से बाहर निकलने का तरीका: जब FIX_STATUS = "SUCCESSFUL" होता है, तब एस्केलेशन के ज़रिए लूप खत्म हो जाता है
इससे क्या पता चलता है:
- लूप आर्किटेक्चर सही तरीके से काम कर रहा हो
- ज़्यादातर गड़बड़ियां एक से दो इटरेशन में ठीक हो जाती हैं (अच्छा डिज़ाइन)
- हर इटरेशन में ये शामिल होते हैं: फ़िक्स जनरेट करना → टेस्टिंग → पुष्टि करना
- कोड को लागू करने में हर इटरेशन में ज़्यादा समय लगता है (1.5 से 1.7 सेकंड)
- शर्तें पूरी होने पर लूप सही तरीके से बंद हो जाता है
शुल्क का ब्रेकडाउन:
- पहली बार: 3.2 सेकंड
- दूसरा जवाब: 4.5 सेकंड (कॉन्टेक्स्ट इकट्ठा होने की वजह से ज़्यादा समय लगा)
- कुल लूप: 7.8 सेकंड
- सिंथेसिस: 0.7 सेकंड
- पाइपलाइन को ठीक करने में लगा कुल समय: 8.5 सेकंड
समीक्षा पाइपलाइन से तुलना करना
समीक्षा पाइपलाइन: ~2.3 सेकंड
गड़बड़ी ठीक करने वाली पाइपलाइन: ~8.5 सेकंड (दो बार दोहराने पर)
फ़िक्स पाइपलाइन में ~3.7 गुना ज़्यादा समय लगता है, जो कि सही है:
- इसमें बार-बार सुधार करना शामिल है
- यह कोड को कई बार चलाता है (हर इटरेशन में एक बार)
- यह पिछले अनुरोधों से कॉन्टेक्स्ट इकट्ठा करता है
चौथा चरण: आपको क्या पता चला
परफ़ॉर्मेंस पैटर्न
ट्रेस की जांच करने के बाद, अब आपको यह पता है:
पाइपलाइन की समीक्षा करना:
- आम तौर पर अवधि: दो से तीन सेकंड
- ज़्यादा समय लेने वाला मुख्य प्रोसेस: TestRunner (कोड एक्ज़ीक्यूट करना)
- एलएलएम कॉल: तेज़ (हर कॉल में 100 से 300 मि॰से॰ लगते हैं)
- स्टेट ऑपरेशन: न के बराबर (10 मि॰से॰)
पाइपलाइन ठीक करना:
- आम तौर पर लगने वाला समय: हर इटरेशन के लिए 4 से 5 सेकंड
- ज़्यादातर समस्याएं ठीक करने के लिए: 1-2 इटरेशन
- कोड को लागू करना: हर इटरेशन में 1.5 से 2.0 सेकंड
- प्रोग्रेसिव लागत: बाद के इटरेशन में ज़्यादा समय लगता है
तेज़ी से क्या होता है:
- स्टेट को पढ़ना/लिखना (10 मि॰से॰)
- विश्लेषण के लिए टूल का इस्तेमाल (100 मि॰से॰)
- एलएलएम को अलग-अलग कॉल (100 से 300 मि॰से॰)
कौनसी प्रोसेस धीमी है, लेकिन ज़रूरी है:
- टेस्ट के साथ कोड लागू करना (0.9 से 2.0 सेकंड)
- लूप के कई इटरेशन (संचयी)
समस्याएं कहां देखें
प्रोडक्शन में ट्रेस की समीक्षा करते समय, इन बातों का ध्यान रखें:
- सामान्य से ज़्यादा समय तक चलने वाले ट्रेस (>15 सेकंड) - जांच करें कि क्या गड़बड़ी हुई
- फ़ेल हुए स्पैन (status != OK) - एक्ज़ीक्यूशन में गड़बड़ियां
- लूप के बहुत ज़्यादा इटरेशन (>2) - क्वालिटी से जुड़ी समस्याएं ठीक करें
- टोकन की संख्या बहुत ज़्यादा है - प्रॉम्प्ट को ऑप्टिमाइज़ करने के अवसर
आपने क्या सीखा
Cloud Trace की मदद से, अब आपको यह जानकारी मिलती है:
✅ अनुरोध का फ़्लो: आपकी पाइपलाइन के ज़रिए अनुरोध पूरा होने का पाथ
✅ परफ़ॉर्मेंस की विशेषताएं: कौनसी चीज़ तेज़ी से होती है, कौनसी धीमी गति से होती है, और क्यों
✅ लूप का व्यवहार: लूप कैसे काम करते हैं और कब बंद होते हैं
✅ स्पैन का क्रम: ऑपरेशन एक-दूसरे में कैसे नेस्ट होते हैं
✅ ट्रेस नेविगेशन: वॉटरफ़ॉल चार्ट को असरदार तरीके से पढ़ना
✅ टोकन की जानकारी: एलएलएम की लागत कहां बढ़ती है
मास्टर किए गए मुख्य कॉन्सेप्ट
ट्रेस और स्पैन:
- ट्रेस = अनुरोध की पूरी टाइमलाइन
- स्पैन = ट्रेस में मौजूद अलग-अलग कार्रवाइयाँ
- वॉटरफ़ॉल व्यू में, एक्ज़ीक्यूशन हैरारकी दिखाई जाती है
- ADK के ज़रिए अपने-आप इंस्ट्रूमेंटेशन की सुविधा
परफ़ॉर्मेंस का विश्लेषण:
- Gantt चार्ट विज़ुअलाइज़ेशन को पढ़ना
- क्रिटिकल पाथ की पहचान करना
- अवधि के डिस्ट्रिब्यूशन को समझना
- बॉटलनेक का पता लगाना
प्रोडक्शन दिखने की सेटिंग:
- हर ऑपरेशन अपने-आप ट्रेस हो जाता है
- एलएलएम के हर कॉल के लिए, टोकन के इस्तेमाल की जानकारी कैप्चर की गई
- स्टेट में हुए बदलावों को देखा और ट्रैक किया जा सकता है
- लूप के हर वर्शन को अलग-अलग ट्रैक किया जाता है
आगे क्या करना है
Cloud Trace के बारे में ज़्यादा जानें:
- समस्याओं का पता लगाने के लिए, ट्रेस की नियमित तौर पर निगरानी करना
- परफ़ॉर्मेंस में गिरावट का पता लगाने के लिए, ट्रेस की तुलना करना
- ऑप्टिमाइज़ेशन से जुड़े फ़ैसले लेने के लिए, ट्रेस डेटा का इस्तेमाल करना
- धीमे अनुरोधों को ढूंढने के लिए, अवधि के हिसाब से फ़िल्टर करना
बेहतर मॉनिटरिंग (ज़रूरी नहीं):
- ज़्यादा बेहतर तरीके से विश्लेषण करने के लिए, ट्रेस को BigQuery में एक्सपोर्ट करें (दस्तावेज़)
- Cloud Monitoring में कस्टम डैशबोर्ड बनाना
- परफ़ॉर्मेंस में गिरावट आने पर सूचनाएं पाने की सुविधा सेट अप करना
- ऐप्लिकेशन लॉग के साथ ट्रेस को कोरिलेट करना
9. नतीजा: प्रोटोटाइप से प्रोडक्शन तक
आपने क्या बनाया है
आपने सिर्फ़ सात लाइनों के कोड से शुरुआत की और प्रोडक्शन-ग्रेड एआई एजेंट सिस्टम बनाया:
# Where we started (7 lines)
agent = Agent(
model="gemini-2.5-flash",
instruction="Review Python code for issues"
)
# Where we ended (production system)
- Two distinct multi-agent pipelines (review and fix) built from 8 specialized agents.
- An iterative fix loop architecture for automated validation and retries.
- Real AST-based code analysis tools for deterministic, accurate feedback.
- Robust state management using the "constants pattern" for type-safe communication.
- Fully automated deployment to a managed, scalable cloud infrastructure.
- Complete, built-in observability with Cloud Trace for production monitoring.
आर्किटेक्चर के मुख्य पैटर्न की जानकारी
पैटर्न | लागू करना | प्रोडक्शन पर असर |
टूल इंटिग्रेशन | AST विश्लेषण, स्टाइल की जांच करना | एलएलएम की राय के बजाय, असल पुष्टि |
सीक्वेंशियल पाइपलाइन | समीक्षा करें → वर्कफ़्लो ठीक करें | अनुमान लगाया जा सकने वाला और डीबग किया जा सकने वाला एक्ज़ीक्यूशन |
लूप आर्किटेक्चर | शर्तों के साथ बार-बार ठीक करना | जब तक लक्ष्य हासिल नहीं हो जाता, तब तक खुद को बेहतर बनाता रहता है |
स्टेट मैनेजमेंट | कॉन्स्टेंट पैटर्न, तीन लेयर वाली मेमोरी | टाइप-सेफ़ और बनाए रखने लायक स्टेट हैंडलिंग |
प्रोडक्शन डिप्लॉयमेंट | deploy.sh के ज़रिए एजेंट इंजन | मैनेज किया गया, बड़े पैमाने पर इस्तेमाल किया जा सकने वाला इन्फ़्रास्ट्रक्चर |
जांचने की क्षमता | Cloud Trace इंटिग्रेशन | प्रोडक्शन के व्यवहार के बारे में पूरी जानकारी |
Traces से प्रोडक्शन की अहम जानकारी
Cloud Trace के डेटा से अहम जानकारी मिली है:
✅ बॉटलनैक की पहचान की गई: TestRunner के एलएलएम कॉल में ज़्यादा समय लगता है
✅ टूल की परफ़ॉर्मेंस: AST विश्लेषण 100 मि॰से॰ में पूरा होता है (बहुत अच्छा)
✅ सफलता की दर: गड़बड़ियों को ठीक करने के लिए, दो से तीन बार दोहराने की ज़रूरत होती है
✅ टोकन का इस्तेमाल: हर समीक्षा के लिए ~600 टोकन और गड़बड़ियों को ठीक करने के लिए ~1800 टोकन
इन अहम जानकारी से, लगातार सुधार करने में मदद मिलती है.
संसाधन मिटाना (ज़रूरी नहीं)
अगर आपको एक्सपेरिमेंट नहीं करना है और शुल्क से बचना है, तो:
एजेंट इंजन का डिप्लॉयमेंट मिटाना:
import vertexai
client = vertexai.Client( # For service interactions via client.agent_engines
project="PROJECT_ID",
location="LOCATION",
)
RESOURCE_NAME = "projects/{PROJECT_ID}/locations/{LOCATION}/reasoningEngines/{RESOURCE_ID}"
client.agent_engines.delete(
name=RESOURCE_NAME,
force=True, # Optional, if the agent has resources (e.g. sessions, memory)
)
Cloud Run सेवा को मिटाएं (अगर बनाई गई है):
gcloud run services delete code-review-assistant \
--region=$GOOGLE_CLOUD_LOCATION \
--quiet
Cloud SQL इंस्टेंस मिटाएं (अगर बनाया गया है):
gcloud sql instances delete your-project-db \
--quiet
स्टोरेज बकेट खाली करें:
gsutil -m rm -r gs://your-project-staging
gsutil -m rm -r gs://your-project-artifacts
अगले चरण
बुनियादी सेटअप पूरा होने के बाद, इन बेहतर सुविधाओं का इस्तेमाल करें:
- ज़्यादा भाषाएँ जोड़ें: JavaScript, Go, Java के लिए टूल उपलब्ध कराएँ
- GitHub के साथ इंटिग्रेट करें: पीआर की अपने-आप समीक्षाएं
- कैशिंग लागू करना: सामान्य पैटर्न के लिए इंतज़ार का समय कम करना
- खास एजेंट जोड़ना: सुरक्षा स्कैनिंग, परफ़ॉर्मेंस का विश्लेषण
- A/B टेस्टिंग चालू करें: अलग-अलग मॉडल और प्रॉम्प्ट की तुलना करें
- मेट्रिक एक्सपोर्ट करना: खास निगरानी वाले प्लैटफ़ॉर्म पर ट्रेस भेजना
सीखने लायक ज़रूरी बातें
- आसान तरीके से शुरू करें और तेज़ी से दोहराएं: मैनेज किए जा सकने वाले चरणों में प्रोडक्शन के लिए सात लाइनें
- प्रॉम्प्ट के बजाय टूल का इस्तेमाल करना: "कृपया गड़बड़ियों की जांच करें" के बजाय, एएसटी का रीयल विश्लेषण करना ज़्यादा बेहतर होता है
- स्टेट मैनेजमेंट ज़रूरी है: कॉन्स्टेंट पैटर्न से टाइपो की वजह से होने वाली गड़बड़ियों को रोका जा सकता है
- लूप के लिए, बाहर निकलने की शर्तें ज़रूरी हैं: हमेशा ज़्यादा से ज़्यादा दोहराव और एस्केलेशन सेट करें
- ऑटोमेशन की मदद से डिप्लॉय करना: deploy.sh, सभी जटिलताओं को मैनेज करता है
- निगरानी करना ज़रूरी है: मेज़रमेंट के बिना, किसी भी चीज़ को बेहतर नहीं बनाया जा सकता
सीखते रहने के लिए संसाधन
- ADK से जुड़े दस्तावेज़
- ADK के ऐडवांस पैटर्न
- एजेंट इंजन गाइड
- Cloud Run के दस्तावेज़
- Cloud Trace के बारे में दस्तावेज़
आपका सफ़र जारी है
आपने कोड की समीक्षा करने वाले एक असिस्टेंट से ज़्यादा कुछ बनाया है. आपने किसी भी प्रोडक्शन एआई एजेंट को बनाने के पैटर्न में महारत हासिल कर ली है:
✅ कई खास एजेंटों के साथ जटिल वर्कफ़्लो
✅ असली टूल इंटिग्रेशन, ताकि सही तरीके से काम किया जा सके
✅ सही तरीके से निगरानी रखने के लिए, प्रोडक्शन डिप्लॉयमेंट
✅ सिस्टम को बनाए रखने के लिए, स्टेट मैनेजमेंट
ये पैटर्न, सामान्य असिस्टेंट से लेकर जटिल ऑटोनॉमस सिस्टम तक होते हैं. यहां बनाई गई आपकी बुनियाद, आपको ज़्यादा बेहतर एजेंट आर्किटेक्चर को समझने में मदद करेगी.
एआई एजेंट को प्रोडक्शन के लिए तैयार करने की प्रोसेस में आपका स्वागत है. कोड की समीक्षा करने में मदद करने वाली सुविधा, सिर्फ़ एक शुरुआत है.