Batch Processing with AI APIs

Processing thousands of documents, summarizing hundreds of articles, or translating entire datasets — batch processing is essential for scaling AI workloads. This guide covers efficient batch processing patterns with Chinese AI models through TokenEase, reducing costs by up to 70% compared to sequential API calls.

Why Batch Processing?

Metric	Sequential	Parallel Batch	Improvement
100 requests	~300s	~15s	20x faster
Connection overhead	100 handshakes	10 handshakes	10x less
Queue wait time	100x latency	10x latency	90% reduction
Token cost	Same tokens	Same tokens	No change

Pattern 1: ThreadPool Parallel Execution

The simplest approach: send multiple requests simultaneously using a thread pool:

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

API_KEY = "your-tokenease-api-key"
BASE_URL = "https://tokenease.io/v1"

def process_single(item):
    response = requests.post(
        f"{BASE_URL}/chat/completions",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "model": "qwen",
            "messages": [
                {"role": "system", "content": "Summarize the following text in 2 sentences."},
                {"role": "user", "content": item["text"][:4000]}
            ],
            "max_tokens": 100
        },
        timeout=30
    )
    return {
        "id": item["id"],
        "summary": response.json()["choices"][0]["message"]["content"]
    }

def batch_process(items, max_workers=10):
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(process_single, item): item for item in items}
        
        for future in as_completed(futures):
            try:
                result = future.result()
                results.append(result)
            except Exception as e:
                item = futures[future]
                results.append({"id": item["id"], "error": str(e)})
    
    return results

# Process 100 documents
items = [{"id": i, "text": doc} for i, doc in enumerate(documents)]
start = time.time()
results = batch_process(items, max_workers=10)
print(f"Processed {len(results)} items in {time.time() - start:.1f}s")

Pattern 2: AsyncIO for Maximum Throughput

For even higher throughput, use asyncio with aiohttp:

import asyncio
import aiohttp

async def process_single_async(session, item):
    async with session.post(
        f"{BASE_URL}/chat/completions",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "model": "qwen",
            "messages": [
                {"role": "system", "content": "Classify sentiment: positive, negative, or neutral."},
                {"role": "user", "content": item["text"][:2000]}
            ],
            "max_tokens": 10
        }
    ) as response:
        data = await response.json()
        return {
            "id": item["id"],
            "sentiment": data["choices"][0]["message"]["content"]
        }

async def batch_process_async(items, concurrency=20):
    async with aiohttp.ClientSession() as session:
        semaphore = asyncio.Semaphore(concurrency)
        
        async def bounded_process(item):
            async with semaphore:
                return await process_single_async(session, item)
        
        tasks = [bounded_process(item) for item in items]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Handle errors
        processed = []
        for item, result in zip(items, results):
            if isinstance(result, Exception):
                processed.append({"id": item["id"], "error": str(result)})
            else:
                processed.append(result)
        
        return processed

# Usage
results = asyncio.run(batch_process_async(items, concurrency=20))

Pattern 3: Chunked Batching with Rate Limiting

For very large batches, process in chunks to avoid overwhelming the API:

import time

def chunked_batch_process(items, chunk_size=50, delay_between_chunks=1):
    all_results = []
    
    for i in range(0, len(items), chunk_size):
        chunk = items[i:i + chunk_size]
        print(f"Processing chunk {i//chunk_size + 1}/{(len(items)-1)//chunk_size + 1}")
        
        # Process chunk in parallel
        results = batch_process(chunk, max_workers=10)
        all_results.extend(results)
        
        # Rate limit between chunks
        if i + chunk_size < len(items):
            time.sleep(delay_between_chunks)
    
    return all_results

# Process 10,000 items in chunks
results = chunked_batch_process(large_dataset, chunk_size=50, delay_between_chunks=2)

Pattern 4: Priority Queue for Mixed Workloads

When you have both urgent and background tasks:

import queue
import threading

class PriorityBatchProcessor:
    def __init__(self, max_workers=10):
        self.task_queue = queue.PriorityQueue()
        self.results = {}
        self.lock = threading.Lock()
    
    def add_task(self, priority, item):
        # Lower priority number = higher priority
        self.task_queue.put((priority, time.time(), item))
    
    def worker(self):
        while True:
            try:
                priority, _, item = self.task_queue.get(timeout=5)
                result = process_single(item)
                
                with self.lock:
                    self.results[item["id"]] = result
                
                self.task_queue.task_done()
            except queue.Empty:
                break
    
    def process_all(self):
        threads = []
        for _ in range(10):
            t = threading.Thread(target=self.worker)
            t.start()
            threads.append(t)
        
        self.task_queue.join()
        
        for t in threads:
            t.join()
        
        return self.results

