AI API Cost Optimization: Token Compression & Prompt Engineering实战技巧 (Practical Techniques)

Understanding Token Economics

Technique 1: Structured Prompt Compression

You are a helpful customer service assistant for an e-commerce company.
Your job is to respond to customer inquiries about products, orders,
shipping, returns, and general questions. Please be polite, professional,
and helpful in all your responses. If you don't know something, say so
honestly rather than making up information. Always try to be as helpful
as possible while maintaining a friendly tone...

Role: E-commerce customer support. Respond concisely to inquiries about
products, orders, shipping, returns. Be honest if unsure. Max 3 sentences.

Technique 2: Few-Shot Example Optimization

# BEFORE: Verbose few-shot (6 examples × 200 tokens = 1,200 tokens)
examples = """
Example 1:
Customer: Where is my order?
Assistant: I can help! Your order #12345 was shipped on March 15 via FedEx...
"""

AFTER: Compressed few-shot (3 examples × 45 tokens = 135 tokens)
examples = [
    {"q": "order status #12345", "a": "Shipped Mar 15, FedEx, ETA Mar 18"},
    {"q": "return policy", "a": "30 days, original packaging, free return shipping"},
    {"q": "track package", "a": "Use tracking #FX123456 at fedex.com or text TRACK to 12345"}
]

Production Code: Complete Optimization Pipeline

import requests
import json
import re

class TokenOptimizer:
    """Compress prompts and messages to reduce API costs by 60-80%."""
    
    def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
    
    def compress_message(self, content: str) -> str:
        """Remove redundant whitespace, shorten phrases."""
        # Remove extra whitespace
        content = re.sub(r'\s+', ' ', content).strip()
        
        # Dictionary of common phrase replacements
        replacements = {
            "please provide": "give",
            "could you please": "please",
            "I would like to": "I want",
            "in order to": "to",
            "at this point in time": "now",
            "due to the fact that": "because",
            "in the event that": "if",
            "for the purpose of": "for",
        }
        
        for old, new in replacements.items():
            content = content.replace(old, new)
        
        return content
    
    def build_efficient_messages(self, system: str, user: str, 
                                  context: dict = None) -> list:
        """Build compressed message structure."""
        messages = []
        
        # Compress system prompt
        if system:
            messages.append({
                "role": "system",
                "content": self.compress_message(system)
            })
        
        # Add context as system instruction if available (cheaper than user messages)
        if context:
            context_str = "Context: " + json.dumps(context, separators=(',', ':'))
            messages.append({
                "role": "system", 
                "content": self.compress_message(context_str)
            })
        
        # Compress user message
        messages.append({
            "role": "user",
            "content": self.compress_message(user)
        })
        
        return messages
    
    def chat(self, messages: list, model: str = "deepseek-v3",
             max_tokens: int = 500, temperature: float = 0.7) -> dict:
        """Send optimized request to HolySheep API."""
        payload = {
            "model": model,
            "messages": messages,
            "max_tokens": max_tokens,
            "temperature": temperature
        }
        
        response = self.session.post(
            f"{self.base_url}/chat/completions",
            json=payload,
            timeout=30
        )
        
        if response.status_code != 200:
            raise Exception(f"API Error {response.status_code}: {response.text}")
        
        result = response.json()
        
        # Calculate token savings
        usage = result.get("usage", {})
        input_tokens = usage.get("prompt_tokens", 0)
        output_tokens = usage.get("completion_tokens", 0)
        
        # Estimate savings (comparing to average unoptimized prompt)
        baseline_tokens = input_tokens * 3  # Rough estimate
        savings = ((baseline_tokens - input_tokens) / baseline_tokens) * 100
        
        result["cost_analysis"] = {
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "total_tokens": input_tokens + output_tokens,
            "estimated_savings_percent": round(savings, 1)
        }
        
        return result


Usage example
if __name__ == "__main__":
    optimizer = TokenOptimizer(api_key="YOUR_HOLYSHEEP_API_KEY")
    
    # Before optimization: ~2,500 tokens
    old_system = """
    You are an intelligent AI assistant helping customers with their 
    shopping experience. Your responsibilities include answering 
    product questions, helping customers find items, explaining 
    pricing and promotions, assisting with order tracking...
    """
    
    # After optimization: ~180 tokens
    new_system = """
    Shopping assistant: answer product questions, find items, 
    explain pricing/promos, track orders. Be concise.
    """
    
    messages = optimizer.build_efficient_messages(
        system=new_system,
        user="What's the price of the wireless headphones in blue?",
        context={"department": "electronics", "currency": "USD"}
    )
    
    result = optimizer.chat(messages, model="deepseek-v3", max_tokens=300)
    
    print(f"Response: {result['choices'][0]['message']['content']}")
    print(f"Token analysis: {result['cost_analysis']}")

