Building Autonomous AI systems with Looping Agents from Google’s Agent Development Kit(ADK) | by Ravi Manjunatha | Google Cloud – Community | Jun, 2025

# — 2. Import necessary libraries —
import asyncio
from google.adk.agents import LoopAgent, LlmAgent, SequentialAgent
from google.adk.runners import Runner, InMemorySessionService
from google.adk.tools.tool_context import ToolContext
from google.genai.types import Content, Part
from google.colab import userdata
import time
import random

print(“✅ Libraries imported.”)

# — 3. Set up your Gemini API Key —
try:
# Recommended: Store your API key securely using Colab Secrets.
from google.colab import userdata
import google.generativeai as genai

# Get the API key
api_key = userdata.get(‘GEMINI_API_KEY’)

# Configure the API key for both genai and ADK
genai.configure(api_key=api_key)

# Set environment variable for ADK to use
import os
os.environ[‘GOOGLE_API_KEY’] = api_key

print(“✅ Gemini API Key configured successfully.”)
except Exception as e:
print(f”❌ Error setting up API Key: {e}”)
print(“Please ensure ‘GEMINI_API_KEY’ is set in Colab Secrets.”)

# — 4. Define Sample Customer Data —
sample_customers = [
{
“customer_id”: “CUST001”,
“name”: “Priya Sharma”,
“age”: 28,
“location”: “Mumbai, Maharashtra”,
“purchase_history”: “ethnic wear, kurtas, sarees”,
“preferred_style”: “traditional with modern touch”
},
{
“customer_id”: “CUST002”,
“name”: “Rajesh Kumar”,
“age”: 35,
“location”: “Bangalore, Karnataka”,
“purchase_history”: “formal shirts, trousers, blazers”,
“preferred_style”: “professional and formal”
},
{
“customer_id”: “CUST003”,
“name”: “Ananya Patel”,
“age”: 22,
“location”: “Delhi, NCR”,
“purchase_history”: “casual tops, jeans, dresses”,
“preferred_style”: “trendy and casual”
}
]

# New apparel collections to promote
apparel_collections = [
{
“collection_name”: “Festive Fusion 2025”,
“category”: “ethnic wear”,
“description”: “Contemporary ethnic wear perfect for festivals and celebrations”,
“key_pieces”: “designer kurtas, embellished lehengas, silk sarees”,
“target_occasion”: “festivals and weddings”
},
{
“collection_name”: “Corporate Elite”,
“category”: “formal wear”,
“description”: “Premium formal wear for the modern professional”,
“key_pieces”: “tailored suits, formal shirts, leather accessories”,
“target_occasion”: “office and business meetings”
},
{
“collection_name”: “Gen-Z Street Style”,
“category”: “casual wear”,
“description”: “Trendy casual wear inspired by street fashion”,
“key_pieces”: “oversized hoodies, ripped jeans, crop tops”,
“target_occasion”: “casual outings and college”
}
]

print(f”✅ Sample data created: {len(sample_customers)} customers, {len(apparel_collections)} collections.”)

# — 5. Define Constants —
APP_NAME = “push_notification_quality_app”
USER_ID = “marketing_team”
GEMINI_MODEL = “gemini-2.5-flash”
GEMINI_PRO_MODEL = “gemini-2.5-pro” # Use Pro model for quality checking

# — State Keys —
STATE_CUSTOMER_DATA = “customer_data”
STATE_APPAREL_DATA = “apparel_data”
STATE_PUSH_NOTIFICATION = “push_notification”
STATE_QUALITY_FEEDBACK = “quality_feedback”
COMPLETION_PHRASE = “Push notification approved for sending.”

# — Tool Definition —
def approve_notification(tool_context: ToolContext):
“””Call this function ONLY when the push notification meets all quality standards.”””
print(f” [Tool Call] approve_notification triggered by ‘{tool_context.agent_name}’. Notification approved!”)
tool_context.actions.escalate = True # This signals the LoopAgent to stop.
return {}

# — Agent Definitions —

# Agent 1: Initial Push Notification Generator
initial_notification_generator = LlmAgent(
name=”NotificationGeneratorAgent”,
model=GEMINI_MODEL,
instruction=f”””You are a skilled e-commerce marketing specialist creating personalized push notifications for Indian customers.

**Customer Profile:**
– Name: {{customer_data.name}}
– Age: {{customer_data.age}}
– Location: {{customer_data.location}}
– Purchase History: {{customer_data.purchase_history}}
– Style Preference: {{customer_data.preferred_style}}

**New Apparel Collection:**
– Collection: {{apparel_data.collection_name}}
– Category: {{apparel_data.category}}
– Description: {{apparel_data.description}}
– Key Pieces: {{apparel_data.key_pieces}}
– Target Occasion: {{apparel_data.target_occasion}}

CRITICAL REQUIREMENTS:
1. Use EXACTLY this customer name: {{customer_data.name}} (not any other name)
2. Keep notification between 50-80 characters total
3. Reference the collection: {{apparel_data.collection_name}}
4. Make it relevant to {{customer_data.age}} year old from {{customer_data.location}}
5. Include appropriate emojis for Indian audience
6. Add clear call-to-action (Shop now, Explore, Browse, etc.)
7. NO template variables like {{}} should appear in final text

Create a personalized push notification that meets ALL requirements above.

Output ONLY the push notification text (no quotes, no extra text).”””,
output_key=STATE_PUSH_NOTIFICATION
)