# Usage: urgent tasks priority 1, background tasks priority 5
processor = PriorityBatchProcessor()
for urgent_item in urgent_items:
    processor.add_task(1, urgent_item)
for bg_item in background_items:
    processor.add_task(5, bg_item)

results = processor.process_all()

Cost Optimization Strategies

1. Model Selection per Task

Not every task needs the most powerful model:

Task Type	Recommended Model	Reason
Simple classification	Qwen-Plus	Fastest, cheapest
Document summarization	GLM-5.1	Great Chinese comprehension
Complex analysis	DeepSeek-V4	Best reasoning
Translation	Qwen-Plus	Fast, accurate

2. Input Truncation

Truncate inputs to reduce token costs:

def smart_truncate(text, max_tokens=3000, tokenizer=None):
    # Rough estimation: 1 token ≈ 1.5 Chinese characters
    max_chars = int(max_tokens * 1.5)
    
    if len(text) <= max_chars:
        return text
    
    # Truncate at sentence boundary
    truncated = text[:max_chars]
    last_period = max(truncated.rfind('。'), truncated.rfind('！'), truncated.rfind('？'))
    
    if last_period > max_chars * 0.8:
        return truncated[:last_period + 1]
    
    return truncated + "..."

# Apply before batch processing
for item in items:
    item["text"] = smart_truncate(item["text"], max_tokens=2000)

3. Response Caching

Cache identical inputs to avoid redundant API calls:

import hashlib
import json

class ResponseCache:
    def __init__(self):
        self.cache = {}
    
    def _get_key(self, model, messages):
        content = json.dumps({"model": model, "messages": messages}, sort_keys=True)
        return hashlib.md5(content.encode()).hexdigest()
    
    def get(self, model, messages):
        key = self._get_key(model, messages)
        return self.cache.get(key)
    
    def set(self, model, messages, response):
        key = self._get_key(model, messages)
        self.cache[key] = response

# Usage
cache = ResponseCache()

def process_with_cache(item):
    messages = build_messages(item)
    cached = cache.get("qwen", messages)
    if cached:
        return cached
    
    result = process_single(item)
    cache.set("qwen", messages, result)
    return result

Error Handling at Scale

In batch processing, some requests will fail. Handle gracefully:

def process_with_fallback(item, max_retries=2):
    models = ["qwen", "zhipu", "deepseek"]
    
    for model in models:
        for attempt in range(max_retries):
            try:
                result = process_single_with_model(item, model)
                return {"success": True, "model": model, "result": result}
            except Exception as e:
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)
                continue
    
    return {"success": False, "error": "All models failed"}

# After batch processing, identify failures
failures = [r for r in results if not r.get("success")]
print(f"Success rate: {(len(results) - len(failures)) / len(results):.1%}")

Monitoring Batch Jobs

class BatchMonitor:
    def __init__(self, total_items):
        self.total = total_items
        self.completed = 0
        self.failed = 0
        self.start_time = time.time()
    
    def update(self, success=True):
        if success:
            self.completed += 1
        else:
            self.failed += 1
        
        elapsed = time.time() - self.start_time
        rate = (self.completed + self.failed) / elapsed
        remaining = (self.total - self.completed - self.failed) / rate if rate > 0 else 0
        
        print(f"Progress: {self.completed}/{self.total} "
              f"({self.completed/self.total:.1%}) | "
              f"Failed: {self.failed} | "
              f"ETA: {remaining/60:.1f}m")

# Usage
monitor = BatchMonitor(len(items))
for item in items:
    try:
        result = process_single(item)
        monitor.update(success=True)
    except:
        monitor.update(success=False)

Production Checklist

• Set appropriate concurrency (10-20 for most APIs, test first)
• Implement chunked processing for datasets > 1000 items
• Add retry logic with exponential backoff for failed requests
• Use model fallback to maximize success rate
• Cache responses for identical or near-identical inputs
• Truncate inputs to control token costs
• Monitor progress with ETA estimates for long jobs
• Save intermediate results to avoid losing progress on crashes

Conclusion

Batch processing transforms AI APIs from interactive tools into production data pipelines. With the right patterns — ThreadPool for simplicity, asyncio for maximum throughput, chunked processing for scale — you can process thousands of items in minutes rather than hours.

Start with parallel ThreadPool execution for quick wins. Add chunked processing and caching as your volume grows. Monitor success rates and implement model fallback for reliability. With Chinese AI models costing a fraction of OpenAI, large-scale batch processing is now economically viable for any team.