Optimizing LLM API Costs: A Deep Dive into Prompt Compression and Dynamic Batching

It started, as many painful lessons do, with a billing alert. A big, red, flashing number in my cloud provider dashboard. My heart sank. What was supposed to be a lean, efficient service for generating content had suddenly developed an insatiable appetite for tokens, and my wallet was feeling the burn. We’re talking about a 3x increase in LLM API costs over a single month. This wasn't just a blip; it was a crisis that threatened the sustainability of our entire operation.

My first thought was, "Did someone crank up the generation volume?" A quick check of our analytics showed a modest increase, but nowhere near enough to explain the exponential cost jump. The culprit, I quickly realized, wasn't just *how many* API calls we were making, but *how* we were making them and, crucially, *what* we were sending in each call. We were effectively paying premium prices for sending verbose, often redundant, information to large language models, and then paying again for each token generated in response. It was a classic case of unoptimized resource utilization, and I knew I had to fix it.

The Problem: Verbosity and Inefficiency in LLM Interactions

Our core process involved feeding user-provided inputs, augmented with context from various data sources, into an LLM to generate drafts. For example, a user might provide a short topic, and we'd pull in several related articles, product descriptions, or internal knowledge base entries to provide the LLM with sufficient context to generate a comprehensive blog post or article. The problem was, this "context" often grew unwieldy. We were sending entire documents, sometimes multiple, to the LLM, assuming "more context is always better."

Here’s a simplified look at what a typical API request payload might resemble, before my intervention:


{
    "model": "gpt-4-turbo",
    "messages": [
        {
            "role": "system",
            "content": "You are an expert content creator. Generate a compelling blog post based on the provided topic and context."
        },
        {
            "role": "user",
            "content": "Topic: Latest trends in AI ethics.\n\nContext:\nArticle 1: 'Quantum AI Nexus: Revolutionizing Machine Learning and Data Processing' (Full text, ~2000 words)\nArticle 2: 'My Battle with the Bots: Taming Hallucinations and Bias in My Generative AI' (Full text, ~1500 words)\nRecent news snippets: (Several paragraphs of unstructured text)\nUser's specific requirements: (Detailed list of requirements, tone, keywords, etc.)\n\nDraft a 1000-word blog post covering the topic, incorporating insights from the context, and ensuring a balanced perspective."
        }
    ],
    "max_tokens": 1200,
    "temperature": 0.7
}

Notice the "Full text" entries. Each of those represented thousands of tokens. When you combine that with the system prompt, the user's requirements, and the desired output length, we were regularly hitting context window limits and incurring massive input token costs. This wasn't just inefficient; it was financially unsustainable.

Strategy 1: Prompt Compression – Getting Leaner with Context

My first attack vector was the sheer volume of input tokens. I realized that while the LLM needed context, it didn't necessarily need *all* of it, verbatim. Much like a human, an LLM can often work with a summary or key points. This led me down the path of **prompt compression**.

Implementing Contextual Summarization

Instead of sending entire articles, I decided to pre-process the contextual documents. For each document, I’d extract key information or generate a concise summary. This required a separate, less expensive LLM call (or even a classical NLP approach for simpler cases) for each context document *before* constructing the main generation prompt.

Here’s a conceptual Python snippet illustrating the summarization approach:


import openai # Assuming OpenAI API, but concept applies universally
import tiktoken # For token counting

# --- Configuration ---
COMPRESSION_MODEL = "gpt-3.5-turbo-0125" # Cheaper, faster model for summarization
GENERATION_MODEL = "gpt-4-turbo"        # More capable model for final generation
MAX_COMPRESSED_TOKENS = 500             # Target token count for compressed context

def get_llm_response(model: str, messages: list, max_tokens: int, temperature: float = 0.7):
    """Helper function to get LLM response."""
    try:
        response = openai.chat.completions.create(
            model=model,
            messages=messages,
            max_tokens=max_tokens,
            temperature=temperature
        )
        return response.choices.message.content
    except Exception as e:
        print(f"LLM API error: {e}")
        return None

def summarize_document(document_text: str) -> str:
    """Summarizes a long document using a cheaper LLM."""
    prompt_messages = [
        {"role": "system", "content": "You are an expert summarizer. Condense the following document into a concise summary, focusing on key arguments, findings, and conclusions. Aim for no more than 300 words."},
        {"role": "user", "content": f"Document to summarize:\n\n{document_text}"}
    ]
    # We set a max_tokens for the summary itself to control output length
    summary = get_llm_response(COMPRESSION_MODEL, prompt_messages, max_tokens=350, temperature=0.3)
    return summary if summary else document_text # Fallback in case of error

def count_tokens(text: str, model_name: str) -> int:
    """Counts tokens for a given text and model."""
    encoding = tiktoken.encoding_for_model(model_name)
    return len(encoding.encode(text))

