Workflows & Orchestration
🎯 Purpose: Master workflow patterns to orchestrate complex agent behaviors and multi-step processes.
📚 Source of Truth: google/adk-python/src/google/adk/agents/workflow_agents/ (ADK 1.15)
[FLOW] Workflow Patterns Overview
Mental Model: Workflows are like assembly line strategies for agent orchestration:
┌──────────────────────────────────────────────────────────────┐
│ WORKFLOW PATTERNS │
├──────────────────────────────────────────────────────────────┤
│ │
│ [INSTR] SEQUENTIAL (Assembly Line) │
│ "One step after another, in order order" │
│ │
│ Step 1 → Step 2 → Step 3 → Step 4 │
│ Write Review Refactor Test │
│ │
│ Use: Pipelines, dependencies, order matters │
│ Pattern: Each step uses output from previous │
│ Source: agents/workflow_agents/sequential_agent.py │
│ │
│ [PARALLEL] PARALLEL (Fan-out/Gather) │
│ "Multiple tasks at once, then combine" │
│ │
│ ┌─── Task A ───┐ │
│ ├─── Task B ───┤ → Merge Results │
│ └─── Task C ───┘ │
│ Research Research Synthesis │
│ Source 1 Source 2 │
│ │
│ Use: Independent tasks, speed critical │
│ Pattern: Fan-out → Execute → Gather │
│ Source: agents/workflow_agents/parallel_agent.py │
│ │
│ [LOOP] LOOP (Iterative Refinement) │
│ "Repeat until good enough or max iterations" │
│ │
│ ┌──────────────────┐ │
│ │ ┌──► Critic ───┐│ │
│ │ │ ││ │
│ │ └─── Refiner ◄─┘│ │
│ └──────────────────┘ │
│ (Repeat 5x or until exit_loop) │
│ │
│ Use: Quality improvement, retry logic │
│ Pattern: Generate → Critique → Improve → Repeat │
│ Source: agents/workflow_agents/loop_agent.py │
│ │
└──────────────────────────────────────────────────────────────┘
[INSTR] Sequential Workflows (Assembly Line)
Basic Sequential Pattern
Mental Model: Steps execute in order, each using the output of the previous:
from google.adk.agents import SequentialAgent
# Define individual agents
research_agent = Agent(
name="researcher",
model="gemini-2.5-flash",
instruction="Research the given topic thoroughly",
output_key="research_results"
)
writer_agent = Agent(
name="writer",
model="gemini-2.5-flash",
instruction="Write a comprehensive article based on the research: {research_results}",
output_key="article_draft"
)
editor_agent = Agent(
name="editor",
model="gemini-2.5-flash",
instruction="Edit and improve the article: {article_draft}",
output_key="final_article"
)
# Create sequential workflow
content_pipeline = SequentialAgent(
name="content_creation_pipeline",
sub_agents=[research_agent, writer_agent, editor_agent],
description="Complete content creation from research to publication"
)
Sequential Workflow Execution
Execution Flow:
User Query → Research Agent → Writer Agent → Editor Agent → Final Result
1. Research agent gets user query
2. Research agent saves results to state['research_results']
3. Writer agent reads {research_results} from instruction
4. Writer agent saves draft to state['article_draft']
5. Editor agent reads {article_draft} from instruction
6. Editor agent produces final output
Advanced Sequential Patterns
Conditional Branching:
# Dynamic routing based on content type
def route_by_topic(context, result):
topic = result.get('topic', '').lower()
if 'technical' in topic:
return 'tech_writer'
elif 'business' in topic:
return 'business_writer'
else:
return 'general_writer'
routing_agent = Agent(
name="router",
model="gemini-2.5-flash",
instruction="Analyze the topic and determine content type",
output_key="topic_analysis"
)
tech_writer = Agent(name="tech_writer", ...)
business_writer = Agent(name="business_writer", ...)
general_writer = Agent(name="general_writer", ...)
# Sequential with dynamic agent selection
content_workflow = SequentialAgent(
sub_agents=[routing_agent], # Start with router
dynamic_agents={
'tech_writer': tech_writer,
'business_writer': business_writer,
'general_writer': general_writer
},
routing_function=route_by_topic
)
⚡ Parallel Workflows (Fan-out/Gather)
Basic Parallel Pattern
Mental Model: Independent tasks execute simultaneously, then results are combined:
from google.adk.agents import ParallelAgent
# Research different aspects in parallel
web_research_agent = Agent(
name="web_researcher",
model="gemini-2.5-flash",
tools=[google_search],
instruction="Research topic using web search",
output_key="web_findings"
)
database_research_agent = Agent(
name="db_researcher",
model="gemini-2.5-flash",
tools=[database_tool],
instruction="Search internal database for relevant data",
output_key="db_findings"
)
expert_opinion_agent = Agent(
name="expert_consultant",
model="gemini-2.5-flash",
tools=[expert_tool],
instruction="Consult domain experts on the topic",
output_key="expert_insights"
)
# Execute all research in parallel
parallel_research = ParallelAgent(
name="comprehensive_research",
sub_agents=[web_research_agent, database_research_agent, expert_opinion_agent],
description="Research topic from multiple sources simultaneously"
)
Parallel Execution Flow
Fan-out → Execute → Gather:
User Query
│
┌─── Fan-out ───┐
│ │
┌───▼──┐ ┌───▼──┐ ┌───▼───┐
│ Web │ │ DB │ │Expert │
│Search│ │Search│ │Consult│
└───┬──┘ └───┬──┘ └───┬───┘
│ │ │
└─── Gather ────────┘
│
Merge Results
│
Final Synthesis
Parallel with Sequential Merger
Complete Research Pipeline:
# Parallel research phase
parallel_research = ParallelAgent(
sub_agents=[web_agent, db_agent, expert_agent]
)
# Sequential synthesis phase
synthesis_agent = Agent(
name="synthesizer",
model="gemini-2.5-flash",
instruction="""
Synthesize findings from multiple sources:
Web: {web_findings}
Database: {db_findings}
Experts: {expert_insights}
Create a comprehensive report.
""",
output_key="final_report"
)
# Complete workflow: Parallel → Sequential
research_pipeline = SequentialAgent(
sub_agents=[parallel_research, synthesis_agent]
)
🔁 Loop Workflows (Iterative Refinement)
Basic Loop Pattern
Mental Model: Repeat until quality standards are met or max iterations reached:
from google.adk.agents import LoopAgent
# Content generator
writer_agent = Agent(
name="content_writer",
model="gemini-2.5-flash",
instruction="Write content on the topic: {topic}",
output_key="content_draft"
)
# Quality critic
critic_agent = Agent(
name="content_critic",
model="gemini-2.5-flash",
instruction="""
Evaluate the content quality: {content_draft}
Rate on scale 1-10 for:
- Accuracy
- Completeness
- Clarity
- Engagement
If score < 8, provide specific improvement suggestions.
""",
output_key="critique"
)
# Improvement refiner
refiner_agent = Agent(
name="content_refiner",
model="gemini-2.5-flash",
instruction="""
Improve the content based on critique: {critique}
Original: {content_draft}
Address all the critic's suggestions.
""",
output_key="improved_content"
)
# Iterative refinement loop
quality_loop = LoopAgent(
sub_agents=[critic_agent, refiner_agent],
max_iterations=5,
description="Iteratively improve content until quality standards are met"
)
Loop Execution Flow
Generate → Critique → Refine → Repeat:
Initial Content
│
┌──▼──┐
│Critic│ ←──┐
└─────┘ │
│ │
Critique │
│ │
┌──▼───┐ │
│Refine│ │
└──────┘ │
│ │
Improved │
│ │
Quality Check │
├─ Good ─────┘
└─ Poor ──────┐
│
Continue Loop
(up to max_iterations)
Advanced Loop Patterns
Conditional Exit:
def should_continue_loop(context, result):
"""Custom exit condition"""
critique = result.get('critique', '')
score = extract_score_from_critique(critique)
return score < 8 # Continue if quality < 8/10
quality_loop = LoopAgent(
sub_agents=[critic_agent, refiner_agent],
max_iterations=5,
exit_condition=should_continue_loop,
description="Iterative refinement with quality threshold"
)
Multi-Agent Loop:
# Complex iterative process
brainstorm_agent = Agent(name="brainstormer", ...)
designer_agent = Agent(name="designer", ...)
developer_agent = Agent(name="developer", ...)
tester_agent = Agent(name="tester", ...)
# Development cycle
development_loop = LoopAgent(
sub_agents=[designer_agent, developer_agent, tester_agent],
max_iterations=10,
description="Iterative product development cycle"
)
[FLOW] Complex Workflow Composition
Nested Workflows
Mental Model: Workflows can contain other workflows for hierarchical organization:
# Level 1: Individual research tasks
web_agent = Agent(name="web_researcher", ...)
api_agent = Agent(name="api_researcher", ...)
file_agent = Agent(name="file_analyzer", ...)
# Level 2: Parallel research
research_team = ParallelAgent(
sub_agents=[web_agent, api_agent, file_agent]
)
# Level 3: Sequential processing
processing_pipeline = SequentialAgent(
sub_agents=[
research_team, # Parallel research
data_cleaner, # Sequential processing
analyzer, # Sequential processing
reporter # Sequential processing
]
)
# Level 4: Quality loop
quality_assurance = LoopAgent(
sub_agents=[processing_pipeline, quality_checker, improver],
max_iterations=3
)
Real-World Example: Content Creation Pipeline
# 1. Research Phase (Parallel)
research_sources = ParallelAgent(
sub_agents=[
web_research_agent,
academic_search_agent,
social_media_monitor
]
)
# 2. Content Generation (Sequential)
content_creation = SequentialAgent(
sub_agents=[
outline_writer,
draft_writer,
fact_checker
]
)
# 3. Review & Editing (Loop)
editing_cycle = LoopAgent(
sub_agents=[
editor_agent,
proofreader_agent,
final_reviewer
],
max_iterations=3
)
# 4. Publication (Sequential)
publication_pipeline = SequentialAgent(
sub_agents=[
seo_optimizer,
formatter_agent,
publisher_agent
]
)
# Complete Content Pipeline
content_workflow = SequentialAgent(
sub_agents=[
research_sources, # Parallel
content_creation, # Sequential
editing_cycle, # Loop
publication_pipeline # Sequential
]
)
🎯 Workflow Decision Framework
When to Use Each Pattern
| Scenario | Sequential | Parallel | Loop |
|---|---|---|---|
| Order matters | ✅ Yes | ❌ No | ❌ No |
| Independent tasks | ❌ No | ✅ Yes | ❌ No |
| Need speed | ❌ No | ✅ Yes | ❌ No |
| Iterative refinement | ❌ No | ❌ No | ✅ Yes |
| Quality > speed | ❌ No | ❌ No | ✅ Yes |
| Dependencies | ✅ Yes | ❌ No | 🤔 Maybe |
Workflow Selection Guide
Need to orchestrate multiple agents?
│
├─ Steps depend on each other?
│ ├─ Simple dependency chain?
│ │ └─ SequentialAgent
│ └─ Complex dependencies?
│ └─ SequentialAgent + state routing
│
├─ Steps are independent?
│ ├─ Need results combined?
│ │ └─ ParallelAgent + Sequential merger
│ └─ Can process separately?
│ └─ ParallelAgent (fire and forget)
│
├─ Need iterative improvement?
│ ├─ Quality refinement?
│ │ └─ LoopAgent (critic + refiner)
│ └─ Progressive enhancement?
│ └─ LoopAgent (multi-stage improvement)
│
└─ Complex combination?
└─ Nested workflows (Parallel + Sequential + Loop)
⚡ Performance Optimization
Parallel Execution Benefits
Speed Improvements:
- Independent Tasks: 3x faster with 3 parallel agents
- I/O Bound: Network requests, API calls, file operations
- CPU Bound: Distribute across agents with different models
Cost Considerations:
- Token Efficiency: Same total tokens, faster execution
- Model Selection: Use smaller models for parallel tasks
- Caching: Cache intermediate results to avoid recomputation
Optimization Strategies
Batch Processing:
# Process multiple items in parallel
batch_processor = ParallelAgent(
sub_agents=[
Agent(name="item_1_processor", ...),
Agent(name="item_2_processor", ...),
Agent(name="item_3_processor", ...)
]
)
# More efficient than sequential processing
sequential_processor = SequentialAgent(
sub_agents=[item_1_processor, item_2_processor, item_3_processor]
)
Early Exit Optimization:
# Stop when good enough
quality_loop = LoopAgent(
sub_agents=[generator, critic, improver],
max_iterations=10,
exit_condition=lambda ctx, res: res.get('quality_score', 0) >= 9
)
🔍 Debugging Workflows
State Inspection
Track Data Flow:
# Enable state logging
import logging
logging.getLogger('google.adk.agents').setLevel(logging.DEBUG)
# Inspect state at each step
result = await runner.run_async(query)
for event in result.events:
if 'state' in event:
print(f"Step: {event.step}")
print(f"State: {event.state}")
Workflow Visualization
Execution Graph:
# Generate workflow diagram
workflow_graph = content_pipeline.get_execution_graph()
print(workflow_graph) # Mermaid diagram
# Analyze bottlenecks
performance_report = content_pipeline.analyze_performance()
print(performance_report) # Timing, bottlenecks, optimization suggestions
Common Issues & Solutions
| Issue | Symptom | Solution |
|---|---|---|
| State not passed | Agent can't access previous results | Check output_key and state interpolation |
| Parallel slowdown | Sequential execution instead of parallel | Verify agents are truly independent |
| Loop never exits | Infinite refinement cycles | Set max_iterations, add exit conditions |
| Memory bloat | State growing too large | Use temp: scope, clean up intermediate data |
| Race conditions | Non-deterministic results | Ensure proper state synchronization |
📚 Related Topics
- Agent Architecture →: Individual agent design
- Tools & Capabilities →: What agents can do
- LLM Integration →: How LLMs drive workflows
🎓 Hands-On Tutorials
- Tutorial 04: Sequential Workflows: Build ordered agent pipelines
- Tutorial 05: Parallel Processing: Run agents simultaneously for speed
- Tutorial 06: Multi-Agent Systems: Complex agent coordination
- Tutorial 07: Loop Agents: Iterative refinement patterns
🎯 Key Takeaways
- Sequential: For ordered, dependent steps (assembly line)
- Parallel: For independent tasks (fan-out/gather)
- Loop: For iterative refinement (critic/refiner pattern)
- Composition: Nest workflows for complex hierarchies
- Performance: Parallel execution for speed, sequential for dependencies
- State Flow: Use
output_keyand interpolation for data passing
🔗 Next: Learn about LLM Integration to understand how language models drive these workflows.