Architecting Lean LLM Pipelines: Dynamic Context Compression for Cost-Efficient Production AI

Implement a multi-stage LLM processing pipeline leveraging context compression and structured output prompting to significantly reduce token costs and improve efficiency for real-time text analysis tasks.

I've been wrestling with a familiar demon in production LLM applications: the spiraling cost of API calls. It's easy to build a prototype where you feed an entire document to a powerful model and get brilliant insights, but scale that to thousands or millions of dynamic content streams, and your cloud bill quickly becomes unsustainable. The core issue often boils down to verbose inputs, unoptimized context windows, and inefficient output formats. This post isn't about theoretical savings; it's about architecting a robust, cost-aware pipeline that processes dynamic content streams efficiently, preventing budget overruns without sacrificing the quality of insights. If you're an engineer or data scientist looking to make your LLM-powered systems fiscally sustainable at scale, you'll learn how to implement a multi-stage approach, using a "scout" LLM to dynamically compress context before engaging a more powerful, expensive model for deep analysis.

Key Takeaways

• Proactively budget LLM token usage by estimating input costs before making API calls, allowing for dynamic adaptation.
• Leverage a cheaper, faster "scout" LLM to perform initial context compression, reducing the input size for subsequent, more expensive models.
• Design structured output prompts to guide LLMs towards concise, parseable responses, further minimizing output tokens and improving downstream processing.
• Implement robust error handling and fallback mechanisms to maintain pipeline stability even when LLM calls fail or encounter unexpected content.
• Optimize for cost at every stage of the pipeline, recognizing that token count directly translates to operational expenses in production.

The Problem

In a previous post, Guardian AI: Architecting Multi-Stage Hallucination Mitigation for Production Chatbots, I discussed how to build robust, multi-stage systems for LLMs. But even with the best guardrails, the fundamental unit of cost for most commercial LLMs is the token. When you're processing dynamic, unstructured text streams—like news articles, blog posts, or social media feeds—the sheer volume and verbosity can quickly blow through your budget. Sending a 5,000-token article to an LLM just to extract a few key entities or a short sentiment can be a massive waste if a 500-token summary would suffice. My team faced this exact challenge: how do we extract meaningful, structured insights from a continuous stream of technical blog posts without incurring prohibitive costs or sacrificing the depth of analysis when it truly matters?

Data and Sources

For this demonstration, we'll use a real-world, dynamic data source: the Stripe Engineering Blog RSS feed. This provides a continuous stream of technical articles, varying in length and complexity, which is perfect for illustrating dynamic context handling. We'll parse this feed to get the article titles and descriptions/summaries.

• Stripe Engineering Blog RSS Feed: https://stripe.com/blog/feed.rss
• Python feedparser library: https://pypi.org/project/feedparser/
• OpenAI tiktoken library: https://github.com/openai/tiktoken
• OpenAI Python Client: https://github.com/openai/openai-python

Data accessed on 2024-07-29.

Step 1 — Ingesting Dynamic Text Streams

The first hurdle is bringing in the data. Our source is an RSS feed, which means we need a reliable way to fetch and parse XML. Python's feedparser library is excellent for this, handling the intricacies of different RSS/Atom versions and providing a consistent dictionary-like structure for entries. The sub-problem here is transforming raw RSS XML into usable text content for our LLM pipeline.

Here's how I set up the ingestion:

import feedparser
import os
import tiktoken
from openai import OpenAI
import json

# Placeholder for API Key. In production, use a secure secret management system.
# For local testing, set OPENAI_API_KEY environment variable.
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if not OPENAI_API_KEY:
    raise ValueError("OPENAI_API_KEY environment variable not set.")

client = OpenAI(api_key=OPENAI_API_KEY)
# Using a common model for tokenization. Adjust if using a different core model.
TOKENIZER = tiktoken.encoding_for_model("gpt-3.5-turbo")

def get_token_count(text: str) -> int:
    """Calculates the token count for a given text using the global tokenizer."""
    return len(TOKENIZER.encode(text))

def fetch_and_parse_rss(url: str, num_entries: int = 5) -> list[dict]:
    """
    Fetches and parses an RSS feed, returning a list of dictionaries
    with relevant article information.
    """
    try:
        feed = feedparser.parse(url)
        if feed.bozo:
            print(f"Warning: RSS feed has parse errors: {feed.bozo_exception}")

        articles = []
        for entry in feed.entries[:num_entries]:
            title = entry.get('title', 'No Title')
            # Prefer 'summary' over 'description' if available, as it's often cleaner.
            description = entry.get('summary', entry.get('description', 'No Description'))
            link = entry.get('link', 'No Link')
            articles.append({
                'title': title,
                'description': description,
                'link': link
            })
        return articles
    except Exception as e:
        print(f"Error fetching or parsing RSS feed: {e}")
        return []

This snippet defines a function fetch_and_parse_rss that takes a URL and returns a list of dictionaries, each representing an article. I prioritize summary over description because, in my experience, summary fields are often more concise and pre-processed, making them better candidates for initial LLM input. It also includes basic error handling for network issues or malformed feeds.

Step 2 — Proactive Token Budgeting & Monitoring

Before we even think about calling an LLM, we need to understand the cost implication. The sub-problem here is knowing how many tokens an input text will consume and making a decision based on that. Simply sending everything to an LLM is a recipe for budget disaster. We need to budget proactively.

I integrate tiktoken to get an accurate token count for any given text. This allows us to implement conditional logic: if an article's description is short enough, we can send it directly to our main analysis LLM. If it's too long, that's when our "scout" LLM comes into play.

# ... (previous code) ...

# Define token limits for different stages
MAX_TOKENS_FOR_DIRECT_ANALYSIS = 300 # If description is under this, no compression needed
MAX_TOKENS_FOR_SCOUT_LLM_INPUT = 1500 # Max tokens scout LLM can handle effectively
TARGET_TOKENS_AFTER_COMPRESSION = 150 # Aim for this many tokens from scout LLM

def analyze_article_cost_effectively(article: dict) -> dict:
    """
    Processes an article, applying context compression if necessary,
    and then performing a deeper analysis.
    """
    title = article['title']
    original_description = article['description']
    link = article['link']

    original_description_tokens = get_token_count(original_description)
    print(f"\nProcessing '{title}' (Original Tokens: {original_description_tokens})")

    compressed_description = original_description
    compressed_description_tokens = original_description_tokens

    if original_description_tokens > MAX_TOKENS_FOR_DIRECT_ANALYSIS:
        print(f"