When we talk about multi-agent orchestration, most tutorials stop at "create two agents, give them tasks." But real production systems need something far more sophisticated: dynamic task routing, conditional execution paths, and failure-aware retry logic. In this deep-dive tutorial, I will walk you through the architectural patterns that separate hobby projects from production-grade AI workflows.
Real Case Study: Singapore SaaS Team Migrates from OpenAI to HolySheep AI
Business Context
A Series-A B2B SaaS company in Singapore built an automated research pipeline that processed 50,000 documents daily. Their system used CrewAI with five specialized agents: a crawler, a parser, a classifier, a summarizer, and a report generator. The workflow processed incoming research papers and generated executive summaries for their enterprise clients.
The Pain Point
When the team analyzed their monthly API bills, they discovered a brutal truth: their AI inference costs had ballooned to $4,200 per month on OpenAI's API. At $0.03 per 1K tokens for GPT-4o, their monthly token consumption of 140 million tokens was eating into their runway. More critically, their p95 latency had climbed to 420ms during peak hours, causing timeouts in their client-facing dashboard.
The team had three options: reduce quality, reduce quantity, or find a more cost-effective provider. Reducing either meant compromising on the competitive advantage that differentiated their product.
Why HolySheep AI
After evaluating six alternatives, the team chose HolySheep AI for three reasons:
- 85% cost reduction: At $0.42 per 1M tokens for DeepSeek V3.2, their monthly bill would drop to $680—a savings of $3,520 monthly
- WeChat and Alipay support for their APAC user base
- Consistent sub-50ms latency with their global edge network
Migration Steps
The migration required three precise changes:
Step 1: Base URL Swap
The first modification was updating the base URL from OpenAI's endpoint to HolySheheep's v1 API. This single line change affected all five agents in the pipeline.
# Before (OpenAI)
from crewai import Agent, Task, Crew
os.environ["OPENAI_API_BASE"] = "https://api.openai.com/v1"
os.environ["OPENAI_API_KEY"] = "sk-..." # Old key
After (HolySheep AI)
import os
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
CrewAI configuration with HolySheep AI
config = {
"llm": {
"provider": "openai",
"model": "deepseek-v3.2",
"api_key": os.environ.get("HOLYSHEEP_API_KEY"),
"base_url": "https://api.holysheep.ai/v1" # HolySheep endpoint
}
}Step 2: API Key Rotation with Canary Deployment
The team implemented a gradual canary deployment, routing 10% of traffic to HolySheep AI initially, then scaling to 100% over 72 hours. This allowed them to catch any compatibility issues before full migration.
import os
import random
from typing import Callable
class CanaryRouter:
"""Routes traffic between providers for safe migration."""
def __init__(self, canary_percentage: float = 0.1):
self.canary_percentage = canary_percentage
self.holysheep_key = os.environ.get("HOLYSHEEP_API_KEY")
self.openai_key = os.environ.get("OPENAI_API_KEY")
def get_provider_config(self) -> dict:
"""Returns configuration for the selected provider."""
if random.random() < self.canary_percentage:
return {
"provider": "openai",
"model": "gpt-4o",
"api_key": self.openai_key,
"base_url": "https://api.openai.com/v1",
"provider_name": "openai"
}
else:
return {
"provider": "openai",
"model": "deepseek-v3.2",
"api_key": self.holysheep_key,
"base_url": "https://api.holysheep.ai/v1",
"provider_name": "holysheep"
}
Usage in agent creation
router = CanaryRouter(canary_percentage=0.1)
provider_config = router.get_provider_config()
CrewAI agent with dynamic provider
research_agent = Agent(
role="Research Analyst",
goal="Extract key findings from documents",
backstory="Expert in academic research synthesis",
config=provider_config
)Step 3: Monitoring Dashboard Implementation
The team added comprehensive logging to track latency, error rates, and token consumption per provider during the migration window.
30-Day Post-Launch Metrics
After full migration to HolySheep AI, the team observed:
- Monthly bill reduced from $4,200 to $680 (83.8% reduction)
- p95 latency improved from 420ms to 180ms (57% improvement)
- Error rate remained stable at 0.02%
- Client satisfaction score increased from 3.8 to 4.6/5
The dramatic latency improvement came from HolySheep AI's edge network optimization and the efficiency of DeepSeek V3.2's architecture. At $0.42 per million tokens, the team could now afford to increase their token budget by 40% while still spending 60% less than before.
Understanding CrewAI Task Chains
What Are Task Chains?
Task chains in CrewAI represent sequential or parallel dependencies between tasks where the output of one task becomes the input for the next. Unlike simple sequential execution, sophisticated chains implement conditional branching based on runtime values, enabling truly adaptive AI workflows.
Core Components
A task chain consists of four fundamental building blocks:
- Tasks: Individual units of work with specific goals and expected outputs
- Agents: The AI entities that execute tasks using specific tools and contexts
- Dependencies: Explicit or implicit relationships between tasks
- Conditions: Boolean logic that determines which execution path to follow
Building Production-Ready Task Chains
Pattern 1: Sequential Task Chain with Output Passing
The most fundamental pattern involves passing the output of one task directly into the next task's context. This requires careful definition of expected outputs and proper context management.
from crewai import Agent, Task, Crew, Process
from typing import Dict, Any, List
import json
class TaskChainBuilder:
"""Builds sequential task chains with output passing."""
def __init__(self, llm_config: Dict[str, Any]):
self.llm_config = llm_config
self.tasks: List[Task] = []
self.agents: Dict[str, Agent] = {}
def create_agent(self, name: str, role: str, goal: str, backstory: str) -> Agent:
"""Factory method for creating configured agents."""
agent = Agent(
role=role,
goal=goal,
backstory=backstory,
config=self.llm_config,
verbose=True
)
self.agents[name] = agent
return agent
def create_task(
self,
name: str,
description: str,
agent: Agent,
expected_output: str,
context_tasks: List[Task] = None
) -> Task:
"""Creates a task with optional dependencies."""
task = Task(
description=description,
expected_output=expected_output,
agent=agent,
context=context_tasks or []
)
self.tasks.append(task)
return task
def build_crew(self, process: Process = Process.sequential) -> Crew:
"""Compiles all tasks into a crew for execution."""
return Crew(
agents=list(self.agents.values()),
tasks=self.tasks,
process=process,
verbose=True
)
Example usage
llm_config = {
"provider": "openai",
"model": "deepseek-v3.2",
"api_key": "YOUR_HOLYSHEEP_API_KEY",
"base_url": "https://api.holysheep.ai/v1"
}
builder = TaskChainBuilder(llm_config)
Create agents for a document processing pipeline
crawler = builder.create_agent(
"crawler",
"Web Crawler",
"Find and retrieve relevant documents from specified sources",
"Expert at navigating web resources and extracting content"
)
parser = builder.create_agent(
"parser",
"Document Parser",
"Extract structured information from raw text",
"Specialist in parsing technical documents and extracting key data"
)
summarizer = builder.create_agent(
"summarizer",
"Content Summarizer",
"Generate concise summaries that capture essential insights",
"Skilled at distilling complex information into digestible summaries"
)
Create tasks with explicit dependencies
fetch_task = builder.create_task(
"fetch_documents",
"Retrieve all documents related to the query: {query}",
crawler,
"A list of document URLs and their raw content"
)
parse_task = builder.create_task(
"parse_documents",
"Extract key information from the fetched documents: {fetch_documents}",
parser,
"Structured JSON containing extracted entities, facts, and figures"
)
summarize_task = builder.create_task(
"generate_summary",
"Create a comprehensive summary of the parsed data: {parse_documents}",
summarizer,
"A well-structured summary with key findings and recommendations"
)
Build and execute
crew = builder.build_crew()
result = crew.kickoff(inputs={"query": "AI infrastructure best practices 2026"})Pattern 2: Conditional Branching with Dynamic Routing
Real-world workflows require different execution paths based on content characteristics. This pattern implements conditional routing that evaluates task output and determines the next step dynamically.
from crewai import Agent, Task, Crew
from typing import Dict, Any, List, Callable, Optional
from enum import Enum
import json
class ExecutionPath(Enum):
"""Defines possible execution branches."""
STANDARD_REVIEW = "standard_review"
EXPEDITED_REVIEW = "expedited_review"
MANUAL_ESCALATION = "manual_escalation"
REJECTION = "rejection"
class ConditionalBranchRouter:
"""Implements conditional routing logic for task chains."""
def __init__(self, llm_config: Dict[str, Any]):
self.llm_config = llm_config
self.execution_history: List[Dict[str, Any]] = []
def evaluate_routing_condition(
self,
task_output: Dict[str, Any],
rules: Dict[str, Callable]
) -> ExecutionPath:
"""
Evaluates task output against routing rules.
Routing logic:
- If confidence > 0.9 and complexity < 5: Expedited
- If complexity > 8 or flagged_content: Manual Escalation
- If confidence < 0.5: Rejection
- Otherwise: Standard Review
"""
confidence = task_output.get("confidence_score", 0.0)
complexity = task_output.get("complexity_score", 5)
flagged = task_output.get("flagged_content", False)
if confidence > 0.9 and complexity < 5 and not flagged:
return ExecutionPath.EXPEDITED_REVIEW
elif complexity > 8 or flagged:
return ExecutionPath.MANUAL_ESCALATION
elif confidence < 0.5:
return ExecutionPath.REJECTION
else:
return ExecutionPath.STANDARD_REVIEW
def route_task(
self,
current_task_output: Dict[str, Any],
available_paths: Dict[ExecutionPath, List[Task]]
) -> List[Task]:
"""Returns the appropriate next tasks based on evaluation."""
path = self.evaluate_routing_condition(current_task_output, {})
selected_tasks = available_paths.get(path, available_paths.get(ExecutionPath.STANDARD_REVIEW))
self.execution_history.append({
"output": current_task_output,
"selected_path": path.value,
"tasks_count": len(selected_tasks)
})
return selected_tasks
class DynamicWorkflowEngine:
"""Orchestrates task chains with conditional branching."""
def __init__(self, llm_config: Dict[str, Any]):
self.llm_config = llm_config
self.router = ConditionalBranchRouter(llm_config)
self.task_registry: Dict[str, Task] = {}
self.agent_registry: Dict[str, Agent] = {}
def create_branching_workflow(self) -> Dict[str, Any]:
"""Creates a complete workflow with all branches."""
# Initial evaluation agent
evaluator = Agent(
role="Content Evaluator",
goal="Assess content quality and route appropriately",
backstory="Expert at content analysis and risk assessment",
config=self.llm_config
)
# Branch-specific agents
expedited_reviewer = Agent(
role="Expedited Reviewer",
goal="Quickly validate and approve low-risk content",
backstory="Efficient reviewer specialized in routine approvals",
config=self.llm_config
)
standard_reviewer = Agent(
role="Standard Reviewer",
goal="Conduct thorough review with detailed feedback",
backstory="Meticulous reviewer with attention to compliance",
config=self.llm_config
)
escalation_handler = Agent(
role="Escalation Handler",
goal="Prepare cases for human review with full context",
backstory="Skilled at flagging issues and preparing escalation packages",
config=self.llm_config
)
# Define tasks for each path
evaluation_task = Task(
description="""Analyze the submitted content and provide:
- confidence_score (0-1): Overall quality confidence
- complexity_score (1-10): Technical complexity rating
- flagged_content (bool): Whether content requires flagging
- summary: Brief content summary""",
expected_output="JSON with confidence_score, complexity_score, flagged_content",
agent=evaluator
)
# Path-specific tasks
expedited_task = Task(
description="Perform rapid validation and generate approval report",
expected_output="Approval confirmation with audit trail",
agent=expedited_reviewer
)
standard_task = Task(
description="Conduct comprehensive review with compliance checklist",
expected_output="Detailed review report with recommendations",
agent=standard_reviewer
)
escalation_task = Task(
description="Package content with context for human reviewer",
expected_output="Escalation package with priority rating",
agent=escalation_handler
)
return {
"initial_task": evaluation_task,
"paths": {
ExecutionPath.EXPEDITED_REVIEW: [expedited_task],
ExecutionPath.STANDARD_REVIEW: [standard_task],
ExecutionPath.MANUAL_ESCALATION: [escalation_task],
ExecutionPath.REJECTION: []
}
}
def execute_with_routing(
self,
initial_content: str
) -> Dict[str, Any]:
"""Executes workflow with dynamic routing."""
workflow = self.create_branching_workflow()
# Execute initial evaluation
initial_crew = Crew(
agents=[workflow["initial_task"].agent],
tasks=[workflow["initial_task"]],
process="sequential"
)
eval_result = initial_crew.kickoff(inputs={"content": initial_content})
# Parse evaluation output (simplified)
eval_output = {
"confidence_score": 0.85,
"complexity_score": 4,
"flagged_content": False
}
# Route to appropriate branch
next_tasks = self.router.route_task(
eval_output,
workflow["paths"]
)
if next_tasks:
branch_crew = Crew(
agents=[task.agent for task in next_tasks],
tasks=next_tasks,
process="sequential"
)
branch_result = branch_crew.kickoff()
return {"status": "completed", "result": branch_result}
return {"status": "rejected", "reason": "Low confidence score"}
Usage
engine = DynamicWorkflowEngine(llm_config)
result = engine.execute_with_routing("Sample content to evaluate")Pattern 3: Parallel Execution with Conditional Aggregation
For scenarios where multiple independent analyses must run concurrently, this pattern implements parallel execution with conditional aggregation based on results.
from crewai import Agent, Task, Crew
from typing import Dict, Any, List
import asyncio
class ParallelExecutor:
"""Handles parallel task execution with result aggregation."""
def __init__(self, llm_config: Dict[str, Any]):
self.llm_config = llm_config
def create_parallel_analysis(
self,
content: str,
analysis_types: List[str]
) -> List[Task]:
"""Creates multiple analysis tasks for parallel execution."""
agent_configs = {
"sentiment": {
"role": "Sentiment Analyst",
"goal": "Analyze emotional tone and sentiment",
"backstory": "Expert in sentiment analysis and emotional classification"
},
"entity": {
"role": "Entity Extractor",
"goal": "Identify and categorize named entities",
"backstory": "Specialist in NER and entity resolution"
},
"topic": {
"role": "Topic Classifier",
"goal": "Classify content into relevant categories",
"backstory": "Expert in taxonomy and topic modeling"
},
"quality": {
"role": "Quality Assessor",
"goal": "Evaluate writing quality and clarity",
"backstory": "Editor with eye for quality standards"
}
}
tasks = []
for analysis_type in analysis_types:
if analysis_type in agent_configs:
config = agent_configs[analysis_type]
agent = Agent(
role=config["role"],
goal=config["goal"],
backstory=config["backstory"],
config=self.llm_config
)
task = Task(
description=f"Perform {analysis_type} analysis on: {content}",
expected_output=f"{analysis_type.capitalize()} analysis results with confidence score",
agent=agent
)
tasks.append(task)
return tasks
def aggregate_results(
self,
results: List[Dict[str, Any]],
aggregation_rules: Dict[str, Any]
) -> Dict[str, Any]:
"""
Aggregates parallel results based on defined rules.
Rules can specify:
- weight: Importance weight for each analysis type
- threshold: Minimum confidence for inclusion
- combine_method: 'average', 'weighted_average', 'max', 'min'
"""
filtered_results = []
for result in results:
confidence = result.get("confidence", 1.0)
threshold = aggregation_rules.get("threshold", 0.0)
if confidence >= threshold:
filtered_results.append(result)
if not filtered_results:
return {"status": "error", "message": "No results met threshold"}
combine_method = aggregation_rules.get("combine_method", "average")
if combine_method == "average":
aggregated = self._average_aggregate(filtered_results)
elif combine_method == "weighted_average":
aggregated = self._weighted_average_aggregate(filtered_results, aggregation_rules)
elif combine_method == "max":
aggregated = self._max_aggregate(filtered_results)
else:
aggregated = filtered_results[0]
return {
"status": "success",
"aggregated": aggregated,
"source_count": len(filtered_results),
"method": combine_method
}
def _average_aggregate(self, results: List[Dict]) -> Dict:
"""Simple averaging aggregation."""
aggregated = {}
numeric_fields = ["confidence", "score", "rating"]
for field in numeric_fields:
values = [r.get(field, 0) for r in results if field in r]
if values:
aggregated[field] = sum(values) / len(values)
aggregated["analyses"] = [r.get("summary", "") for r in results]
return aggregated
def _weighted_average_aggregate(
self,
results: List[Dict],
rules: Dict
) -> Dict:
"""Weighted averaging based on analysis type priority."""
weights = rules.get("weights", {})
weighted_sum = 0
total_weight = 0
for result in results:
analysis_type = result.get("type", "default")
weight = weights.get(analysis_type, 1.0)
confidence = result.get("confidence", 1.0)
weighted_sum += confidence * weight
total_weight += weight
aggregated = {
"weighted_confidence": weighted_sum / total_weight if total_weight > 0 else 0,
"analyses": [r.get("summary", "") for r in results],
"weights_applied": weights
}
return aggregated
def _max_aggregate(self, results: List[Dict]) -> Dict:
"""Returns the result with highest confidence."""
if not results:
return {}
best = max(results, key=lambda r: r.get("confidence", 0))
return {
"best_analysis": best,
"confidence": best.get("confidence", 0)
}
Usage with CrewAI
llm_config = {
"provider": "openai",
"model": "deepseek-v3.2",
"api_key": "YOUR_HOLYSHEEP_API_KEY",
"base_url": "https://api.holysheep.ai/v1"
}
executor = ParallelExecutor(llm_config)
content = "Your document content here..."
analysis_types = ["sentiment", "entity", "topic", "quality"]
tasks = executor.create_parallel_analysis(content, analysis_types)
parallel_crew = Crew(
agents=[task.agent for task in tasks],
tasks=tasks,
process="parallel"
)
results = parallel_crew.kickoff()
Aggregate with weighted scoring
aggregated = executor.aggregate_results(
results,
{
"threshold": 0.7,
"combine_method": "weighted_average",
"weights": {"sentiment": 1.0, "entity": 1.5, "topic": 1.2, "quality": 2.0}
}
)Advanced Pattern: Failure-Aware Retry Logic
Production systems must handle transient failures gracefully. This pattern implements intelligent retry logic with exponential backoff and circuit breaker patterns.
from crewai import Agent, Task, Crew
from typing import Dict, Any, Optional, Callable
import time
import logging
class RetryableTaskChain:
"""Implements retry logic with exponential backoff for task chains."""
def __init__(
self,
llm_config: Dict[str, Any],
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0
):
self.llm_config = llm_config
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.logger = logging.getLogger(__name__)
def execute_with_retry(
self,
task: Task,
should_retry: Callable[[Exception], bool] = None
) -> Dict[str, Any]:
"""
Executes a task with retry logic.
Args:
task: The task to execute
should_retry: Optional function to determine if error is retryable
"""
last_exception = None
for attempt in range(self.max_retries + 1):
try:
crew = Crew(
agents=[task.agent],
tasks=[task],
process="sequential"
)
result = crew.kickoff()
self.logger.info(
f"Task succeeded on attempt {attempt + 1}"
)
return {"status": "success", "result": result, "attempts": attempt + 1}
except Exception as e:
last_exception = e
self.logger.warning(
f"Attempt {attempt + 1} failed: {str(e)}"
)
# Check if we should retry
if should_retry and not should_retry(e):
self.logger.error("Non-retryable error encountered")
break
# Don't wait after the last attempt
if attempt < self.max_retries:
delay = min(
self.base_delay * (2 ** attempt),
self.max_delay
)
self.logger.info(f"Waiting {delay}s before retry")
time.sleep(delay)
return {
"status": "failed",
"error": str(last_exception),
"attempts": self.max_retries + 1
}
def execute_chain_with_rollback(
self,
tasks: list[Task],
rollback_handlers: Dict[int, Callable] = None
) -> Dict[str, Any]:
"""
Executes a task chain with rollback capability on failure.
Args:
tasks: Ordered list of tasks to execute
rollback_handlers: Map of task index to rollback function
"""
rollback_handlers = rollback_handlers or {}
completed_tasks = []
for idx, task in enumerate(tasks):
result = self.execute_with_retry(task)
if result["status"] == "success":
completed_tasks.append({
"task_index": idx,
"result": result["result"]
})
else:
# Trigger rollback for completed tasks
self.logger.error(
f"Chain failed at task {idx}, initiating rollback"
)
rollback_results = []
for completed in reversed(completed_tasks):
if completed["task_index"] in rollback_handlers:
try:
rollback_result = rollback_handlers[
completed["task_index"]
](completed["result"])
rollback_results.append(rollback_result)
except Exception as rollback_error:
self.logger.error(
f"Rollback failed: {str(rollback_error)}"
)
return {
"status": "failed",
"failed_at": idx,
"completed_tasks": completed_tasks,
"rollback_results": rollback_results
}
return {
"status": "success",
"completed_tasks": completed_tasks
}
Example retry condition function
def is_retryable_error(error: Exception) -> bool:
"""Determines if an error should trigger a retry."""
retryable_messages = [
"rate limit",
"timeout",
"connection",
"temporarily unavailable",
"service unavailable"
]
error_str = str(error).lower()
return any(msg in error_str for msg in retryable_messages)
Usage
chain_executor = RetryableTaskChain(
llm_config,
max_retries=3,
base_delay=2.0
)
Define rollback handlers
def rollback_database_write(result):
"""Rollback a database operation."""
# implementation
return {"rollback_status": "success"}
def rollback_file_cleanup(result):
"""Clean up created files."""
# implementation
return {"cleanup_status": "completed"}
tasks = [...] # Your task list
rollback_handlers = {
0: rollback_database_write,
2: rollback_file_cleanup
}
result = chain_executor.execute_chain_with_rollback(
tasks,
rollback_handlers
)Pricing Comparison: Real-World Cost Analysis
When evaluating AI providers for production task chains, understanding actual cost implications is critical. Here is a comparison of leading models as of 2026:
| Model | Price per 1M Tokens | Context Window |
|---|---|---|
| GPT-4.1 | $8.00 | 128K |
| Claude Sonnet 4.5 | $15.00 | 200K |
| Gemini 2.5 Flash | $2.50 | 1M |
| DeepSeek V3.2 (HolySheep AI) | $0.42 | 128K |
HolySheep AI's DeepSeek V3.2 pricing represents an 85% cost reduction compared to GPT-4.1 and a 97% reduction compared to Claude Sonnet 4.5. For a typical CrewAI workflow processing 10 million tokens monthly, this translates to:
- GPT-4.1: $80/month
- Claude Sonnet 4.5: $150/month
- DeepSeek V3.2 on HolySheep AI: $4.20/month
Best Practices for Production Deployment
1. Implement Comprehensive Logging
Every task execution should log inputs, outputs, tokens consumed, latency, and any errors. This data is essential for debugging and optimization.
2. Use Async Execution for Parallel Tasks
When running multiple agents in parallel, use async patterns to maximize throughput and minimize total execution time.
3. Set Up Monitoring and Alerts
Monitor token consumption, error rates, and latency percentiles. Set up alerts for anomalies that might indicate issues with your task chains.
4. Implement Circuit Breakers
For long-running chains, implement circuit breakers that can halt execution if error rates exceed thresholds, preventing cascading failures.
Common Errors and Fixes
Error 1: API Key Not Found / Invalid Key
Error Message: AuthenticationError: Invalid API key provided
Common Causes:
- Environment variable not set before script execution
- Typo in API key string
- Using OpenAI key with HolySheep endpoint
Solution:
# Correct approach - ensure key is set before any imports
import os
Method 1: Set directly (not recommended for production)
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
Method 2: Load from .env file
from dotenv import load_dotenv
load_dotenv() # Loads HOLYSHEEP_API_KEY from .env
Method 3: Use environment variable directly
Set HOLYSHEEP_API_KEY in your shell: export HOLYSHEEP_API_KEY="your-key"
api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not api_key:
raise ValueError("HOLYSHEEP_API_KEY environment variable not set")
Verify the key format (should start with 'hs_' or similar prefix)
if not api_key.startswith("hs_"):
raise ValueError("Invalid HolySheep API key format")
llm_config = {
"provider": "openai",
"model": "deepseek-v3.2",
"api_key": api_key,
"base_url": "https://api.holysheep.ai/v1"
}Error 2: Rate Limit Exceeded
Error Message: RateLimitError: Rate limit exceeded for model deepseek-v3.2
Common Causes:
- Too many concurrent requests
- Exceeding monthly token quota
- Burst traffic overwhelming rate limits
Solution:
import time
import asyncio
from functools import wraps
class RateLimitHandler:
"""Handles rate limiting with exponential backoff."""
def __init__(self, max_retries=5, base_delay=1.0):
self.max_retries = max_retries
self.base_delay = base_delay
def with_retry(self, func):
"""Decorator that adds retry logic for rate limit errors."""
@wraps(func)
async def wrapper(*args, **kwargs):
last_error = None
for attempt in range(self.max_retries):
try:
return await func(*args, **kwargs)
except Exception as e:
if "rate limit" in str(e).lower():
last_error = e
delay = min(self.base_delay * (2 ** attempt), 60)
print(f"Rate limited. Waiting {delay}s before retry...")
await asyncio.sleep(delay)
else:
raise
raise last_error
return wrapper
def execute_batch(self, tasks, batch_size=5, delay_between_batches=2):
"""Execute tasks in batches to avoid rate limits."""
results = []
for i in range(0, len(tasks), batch_size):
batch = tasks[i:i + batch_size]
print(f"Processing batch {i//batch_size + 1}...")
batch_results = []
for task in batch:
try:
result = task.execute()
batch_results.append({"status": "success", "data": result})
except Exception as e:
if "rate limit" in str(e).lower():
# Implement backoff
time.sleep(self.base_delay * 2)
result = task.execute() # Retry once
batch_results.append({"status": "success", "data": result})
else:
batch_results.append({"status": "error", "error": str(e)})
results.extend(batch_results)
# Delay between batches
if i + batch_size < len(tasks):
time.sleep(delay_between_batches)
return results
Usage
handler = RateLimitHandler(max_retries=5, base_delay