# --- Main process with compression ---
user_topic = "Latest trends in AI ethics"
raw_context_docs = {
    "Article 1": "Full text of Quantum AI Nexus...",
    "Article 2": "Full text of My Battle with the Bots...",
    "News Snippets": "Several paragraphs of unstructured news..."
}
user_requirements = "Draft a 1000-word blog post covering the topic, incorporating insights from the context, and ensuring a balanced perspective."

compressed_context_parts = []
total_compressed_tokens = 0

for title, doc_text in raw_context_docs.items():
    summary = summarize_document(doc_text)
    token_count = count_tokens(summary, GENERATION_MODEL)

    if total_compressed_tokens + token_count <= MAX_COMPRESSED_TOKENS:
        compressed_context_parts.append(f"{title} Summary:\n{summary}")
        total_compressed_tokens += token_count
    else:
        print(f"Skipping {title} due to token limit.")
        break # Or implement more sophisticated truncation/prioritization

final_context = "\n\n".join(compressed_context_parts)

# Constructing the final generation prompt
generation_messages = [
    {"role": "system", "content": "You are an expert content creator. Generate a compelling blog post based on the provided topic and context."},
    {"role": "user", "content": f"Topic: {user_topic}\n\nContext:\n{final_context}\n\nUser's specific requirements: {user_requirements}\n\nDraft a 1000-word blog post covering the topic, incorporating insights from the context, and ensuring a balanced perspective."}
]

# Calculate tokens for the final prompt (input + output)
input_tokens_for_generation = count_tokens(str(generation_messages), GENERATION_MODEL)
output_tokens_expected = 1200 # For a 1000-word post

print(f"Total input tokens for generation (after compression): {input_tokens_for_generation}")
print(f"Estimated total tokens for this call: {input_tokens_for_generation + output_tokens_expected}")

# final_blog_post = get_llm_response(GENERATION_MODEL, generation_messages, max_tokens=output_tokens_expected)
# print(final_blog_post)

This approach immediately yielded significant savings. By using a cheaper model (like gpt-3.5-turbo) for the summarization step, and then feeding only the *summaries* to the more expensive gpt-4-turbo for final generation, I dramatically reduced the input token count for the most costly API calls. On average, I saw a 40-50% reduction in input token costs for these multi-document generation tasks.

One crucial aspect here is balancing compression with context fidelity. If you over-compress, you risk losing important nuances, which can lead to hallucinations or less relevant outputs. This is where careful prompt engineering for the summarization step comes in. I found that iteratively refining the summarization prompt and comparing the output quality was essential. My previous struggles with model reliability, detailed in My Battle with the Bots: Taming Hallucinations and Bias in My Generative AI, definitely helped me in this iterative process.

Strategy 2: Dynamic Batching – Amortizing API Overhead

While prompt compression tackled the "what" we were sending, the next challenge was the "how." Even with leaner prompts, making individual API calls for every single generation request was inefficient, especially during periods of high concurrency. Each API call incurs a certain amount of overhead – network latency, API gateway processing, model loading, etc. For many LLM providers, pricing also often has a per-call component or a minimum charge, making numerous small calls more expensive than fewer large ones. This led me to explore **dynamic batching**.

The idea is simple: instead of sending an API request as soon as a generation task comes in, we accumulate multiple requests in a queue and send them together in a single batch call to the LLM API. The challenge is making it "dynamic" – balancing latency (how long a user waits for their request to be batched) with throughput and cost savings.

Designing a Dynamic Batching System

I implemented a simple in-memory queue and a dedicated worker thread. This worker would continuously monitor the queue. When either a maximum batch size was reached, or a certain timeout period elapsed since the first item was added to the batch, it would trigger a batched API call. This allowed us to process requests efficiently even during sporadic traffic, without excessive delays.

Here’s a conceptual outline of the batching mechanism:


import threading
import time
import queue
import uuid

# --- Configuration for Batching ---
MAX_BATCH_SIZE = 5     # Max number of requests per batch
BATCH_TIMEOUT_SECONDS = 1.0 # Max time to wait for a batch to fill
BATCH_MODEL = "gpt-3.5-turbo" # Example model for batching

class GenerationRequest:
    def __init__(self, request_id: str, messages: list, max_tokens: int, temperature: float, callback):
        self.request_id = request_id
        self.messages = messages
        self.max_tokens = max_tokens
        self.temperature = temperature
        self.callback = callback # Function to call with result or error