Technique 3: Dynamic Context Injection

def build_dynamic_context(user_query: str, database_records: list) -> dict:
    """
    Select only relevant context based on query keywords.
    Reduces context from 2,000+ tokens to 200-400 tokens.
    """
    query_keywords = set(user_query.lower().split())
    relevant_records = []
    
    for record in database_records:
        # Score relevance based on keyword matches
        record_text = json.dumps(record).lower()
        relevance_score = sum(1 for kw in query_keywords if kw in record_text)
        
        if relevance_score >= 2:  # Threshold for inclusion
            relevant_records.append(record)
        
        # Stop if we have enough context
        if len(relevant_records) >= 5:
            break
    
    return {"relevant_context": relevant_records, "match_count": len(relevant_records)}


def optimized_rag_query(vector_db, user_question: str, api_key: str):
    """
    Retrieval-Augmented Generation with token optimization.
    Fetches minimal relevant context instead of entire document chunks.
    """
    optimizer = TokenOptimizer(api_key)
    
    # Fetch top 3 most relevant chunks (not top 20)
    relevant_chunks = vector_db.similarity_search(
        user_question, 
        k=3  # Reduced from typical k=20
    )
    
    # Truncate each chunk to first 200 characters
    truncated_chunks = [
        chunk.page_content[:200] + "..." 
        for chunk in relevant_chunks
    ]
    
    context = {"sources": truncated_chunks}
    
    messages = optimizer.build_efficient_messages(
        system="Answer based ONLY on provided context. Be concise.",
        user=user_question,
        context=context
    )
    
    result = optimizer.chat(messages, max_tokens=400)
    return result

Technique 4: Response Length Control

def estimate_response_tokens(task: str) -> int:
    """Estimate minimum tokens needed based on task type."""
    task_lengths = {
        "classification": 5,
        "sentiment": 10,
        "extraction": 50,
        "summary": 150,
        "analysis": 300,
        "creative": 500
    }
    
    for task_type, tokens in task_lengths.items():
        if task_type in task.lower():
            return tokens
    
    return 200  # Default


def smart_completion_request(prompt: str, task_type: str, api_key: str):
    """
    Automatically tune max_tokens based on task requirements.
    Prevents paying for 2,048 tokens when you only need 50.
    """
    optimizer = TokenOptimizer(api_key)
    
    # Calculate optimal max_tokens
    base_tokens = estimate_response_tokens(task_type)
    prompt_overhead = len(prompt.split()) * 1.3  # Rough input token estimate
    optimal_max = max(50, min(base_tokens, 1000))  # Clamp between 50-1000
    
    messages = optimizer.build_efficient_messages(
        system=f"Task: {task_type}. Respond concisely in {optimal_max} words max.",
        user=prompt
    )
    
    # Use lower temperature for extraction/classification tasks
    temperature = 0.3 if task_type in ["classification", "extraction"] else 0.7
    
    result = optimizer.chat(
        messages, 
        max_tokens=optimal_max,
        temperature=temperature
    )
    
    print(f"Requested max: {optimal_max} tokens")
    print(f"Actual usage: {result['cost_analysis']['output_tokens']} tokens")
    print(f"Waste prevented: {optimal_max - result['cost_analysis']['output_tokens']} tokens")
    
    return result

Real-World Cost Comparison

Common Errors and Fixes

Error 1: 401 Unauthorized - Invalid API Key

requests.exceptions.HTTPError: 401 Client Error: UNAUTHORIZED

# WRONG - Missing or malformed authorization
headers = {
    "Authorization": api_key  # Missing "Bearer " prefix
}

CORRECT - Proper Bearer token format
import os

api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not api_key:
    raise ValueError("HOLYSHEEP_API_KEY environment variable not set")

headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json"
}

Verify key format (should start with "hs_" or be 32+ characters)
if not api_key.startswith("hs_") and len(api_key) < 32:
    raise ValueError("Invalid API key format. Check your dashboard at https://www.holysheep.ai/register")

Error 2: 429 Rate Limit Exceeded

requests.exceptions.HTTPError: 429 Client Error: TOO MANY REQUESTS

import time
from collections import deque
from threading import Lock

