1. লেট নাইট কোড রিভিউ
দুপুর ২টা বাজে
আপনি ঘন্টার জন্য ডিবাগ করা হয়েছে. ফাংশন সঠিক দেখায়, কিন্তু কিছু ভুল. আপনি সেই অনুভূতিটি জানেন - যখন কোডটি কাজ করা উচিত কিন্তু কাজ করে না, এবং আপনি কেন আর দেখতে পাচ্ছেন না কারণ আপনি এটির দিকে অনেকক্ষণ তাকিয়ে আছেন৷
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
এআই ডেভেলপারের যাত্রা
আপনি যদি এটি পড়ছেন, আপনি সম্ভবত AI কোডিং-এ নিয়ে আসা রূপান্তরটি অনুভব করেছেন। Gemini Code Assist , Claude Code , এবং Cursor এর মত টুলগুলো পরিবর্তন করেছে যে আমরা কিভাবে কোড লিখি। এগুলি বয়লারপ্লেট তৈরি করা, বাস্তবায়নের পরামর্শ দেওয়া এবং বিকাশকে ত্বরান্বিত করার জন্য অবিশ্বাস্য।
কিন্তু আপনি এখানে আছেন কারণ আপনি আরও গভীরে যেতে চান। আপনি এই AI সিস্টেমগুলি কীভাবে তৈরি করবেন তা বুঝতে চান, কেবল সেগুলি ব্যবহার করবেন না। আপনি এমন কিছু তৈরি করতে চান যা:
- অনুমানযোগ্য, সনাক্তযোগ্য আচরণ আছে
- আত্মবিশ্বাসের সাথে উৎপাদনে স্থাপন করা যেতে পারে
- আপনি নির্ভর করতে পারেন এমন ধারাবাহিক ফলাফল প্রদান করে
- এটা ঠিক কিভাবে সিদ্ধান্ত নেয় আপনাকে দেখায়
ভোক্তা থেকে সৃষ্টিকর্তা পর্যন্ত
আজ, আপনি এআই সরঞ্জামগুলি ব্যবহার করে সেগুলি তৈরি করতে লাফিয়ে উঠবেন। আপনি একটি মাল্টি-এজেন্ট সিস্টেম তৈরি করবেন যা:
- কোড গঠন নির্ধারকভাবে বিশ্লেষণ করে
- আচরণ যাচাই করার জন্য প্রকৃত পরীক্ষা চালায়
- বাস্তব লিন্টারের সাথে শৈলী সম্মতি যাচাই করে
- ফলাফলগুলিকে কার্যযোগ্য প্রতিক্রিয়ার মধ্যে সংশ্লেষিত করে
- সম্পূর্ণ পর্যবেক্ষণযোগ্যতার সাথে Google ক্লাউডে স্থাপন করে
2. আপনার প্রথম এজেন্ট স্থাপনা
বিকাশকারীর প্রশ্ন
"আমি LLM বুঝতে পারি, আমি API ব্যবহার করেছি, কিন্তু কিভাবে আমি পাইথন স্ক্রিপ্ট থেকে এমন একটি প্রোডাকশন এআই এজেন্টে যাবো যা স্কেল করে?"
আসুন আপনার পরিবেশকে সঠিকভাবে সেট আপ করে এর উত্তর দিই, তারপর উত্পাদনের ধরণগুলিতে ডুব দেওয়ার আগে মূল বিষয়গুলি বোঝার জন্য একটি সাধারণ এজেন্ট তৈরি করুন।
অপরিহার্য সেটআপ প্রথম
আমরা কোনো এজেন্ট তৈরি করার আগে, আপনার Google ক্লাউড পরিবেশ প্রস্তুত আছে তা নিশ্চিত করি।
Google ক্লাউড ক্রেডিট প্রয়োজন?
গুগল ক্লাউড কনসোলের শীর্ষে ক্লাউড শেল সক্রিয় করুন ক্লিক করুন (এটি ক্লাউড শেল প্যানের শীর্ষে টার্মিনাল আকারের আইকন),
আপনার Google ক্লাউড প্রকল্প আইডি খুঁজুন:
- Google ক্লাউড কনসোল খুলুন: https://console.cloud.google.com
- পৃষ্ঠার শীর্ষে প্রজেক্ট ড্রপডাউন থেকে আপনি এই কর্মশালার জন্য যে প্রকল্পটি ব্যবহার করতে চান তা নির্বাচন করুন।
- আপনার প্রজেক্ট আইডি ড্যাশবোর্ডে প্রজেক্ট ইনফো কার্ডে প্রদর্শিত হয়
ধাপ 1: আপনার প্রকল্প আইডি সেট করুন
ক্লাউড শেলে, gcloud কমান্ড-লাইন টুলটি ইতিমধ্যেই কনফিগার করা আছে। আপনার সক্রিয় প্রকল্প সেট করতে নিম্নলিখিত কমান্ড চালান। এটি $GOOGLE_CLOUD_PROJECT এনভায়রনমেন্ট ভেরিয়েবল ব্যবহার করে, যা আপনার ক্লাউড শেল সেশনে স্বয়ংক্রিয়ভাবে আপনার জন্য সেট করা হয়।
gcloud config set project $GOOGLE_CLOUD_PROJECT
ধাপ 2: আপনার সেটআপ যাচাই করুন
এর পরে, আপনার প্রকল্প সঠিকভাবে সেট করা হয়েছে এবং আপনি প্রমাণীকরণ করেছেন তা নিশ্চিত করতে নিম্নলিখিত কমান্ডগুলি চালান।
# Confirm project is set
echo "Current project: $(gcloud config get-value project)"
# Check authentication status
gcloud auth list
আপনি আপনার প্রজেক্ট আইডি প্রিন্ট করা দেখতে পাবেন এবং আপনার ব্যবহারকারী অ্যাকাউন্টের পাশে (ACTIVE) তালিকাভুক্ত দেখতে হবে।
যদি আপনার অ্যাকাউন্ট সক্রিয় হিসাবে তালিকাভুক্ত না হয়, বা আপনি যদি একটি প্রমাণীকরণ ত্রুটি পান, লগ ইন করতে নিম্নলিখিত কমান্ডটি চালান:
gcloud auth application-default login
ধাপ 3: এসেনশিয়াল এপিআই সক্ষম করুন
মৌলিক এজেন্টের জন্য আমাদের অন্তত এই API গুলি প্রয়োজন:
gcloud services enable \
aiplatform.googleapis.com \
compute.googleapis.com
এতে এক বা দুই মিনিট সময় লাগতে পারে। আপনি দেখতে পাবেন:
Operation "operations/..." finished successfully.
ধাপ 4: ADK ইনস্টল করুন
# Install the ADK CLI
pip install google-adk --upgrade
# Verify installation
adk --version
আপনার 1.15.0 বা উচ্চতর সংস্করণের সংখ্যা দেখতে হবে।
এখন আপনার বেসিক এজেন্ট তৈরি করুন
পরিবেশ প্রস্তুত সহ, আসুন সেই সাধারণ এজেন্ট তৈরি করি।
ধাপ 5: ADK তৈরি ব্যবহার করুন
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]:
ধাপ 6: কি তৈরি করা হয়েছিল তা পরীক্ষা করুন
cd my_first_agent
ls -la
আপনি তিনটি ফাইল পাবেন:
.env # Configuration (auto-populated with your project)
__init__.py # Package marker
agent.py # Your agent definition
ধাপ 7: দ্রুত কনফিগারেশন চেক
# 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
ধাপ 8: এজেন্ট কোড দেখুন
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',
)
সহজ, পরিষ্কার, ন্যূনতম। এটি আপনার এজেন্টদের "হ্যালো ওয়ার্ল্ড"।
আপনার মৌলিক এজেন্ট পরীক্ষা
ধাপ 9: আপনার এজেন্ট চালান
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]:
ধাপ 10: কিছু প্রশ্ন চেষ্টা করুন
টার্মিনালে যেখানে 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
এজেন্ট কোড নিয়ে আলোচনা করতে পারে, কিন্তু এটা করতে পারে:
- আসলে গঠন বুঝতে AST পার্স?
- এটি কাজ করে যাচাই করতে পরীক্ষা চালান?
- শৈলী সম্মতি পরীক্ষা করুন?
- আপনার আগের পর্যালোচনা মনে আছে?
না। এখানেই আমাদের আর্কিটেকচার দরকার।
🏃🚪 দিয়ে প্রস্থান করুন
Ctrl+C
যখন অন্বেষণ করা হয়.
3. আপনার উৎপাদন কর্মক্ষেত্র প্রস্তুত করা
সমাধান: একটি উৎপাদন-প্রস্তুত আর্কিটেকচার
সেই সাধারণ এজেন্টটি সূচনা বিন্দু প্রদর্শন করেছে, কিন্তু একটি উৎপাদন ব্যবস্থার জন্য শক্ত কাঠামো প্রয়োজন। আমরা এখন একটি সম্পূর্ণ প্রকল্প সেট আপ করব যা উত্পাদন নীতিগুলিকে মূর্ত করে।
ফাউন্ডেশন স্থাপন করা
আপনি ইতিমধ্যেই মৌলিক এজেন্টের জন্য আপনার Google ক্লাউড প্রকল্প কনফিগার করেছেন৷ এখন একটি বাস্তব সিস্টেমের জন্য প্রয়োজনীয় সমস্ত সরঞ্জাম, নিদর্শন এবং অবকাঠামো সহ সম্পূর্ণ উত্পাদন কর্মক্ষেত্র প্রস্তুত করা যাক।
ধাপ 1: স্ট্রাকচার্ড প্রজেক্ট পান
প্রথমে, 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
ধাপ 2: ভার্চুয়াল পরিবেশ তৈরি এবং সক্রিয় করুন
# Create the virtual environment
python -m venv .venv
# Activate it
# On macOS/Linux:
source .venv/bin/activate
# On Windows:
# .venv\Scripts\activate
যাচাইকরণ : আপনার প্রম্পটটি এখন শুরুতে (.venv) দেখাতে হবে।
ধাপ 3: নির্ভরতা ইনস্টল করুন
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 মডিউল আমদানি করতে দেয়।
ধাপ 4: আপনার পরিবেশ কনফিগার করুন
# 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
ধাপ 5: প্রমাণীকরণ নিশ্চিত করুন
যেহেতু আপনি আগে থেকেই gcloud auth চালিয়েছেন, আসুন শুধু যাচাই করি:
# Check current authentication
gcloud auth list
# Should show your account with (ACTIVE)
# If not, run:
gcloud auth application-default login
ধাপ 6: অতিরিক্ত উত্পাদন API সক্ষম করুন
আমরা ইতিমধ্যেই মৌলিক API সক্ষম করেছি৷ এখন উৎপাদন যোগ করুন:
gcloud services enable \
sqladmin.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
artifactregistry.googleapis.com \
storage.googleapis.com \
cloudtrace.googleapis.com
এটি সক্ষম করে:
- এসকিউএল অ্যাডমিন : ক্লাউড রান ব্যবহার করলে ক্লাউড এসকিউএল-এর জন্য
- ক্লাউড রান : সার্ভারহীন স্থাপনার জন্য
- ক্লাউড বিল্ড : স্বয়ংক্রিয় স্থাপনার জন্য
- আর্টিফ্যাক্ট রেজিস্ট্রি : কন্টেইনার ইমেজ জন্য
- ক্লাউড স্টোরেজ : শিল্পকর্ম এবং মঞ্চায়নের জন্য
- ক্লাউড ট্রেস : পর্যবেক্ষণযোগ্যতার জন্য
ধাপ 7: আর্টিফ্যাক্ট রেজিস্ট্রি সংগ্রহস্থল তৈরি করুন
আমাদের স্থাপনা কন্টেইনার ইমেজ তৈরি করবে যার জন্য একটি বাড়ির প্রয়োজন:
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].
যদি এটি ইতিমধ্যেই বিদ্যমান থাকে (সম্ভবত পূর্ববর্তী প্রচেষ্টা থেকে), এটি ঠিক আছে - আপনি একটি ত্রুটি বার্তা দেখতে পাবেন যা আপনি উপেক্ষা করতে পারেন৷
ধাপ 8: 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 ক্লাউড প্রকল্প কনফিগার করা এবং প্রমাণীকৃত
✅ বেসিক এজেন্ট সীমাবদ্ধতা বোঝার জন্য পরীক্ষা করা হয়েছে
✅ কৌশলগত স্থানধারক সহ প্রজেক্ট কোড প্রস্তুত
✅ নির্ভরতা ভার্চুয়াল পরিবেশে বিচ্ছিন্ন
✅ সমস্ত প্রয়োজনীয় API সক্রিয়
✅ কন্টেইনার রেজিস্ট্রি স্থাপনের জন্য প্রস্তুত
✅ IAM অনুমতি সঠিকভাবে কনফিগার করা হয়েছে
✅ এনভায়রনমেন্ট ভেরিয়েবল সঠিকভাবে সেট করা হয়েছে
এখন আপনি নির্ধারক সরঞ্জাম, রাষ্ট্র পরিচালনা এবং সঠিক স্থাপত্য সহ একটি বাস্তব AI সিস্টেম তৈরি করতে প্রস্তুত।
4. আপনার প্রথম এজেন্ট নির্মাণ
এলএলএম থেকে টুলগুলিকে কী আলাদা করে তোলে
আপনি যখন একটি LLM কে জিজ্ঞাসা করেন "এই কোডটিতে কতগুলি ফাংশন রয়েছে?", এটি প্যাটার্ন ম্যাচিং এবং অনুমান ব্যবহার করে। যখন আপনি একটি টুল ব্যবহার করেন যা পাইথনের ast.parse() কল করে, এটি প্রকৃত সিনট্যাক্স ট্রি পার্স করে - অনুমান করা যায় না, প্রতিবার একই ফলাফল।
এই বিভাগটি এমন একটি টুল তৈরি করে যা কোড স্ট্রাকচার ডিটারমিনিস্টিকভাবে বিশ্লেষণ করে, তারপর এটিকে এমন একটি এজেন্টের সাথে কানেক্ট করে যে কখন এটি চালু করতে হবে।
ধাপ 1: ভারা বোঝা
আপনি যে কাঠামোটি পূরণ করবেন তা পরীক্ষা করা যাক।
👉 খুলুন
code_review_assistant/tools.py
আপনি যেখানে কোড যোগ করবেন সেখানে স্থানধারক মন্তব্য চিহ্নিত করে analyze_code_structure ফাংশন দেখতে পাবেন। ফাংশনটির ইতিমধ্যে মৌলিক কাঠামো রয়েছে - আপনি ধাপে ধাপে এটিকে উন্নত করবেন।
ধাপ 2: স্টেট স্টোরেজ যোগ করুন
স্টেট স্টোরেজ পাইপলাইনে থাকা অন্যান্য এজেন্টদের বিশ্লেষণ পুনরায় না চালিয়ে আপনার টুলের ফলাফল অ্যাক্সেস করতে দেয়।
👉 খুঁজুন:
# 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())
ধাপ 3: থ্রেড পুলের সাথে Async পার্সিং যোগ করুন
আমাদের টুলকে অন্যান্য অপারেশন ব্লক না করে AST পার্স করতে হবে। থ্রেড পুলের সাথে অ্যাসিঙ্ক এক্সিকিউশন যোগ করা যাক।
👉 খুঁজুন:
# 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)
ধাপ 4: ব্যাপক তথ্য বের করুন
এখন ক্লাস, আমদানি এবং বিস্তারিত মেট্রিক্স বের করা যাক - একটি সম্পূর্ণ কোড পর্যালোচনার জন্য আমাদের যা প্রয়োজন।
👉 খুঁজুন:
# 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
ধাপ 5: একটি এজেন্টের সাথে সংযোগ করুন
এখন আমরা একটি এজেন্টের কাছে টুলটি ওয়্যার করি যে কখন এটি ব্যবহার করতে হবে এবং কীভাবে এর ফলাফলগুলি ব্যাখ্যা করতে হবে তা জানে।
👉 খুলুন
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ধ্রুবক ব্যবহার করে ফলাফলগুলি সেশন অবস্থায় সংরক্ষণ করা হয়েছিল - কোড বিশ্লেষক এজেন্ট ফলাফল ব্যাখ্যা করেছে এবং একটি সারসংক্ষেপ প্রদান করেছে
সমস্যা সমাধান:
- "কোড_রিভিউ_অ্যাসিস্ট্যান্ট' নামে কোনো মডিউল নেই" :
pip install -e .প্রকল্প রুট থেকে - "কি ইনপুট আর্গুমেন্ট অনুপস্থিত" : যাচাই করুন আপনার
.envএGOOGLE_CLOUD_PROJECT,GOOGLE_CLOUD_LOCATION, এবংGOOGLE_GENAI_USE_VERTEXAI=true
আপনি কি নির্মাণ করেছেন
আপনার কাছে এখন একটি উত্পাদন-প্রস্তুত কোড বিশ্লেষক রয়েছে যা:
✅ প্রকৃত পাইথন AST পার্স করে - নির্ধারক, প্যাটার্ন ম্যাচিং নয়
✅ ফলাফল রাজ্যে স্টোর করে - অন্যান্য এজেন্ট বিশ্লেষণ অ্যাক্সেস করতে পারে
✅ অ্যাসিঙ্ক্রোনাসভাবে চলে - অন্যান্য টুল ব্লক করে না
✅ ব্যাপক তথ্য বের করে - ফাংশন, ক্লাস, আমদানি, মেট্রিক্স
✅ ত্রুটিগুলি সুন্দরভাবে পরিচালনা করে - লাইন নম্বর সহ সিনট্যাক্স ত্রুটিগুলি রিপোর্ট করে৷
✅ একটি এজেন্টের সাথে সংযোগ করে - এলএলএম জানে কখন এবং কিভাবে এটি ব্যবহার করতে হবে
মূল ধারণা আয়ত্ত
সরঞ্জাম বনাম এজেন্ট:
- সরঞ্জামগুলি নির্ধারণমূলক কাজ করে (AST পার্সিং)
- এজেন্ট কখন টুল ব্যবহার করবেন এবং ফলাফল ব্যাখ্যা করবেন তা নির্ধারণ করে
রিটার্ন মান বনাম রাজ্য:
- প্রত্যাবর্তন: এলএলএম অবিলম্বে যা দেখে
- রাজ্য: অন্যান্য এজেন্টদের জন্য কি অব্যাহত থাকে
রাজ্য কী ধ্রুবক:
- মাল্টি-এজেন্ট সিস্টেমে টাইপো প্রতিরোধ করুন
- এজেন্টদের মধ্যে চুক্তি হিসাবে কাজ করুন
- এজেন্টরা ডেটা শেয়ার করার সময় গুরুত্বপূর্ণ
অ্যাসিঙ্ক + থ্রেড পুল:
-
async defটুলগুলিকে এক্সিকিউশন পজ করার অনুমতি দেয় - থ্রেড পুলগুলি পটভূমিতে CPU- আবদ্ধ কাজ চালায়
- একসঙ্গে তারা ইভেন্ট লুপ প্রতিক্রিয়াশীল রাখা
হেল্পার ফাংশন:
- অ্যাসিঙ্ক টুল থেকে সিঙ্ক সহায়ককে আলাদা করুন
- কোড পরীক্ষাযোগ্য এবং পুনরায় ব্যবহারযোগ্য করে তোলে
এজেন্ট নির্দেশাবলী:
- বিশদ নির্দেশাবলী সাধারণ এলএলএম ভুলগুলি প্রতিরোধ করে
- কী করা উচিত নয় সে সম্পর্কে স্পষ্ট (কোড ঠিক করবেন না)
- সামঞ্জস্যের জন্য কর্মপ্রবাহের ধাপগুলি পরিষ্কার করুন
পরবর্তী কি
মডিউল 5 এ, আপনি যোগ করবেন:
- স্টাইল চেকার যা রাজ্য থেকে কোড পড়ে
- টেস্ট রানার যে আসলে পরীক্ষা চালায়
- ফিডব্যাক সিন্থেসাইজার যা সমস্ত বিশ্লেষণকে একত্রিত করে
আপনি দেখতে পাবেন কিভাবে একটি ক্রমিক পাইপলাইনের মধ্য দিয়ে রাষ্ট্র প্রবাহিত হয় এবং যখন একাধিক এজেন্ট একই ডেটা পড়ে এবং লেখে তখন কেন ধ্রুবক প্যাটার্ন গুরুত্বপূর্ণ।
5. একটি পাইপলাইন নির্মাণ: একাধিক এজেন্ট একসাথে কাজ করছে
ভূমিকা
মডিউল 4 এ, আপনি একটি একক এজেন্ট তৈরি করেছেন যা কোড গঠন বিশ্লেষণ করে। কিন্তু বিস্তৃত কোড পর্যালোচনার জন্য শুধু পার্সিংয়ের চেয়ে আরও বেশি কিছু প্রয়োজন - আপনার শৈলী পরীক্ষা, পরীক্ষা সম্পাদন এবং বুদ্ধিমান প্রতিক্রিয়া সংশ্লেষণ প্রয়োজন।
এই মডিউলটি 4টি এজেন্টের একটি পাইপলাইন তৈরি করে যা ক্রমানুসারে একসাথে কাজ করে, প্রতিটি বিশেষ বিশ্লেষণে অবদান রাখে:
- কোড বিশ্লেষক (মডিউল 4 থেকে) - কাঠামো পার্স করে
- স্টাইল পরীক্ষক - শৈলী লঙ্ঘন সনাক্ত করে
- টেস্ট রানার - পরীক্ষা চালায় এবং যাচাই করে
- ফিডব্যাক সিন্থেসাইজার - সবকিছুকে অ্যাকশনেবল ফিডব্যাকে একত্রিত করে
মূল ধারণা: যোগাযোগ চ্যানেল হিসাবে রাষ্ট্র। প্রতিটি এজেন্ট পূর্ববর্তী এজেন্টরা রাজ্যে যা লিখেছিল তা পড়ে, তার নিজস্ব বিশ্লেষণ যোগ করে এবং পরবর্তী এজেন্টকে সমৃদ্ধ রাষ্ট্রটি পাস করে। মডিউল 4 থেকে ধ্রুবক প্যাটার্ন গুরুত্বপূর্ণ হয়ে ওঠে যখন একাধিক এজেন্ট ডেটা ভাগ করে।
আপনি যা তৈরি করবেন তার পূর্বরূপ: অগোছালো কোড জমা দিন → 4টি এজেন্টের মাধ্যমে রাজ্যের প্রবাহ দেখুন → অতীতের নিদর্শনগুলির উপর ভিত্তি করে ব্যক্তিগতকৃত প্রতিক্রিয়া সহ ব্যাপক প্রতিবেদন পান৷
ধাপ 1: স্টাইল চেকার টুল + এজেন্ট যোগ করুন
স্টাইল চেকার পাইকোডস্টাইল ব্যবহার করে 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"
)
ধাপ 2: টেস্ট রানার এজেন্ট যোগ করুন
টেস্ট রানার ব্যাপক পরীক্ষা তৈরি করে এবং অন্তর্নির্মিত কোড নির্বাহক ব্যবহার করে সেগুলি সম্পাদন করে।
👉 খুলুন
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"
)
ধাপ 3: ক্রস-সেশন শেখার জন্য মেমরি বোঝা
ফিডব্যাক সিন্থেসাইজার তৈরি করার আগে, আপনাকে স্টেট এবং মেমরির মধ্যে পার্থক্য বুঝতে হবে - দুটি ভিন্ন উদ্দেশ্যে দুটি আলাদা স্টোরেজ মেকানিজম।
রাজ্য বনাম মেমরি: মূল পার্থক্য
কোড পর্যালোচনা থেকে একটি কংক্রিট উদাহরণ দিয়ে স্পষ্ট করা যাক:
রাজ্য (শুধুমাত্র বর্তমান অধিবেশন):
# 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"}
]
- সুযোগ: এই কথোপকথন শুধুমাত্র
- উদ্দেশ্য: বর্তমান পাইপলাইনে এজেন্টদের মধ্যে ডেটা পাস করুন
- বাস করে:
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."
ব্যক্তিগতকৃত, প্রাসঙ্গিক, সময়ের সাথে রেফারেন্সের উন্নতি।
উত্পাদন স্থাপনার জন্য, আপনার কাছে বিকল্প রয়েছে :
বিকল্প 1: VertexAiMemoryBankService (উন্নত)
- এটি কী করে: কথোপকথন থেকে অর্থপূর্ণ তথ্যের এলএলএম-চালিত নিষ্কাশন
- অনুসন্ধান: শব্দার্থিক অনুসন্ধান (অর্থ বোঝে, শুধু কীওয়ার্ড নয়)
- স্মৃতি ব্যবস্থাপনা: সময়ের সাথে সাথে স্বয়ংক্রিয়ভাবে স্মৃতিগুলিকে একত্রিত করে এবং আপডেট করে
- প্রয়োজন: গুগল ক্লাউড প্রজেক্ট + এজেন্ট ইঞ্জিন সেটআপ
- যখন ব্যবহার করুন: আপনি পরিশীলিত, বিকশিত, ব্যক্তিগতকৃত স্মৃতি চান
- উদাহরণ: "ব্যবহারকারী কার্যকরী প্রোগ্রামিং পছন্দ করে" (কোড শৈলী সম্পর্কে 10টি কথোপকথন থেকে নেওয়া)
বিকল্প 2: InMemoryMemoryService + স্থায়ী সেশনের সাথে চালিয়ে যান
- এটি কী করে: কীওয়ার্ড অনুসন্ধানের জন্য সম্পূর্ণ কথোপকথনের ইতিহাস সংরক্ষণ করে
- অনুসন্ধান: বিগত সেশন জুড়ে মৌলিক কীওয়ার্ড মিল
- মেমরি ম্যানেজমেন্ট: আপনি কী সংরক্ষণ করবেন তা নিয়ন্ত্রণ করেন (
add_session_to_memoryএর মাধ্যমে) - প্রয়োজন: শুধুমাত্র একটি স্থায়ী
SessionService(যেমনVertexAiSessionServiceবাDatabaseSessionService) - যখন ব্যবহার করুন: আপনার LLM প্রক্রিয়াকরণ ছাড়াই অতীতের কথোপকথন জুড়ে সহজ অনুসন্ধানের প্রয়োজন
- উদাহরণ: অনুসন্ধান "ডকস্ট্রিং" সেই শব্দটি উল্লেখ করে সমস্ত সেশন প্রদান করে
কিভাবে মেমরি জনবহুল হয়
প্রতিটি কোড পর্যালোচনা শেষ হওয়ার পরে:
# At the end of a session (typically in your application code)
await memory_service.add_session_to_memory(session)
কি হয়:
- InMemoryMemoryService: কীওয়ার্ড অনুসন্ধানের জন্য সম্পূর্ণ সেশন ইভেন্ট সংরক্ষণ করে
- VertexAiMemoryBankService: LLM মূল তথ্য বের করে, বিদ্যমান স্মৃতির সাথে একত্রিত করে
ভবিষ্যতের সেশনগুলি তখন প্রশ্ন করতে পারে:
# In a tool, search for relevant past feedback
results = tool_context.search_memory("feedback about docstrings")
ধাপ 4: ফিডব্যাক সিন্থেসাইজার টুল এবং এজেন্ট যোগ করুন
ফিডব্যাক সিন্থেসাইজার হল পাইপলাইনের সবচেয়ে পরিশীলিত এজেন্ট। এটি তিনটি টুল অর্কেস্ট্রেট করে, গতিশীল নির্দেশাবলী ব্যবহার করে এবং রাষ্ট্র, মেমরি এবং শিল্পকর্মকে একত্রিত করে।
তিনটি সিন্থেসাইজার টুল যোগ করুন
👉 খুলুন
code_review_assistant/tools.py
👉 খুঁজুন:
# MODULE_5_STEP_4_SEARCH_PAST_FEEDBACK
👉 টুল 1 দিয়ে প্রতিস্থাপন করুন - মেমরি অনুসন্ধান (উৎপাদন সংস্করণ):
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
👉 টুল 3 দিয়ে প্রতিস্থাপন করুন - আর্টিফ্যাক্ট সেভার (উৎপাদন সংস্করণ):
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"
)
ধাপ 5: পাইপলাইন তারের
এখন চারটি এজেন্টকে একটি অনুক্রমিক পাইপলাইনে সংযুক্ত করুন এবং রুট এজেন্ট তৈরি করুন।
👉 খুলুন
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"
)
ধাপ 6: সম্পূর্ণ পাইপলাইন পরীক্ষা করুন
চারটি এজেন্ট একসাথে কাজ করার সময়।
👉 সিস্টেমটি শুরু করুন:
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 অ্যাক্সেস করতে:
ক্লাউড শেল টুলবারে (সাধারণত উপরের ডানদিকে) ওয়েব প্রিভিউ আইকন থেকে (প্রায়শই একটি চোখ বা তীরযুক্ত একটি বর্গক্ষেত্রের মতো দেখায়), পোর্ট পরিবর্তন করুন নির্বাচন করুন। পপ-আপ উইন্ডোতে, পোর্টটি 8000 এ সেট করুন এবং "পরিবর্তন এবং পূর্বরূপ" ক্লিক করুন। ক্লাউড শেল তারপর একটি নতুন ব্রাউজার ট্যাব বা উইন্ডো খুলবে যা ADK Dev UI প্রদর্শন করবে।
👉 এজেন্ট এখন চলছে। আপনার ব্রাউজারে ADK Dev UI হল এজেন্টের সাথে আপনার সরাসরি সংযোগ।
- আপনার লক্ষ্য নির্বাচন করুন: 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 এজেন্ট কাঁচা কোড পায়। কোডটি কী করে তা অনুমান করে না; এটি একটি নির্ধারক বিমূর্ত সিনট্যাক্স ট্রি (AST) পার্স সম্পাদন করতে 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.
⭐ মান: এই প্রাথমিক ধাপটি অন্যান্য এজেন্টদের জন্য একটি পরিষ্কার, নির্ভরযোগ্য ভিত্তি প্রদান করে। এটি নিশ্চিত করে যে কোডটি বৈধ পাইথন এবং সঠিক উপাদানগুলি সনাক্ত করে যা পর্যালোচনা করা দরকার৷
2. স্টাইল চেকারের PEP 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
⭐ মান: এই এজেন্টটি প্রতিষ্ঠিত সম্প্রদায়ের মান (PEP 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.
⭐ মান: এই এজেন্ট প্রযুক্তিগত ডেটাকে একটি সহায়ক, শিক্ষামূলক অভিজ্ঞতায় রূপান্তরিত করে। এটি সবচেয়ে গুরুত্বপূর্ণ সমস্যাটিকে (বাগ) অগ্রাধিকার দেয়, এটি পরিষ্কারভাবে ব্যাখ্যা করে, সঠিক সমাধান প্রদান করে এবং এটি একটি উত্সাহজনক সুরে করে। এটি সফলভাবে সমস্ত পূর্ববর্তী পর্যায়ের ফলাফলগুলিকে একটি সমন্বিত এবং মূল্যবান সমগ্রের মধ্যে একত্রিত করে।
এই বহু-পর্যায়ের প্রক্রিয়াটি একটি এজেন্টিক পাইপলাইনের শক্তি প্রদর্শন করে। Instead of a single, monolithic response, you get a layered analysis where each agent performs a specialized, verifiable task. This leads to a review that is not only insightful but also deterministic, reliable, and deeply educational.
👉💻 Once you're done testing, return to your Cloud Shell Editor terminal and press Ctrl+C to stop the ADK Dev UI.
What You've Built
You now have a complete code review pipeline that:
✅ Parses code structure - deterministic AST analysis with helper functions
✅ Checks style - weighted scoring with naming conventions
✅ Runs tests - comprehensive test generation with structured JSON output
✅ Synthesizes feedback - integrates state + memory + artifacts
✅ Tracks progress - multi-tier state across invocations/sessions/users
✅ Learns over time - memory service for cross-session patterns
✅ Provides artifacts - downloadable JSON reports with complete audit trail
মূল ধারণা আয়ত্ত
Sequential Pipelines:
- Four agents executing in strict order
- Each enriches state for the next
- Dependencies determine execution sequence
Production Patterns:
- Helper function separation (sync in thread pools)
- Graceful degradation (fallback strategies)
- Multi-tier state management (temp/session/user)
- Dynamic instruction providers (context-aware)
- Dual storage (artifacts + state redundancy)
State as Communication:
- Constants prevent typos across agents
-
output_keywrites agent summaries to state - Later agents read via StateKeys
- State flows linearly through pipeline
Memory vs State:
- State: current session data
- Memory: patterns across sessions
- Different purposes, different lifetimes
Tool Orchestration:
- Single-tool agents (analyzer, style_checker)
- Built-in executors (test_runner)
- Multi-tool coordination (synthesizer)
Model Selection Strategy:
- Worker model: mechanical tasks (parsing, linting, routing)
- Critic model: reasoning tasks (testing, synthesis)
- Cost optimization through appropriate selection
পরবর্তী কি
In Module 6, you'll build the fix pipeline :
- LoopAgent architecture for iterative fixing
- Exit conditions via escalation
- State accumulation across iterations
- Validation and retry logic
- Integration with review pipeline to offer fixes
You'll see how the same state patterns scale to complex iterative workflows where agents attempt multiple times until success, and how to coordinate multiple pipelines in a single application.
6. Adding the Fix Pipeline: Loop Architecture
ভূমিকা
In Module 5, you built a sequential review pipeline that analyzes code and provides feedback. But identifying problems is only half the solution - developers need help fixing them.
This module builds an automated fix pipeline that:
- Generates fixes based on review results
- Validates fixes by running comprehensive tests
- Retries automatically if fixes don't work (up to 3 attempts)
- Reports results with before/after comparisons
Key concept: LoopAgent for automatic retry. Unlike sequential agents that run once, a LoopAgent repeats its sub-agents until an exit condition is met or max iterations reached. Tools signal success by setting tool_context.actions.escalate = True .
Preview of what you'll build: Submit buggy code → review identifies issues → fix loop generates corrections → tests validate → retries if needed → final comprehensive report.
Core Concepts: LoopAgent vs Sequential
Sequential Pipeline (Module 5):
SequentialAgent(agents=[A, B, C])
# Executes: A → B → C → Done
- One-way flow
- Each agent runs exactly once
- No retry logic
Loop Pipeline (Module 6):
LoopAgent(agents=[A, B, C], max_iterations=3)
# Executes: A → B → C → (check exit) → A → B → C → (check exit) → ...
- Cyclic flow
- Agents can run multiple times
- Exits when:
- A tool sets
tool_context.actions.escalate = True(success) -
max_iterationsreached (safety limit) - Unhandled exception occurs (error)
- A tool sets
Why loops for code fixing:
Code fixes often need multiple attempts:
- First attempt : Fix obvious bugs (wrong variable types)
- Second attempt : Fix secondary issues revealed by tests (edge cases)
- Third attempt : Fine-tune and validate all tests pass
Without a loop, you'd need complex conditional logic in agent instructions. With LoopAgent , retry is automatic.
Architecture comparison:
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
Step 1: Add Code Fixer Agent
The code fixer generates corrected Python code based on review results.
👉 Open
code_review_assistant/sub_agents/fix_pipeline/code_fixer.py
👉 Find:
# MODULE_6_STEP_1_CODE_FIXER_INSTRUCTION_PROVIDER
👉 Replace that single line with:
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)
👉 Find:
# MODULE_6_STEP_1_CODE_FIXER_AGENT
👉 Replace that single line with:
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"
)
Step 2: Add Fix Test Runner Agent
The fix test runner validates corrections by executing comprehensive tests on the fixed code.
👉 Open
code_review_assistant/sub_agents/fix_pipeline/fix_test_runner.py
👉 Find:
# MODULE_6_STEP_2_FIX_TEST_RUNNER_INSTRUCTION_PROVIDER
👉 Replace that single line with:
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)
👉 Find:
# MODULE_6_STEP_2_FIX_TEST_RUNNER_AGENT
👉 Replace that single line with:
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"
)
ধাপ 3: ফিক্স ভ্যালিডেটর এজেন্ট যোগ করুন
The validator checks if fixes were successful and decides whether to exit the loop.
টুলস বোঝা
First, add the three tools the validator needs.
👉 Open
code_review_assistant/tools.py
👉 Find:
# MODULE_6_STEP_3_VALIDATE_FIXED_STYLE
👉 Replace with Tool 1 - Style Validator:
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)
}
👉 Find:
# MODULE_6_STEP_3_COMPILE_FIX_REPORT
👉 Replace with Tool 2 - Report Compiler:
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)
}
👉 Find:
# MODULE_6_STEP_3_EXIT_FIX_LOOP
👉 Replace with Tool 3 - Loop Exit Signal:
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"
}
Create the Validator Agent
👉 Open
code_review_assistant/sub_agents/fix_pipeline/fix_validator.py
👉 Find:
# MODULE_6_STEP_3_FIX_VALIDATOR_INSTRUCTION_PROVIDER
👉 Replace that single line with:
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)
👉 Find:
# MODULE_6_STEP_3_FIX_VALIDATOR_AGENT
👉 Replace that single line with:
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"
)
Step 4: Understanding LoopAgent Exit Conditions
The LoopAgent has three ways to exit:
1. Success Exit (via escalate)
# 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. Error Exit
# 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
State Evolution Across Iterations:
Each iteration sees updated state from the previous attempt:
# 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
Instead of Return Values:
# 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}
সুবিধা:
- Works from any tool, not just the last one
- Doesn't interfere with return data
- Clear semantic meaning
- Framework handles the exit logic
Step 5: Wire the Fix Pipeline
👉 Open
code_review_assistant/agent.py
👉 Add the fix pipeline imports (after the existing imports):
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
Your imports should now be:
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
👉 Find:
# MODULE_6_STEP_5_CREATE_FIX_LOOP
👉 Replace that single line with:
# 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)
]
)
👉 Remove the existing
root_agent
সংজ্ঞা:
root_agent = Agent(...)
👉 Find:
# MODULE_6_STEP_5_UPDATE_ROOT_AGENT
👉 Replace that single line with:
# 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"
)
ধাপ 6: ফিক্স সিন্থেসাইজার এজেন্ট যোগ করুন
The synthesizer creates a user-friendly presentation of fix results after the loop completes.
👉 Open
code_review_assistant/sub_agents/fix_pipeline/fix_synthesizer.py
👉 Find:
# MODULE_6_STEP_6_FIX_SYNTHESIZER_INSTRUCTION_PROVIDER
👉 Replace that single line with:
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)
👉 Find:
# MODULE_6_STEP_6_FIX_SYNTHESIZER_AGENT
👉 Replace that single line with:
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"
)
👉 Add
save_fix_report
টুল
tools.py
:
👉 Find:
# MODULE_6_STEP_6_SAVE_FIX_REPORT
👉 Replace with:
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)
}
Step 7: Test Complete Fix Pipeline
Time to see the entire loop in action.
👉 Start the system:
adk web code_review_assistant
After running the adk web command, you should see output in your terminal indicating that the ADK Web Server has started, similar to this:
+-----------------------------------------------------------------------------+
| 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)
👉 Test Prompt:
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
First, submit the buggy code to trigger the review pipeline . After it identifies the flaws, you will ask the agent to "Please fix the code" which triggers the powerful, iterative fix pipeline .
1. The Initial Review (Finding the Flaws)
This is the first half of the process. The four-agent review pipeline analyzes the code, checks its style, and runs a generated test suite. It correctly identifies a critical AttributeError and other issues, delivering a verdict: the code is BROKEN , with a test pass rate of only 84.21% .
2. অটোমেটেড ফিক্স (দ্যা লুপ ইন অ্যাকশন)
This is the most impressive part. When you ask the agent to fix the code, it doesn't just make one change. It kicks off an iterative Fix and Validate Loop that works just like a diligent developer: it tries a fix, tests it thoroughly, and if it's not perfect, it tries again.
Iteration #1: The First Attempt (Partial Success)
- The Fix: The
CodeFixeragent reads the initial report and makes the most obvious corrections. It changesstack = starttostack = [start]and usesgraph.get()to preventKeyErrorexceptions. - The Validation: The
TestRunnerimmediately re-runs the full test suite against this new code. - The Result: The pass rate improves significantly to 88.89% ! The critical bugs are gone. However, the tests are so comprehensive that they reveal two new, subtle bugs (regressions) related to handling
Noneas a graph or non-list neighbor values. The system marks the fix as PARTIAL .
Iteration #2: The Final Polish (100% Success)
- The Fix: Because the loop's exit condition (100% pass rate) was not met, it runs again. The
CodeFixernow has more information—the two new regression failures. It generates a final, more robust version of the code that explicitly handles those edge cases. - The Validation: The
TestRunnerexecutes the test suite one last time against the final version of the code. - The Result: A perfect 100% pass rate . All original bugs and all regressions are resolved. The system marks the fix as SUCCESSFUL and the loop exits.
3. The Final Report: A Perfect Score
With a fully validated fix, the FixSynthesizer agent takes over to present the final report, transforming the technical data into a clear, educational summary.
মেট্রিক | আগে | পরে | উন্নতি |
Test Pass Rate | 84.21% | 100% | ▲ 15.79% |
Style Score | 88/100 | 98/100 | ▲ 10 pts |
বাগ সংশোধন করা হয়েছে | 3 এর 0 | 3 এর 3 | ✅ |
✅ The Final, Validated Code
Here is the complete, corrected code that now passes all 19 tests, demonstrating the successful fix:
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
👉💻 Once you're done testing, return to your Cloud Shell Editor terminal and press Ctrl+C to stop the ADK Dev UI.
What You've Built
You now have a complete automated fix pipeline that:
✅ Generates fixes - Based on review analysis
✅ Validates iteratively - Tests after each fix attempt
✅ Retries automatically - Up to 3 attempts for success
✅ Exits intelligently - Via escalate when successful
✅ Tracks improvements - Compares before/after metrics
✅ Provides artifacts - Downloadable fix reports
মূল ধারণা আয়ত্ত
LoopAgent vs Sequential:
- Sequential: One pass through agents
- LoopAgent: Repeats until exit condition or max iterations
- Exit via
tool_context.actions.escalate = True
State Evolution Across Iterations:
-
CODE_FIXESupdated each iteration - Test results show improvement over time
- Validator sees cumulative changes
Multi-Pipeline Architecture:
- Review pipeline: Read-only analysis (Module 5)
- Fix loop: Iterative correction (Module 6 inner loop)
- Fix pipeline: Loop + synthesizer (Module 6 outer)
- Root agent: Orchestrates based on user intent
Tools Controlling Flow:
-
exit_fix_loop()sets escalate - Any tool can signal loop completion
- Decouples exit logic from agent instructions
Max Iterations Safety:
- Prevents infinite loops
- Ensures system always responds
- Presents best attempt even if not perfect
পরবর্তী কি
In the final module, you'll learn how to deploy your agent to production :
- Setting up persistent storage with VertexAiSessionService
- Deploying to Agent Engine on Google Cloud
- Monitoring and debugging production agents
- Best practices for scaling and reliability
You've built a complete multi-agent system with sequential and loop architectures. The patterns you've learned - state management, dynamic instructions, tool orchestration, and iterative refinement - are production-ready techniques used in real agentic systems.
7. Deploying to Production
ভূমিকা
Your code review assistant is now complete with review and fix pipelines working locally. The missing piece: it only runs on your machine. In this module, you'll deploy your agent to Google Cloud, making it accessible to your team with persistent sessions and production-grade infrastructure.
আপনি যা শিখবেন:
- Three deployment paths: Local, Cloud Run, and Agent Engine
- Automated infrastructure provisioning
- Session persistence strategies
- Testing deployed agents
Understanding Deployment Options
The ADK supports multiple deployment targets, each with different tradeoffs:
Deployment Paths
ফ্যাক্টর | Local ( | Cloud Run ( | Agent Engine ( |
জটিলতা | ন্যূনতম | মাঝারি | কম |
অধিবেশন অধ্যবসায় | In-memory only (lost on restart) | Cloud SQL (PostgreSQL) | Vertex AI managed (automatic) |
অবকাঠামো | None (dev machine only) | Container + Database | সম্পূর্ণরূপে পরিচালিত |
কোল্ড স্টার্ট | N/A | 100-2000ms | 100-500ms |
স্কেলিং | একক উদাহরণ | Automatic (to zero) | স্বয়ংক্রিয় |
খরচ মডেল | Free (local compute) | Request-based + free tier | Compute-based |
UI Support | Yes (via | Yes (via | No (API only) |
জন্য সেরা | Development/testing | Variable traffic, cost control | Production agents |
Additional deployment option: Google Kubernetes Engine (GKE) is available for advanced users requiring Kubernetes-level control, custom networking, or multi-service orchestration. GKE deployment is not covered in this codelab but is documented in the ADK deployment guide .
What Gets Deployed
When deploying to Cloud Run or Agent Engine, the following is packaged and deployed:
- Your agent code (
agent.py, all sub-agents, tools) - Dependencies (
requirements.txt) - ADK API server (automatically included)
- Web UI (Cloud Run only, when
--with_uispecified)
Important differences:
- Cloud Run : Uses
adk deploy cloud_runCLI (builds container automatically) orgcloud run deploy(requires custom Dockerfile) - Agent Engine : Uses
adk deploy agent_engineCLI (no container building needed, directly packages Python code)
ধাপ 1: আপনার পরিবেশ কনফিগার করুন
Configure Your .env File
Your .env file (created in Module 3) needs updates for cloud deployment. Open .env and verify/update these settings:
Required for all cloud deployments:
# 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
Set bucket names (REQUIRED before running deploy.sh):
The deployment script creates buckets based on these names. Set them now:
# 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
Replace your-project-id with your actual project ID in both bucket names. The script will create these buckets if they don't exist.
Optional variables (created automatically if blank):
# 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=
প্রমাণীকরণ চেক
If you encounter authentication errors during deployment:
gcloud auth application-default login
gcloud config set project $GOOGLE_CLOUD_PROJECT
Step 2: Understanding the Deployment Script
The deploy.sh script provides a unified interface for all deployment modes:
./deploy.sh {local|cloud-run|agent-engine}
Script Capabilities
Infrastructure provisioning:
- API enablement (AI Platform, Storage, Cloud Build, Cloud Trace, Cloud SQL)
- IAM permission configuration (service accounts, roles)
- Resource creation (buckets, databases, instances)
- Deployment with proper flags
- Post-deployment verification
Key Script Sections
- Configuration (lines 1-35) : Project, region, service names, defaults
- Helper Functions (lines 37-200) : API enablement, bucket creation, IAM setup
- Main Logic (lines 202-400) : Mode-specific deployment orchestration
Step 3: Prepare Agent for Agent Engine
Before deploying to Agent Engine, an agent_engine_app.py file is needed that wraps your agent for the managed runtime. This has been created for you already.
View code_review_assistant/agent_engine_app.py
👉 Open file:
"""
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,
)
Step 4: Deploy to Agent Engine
Agent Engine is the recommended production deployment for ADK agents because it provides:
- Fully managed infrastructure (no containers to build)
- Built-in session persistence via
VertexAiSessionService - Automatic scaling from zero
- Cloud Trace integration enabled by default
How Agent Engine Differs from Other Deployments
ফণার নিচে,
deploy.sh agent-engine
ব্যবহার করে:
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
এই আদেশ:
- Packages your Python code directly (no Docker build)
- Uploads to the staging bucket you specified in
.env - Creates a managed Agent Engine instance
- Enables Cloud Trace for observability
- Uses
agent_engine_app.pyto configure the runtime
Unlike Cloud Run which containerizes your code, Agent Engine runs your Python code directly in a managed runtime environment, similar to serverless functions.
Run the Deployment
From your project root:
./deploy.sh agent-engine
Deployment Phases
Watch the script execute these phases:
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
This process takes 5-10 minutes as it packages the agent and deploys it to Vertex AI infrastructure.
Save Your Agent Engine ID
Upon successful deployment:
✅ 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
file immediately:
echo "AGENT_ENGINE_ID=7917477678498709504" >> .env
This ID is required for:
- Testing the deployed agent
- Updating the deployment later
- Accessing logs and traces
কি স্থাপন করা হয়েছিল
Your Agent Engine deployment now includes:
✅ Complete review pipeline (4 agents)
✅ Complete fix pipeline (loop + synthesizer)
✅ All tools (AST analysis, style checking, artifact generation)
✅ Session persistence (automatic via VertexAiSessionService )
✅ State management (session/user/lifetime tiers)
✅ Observability (Cloud Trace enabled)
✅ Auto-scaling infrastructure
Step 5: Test Your Deployed Agent
Update Your .env File
After deployment, verify your .env includes:
AGENT_ENGINE_ID=7917477678498709504 # From deployment output
GOOGLE_CLOUD_PROJECT=your-project-id
GOOGLE_CLOUD_LOCATION=us-central1
Run the Test Script
The project includes tests/test_agent_engine.py specifically for testing Agent Engine deployments:
python tests/test_agent_engine.py
টেস্ট কি করে
- Authenticates with your Google Cloud project
- Creates a session with the deployed agent
- Sends a code review request (the DFS bug example)
- Streams the response back via Server-Sent Events (SSE)
- Verifies session persistence and state management
প্রত্যাশিত আউটপুট
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.
যাচাইকরণ চেকলিস্ট
- ✅ Full review pipeline executes (all 4 agents)
- ✅ Streaming response shows progressive output
- ✅ Session state persists across requests
- ✅ No authentication or connection errors
- ✅ Tool calls execute successfully (AST analysis, style checking)
- ✅ Artifacts are saved (grading report accessible)
Alternative: Deploy to Cloud Run
While Agent Engine is recommended for streamlined production deployment, Cloud Run offers more control and supports the ADK web UI. এই বিভাগে একটি ওভারভিউ প্রদান করে.
When to Use Cloud Run
Choose Cloud Run if you need:
- The ADK web UI for user interaction
- Full control over the container environment
- Custom database configurations
- Integration with existing Cloud Run services
ক্লাউড রান ডিপ্লয়মেন্ট কিভাবে কাজ করে
ফণার নিচে,
deploy.sh cloud-run
ব্যবহার করে:
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
এই আদেশ:
- Builds a Docker container with your agent code
- Pushes to Google Artifact Registry
- Deploys as a Cloud Run service
- Includes the ADK web UI (
--with_ui) - Configures Cloud SQL connection (added by script after initial deployment)
The key difference from Agent Engine: Cloud Run containerizes your code and requires a database for session persistence, while Agent Engine handles both automatically.
Cloud Run Deployment Command
./deploy.sh cloud-run
What's Different
অবকাঠামো:
- Containerized deployment (Docker built automatically by ADK)
- Cloud SQL (PostgreSQL) for session persistence
- Database auto-created by script or uses existing instance
Session Management:
- Uses
DatabaseSessionServiceinstead ofVertexAiSessionService - Requires database credentials in
.env(or auto-generated) - State persists in PostgreSQL database
UI Support:
- Web UI available via
--with_uiflag (handled by script) - Access at:
https://code-review-assistant-xyz.a.run.app
What You've Accomplished
Your production deployment includes:
✅ Automated provisioning via deploy.sh script
✅ Managed infrastructure (Agent Engine handles scaling, persistence, monitoring)
✅ Persistent state across all memory tiers (session/user/lifetime)
✅ Secure credential management (automatic generation and IAM setup)
✅ Scalable architecture (zero to thousands of concurrent users)
✅ Built-in observability (Cloud Trace integration enabled)
✅ Production-grade error handling and recovery
মূল ধারণা আয়ত্ত
Deployment Preparation:
-
agent_engine_app.py: Wraps agent withAdkAppfor Agent Engine -
AdkAppautomatically configuresVertexAiSessionServicefor persistence - Tracing enabled via
enable_tracing=True
Deployment Commands:
-
adk deploy agent_engine: Packages Python code, no containers -
adk deploy cloud_run: Builds Docker container automatically -
gcloud run deploy: Alternative with custom Dockerfile
স্থাপনার বিকল্প:
- Agent Engine: Fully managed, fastest to production
- Cloud Run: More control, supports web UI
- GKE: Advanced Kubernetes control (see GKE deployment guide )
পরিচালিত পরিষেবা:
- Agent Engine handles session persistence automatically
- Cloud Run requires database setup (or auto-created)
- Both support artifact storage via GCS
Session Management:
- Agent Engine:
VertexAiSessionService(automatic) - Cloud Run:
DatabaseSessionService(Cloud SQL) - Local:
InMemorySessionService(ephemeral)
Your Agent Is Live
Your code review assistant is now:
- Accessible via HTTPS API endpoints
- Persistent with state surviving restarts
- Scalable to handle team growth automatically
- Observable with complete request traces
- Maintainable through scripted deployments
পরবর্তী কি? In Module 8, you'll learn to use Cloud Trace to understand your agent's performance, identify bottlenecks in the review and fix pipelines, and optimize execution times.
8. Production Observability
ভূমিকা
Your code review assistant is now deployed and running in production on Agent Engine. But how do you know it's working well? Can you answer these critical questions:
- Is the agent responding quickly enough?
- Which operations are slowest?
- Are the fix loops completing efficiently?
- Where are performance bottlenecks?
Without observability, you're operating blind. The --trace-to-cloud flag you used during deployment automatically enabled Cloud Trace, giving you complete visibility into every request your agent processes.
In this module, you'll learn to read traces, understand your agent's performance characteristics, and identify areas for optimization.
Understanding Traces and Spans
What is a Trace?
A trace is the complete timeline of your agent handling a single request. It captures everything from when a user sends a query until the final response is delivered. প্রতিটি ট্রেস দেখায়:
- Total duration of the request
- All operations that executed
- How operations relate to each other (parent-child relationships)
- When each operation started and ended
একটি স্প্যান কি?
A span represents a single unit of work within a trace. Common span types in your code review assistant:
-
agent_run: Execution of an agent (root agent or sub-agent) -
call_llm: Request to a language model -
execute_tool: Tool function execution -
state_read/state_write: State management operations -
code_executor: Running code with tests
Spans have:
- Name : What operation this represents
- Duration : How long it took
- Attributes : Metadata like model name, token counts, inputs/outputs
- Status : Success or failure
- Parent/child relationships : Which operations triggered which
স্বয়ংক্রিয় ইন্সট্রুমেন্টেশন
When you deployed with --trace-to-cloud , ADK automatically instruments:
- Every agent invocation and sub-agent call
- All LLM requests with token counts
- Tool executions with inputs/outputs
- State operations (read/write)
- Loop iterations in your fix pipeline
- Error conditions and retries
No code changes required - tracing is built into ADK's runtime.
Step 1: Access Cloud Trace Explorer
Open Cloud Trace in your Google Cloud Console:
- Navigate to Cloud Trace Explorer
- Select your project from the dropdown (should be pre-selected)
- You should see traces from your test in Module 7
If you don't see traces yet:
The test you ran in Module 7 should have generated traces. If the list is empty, generate some trace data:
python tests/test_agent_engine.py
Wait 1-2 minutes for traces to appear in the console.
আপনি যা খুঁজছেন
The Trace Explorer shows:
- List of traces : Each row represents one complete request
- Timeline : When requests occurred
- Duration : How long each request took
- Request details : Timestamp, latency, span count
This is your production traffic log - every interaction with your agent creates a trace.
Step 2: Examine a Review Pipeline Trace
Click on any trace in the list to open the waterfall view
You'll see a Gantt chart showing the complete execution timeline. Here's what a typical review pipeline trace looks like:
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) ────►
Reading the Waterfall
Each bar represents a span. The horizontal position shows when it started, and the length shows how long it took.
Key insights from this trace:
- Total latency : 2.3 seconds from request to response
- Critical path : TestRunner takes 1.2s (52% of total time)
- Bottleneck : Code execution within TestRunner takes 0.9s (75% of TestRunner's time)
- State operations : Very fast (10ms each) - not a concern
- Pipeline structure : Sequential execution - CodeAnalyzer → StyleChecker → TestRunner → FeedbackSynthesizer
Inspecting Span Details
এ ক্লিক করুন
call_llm: gemini-2.5-flash
span under FeedbackSynthesizer
You'll see detailed attributes for this LLM call:
{
"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"
}
}
এটি দেখায়:
- Which model was used
- How many tokens were consumed (input + output)
- Request duration
- Success/failure status
- The full prompt is also visible in the attributes (scroll to see it)
Understanding the Pipeline Flow
Notice how the trace reveals your architecture:
- Root agent (CodeReviewAssistant) receives the request
- State read retrieves the code to review
- Review pipeline orchestrates four sub-agents sequentially
- Each sub-agent uses tools and LLM calls to complete its work
- Final response flows back up through the hierarchy
This visibility helps you understand exactly what happens during each request.
Step 3: Analyze a Fix Pipeline Trace
The fix pipeline is more complex because it includes loops. Let's examine how traces capture iterative behavior.
Find a trace that includes "CodeFixPipeline" in the span names
You may need to scroll through your traces or send a request that triggers the fix pipeline. If you don't have one, you can generate it:
# In your test script, respond "yes" when asked to fix issues
python tests/test_agent_engine.py
Examining Loop Structure
Here's what a fix pipeline trace with 2 iterations looks like:
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) ────►
Key Observations About Loops
পুনরাবৃত্তি নিদর্শন:
- Two iterations : First attempt achieved partial success, second completed fully
- Progressive cost : Iteration 2 takes longer (4.5s vs 3.2s)
- State tracking : Each iteration writes FIX_STATUS to state
- Exit mechanism : Loop terminates via escalation when FIX_STATUS = "SUCCESSFUL"
What this reveals:
- Your loop architecture is working correctly
- Most fixes complete in 1-2 iterations (good design)
- Each iteration includes: fix generation → testing → validation
- Code execution dominates each iteration (1.5-1.7s)
- The loop properly exits when conditions are met
খরচ ভাঙ্গন:
- Iteration 1: 3.2s
- Iteration 2: 4.5s (longer due to accumulated context)
- Total loop: 7.8s
- Synthesis: 0.7s
- Total fix pipeline: 8.5s
Comparing to Review Pipeline
Review pipeline: ~2.3s
Fix pipeline: ~8.5s (with 2 iterations)
The fix pipeline takes ~3.7x longer, which makes sense:
- It includes iterative refinement
- It runs code multiple times (once per iteration)
- It accumulates context from previous attempts
ধাপ 4: আপনি যা আবিষ্কার করেছেন
Performance Patterns
From examining traces, you now know:
Review pipeline:
- Typical duration: 2-3 seconds
- Main time consumer: TestRunner (code execution)
- LLM calls: Fast (100-300ms each)
- State operations: Negligible (10ms)
Fix pipeline:
- Typical duration: 4-5 seconds per iteration
- Most fixes: 1-2 iterations
- Code execution: 1.5-2.0 seconds per iteration
- Progressive cost: Later iterations take longer
What's fast:
- State reads/writes (10ms)
- Tool executions for analysis (100ms)
- Individual LLM calls (100-300ms)
What's slow (but necessary):
- Code execution with tests (0.9-2.0s)
- Multiple loop iterations (cumulative)
Where to Look for Issues
When reviewing traces in production, watch for:
- Unusually long traces (>15 seconds) - investigate what went wrong
- Failed spans (status != OK) - errors in execution
- Excessive loop iterations (>2) - fix quality issues
- Very high token counts - prompt optimization opportunities
আপনি কি শিখেছি
Through Cloud Trace, you now understand:
✅ Request flow : Complete execution path through your pipelines
✅ Performance characteristics : What's fast, what's slow, and why
✅ Loop behavior : How iterations execute and terminate
✅ Span hierarchy : How operations nest within each other
✅ Trace navigation : Reading waterfall charts effectively
✅ Token visibility : Where LLM costs accumulate
মূল ধারণা আয়ত্ত
Traces and Spans:
- Traces = complete request timelines
- Spans = individual operations within traces
- Waterfall view shows execution hierarchy
- Automatic instrumentation via ADK
কর্মক্ষমতা বিশ্লেষণ:
- Reading Gantt chart visualizations
- Identifying critical paths
- Understanding duration distributions
- Spotting bottlenecks
Production Visibility:
- Every operation traced automatically
- Token usage captured per LLM call
- State changes visible and traceable
- Loop iterations individually tracked
পরবর্তী কি
Continue exploring Cloud Trace:
- Monitor traces regularly to catch issues early
- Compare traces to identify performance regressions
- Use trace data to inform optimization decisions
- Filter by duration to find slow requests
Advanced observability (optional):
- Export traces to BigQuery for complex analysis ( docs )
- Create custom dashboards in Cloud Monitoring
- Set up alerts for performance degradation
- Correlate traces with application logs
9. Conclusion: From Prototype to Production
What You've Built
You started with just seven lines of code and built a production-grade AI agent system:
# 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.
মূল স্থাপত্য নিদর্শন আয়ত্ত
প্যাটার্ন | বাস্তবায়ন | Production Impact |
টুল ইন্টিগ্রেশন | AST analysis, style checking | Real validation, not just LLM opinions |
Sequential Pipelines | Review → Fix workflows | Predictable, debuggable execution |
লুপ আর্কিটেকচার | Iterative fixing with exit conditions | Self-improving until success |
রাজ্য ব্যবস্থাপনা | Constants pattern, three-tier memory | Type-safe, maintainable state handling |
উৎপাদন স্থাপনা | Agent Engine via deploy.sh | Managed, scalable infrastructure |
পর্যবেক্ষণযোগ্যতা | Cloud Trace integration | Full visibility into production behavior |
Production Insights from Traces
Your Cloud Trace data revealed critical insights:
✅ Bottleneck identified : TestRunner's LLM calls dominate latency
✅ Tool performance : AST analysis executes in 100ms (excellent)
✅ Success rate : Fix loops converge within 2-3 iterations
✅ Token usage : ~600 tokens per review, ~1800 for fixes
These insights drive continuous improvement.
Clean Up Resources (Optional)
If you're done experimenting and want to avoid charges:
Delete Agent Engine deployment:
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)
)
Delete Cloud Run service (if created):
gcloud run services delete code-review-assistant \
--region=$GOOGLE_CLOUD_LOCATION \
--quiet
Delete Cloud SQL instance (if created):
gcloud sql instances delete your-project-db \
--quiet
Clean up storage buckets:
gsutil -m rm -r gs://your-project-staging
gsutil -m rm -r gs://your-project-artifacts
পরবর্তী পদক্ষেপ
With your foundation complete, consider these enhancements:
- Add more languages : Extend tools to support JavaScript, Go, Java
- Integrate with GitHub : Automatic PR reviews
- Implement caching : Reduce latency for common patterns
- Add specialized agents : Security scanning, performance analysis
- Enable A/B testing : Compare different models and prompts
- Export metrics : Send traces to specialized observability platforms
মূল গ্রহণ
- Start simple, iterate fast : Seven lines to production in manageable steps
- Tools over prompts : Real AST analysis beats "please check for bugs"
- State management matters : Constants pattern prevents typo bugs
- Loops need exit conditions : Always set max iterations and escalation
- Deploy with automation : deploy.sh handles all the complexity
- Observability is non-negotiable : You can't improve what you can't measure
অবিরত শেখার জন্য সম্পদ
- ADK Documentation
- Advanced ADK Patterns
- Agent Engine Guide
- Cloud Run Documentation
- Cloud Trace Documentation
Your Journey Continues
You've built more than a code review assistant—you've mastered the patterns for building any production AI agent:
✅ Complex workflows with multiple specialized agents
✅ Real tool integration for genuine capabilities
✅ Production deployment with proper observability
✅ State management for maintainable systems
These patterns scale from simple assistants to complex autonomous systems. The foundation you've built here will serve you well as you tackle increasingly sophisticated agent architectures.
Welcome to production AI agent development. Your code review assistant is just the beginning.