Skip to main content

Simplified Sequential Processor for Google Gemini API

Project description

Gemini Parallel

A clean, simple Python library for Google Gemini API with intelligent key management and automatic rate limiting.

Features

  • Simple Sequential Processing - No threading complexity, just straightforward API calls
  • Smart Key Management - Automatic key rotation, cooldown, and exhaustion recovery
  • CLI Key Manager - Easy command-line tool to manage your API keys
  • Multi-Modal Support - Text, audio, and video inputs with flexible positioning
  • Auto Rate Limiting - Built-in protection against IP bans
  • Adaptive Cooldown - Automatically adjusts to API rate limits

Installation

pip install gemini-parallel

Or install from source:

git clone https://github.com/your-repo/gemini-parallel
cd gemini-parallel
pip install -e .

Quick Start

1. Set up your API keys

# Initialize .env file
geminiparallel init

# Add your API key(s)
geminiparallel add YOUR_GEMINI_API_KEY_HERE

# List your keys (masked)
geminiparallel list

# Test your keys
geminiparallel test

2. Use in your code

from gemini_parallel import GeminiSequentialProcessor, AdvancedApiKeyManager

# Initialize key manager (reads from .env)
key_manager = AdvancedApiKeyManager(keylist_names="all")

# Create processor
processor = GeminiSequentialProcessor(
    key_manager=key_manager,
    model_name="gemini-3-flash-preview",
    api_call_interval=4.0  # IP ban protection (default: 4.0)
)

# Process a single request
result = processor.process_single({
    'prompt': 'What is the capital of France?',
    'metadata': {'task_id': 'question_1'}
})

metadata, response, error = result
if error is None:
    print(response)

3. Process multiple requests

prompts = [
    {'prompt': 'What is AI?', 'metadata': {'task_id': 'q1'}},
    {'prompt': 'Explain quantum computing', 'metadata': {'task_id': 'q2'}},
    {'prompt': 'What is Python?', 'metadata': {'task_id': 'q3'}}
]

results = processor.process_batch(prompts)

for metadata, response, error in results:
    if error:
        print(f"{metadata['task_id']} failed: {error}")
    else:
        print(f"{metadata['task_id']}: {response[:100]}...")

CLI Commands

Manage your API keys easily with the geminiparallel command:

Command Description
geminiparallel init Initialize .env file
geminiparallel add KEY Add a new API key
geminiparallel list List all keys (masked)
geminiparallel remove KEY_NAME Remove a specific key
geminiparallel test Test all keys

Example workflow:

# Set up new project
cd my-project
geminiparallel init

# Add your keys
geminiparallel add AIza...your_key_1
geminiparallel add AIza...your_key_2

# Verify they work
geminiparallel test

# Check what you have
geminiparallel list

Multi-Modal Usage

Image Processing

# Single image from file
result = processor.process_single({
    'prompt': 'Describe this image: <image>',
    'image_path': '/path/to/image.jpg',
    'metadata': {'task_id': 'image_1'}
})

# Image from URL
result = processor.process_single({
    'prompt': 'What is in this image?',
    'image_url': 'https://example.com/image.jpg',
    'metadata': {'task_id': 'image_2'}
})

# Image from bytes
with open('/path/to/image.png', 'rb') as f:
    image_bytes = f.read()

result = processor.process_single({
    'prompt': 'Analyze this image: <image>',
    'image_bytes': image_bytes,
    'image_mime_type': 'image/png',
    'metadata': {'task_id': 'image_3'}
})

# Multiple images
result = processor.process_single({
    'prompt': 'Compare <image> with <image>',
    'image_path': ['/path/to/image1.jpg', '/path/to/image2.jpg'],
    'metadata': {'task_id': 'multi_image'}
})

Audio Processing

result = processor.process_single({
    'prompt': 'Transcribe and summarize: <audio>',
    'audio_path': '/path/to/audio.mp3',
    'metadata': {'task_id': 'audio_1'}
})

Video Processing

result = processor.process_single({
    'prompt': 'What happens in this video: <video>',
    'video_path': '/path/to/video.mp4',
    'metadata': {'task_id': 'video_1'}
})

Multiple Media Files

result = processor.process_single({
    'prompt': 'Describe <image>, then transcribe <audio>, and summarize <video>',
    'image_path': '/path/to/image.jpg',
    'audio_path': 'audio1.mp3',
    'video_path': 'video1.mp4',
    'metadata': {'task_id': 'multi_1'}
})

Key Management

Basic Setup

# Load all GEMINI_API_KEY_* from .env
key_manager = AdvancedApiKeyManager(keylist_names="all")

# Or specify exact keys
key_manager = AdvancedApiKeyManager(
    keylist_names=["GEMINI_API_KEY_1", "GEMINI_API_KEY_2"]
)

# Or use first N keys
key_manager = AdvancedApiKeyManager(keylist_names=5)

Paid vs Free Keys

# Mark some keys as paid (no cooldown)
key_manager = AdvancedApiKeyManager(
    keylist_names="all",
    paid_keys=["GEMINI_API_KEY_1", "GEMINI_API_KEY_2"]
)

# Or mark all as paid
key_manager = AdvancedApiKeyManager(
    keylist_names="all",
    paid_keys="all"
)

Custom Settings