class RateLimitedClient:
    """Handle rate limiting with exponential backoff."""
    
    def __init__(self, api_key: str, requests_per_minute: int = 60):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.request_times = deque(maxlen=requests_per_minute)
        self.lock = Lock()
        self.max_retries = 5
    
    def throttled_request(self, payload: dict) -> dict:
        """Make request with automatic rate limiting."""
        for attempt in range(self.max_retries):
            with self.lock:
                now = time.time()
                
                # Remove requests older than 1 minute
                while self.request_times and now - self.request_times[0] > 60:
                    self.request_times.popleft()
                
                # Wait if at limit
                if len(self.request_times) >= 60:
                    sleep_time = 60 - (now - self.request_times[0])
                    if sleep_time > 0:
                        time.sleep(sleep_time)
                
                self.request_times.append(time.time())
            
            try:
                response = requests.post(
                    f"{self.base_url}/chat/completions",
                    headers={
                        "Authorization": f"Bearer {self.api_key}",
                        "Content-Type": "application/json"
                    },
                    json=payload,
                    timeout=30
                )
                
                if response.status_code == 429:
                    wait_time = 2 ** attempt  # Exponential backoff: 1, 2, 4, 8, 16 seconds
                    print(f"Rate limited. Waiting {wait_time}s before retry...")
                    time.sleep(wait_time)
                    continue
                
                response.raise_for_status()
                return response.json()
                
            except requests.exceptions.RequestException as e:
                if attempt == self.max_retries - 1:
                    raise
                time.sleep(2 ** attempt)
        
        raise Exception("Max retries exceeded")

Error 3: Connection Timeout - Request Timeout

requests.exceptions.ConnectTimeout: HTTPSConnectionPool(host='api.holysheep.ai', 
port=443): Max retries exceeded with url: /v1/chat/completions

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_resilient_session() -> requests.Session:
    """Create session with automatic retry and timeout handling."""
    session = requests.Session()
    
    # Configure retry strategy
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,  # Wait 1s, 2s, 4s between retries
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["POST"]
    )
    
    # Mount adapter with retry logic
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    
    return session

def robust_api_call(messages: list, api_key: str) -> dict:
    """
    Make API call with comprehensive error handling.
    Handles timeouts, retries, and connection errors gracefully.
    """
    session = create_resilient_session()
    
    payload = {
        "model": "deepseek-v3",
        "messages": messages,
        "max_tokens": 500
    }
    
    try:
        response = session.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json"
            },
            json=payload,
            timeout=(10, 45)  # (connect_timeout, read_timeout)
        )
        
        response.raise_for_status()
        return response.json()
        
    except requests.exceptions.Timeout:
        # Try with shorter max_tokens if timeout
        payload["max_tokens"] = 200
        response = session.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json"
            },
            json=payload,
            timeout=(5, 30)
        )
        response.raise_for_status()
        return response.json()
        
    except requests.exceptions.ConnectionError as e:
        # Check network connectivity
        import socket
        try:
            socket.create_connection(("api.holysheep.ai", 443), timeout=5)
        except OSError:
            raise Exception("Network connectivity issue. Check your internet connection.")
        raise Exception(f"Connection failed: {str(e)}")

Error 4: 400 Bad Request - Context Length Exceeded

requests.exceptions.HTTPError: 400 Client Error: BAD REQUEST
{"error": {"message": "maximum context length exceeded", "type": "invalid_request_error"}}

def truncate_context(context: str, max_tokens: int = 7000, 
                      model: str = "deepseek-v3") -> str:
    """
    Truncate context to fit within model's context window.
    Reserves tokens for response and user message.
    """
    # Model context limits (approximate)
    context_limits = {
        "deepseek-v3": 32000,
        "gpt-4": 8192,
        "claude-3": 200000
    }
    
    limit = context_limits.get(model, 16000)
    reserved = 500  # Tokens for response
    available = limit - max_tokens - reserved
    
    # Convert approximate characters to tokens (rough: 4 chars = 1 token)
    max_chars = available * 4
    
    if len(context) > max_chars:
        return context[:max_chars] + "... [truncated]"
    
    return context

def smart_message_builder(system: str, history: list, 
                          new_message: str, max_response: int = 500) -> list:
    """
    Build messages that fit within context window by truncating history.
    """
    messages = [{"role": "system", "content": system}]
    
    # Add conversation history (newest first, truncated)
    for msg in reversed(history[-10:]):  # Last 10 messages
        truncated_content = truncate_context(
            msg["content"], 
            max_tokens=300
        )
        messages.insert(1, {
            "role": msg["role"],
            "content": truncated_content
        })
    
    # Ensure total fits
    total_estimate = sum(len(m["content"]) for m in messages) // 4 + len(new_message) // 4
    
    if total_estimate > 15000:  # Safety margin
        # Remove oldest messages until it fits
        while len(messages) > 2 and total_estimate > 12000:
            messages.pop(1)
            total_estimate = sum(len(m["content"]) for m in messages) // 4
    
    messages.append({"role": "user", "content": new_message})
    
    return messages

Quick Reference: Token Savings Checklist

Conclusion