中文版 README | English | PyPI Package

Intelligent batch processing tool for Replicate models with automatic fallback mechanisms and concurrent processing.

• 🔄 Intelligent Fallback System - Automatic model switching on incompatibility
• ⚡ Smart Concurrency Control - Adaptive rate limiting and batch processing
• 🎯 Three Usage Modes - Single, batch same-model, and mixed-model processing
• 📝 Custom File Naming - Ordered output with correspondence control
• 🛡️ Error Resilience - Comprehensive retry and recovery mechanisms
• ✅ Model Validation - Automatic detection of unsupported models with clear error messages

• Python: 3.8, 3.9, 3.10, 3.11, or 3.12
• Operating System: Windows, macOS, Linux
• Memory: Minimum 4GB RAM recommended

replicate>=0.15.0      # Replicate API client
requests>=2.25.0       # HTTP library
asyncio-throttle>=1.0.2  # Rate limiting for async operations
python-dotenv>=0.19.0  # Environment variable management

• Replicate API Token: Required (Get one here)
• Network: Stable internet connection for API calls
• Rate Limits: 600 requests/minute (shared across all models)

# First time installation
pip install replicate-batch-process

# Upgrade to latest version
pip install --upgrade replicate-batch-process

# Option 1: Interactive setup
replicate-init

# Option 2: Manual setup
export REPLICATE_API_TOKEN="your-token-here"

# Option 3: .env file
echo "REPLICATE_API_TOKEN=your-token-here" > .env

from replicate_batch_process import replicate_model_calling

file_paths = replicate_model_calling(
    prompt="A beautiful sunset over mountains",
    model_name="qwen/qwen-image",  # Use supported model
    output_filepath="output/sunset.jpg"
)

import asyncio
from replicate_batch_process import intelligent_batch_process

async def main():
    # Process multiple prompts with the same model
    files = await intelligent_batch_process(
        prompts=["sunset", "city", "forest"],
        model_name="qwen/qwen-image",
        max_concurrent=8,
        output_filepath=["output/sunset.png", "output/city.png", "output/forest.png"]
    )
    
    print(f"Generated {len(files)} images:")
    for file in files:
        print(f"  - {file}")
    
    return files

# Run the async function
if __name__ == "__main__":
    results = asyncio.run(main())

import asyncio
from replicate_batch_process import IntelligentBatchProcessor, BatchRequest

async def advanced_batch():
    processor = IntelligentBatchProcessor()
    
    # Create requests with different models and parameters
    requests = [
        BatchRequest(
            prompt="A futuristic city",
            model_name="qwen/qwen-image",
            output_filepath="output/city.png",
            kwargs={"aspect_ratio": "16:9"}
        ),
        BatchRequest(
            prompt="A magical forest",
            model_name="google/imagen-4-ultra",
            output_filepath="output/forest.png",
            kwargs={"output_quality": 90}
        ),
        BatchRequest(
            prompt="Character with red hair",
            model_name="black-forest-labs/flux-kontext-max",
            output_filepath="output/character.png",
            kwargs={"input_image": "reference.jpg"}
        )
    ]
    
    # Process all requests concurrently
    results = await processor.process_intelligent_batch(requests, max_concurrent=5)
    
    # Handle results
    successful = [r for r in results if r.success]
    failed = [r for r in results if not r.success]
    
    print(f"✅ Success: {len(successful)}/{len(results)}")
    for result in failed:
        print(f"❌ Failed: {result.error}")
    
    return results

# Run with proper async context
if __name__ == "__main__":
    asyncio.run(advanced_batch())

⚠️ Note: Using unsupported models will return a clear error message: "Model '{model_name}' is not supported. Please use one of the supported models listed above."

Automatic model switching when issues arise:

# User provides reference image to non-supporting model
replicate_model_calling(
    prompt="Generate based on this image",
    model_name="black-forest-labs/flux-dev",  # Doesn't support reference images
    input_image="path/to/image.jpg"           # → Auto-switches to flux-kontext-max
)

# Unsupported parameters automatically cleaned and model switched
replicate_model_calling(
    prompt="Generate image",
    model_name="black-forest-labs/flux-kontext-max",
    guidance=3.5,        # Unsupported parameter
    num_outputs=2        # → Auto-switches to compatible model
)

v1.0.9 Enhanced Triangular Fallback Loop:

• Flux Dev → Flux Kontext Max (for reference images)
• Flux Kontext Max → Qwen Image (on sensitive content)
• Qwen Image → Flux Dev (with weak censorship disabled)
• Ultimate Fallback: Black image (1600x900) if all models fail

The system automatically selects optimal processing strategy:

