DeepSeek API Batch Content Generator

This tutorial provides a complete, production-ready Python script for generating content in batches using the DeepSeek API. It incorporates best practices for interacting with external APIs, including robust error handling, exponential backoff with retries, and clear configuration.

Problem Statement

Generating a large volume of unique content, descriptions, or creative text snippets using AI models can be a challenging task. When dealing with hundreds or thousands of prompts, developers often face several issues:

1. API Rate Limits: AI providers impose limits on how many requests can be made within a certain timeframe, leading to failed requests if not managed properly.
2. Transient Network Errors: Intermittent network connectivity issues or temporary API server unavailability can cause requests to fail, requiring manual retries.
3. Scalability and Automation: Manually triggering requests and saving responses is inefficient for batch processing. An automated solution is needed to run jobs unattended.
4. Error Handling and Logging: Without proper error handling and logging, it's difficult to diagnose failures, track progress, or determine which prompts succeeded or failed.

This script addresses these problems by providing an automated, resilient solution to generate content from a list of prompts, ensuring high reliability and efficient resource utilization.

How It Works

The script is designed for robustness and ease of use. At its core, it leverages the openai Python client library (which is compatible with DeepSeek's API) to send chat completion requests. For resilience against transient failures and rate limits, it integrates the tenacity library, which automatically handles retries with an exponential backoff strategy.

The workflow is as follows:

1. Configuration: API keys, base URLs, and model names are loaded from environment variables, ensuring secure handling of sensitive credentials and flexible deployment.
2. Prompt Loading: Prompts can be supplied via a text file (one prompt per line) or, for demonstration, a set of hardcoded examples. The script can also process a limited number of prompts using the --limit argument.
3. API Interaction: For each prompt, a dedicated function (generate_content_with_retry) is called. This function is decorated with @retry from tenacity, which means if an openai.APIError occurs (e.g., due to a rate limit or server error), the request will be automatically retried after an increasing delay, up to a configured maximum number of attempts.
4. Output Management: Successfully generated content is saved to individual text files within a specified output directory, making it easy to review and use the results. Each file includes both the original prompt and the generated response.
5. Logging and Reporting: Detailed logs are maintained, both to the console and a file, providing visibility into the progress, successful generations, and any failures. A summary report is presented at the end, detailing total attempts, successes, and failures.

This architecture ensures that the content generation process is largely self-sufficient, capable of recovering from common API interaction issues, and provides clear insights into its operation.

Full Script

#!/usr/bin/env python3
"""
DeepSeek API Batch Content Generator

This script generates content in batches using the DeepSeek API.
It processes a list of prompts, handles API rate limits and transient errors
with exponential backoff and retries, and saves the generated responses
to individual files in a specified output directory.
"""

import os
import argparse
import logging
import json
import time
from typing import List, Dict, Any, Optional

from openai import OpenAI, APIError
from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type

# --- Configuration ---
# DeepSeek API Key (required)
# It's recommended to set this as an environment variable for security.
DEEPSEEK_API_KEY: str = os.environ.get("DEEPSEEK_API_KEY", "")
if not DEEPSEEK_API_KEY:
    raise ValueError("DEEPSEEK_API_KEY environment variable not set. Please set it to your DeepSeek API key.")

# DeepSeek Base URL (default for DeepSeek chat models)
DEEPSEEK_BASE_URL: str = os.environ.get("DEEPSEEK_BASE_URL", "https://api.deepseek.com/v1")

# DeepSeek Model to use for content generation
DEEPSEEK_MODEL: str = os.environ.get("DEEPSEEK_MODEL", "deepseek-chat")

# Default output directory for generated content files
DEFAULT_OUTPUT_DIR: str = os.environ.get("OUTPUT_DIR", "generated_content")

# Retry configuration for API calls using tenacity
MAX_RETRIES: int = int(os.environ.get("MAX_RETRIES", 5))
INITIAL_RETRY_DELAY_SEC: float = float(os.environ.get("INITIAL_RETRY_DELAY_SEC", 1.0))
MAX_RETRY_DELAY_SEC: float = float(os.environ.get("MAX_RETRY_DELAY_SEC", 60.0))
API_TIMEOUT_SEC: float = float(os.environ.get("API_TIMEOUT_SEC", 60.0))

# --- Logging Setup ---
# Configure logging to output to both a file and the console
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("batch_ai_generator.log"), # Log to a file
        logging.StreamHandler()                        # Log to console
    ]
)
logger = logging.getLogger(__name__)

# --- API Client Initialization ---
# Initialize the OpenAI client, pointing its base_url to DeepSeek's API endpoint.
# This client is used for all API interactions.
client = OpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL,
    timeout=API_TIMEOUT_SEC # Set a reasonable timeout for API requests
)

# --- Core Logic ---

@retry(
    wait=wait_exponential(multiplier=INITIAL_RETRY_DELAY_SEC, min=1, max=MAX_RETRY_DELAY_SEC),
    stop=stop_after_attempt(MAX_RETRIES),
    retry=retry_if_exception_type(APIError),
    reraise=True # Re-raise the last exception if all retries fail, for main() to catch
)
def generate_content_with_retry(prompt: str, model: str) -> Optional[str]:
    """
    Generates content using the DeepSeek API, incorporating robust retry