key_manager = AdvancedApiKeyManager(
    keylist_names="all",
    key_settings={
        "free": {
            "key_cooldown_seconds": 30,      # 30s cooldown for free keys
            "exhausted_wait_seconds": 120,   # 2 min wait on rate limit
        },
        "paid": {
            "key_cooldown_seconds": 0,       # No cooldown for paid keys
            "exhausted_wait_seconds": 60,    # 1 min wait on rate limit
        }
    }
)

Generation Configuration

Customize AI responses:

result = processor.process_single({
    'prompt': 'Write a creative story',
    'generation_config': {
        'temperature': 1.0,  # Gemini 3: Keep at 1.0 (default, recommended)
        'top_p': 0.8,
        'max_output_tokens': 1000,
        'thinking_config': {
            'thinking_level': 'high'  # minimal, low, medium, high
        }
    },
    'metadata': {'task_id': 'creative_1'}
})

Gemini 3 Best Practices

Model Selection:

  • gemini-3-flash-preview: Fast, cost-effective, Pro-level intelligence
  • gemini-3-pro-preview: Complex tasks requiring advanced reasoning

Temperature: Keep at 1.0 (default). Gemini 3's reasoning is optimized for this value. Lower values may cause looping or degraded performance.

Thinking Level: Controls reasoning depth

  • minimal: Fastest, minimal thinking (Flash only)
  • low: Simple tasks, low latency
  • medium: Balanced (Flash only)
  • high: Maximum reasoning (default, dynamic)

Prompting Tips:

  • Be concise and direct - Gemini 3 prefers clear instructions
  • For large context (books, codebases): place questions at the end
  • For verbose responses: explicitly request conversational style
  • Anchor reasoning: "Based on the information above..."

Error Handling

The library handles errors automatically:

  • Resource Exhaustion (429) - Tries another key automatically
  • API Errors - Retries with exponential backoff
  • Network Issues - Graceful degradation
metadata, response, error = processor.process_single(prompt_data)

if error:
    if "Fatal: No usable API keys" in error:
        print("All keys are exhausted or invalid")
    elif "Persistent API error" in error:
        print("API call failed after retries")
    else:
        print(f"Error: {error}")
else:
    print(f"Success: {response}")

Rate Limiting

The library protects you from IP bans:

  • API Call Interval (default 4s) - Minimum time between ANY API calls
  • Key Cooldown - Per-key cooldown after use (configurable)
  • Adaptive Cooldown - Automatically increases intervals if too many 429 errors
processor = GeminiSequentialProcessor(
    key_manager=key_manager,
    model_name="gemini-3-flash-preview",
    api_call_interval=4.0,      # Global rate limit (IP ban protection, default: 4.0)
    api_call_retries=3          # Max retries per request
)

Text-to-Speech (TTS)

Generate speech from text:

from gemini_parallel import GeminiTTSProcessor

tts = GeminiTTSProcessor(
    key_manager=key_manager,
    model="flash"
)

# Generate speech
audio = tts.generate_speech(
    text="Hello, world!",
    voice="Kore",
    output_file="hello.wav"
)

# Multi-speaker dialogue
dialogue = [
    ("Speaker1", "Hello there!"),
    ("Speaker2", "Hi, how are you?")
]

audio = tts.generate_multi_speaker_dialogue(
    dialogue=dialogue,
    voices={"Speaker1": "Kore", "Speaker2": "Puck"},
    output_file="conversation.wav"
)

Environment Variables

Your .env file (managed by geminiparallel CLI):

# Gemini API Keys
GEMINI_API_KEY_1=your_first_key_here
GEMINI_API_KEY_2=your_second_key_here
GEMINI_API_KEY_3=your_third_key_here

Important: Add .env to your .gitignore!

echo ".env" >> .gitignore

License

MIT License - See LICENSE file for details.

Contributing

Contributions welcome! Please open an issue or PR on GitHub.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

gemini_parallel-0.8.0.tar.gz (26.7 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

gemini_parallel-0.8.0-py3-none-any.whl (26.4 kB view details)

Uploaded Python 3

File details

Details for the file gemini_parallel-0.8.0.tar.gz.

File metadata

  • Download URL: gemini_parallel-0.8.0.tar.gz
  • Upload date:
  • Size: 26.7 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.1.0 CPython/3.11.11

File hashes

Hashes for gemini_parallel-0.8.0.tar.gz
Algorithm Hash digest
SHA256 aa396c28cbe2537ddfc76be12c4b822e764e105f184b55be54e5cda61f564908
MD5 0f1a7901f9878ebfd49bc8e673958b23
BLAKE2b-256 da913888cb5853a98b2e20fc970a3f81796ececcee8593513081364e98367828

See more details on using hashes here.

File details

Details for the file gemini_parallel-0.8.0-py3-none-any.whl.

File metadata

File hashes

Hashes for gemini_parallel-0.8.0-py3-none-any.whl
Algorithm Hash digest
SHA256 75d3acad14f8e8c9939ebe75b28fe4226ff4d1d2413d38d7f44ca75144ea0f3e
MD5 78afe3982cab64cff55151ee31c022df
BLAKE2b-256 19589811fdcb7b0b5d935cb044d3f0c06757d36f174154d32466da7c7ff0c86f

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page