Batch Processing with Chinese AI Models: Process 10K Prompts for $1

July 20, 2026 · 12 min read · Pricing

When you need to classify 10,000 support tickets, generate 5,000 product descriptions, or analyze 1,000 documents, you can't do it one-by-one. You need batch processing — and Chinese AI models make it affordable at scale.

The Cost Math

Workload	Avg Tokens	GPT-4o	DeepSeek V4 Flash	Savings
10K classifications (200 in, 50 out each)	2M in, 500K out	$10.00	$0.56	94%
5K descriptions (500 in, 200 out each)	2.5M in, 1M out	$16.25	$0.91	94%
1K document summaries (5K in, 500 out each)	5M in, 500K out	$17.50	$0.98	94%

DeepSeek V4 Flash ($0.14/M in, $0.56/M out) is the champion for batch work. It's fast, cheap, and handles classification/summarization tasks well.

Python: Async Batch Processor

"""Batch processor for Chinese AI models with concurrency control"""
import os
import asyncio
import json
import aiohttp
from pathlib import Path

API_KEY = os.environ["AIWAVE_API_KEY"]
BASE_URL = "https://aiwave.live/v1"

async def call_model(session, prompt: str, model: str = "deepseek-v4-flash",
                     max_tokens: int = 200) -> dict:
    """Single API call with retry."""
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "max_tokens": max_tokens,
        "temperature": 0.1,
    }
    
    for attempt in range(3):
        try:
            async with session.post(
                f"{BASE_URL}/chat/completions",
                headers={"Authorization": f"Bearer {API_KEY}"},
                json=payload,
                timeout=aiohttp.ClientTimeout(total=30),
            ) as resp:
                data = await resp.json()
                if "choices" in data:
                    return {
                        "text": data["choices"][0]["message"]["content"],
                        "tokens": data["usage"]["total_tokens"],
                        "cost": (
                            data["usage"]["prompt_tokens"] / 1_000_000 * 0.14 +
                            data["usage"]["completion_tokens"] / 1_000_000 * 0.56
                        ),
                    }
                elif data.get("error"):
                    await asyncio.sleep(2 ** attempt)  # Rate limit backoff
        except Exception as e:
            await asyncio.sleep(2 ** attempt)
    
    return {"text": "FAILED", "tokens": 0, "cost": 0}

async def batch_process(prompts: list[str], concurrency: int = 10,
                        model: str = "deepseek-v4-flash") -> list[dict]:
    """Process prompts with bounded concurrency."""
    semaphore = asyncio.Semaphore(concurrency)
    
    async def bounded_call(session, prompt):
        async with semaphore:
            return await call_model(session, prompt, model)
    
    connector = aiohttp.TCPConnector(limit=concurrency)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [bounded_call(session, p) for p in prompts]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # Summarize
    total_cost = 0
    total_tokens = 0
    failures = 0
    for r in results:
        if isinstance(r, Exception):
            failures += 1
        else:
            total_cost += r["cost"]
            total_tokens += r["tokens"]
    
    print(f"Processed {len(prompts)} prompts")
    print(f"Total tokens: {total_tokens:,}")
    print(f"Total cost: ${total_cost:.4f}")
    print(f"Failures: {failures}")
    return [r for r in results if not isinstance(r, Exception)]

# === Usage ===
if __name__ == "__main__":
    prompts = [f"Classify this text: {text}" for text in open("tickets.txt")]
    results = asyncio.run(batch_process(prompts, concurrency=15))
    
    # Save results
    with open("results.jsonl", "w") as f:
        for r in results:
            f.write(json.dumps(r) + "\n")

Key Parameters

Parameter	Value	Why
temperature	0.1	Consistent, deterministic classification
max_tokens	200-500	Limit output to save cost
concurrency	10-20	AIWave supports 50 req/s burst; stay under that
retries	3	Handle rate limits with exponential backoff

Structured Output for Classification

# Force JSON output for reliable parsing
payload = {
    "model": "deepseek-v4-flash",
    "messages": [{
        "role": "system",
        "content": "Classify the text into one of: billing, technical, feature_request, other. Respond with ONLY a JSON object: {\"category\": \"...\", \"confidence\": 0.0-1.0}"
    }, {
        "role": "user",
        "content": prompt
    }],
    "max_tokens": 50,
    "temperature": 0.0,
    "response_format": {"type": "json_object"},
}

# Parse reliably
result = json.loads(response_text)
category = result["category"]
confidence = result["confidence"]

Common Batch Workloads

1. Sentiment Analysis

SYSTEM = "Rate sentiment from -1 to 1. Reply ONLY with a number."

# 10K reviews at ~150 tokens each = 1.5M tokens
# Cost: $0.14 * 1.5 + $0.56 * 0.1 = $0.27

2. Content Moderation

SYSTEM = "Flag as: safe, questionable, unsafe. Reply ONLY: safe|questionable|unsafe"

# Flag and queue for human review

3. Data Extraction

SYSTEM = """Extract: name, email, phone from text.
Reply ONLY with JSON: {"name": "", "email": "", "phone": ""}"""

# Process 5K forms for $0.50