class LLMBatchProcessor:
    def __init__(self, api_key: str):
        self.api_key = api_key # Not used in this snippet, but for context
        self.request_queue = queue.Queue()
        self.batch_buffer = []
        self.last_batch_time = time.time()
        self.lock = threading.Lock()
        self.worker_thread = threading.Thread(target=self._process_batches, daemon=True)
        self.running = True
        self.worker_thread.start()

    def add_request(self, messages: list, max_tokens: int, temperature: float, callback):
        request_id = str(uuid.uuid4())
        req = GenerationRequest(request_id, messages, max_tokens, temperature, callback)
        self.request_queue.put(req)
        print(f"Added request {request_id} to queue.")

    def _process_batches(self):
        while self.running:
            try:
                # Try to get requests from queue without blocking indefinitely
                req = self.request_queue.get(timeout=0.1) # Short timeout
                with self.lock:
                    self.batch_buffer.append(req)
            except queue.Empty:
                pass # No new requests, check batch conditions

            with self.lock:
                current_time = time.time()
                # Check if batch is ready due to size or timeout
                if (len(self.batch_buffer) >= MAX_BATCH_SIZE) or \
                   (len(self.batch_buffer) > 0 and (current_time - self.last_batch_time >= BATCH_TIMEOUT_SECONDS)):
                    self._execute_batch()
                    self.last_batch_time = current_time

            time.sleep(0.05) # Small sleep to prevent busy-waiting

    def _execute_batch(self):
        if not self.batch_buffer:
            return

        print(f"Executing batch of {len(self.batch_buffer)} requests.")
        batch_to_process = self.batch_buffer[:] # Take a copy
        self.batch_buffer.clear() # Clear for next batch

        # In a real scenario, you'd use a batch API endpoint if available,
        # or make parallel API calls if the provider supports it and it's cheaper.
        # For simplicity, let's simulate individual calls for now,
        # but the actual benefit comes from a single batched API call if supported.

        # --- Simulated batched API call ---
        # In a real system, you'd send a single request with multiple inputs
        # or use an async model. For demonstration, we'll iterate.
        # The true cost benefit comes from the API provider's batching feature
        # or amortizing connection overhead.
        
        results = {}
        for req in batch_to_process:
            print(f"  Processing request {req.request_id} within batch...")
            # Simulate LLM call
            simulated_response = f"Generated content for {req.request_id} based on {req.messages['content'][:50]}..."
            results[req.request_id] = {"content": simulated_response, "tokens_used": 100} # Dummy data

        # Distribute results back to original callers via callbacks
        for req in batch_to_process:
            result = results.get(req.request_id)
            if result:
                req.callback(req.request_id, result["content"], result["tokens_used"])
            else:
                req.callback(req.request_id, None, 0, error="Processing failed")

    def stop(self):
        self.running = False
        self.worker_thread.join()
        print("Batch processor stopped.")

# --- Example Usage ---
if __name__ == "__main__":
    processor = LLMBatchProcessor(api_key="YOUR_API_KEY")

    def my_callback(request_id, content, tokens, error=None):
        if error:
            print(f"Request {request_id} failed: {error}")
        else:
            print(f"Request {request_id} completed. Content snippet: {content[:30]}..., Tokens: {tokens}")

    # Add some requests
    for i in range(7):
        messages_payload = [
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": f"Tell me a joke about {['cats', 'dogs', 'birds', 'fish', 'hamsters', 'snakes', 'pandas'][i % 7]}."}
        ]
        processor.add_request(messages_payload, max_tokens=50, temperature=0.7, callback=my_callback)
        time.sleep(0.3) # Simulate requests coming in over time

    print("Added all requests. Waiting for processing...")
    time.sleep(5) # Give time for batches to process

    processor.stop()
    print("Application finished.")

The real cost benefit of batching comes when the LLM provider offers a specific batch API endpoint, or when the cost model significantly favors fewer, larger calls. For instance, OpenAI recently introduced batching for their API, which can offer significant cost reductions (often 50% or more) and higher throughput compared to individual synchronous calls. Even without a dedicated batching endpoint, simply reducing the number of connection establishments and API call overheads through client-side batching can lead to measurable savings and improved throughput.

After implementing dynamic batching, particularly for our less latency-sensitive tasks like background content generation or bulk processing, I observed an additional 20-30% reduction in overall API transaction costs and a noticeable increase in our system's throughput. The combined effect of prompt compression and dynamic batching was truly transformative for our cost structure.

What I Learned / The Challenge

This whole experience was a stark reminder that simply adopting powerful new technologies like LLMs isn't enough; optimizing their usage is paramount for long-term viability. The biggest challenge was finding the right balance between cost savings and quality. Over-compressing prompts led to degraded output quality, and aggressive batching introduced unacceptable latency for real-time user interactions. It was an iterative process of tweaking parameters, A/B testing outputs, and closely monitoring both cost metrics and qualitative feedback.

Another learning was the importance of observability. Without detailed cost breakdowns by API call type, model, and even input/output tokens, pinpointing the exact source of the cost spike would have been much harder. Investing in robust logging and monitoring for LLM interactions is non-negotiable.

The journey also highlighted that "one size fits all" doesn't apply to LLM usage. Different tasks have different requirements for latency, context size, and output quality. This necessitated a more nuanced approach, where we dynamically chose compression strategies and batching parameters based on the specific use case.

Search This Blog

TechFrontier | AI Engineering, Go & Cloud Cost Optimization