# Agent 2: Quality Checker Agent (Using Pro model for better accuracy)
quality_checker_agent = LlmAgent(
name=”QualityCheckerAgent”,
model=GEMINI_PRO_MODEL, # Use Pro model for more accurate quality checking
instruction=f”””You are a strict marketing quality assurance specialist reviewing push notifications for an Indian e-commerce platform.

**Push Notification to Review:** “`{{push_notification}}“`

**REQUIRED Customer Context:**
– Required Customer Name: {{customer_data.name}}
– Age: {{customer_data.age}}
– Location: {{customer_data.location}}
– Style Preference: {{customer_data.preferred_style}}

**REQUIRED Collection Context:** {{apparel_data.collection_name}} – {{apparel_data.description}}

CRITICAL QUALITY CHECKS – ALL must pass:

1. **CORRECT NAME VALIDATION**:
– The notification MUST use EXACTLY “{{customer_data.name}}”
– Check if the notification contains the EXACT name “{{customer_data.name}}”
– If any other name appears (like “Arjun” when customer is “Rajesh Kumar”), this is a CRITICAL FAILURE
– If “Priya” appears when customer is “Ananya Patel”, this is a CRITICAL FAILURE

2. **NO TEMPLATE VARIABLES**: Check for any {{}} or unresolved variables like “{{customer_data.name}}” in the text

3. **LENGTH COMPLIANCE**: Must be exactly 50-80 characters – count carefully

4. **CORRECT COLLECTION**: Must mention “{{apparel_data.collection_name}}” or be clearly relevant to “{{apparel_data.category}}”

5. **PERSONALIZATION**: Must be appropriate for {{customer_data.age}} year old from {{customer_data.location}}

6. **CULTURAL APPROPRIATENESS**: Suitable for Indian audience and location context

7. **CLEAR CTA**: Must have actionable call-to-action

8. **PROFESSIONAL QUALITY**: No grammatical errors, appropriate emojis

STEP-BY-STEP VALIDATION:
Step 1: Does the notification contain the exact customer name “{{customer_data.name}}”?
Step 2: Are there any wrong names in the notification?
Step 3: Count characters exactly – is it 50-80?
Step 4: Does it reference the correct collection “{{apparel_data.collection_name}}”?

Current notification character count: [Count the exact characters in the notification]

IF ALL 8 criteria are perfectly met, respond with EXACTLY “{COMPLETION_PHRASE}”.

IF ANY criteria fail (especially wrong names), provide specific feedback: “CRITICAL FAILURE: [specific issue]. The notification uses wrong name/wrong collection/wrong length etc.”

Your assessment:”””,
output_key=STATE_QUALITY_FEEDBACK
)

# Agent 3: Notification Refiner Agent (Using Pro model for better corrections)
notification_refiner_agent = LlmAgent(
name=”NotificationRefinerAgent”,
model=GEMINI_PRO_MODEL, # Use Pro model for more accurate refinements
instruction=f”””You are a marketing content optimizer improving push notifications based on quality feedback.

**Current Notification:** “`{{push_notification}}“`
**Quality Feedback:** “`{{quality_feedback}}“`

**REQUIRED Customer Context:**
– Required Name: {{customer_data.name}}
– Age: {{customer_data.age}}
– Location: {{customer_data.location}}
– Style Preference: {{customer_data.preferred_style}}

**REQUIRED Collection:** {{apparel_data.collection_name}} – {{apparel_data.description}}

Analyze the feedback:
– IF the feedback is EXACTLY “{COMPLETION_PHRASE}”, call the ‘approve_notification’ function.
– IF there are improvement suggestions, apply them to create a better notification.

CRITICAL REQUIREMENTS for improved notification:
1. Use EXACTLY this customer name: “{{customer_data.name}}” (not Arjun, not Priya, not any other name)
2. Keep between 50-80 characters exactly
3. Reference collection: “{{apparel_data.collection_name}}”
4. Appropriate for {{customer_data.age}} year old from {{customer_data.location}}
5. NO template variables like {{}} should appear in final text
6. Address the specific feedback given
7. Include clear call-to-action

IMPORTANT: If the feedback mentions “CRITICAL FAILURE” about wrong names, you MUST fix the name to be exactly “{{customer_data.name}}”.

If improving, output ONLY the corrected notification text (no quotes, no extra text).
If approved, call the approval function.”””,
tools=[approve_notification],
output_key=STATE_PUSH_NOTIFICATION
)

