Introduction
As AI applications become more capable, they are often asked to solve problems that involve several different types of reasoning. Trying to solve everything in a single AI request can make the process expensive, difficult to manage, and sometimes less effective.
In this article, we will explore a pattern where a complex problem is divided into smaller, independent tasks. Each task can then be handled separately, and the results can be combined to produce the final response. For example, imagine an AI system that needs to analyze a company. Instead of asking one AI agent to research everything at once, we can divide the work:
Agent 1: Analyze the company's financial performance.
Agent 2: Analyze its competitors.
Agent 3: Analyze recent market trends.
Agent 4: Analyze customer feedback.
Agent 5: Combine all the findings and prepare the final report.
Since the first four tasks do not depend on each other, they can be handled separately. This makes the overall solution easier to manage and allows multiple tasks to be processed at the same time.
Benefits of This Approach
This pattern provides several practical benefits:
Less information in each request: Each AI agent only receives the information it needs for its specific task instead of receiving the entire problem context.
Better handling of complex problems: A large problem can be divided into smaller problems, making each one easier for the AI to process accurately.
Lower cost: Different tasks can use different AI models depending on their complexity. A simple sub-task does not necessarily need the most expensive or powerful model.
Better scalability: When tasks are independent, more tasks can be processed concurrently, improving response speed as user demand scales.
When Should You Use This Pattern?
This approach is particularly effective when a problem can be broken down into structured, independent sub-components.
For example, consider an AI-powered travel planner handling the following request:
"Plan a 7-day trip to Japan, including flights, hotels, food, sightseeing, and a budget."
The workflow can be structured into three defined stages:
Initial Entry & Constraint Setup: The Trip Planner accepts the request and routes it to the Budget Options Agent to establish the financial framework.
Parallel Fan-Out: Using the budget baseline, sub-tasks are dispatched concurrently to specialized agents:
Flight Options Agent
Hotel Options Agent
Visiting Places Agent
Restaurants Agent
Final Synthesis (Fan-In): Results from all specialized agents are gathered and merged into a cohesive, structured itinerary.
By assigning focused responsibilities to specialized agents, complex AI workflows become faster, lower-cost, and far more reliable.
Implementation
Setup Environment
Bash
pip install google-genai
export GEMINI_API_KEY="your-api-key-here" Python Implementation
Python
import asyncio
import json
from google import genai
from google.genai import types
# Initialize the Gemini Client
client = genai.Client()
MODEL_NAME = "gemini-2.0-flash"
async def budget_agent(user_prompt: str) -> dict:
"""Step 1: Parse the user prompt and generate budget constraints."""
system_instruction = (
"You are a Travel Financial Advisor. Analyze the user request and "
"allocate budget ranges (in USD) for flights, hotels, sightseeing, and food. "
"Return output strictly as JSON with keys: total_budget, flight_budget, "
"hotel_budget, sightseeing_budget, food_budget, currency, target_destination."
)
response = await client.aio.models.generate_content(
model=MODEL_NAME,
contents=user_prompt,
config=types.GenerateContentConfig(
system_instruction=system_instruction,
response_mime_type="application/json",
temperature=0.2,
),
)
return json.loads(response.text)
async def flight_agent(destination: str, budget: float) -> str:
"""Parallel Worker: Research flights within budget."""
prompt = f"Find realistic round-trip flight options to {destination} within a ${budget} budget."
response = await client.aio.models.generate_content(
model=MODEL_NAME,
contents=prompt,
config=types.GenerateContentConfig(
system_instruction="You are a Flight Specialist. Provide 2-3 specific flight options with estimated costs.",
temperature=0.5,
),
)
return response.text
async def hotel_agent(destination: str, budget: float) -> str:
"""Parallel Worker: Research lodging options within budget."""
prompt = f"Find hotel options in {destination} for a 7-day stay within a total budget of ${budget}."
response = await client.aio.models.generate_content(
model=MODEL_NAME,
contents=prompt,
config=types.GenerateContentConfig(
system_instruction="You are a Hotel Specialist. Provide 2-3 accommodation options matching the budget.",
temperature=0.5,
),
)
return response.text
async def sightseeing_agent(destination: str, budget: float) -> str:
"""Parallel Worker: Research attractions within budget."""
prompt = f"Suggest top sights and activity packages in {destination} keeping activity costs under ${budget}."
response = await client.aio.models.generate_content(
model=MODEL_NAME,
contents=prompt,
config=types.GenerateContentConfig(
system_instruction="You are a Tour Guide Specialist. Recommend key attractions and day trips.",
temperature=0.5,
),
)
return response.text
async def restaurant_agent(destination: str, budget: float) -> str:
"""Parallel Worker: Research dining options within budget."""
prompt = f"Recommend dining options and food experiences in {destination} for 7 days within a total food budget of ${budget}."
response = await client.aio.models.generate_content(
model=MODEL_NAME,
contents=prompt,
config=types.GenerateContentConfig(
system_instruction="You are a Culinary Travel Expert. Provide restaurant recommendations from budget to mid-range.",
temperature=0.5,
),
)
return response.text
async def synthesis_agent(
original_prompt: str, budget_data: dict, flight_info: str, hotel_info: str, sight_info: str, food_info: str
) -> str:
"""Step 3 (Fan-In): Merge all parallel outputs into a complete trip itinerary."""
synthesis_prompt = f"""
Original Request: {original_prompt}
Budget Allocation: {json.dumps(budget_data, indent=2)}
--- FLIGHT OPTIONS ---
{flight_info}
--- ACCOMMODATION OPTIONS ---
{hotel_info}
--- SIGHTSEEING & ACTIVITIES ---
{sight_info}
--- DINING & RESTAURANTS ---
{food_info}
Task: Combine all the above options into a cohesive, structured 7-day itinerary.
Ensure all selections stay within the total allocated budget.
"""
response = await client.aio.models.generate_content(
model=MODEL_NAME,
contents=synthesis_prompt,
config=types.GenerateContentConfig(
system_instruction="You are a Lead Travel Planner. Synthesize specialized agent recommendations into an executive summary and daily schedule.",
temperature=0.3,
),
)
return response.text
async def main():
user_request = "Plan a 7-day trip to Japan with flights, hotels, food, sightseeing, and a budget of $3,500 total."
print("Step 1: Establishing Budget Baseline...")
budget_data = await budget_agent(user_request)
dest = budget_data.get("target_destination", "Japan")
print(f"Budget Plan Created: {budget_data}\n")
print("Step 2: Launching Parallel Specialized Agents (Fan-Out)...")
# asyncio.gather executes all four sub-agent API calls concurrently
flight_res, hotel_res, sight_res, food_res = await asyncio.gather(
flight_agent(dest, budget_data.get("flight_budget", 1000)),
hotel_agent(dest, budget_data.get("hotel_budget", 1200)),
sightseeing_agent(dest, budget_data.get("sightseeing_budget", 500)),
restaurant_agent(dest, budget_data.get("food_budget", 800)),
)
print("All specialized agents returned results.\n")
print("Step 3: Synthesizing Final Itinerary (Fan-In)...")
final_itinerary = await synthesis_agent(
original_prompt=user_request,
budget_data=budget_data,
flight_info=flight_res,
hotel_info=hotel_res,
sight_info=sight_res,
food_info=food_res,
)
print("\n================ FINAL ITINERARY ================\n")
print(final_itinerary)
if __name__ == "__main__":
asyncio.run(main())Conclusion
Decomposing complex tasks using the Orchestrator-Worker (Fan-Out/Fan-In) architecture transforms monolithic LLM prompts into manageable, parallelized micro-tasks. By leveraging asynchronous execution (asyncio.gather) and domain-focused sub-agents, AI systems achieve significantly faster response times, lower token costs through targeted prompting, and reduced hallucination rates. Adopting this multi-agent pattern provides a modular framework for building scalable and maintainable enterprise AI applications.
Join the conversation! Your thoughts help the community grow.