Batch API and Production Checklist

The Batch API Alternative: 20% Off Tokens

When real-time processing isn’t a constraint, xAI’s Batch API offers a 20% reduction on all tokens for eligible text models (from grok-4.3 to grok-4.7 as of November 2, 2026, according to xAI batch pricing). Instead of sending requests one by one and managing concurrency, you submit your requests in batches via the API or in a JSONL file, and you retrieve results as processing completes—typically within 24 hours. Batch requests don’t count toward rate limits. The Batch API: Bulk Processing course details the workflow.

When to Use the Batch API

• Processing product catalogs (thousands of descriptions)
• Bulk classification of support tickets
• Generating summaries for a document base
• Any task where a delay of a few hours is acceptable

When to Stick with Classic Async

• Interactive chatbot (real-time response)
• Pipeline where each step depends on the previous one
• Real-time monitoring and alerts
• Any case where the Batch API delay is unacceptable

Approach Comparison

Production Checklist

Before deploying your asynchronous request system to production, check every point on this list:

Configuration

• Timeout is set to 3600 seconds for reasoning models
• Semaphore is sized based on your actual rate limits (not estimates)
• Exponential backoff is implemented with jitter for 429 errors
• API keys are loaded from environment variables, never hardcoded in code
• The x-grok-conv-id header is used for multi-turn conversations

Error Handling

• return_exceptions=True is used in asyncio.gather to isolate errors
• 429, timeout, and connection errors trigger a retry
• Invalid request errors (BadRequestError) do not trigger a retry
• Failed requests are identified and can be retried
• Logs include error type, attempt number, and retry delay

Monitoring

• Number of tokens consumed is tracked via response.usage
• Cache hit rate is measured to validate x-grok-conv-id usage
• Costs are estimated in real time with cost_in_usd_ticks
• Remaining rate limits are monitored to adjust the semaphore dynamically

Performance

• Chunking is used for batches larger than 100 requests
• Intermediate results are saved between sub-batches
• Progress is displayed for long-running jobs
• Semaphore is adjusted based on the model used (reasoning = lower)

Security

• API keys are not exposed in logs or error messages
• Responses containing sensitive data are processed in memory, not written to disk
• Remote storage (store_messages) is disabled for confidential data
• use_encrypted_content is enabled when reasoning confidentiality is required

Recommended Architecture

import os
import asyncio
import uuid
from asyncio import Semaphore
from openai import AsyncOpenAI, RateLimitError, APITimeoutError
import httpx

class GrokAsyncProcessor:
    def __init__(self, max_concurrent: int = 5):
        self.client = AsyncOpenAI(
            api_key=os.getenv("XAI_API_KEY"),
            base_url="https://api.x.ai/v1",
            timeout=httpx.Timeout(3600.0)
        )
        self.sem = Semaphore(max_concurrent)
        self.conv_id = str(uuid.uuid4())

    async def request(self, prompt: str, max_retries: int = 5) -> str:
        async with self.sem:
            for attempt in range(max_retries):
                try:
                    response = await self.client.chat.completions.create(
                        model="grok-4.5",
                        messages=[{"role": "user", "content": prompt}],
                        extra_headers={"x-grok-conv-id": self.conv_id}
                    )
                    return response.choices[0].message.content
                except RateLimitError:
                    if attempt == max_retries - 1:
                        raise
                    await asyncio.sleep(2 ** attempt)
                except APITimeoutError:
                    if attempt == max_retries - 1:
                        raise
                    await asyncio.sleep(2 ** attempt)

    async def batch(self, prompts: list[str]) -> list[str | Exception]:
        tasks = [self.request(p) for p in prompts]
        return await asyncio.gather(*tasks, return_exceptions=True)

This class encapsulates all patterns covered in this course: async client, semaphore, backoff, cache, and error handling. It provides a solid starting point for your production projects.

Key Takeaways

• The Batch API offers 20% off eligible models but imposes a processing delay
• Choose between real-time async and batch based on your latency constraints
• The production checklist covers configuration, errors, monitoring, performance, and security
• The GrokAsyncProcessor class encapsulates all course patterns in a single abstraction
• Always test your configuration with a small batch before launching bulk processing

Test Your Knowledge

Async, limits, cache: production patterns in five questions.

1. When does async prevail over sync?

Answer: Whenever handling multiple independent requests: the AsyncClient (or AsyncOpenAI) overlaps network waits—throughput multiplies without multiplying threads.

2. How do you process a batch without hitting limits?

Answer: asyncio.gather for parallelism, a semaphore to cap concurrent requests, and exponential backoff on HTTP 429—respecting the console’s RPM/TPM limits.

3. How do you handle multi-turn context at scale?

Answer: Through server-side response storage and reused encrypted reasoning—state lives on the API side, your code only tracks identifiers.

4. What do you need to know about cache and timeouts?

Answer: Conversation cache (x-grok-conv-id) speeds up followed threads, and reasoning requests require adapted timeouts—two settings that transform production.

5. What does the course production checklist contain?

Answer: Mastered async, respected limits (semaphore + backoff), batch for cold volume, configured timeouts and retries, tracked costs—robustness as a checklist.

Bounded concurrency, planned failures, server-side state: async patterns transform your scripts into services—the final checklist locks them down.