• Immediate Processing: Tasks ≤ available quota → Full concurrency
• Window Processing: Tasks ≤ 600 but > current quota → Wait then batch
• Dynamic Queue: Tasks > 600 → Continuous processing with queue management

Get your Replicate API token: replicate.com/account/api-tokens

Modify config.py:

FALLBACK_MODELS = {
    'your-model': {
        'fail': {
            'fallback_model': 'backup-model',
            'condition': 'api_error'
        }
    }
}

1. Async/Await: Batch functions must be called within async context
2. Model Names: Use exact model names from supported list above
3. File Paths: Ensure output directories exist before processing
4. Rate Limits: Keep max_concurrent ≤ 15 to avoid 429 errors

Tested on: Python 3.9.16, macOS, Replicate API

• Replicate API: 600 requests/minute (shared across all models)
• Recommended Concurrency: 5-8 (conservative) to 12 (aggressive)
• Auto-Retry: Built-in 429 error handling with exponential backoff

✅ FileOutput Handling Bug (v1.0.2-1.0.6)

• Issue: Kontext Max model created 814 empty files when returning single image
• Root Cause: Incorrect iteration over bytes instead of file object
• Fix: Added intelligent output type detection with hasattr(output, 'read') check
• Status: ✅ Fully resolved

✅ Parameter Routing Bug (v1.0.2-1.0.6)

• Issue: output_filepath incorrectly placed in kwargs for batch processing
• Fix: Corrected parameter assignment in BatchRequest
• Status: ✅ Fully resolved

⚠️ Kontext Max Parameter Compatibility

• Issue: Certain parameters cause Kontext Max to fail
• Solution: v1.0.9 implements triangular fallback loop for maximum success
• Impact: Minimal - automatic recovery ensures image generation
• Status: ✅ Enhanced with triangular fallback mechanism

ℹ️ Rate Limiting on Large Batches

• Issue: Batches >50 may hit rate limits even with throttling
• Workaround: Use chunking strategy (see Best Practices)
• Impact: Minor - automatic retry handles most cases

Found a bug? Please report at: https://github.com/preangelleo/replicate_batch_process/issues

pip install --upgrade replicate-batch-process==1.0.9

New Features:

• ✅ Triangular fallback loop for maximum success rate
• ✅ Black image (1600x900) as ultimate fallback
• ✅ Fixed flux-kontext-max parameter compatibility issues
• ✅ Simplified logging output

Action Required: None - fully backward compatible

pip install --upgrade replicate-batch-process==1.0.7

Critical Fixes:

1. intelligent_batch_process parameter bug - Now correctly handles output_filepath
2. FileOutput compatibility - No more empty file creation
3. Model validation - Clear error messages for unsupported models

Code Changes Needed: None, but review error handling for better messages

Breaking Changes:

• Package renamed from replicate_batch to replicate-batch-process
• New import structure:

  # Old (v0.x)
  from replicate_batch import process_batch
  
  # New (v1.0.7)
  from replicate_batch_process import intelligent_batch_process

Migration Steps:

1. Uninstall old package: pip uninstall replicate_batch
2. Install new package: pip install replicate-batch-process
3. Update imports in your code
4. Test with small batches first

# For large batches, use chunking
def process_large_batch(prompts, chunk_size=50):
    for chunk in chunks(prompts, chunk_size):
        files = await intelligent_batch_process(chunk, model_name)
        yield files

# Error handling with complete example
from replicate_batch_process import IntelligentBatchProcessor, BatchRequest

async def batch_with_error_handling():
    processor = IntelligentBatchProcessor()
    requests = [
        BatchRequest(prompt="sunset", model_name="qwen/qwen-image", output_filepath="output/sunset.png"),
        BatchRequest(prompt="city", model_name="qwen/qwen-image", output_filepath="output/city.png"),
    ]
    results = await processor.process_intelligent_batch(requests)
    
    for result in results:
        if result.success:
            print(f"✅ Generated: {result.file_paths}")
        else:
            print(f"❌ Failed: {result.error}")

asyncio.run(batch_with_error_handling())

replicate-batch-process/
├── main.py                      # Single image generation
├── intelligent_batch_processor.py  # Batch processing engine
├── config.py                    # Model configurations & fallbacks
├── init_environment.py          # Environment setup
└── example_usage.py            # Complete examples

# Clone repository
git clone https://github.com/preangelleo/replicate_batch_process.git

# Install in development mode
pip install -e .

# Run examples
python example_usage.py

MIT License - see LICENSE file for details.

1. Fork the repository
2. Create your feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request

• PyPI: https://pypi.org/project/replicate-batch-process/
• GitHub: https://github.com/preangelleo/replicate_batch_process
• Issues: https://github.com/preangelleo/replicate_batch_process/issues

Made with ❤️ for the AI community