# — Create the Pipeline —
quality_refinement_loop = LoopAgent(
name=”QualityRefinementLoop”,
sub_agents=[quality_checker_agent, notification_refiner_agent],
max_iterations=5 # Increased to allow more correction attempts
)

root_agent = SequentialAgent(
name=”PushNotificationQualityPipeline”,
sub_agents=[initial_notification_generator, quality_refinement_loop]
)

print(f”✅ Push notification pipeline ‘{root_agent.name}’ created successfully.”)

# — 6. Setup Session Service and Runner —
session_service = InMemorySessionService()

runner = Runner(
agent=root_agent,
app_name=APP_NAME,
session_service=session_service
)

print(f”✅ Runner created for agent ‘{runner.agent.name}’.”)

# — 7. Process Single Customer Notification —
async def create_notification_for_customer(customer_data: dict, apparel_data: dict):
“””Creates and refines a push notification for a single customer.”””

session_id = f”notification_{customer_data[‘customer_id’]}_{int(time.time())}”

print(“-” * 70)
print(f”\n📱 Creating notification for: {customer_data[‘name’]}”)
print(f” Customer ID: {customer_data[‘customer_id’]}”)
print(f” Location: {customer_data[‘location’]}”)
print(f” Collection: {apparel_data[‘collection_name’]}”)
print(f” Session ID: {session_id}”)
print(“-” * 70)

# Initialize state with customer and apparel data
initial_state = {
STATE_CUSTOMER_DATA: customer_data,
STATE_APPAREL_DATA: apparel_data
}

# Create session with initial state
session = await session_service.create_session(
app_name=APP_NAME,
user_id=USER_ID,
session_id=session_id,
state=initial_state
)

# Create initial message to start the pipeline
initial_message = Content(parts=[Part(text=”Create a personalized push notification for this customer.”)])

final_notification = “Notification generation failed.”

# Process through the pipeline
async for event in runner.run_async(
user_id=USER_ID,
session_id=session_id,
new_message=initial_message
):
author = event.author
content_text = “”

if event.content and event.content.parts:
for part in event.content.parts:
if hasattr(part, ‘text’) and part.text:
content_text = part.text.strip()
break
elif hasattr(part, ‘function_call’) and part.function_call:
func_name = part.function_call.name
content_text = f”[Function Call: {func_name}]”
break

print(f” [Event] {author}”)
if content_text:
# Truncate long content for readability
if len(content_text) > 120:
print(f” >> Content: \”{content_text[:120]}…\””)
else:
print(f” >> Content: \”{content_text}\””)

# Check for approval/completion
if event.actions and event.actions.escalate:
print(f” >> Action: Notification approved and ready for sending!”)
break

# Get final notification from session
final_session = await session_service.get_session(
app_name=APP_NAME,
user_id=USER_ID,
session_id=session_id
)

if final_session and STATE_PUSH_NOTIFICATION in final_session.state:
final_notification = final_session.state[STATE_PUSH_NOTIFICATION]

print(f”\n✅ Final Approved Notification:”)
print(f”📱 \”{final_notification}\””)
print(f” Length: {len(final_notification)} characters”)
print(“-” * 70)

return {
‘customer_id’: customer_data[‘customer_id’],
‘customer_name’: customer_data[‘name’],
‘location’: customer_data[‘location’],
‘collection’: apparel_data[‘collection_name’],
‘final_notification’: final_notification,
‘character_count’: len(final_notification)
}

# — 8. Process All Customers —
async def create_notifications_for_all_customers():
“””Creates personalized notifications for all sample customers.”””

print(“\n🚀 Starting Push Notification Generation for All Customers…”)
print(“=” * 80)

results = []

for i, customer in enumerate(sample_customers):
# Randomly assign an apparel collection for variety
apparel = random.choice(apparel_collections)

try:
result = await create_notification_for_customer(customer, apparel)
results.append(result)

# Small delay between customers
if i < len(sample_customers) – 1:
await asyncio.sleep(2)

except Exception as e:
print(f”❌ Error processing {customer[‘name’]}: {e}”)
results.append({
‘customer_id’: customer[‘customer_id’],
‘customer_name’: customer[‘name’],
‘location’: customer[‘location’],
‘collection’: apparel[‘collection_name’],
‘final_notification’: f”Error: {e}”,
‘character_count’: 0
})

# Display final results summary
print(“\n” + “=” * 80)
print(“🎉 PUSH NOTIFICATION GENERATION COMPLETE!”)
print(“=” * 80)

for result in results:
print(f”\n👤 {result[‘customer_name’]} ({result[‘location’]})”)
print(f”📦 Collection: {result[‘collection’]}”)
print(f”📱 Notification: \”{result[‘final_notification’]}\””)
print(f”📏 Length: {result[‘character_count’]} chars”)

print(f”\n✅ Successfully created {len(results)} personalized notifications!”)
return results

# — 9. Run the Complete Pipeline —
async def main():
“””Main execution function.”””
return await create_notifications_for_all_customers()

# Execute the pipeline
print(“\n🏁 Starting Push Notification Quality Pipeline…”)
await main()