cost-aware-llm-pipeline

Cost-Aware LLM Pipeline

Patterns for controlling LLM API costs while maintaining quality. Combine Model Routing, Budget Tracking, Retry Logic, and Prompt Caching into a reusable pipeline.

When to Enable

• Building applications that call LLM APIs (Claude, GPT, etc.)
• Processing batch items with varying complexity
• Need to keep API spending within budget
• Optimize costs without sacrificing quality on complex tasks

Core Concepts

1. Model Routing Based on Task Complexity

Automatically select cheaper models for simple tasks, reserving expensive models for complex tasks.

MODEL_SONNET = "claude-sonnet-4-6"
MODEL_HAIKU = "claude-haiku-4-5-20251001"

_SONNET_TEXT_THRESHOLD = 10_000  # 字符数阈值
_SONNET_ITEM_THRESHOLD = 30     # 项目数阈值

def select_model(
    text_length: int,
    item_count: int,
    force_model: str | None = None,
) -> str:
    """根据任务复杂度选择模型。"""
    if force_model is not None:
        return force_model
    if text_length >= _SONNET_TEXT_THRESHOLD or item_count >= _SONNET_ITEM_THRESHOLD:
        return MODEL_SONNET  # 复杂任务
    return MODEL_HAIKU  # 简单任务 (便宜 3-4 倍)

2. Immutable Cost Tracking

Track cumulative spending using Frozen Dataclasses. Each API call returns a new tracker—never modify the original state.

from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class CostRecord:
    model: str
    input_tokens: int
    output_tokens: int
    cost_usd: float

@dataclass(frozen=True, slots=True)
class CostTracker:
    budget_limit: float = 1.00
    records: tuple[CostRecord, ...] = ()

    def add(self, record: CostRecord) -> "CostTracker":
        """返回添加了新记录的新跟踪器 (绝不修改自身状态)。"""
        return CostTracker(
            budget_limit=self.budget_limit,
            records=(*self.records, record),
        )

    @property
    def total_cost(self) -> float:
        return sum(r.cost_usd for r in self.records)

    @property
    def over_budget(self) -> bool:
        return self.total_cost > self.budget_limit

3. Narrow Retry Logic

Retry only on Transient Errors. Fail Fast on authentication or bad requests.

from anthropic import (
    APIConnectionError,
    InternalServerError,
    RateLimitError,
)

_RETRYABLE_ERRORS = (APIConnectionError, RateLimitError, InternalServerError)
_MAX_RETRIES = 3

def call_with_retry(func, *, max_retries: int = _MAX_RETRIES):
    """仅在瞬时错误时重试，其他错误立即报错。"""
    for attempt in range(max_retries):
        try:
            return func()
        except _RETRYABLE_ERRORS:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)  # 指数退避 (Exponential backoff)
    # AuthenticationError, BadRequestError 等 -> 立即抛出异常

4. Prompt Caching

Cache longer System Prompts to avoid resending them on every request.

messages = [
    {
        "role": "user",
        "content": [
            {
                "type": "text",
                "text": system_prompt,
                "cache_control": {"type": "ephemeral"},  # 缓存此内容
            },
            {
                "type": "text",
                "text": user_input,  # 变量部分
            },
        ],
    }
]

Combined Usage

Combine all four techniques in a single pipeline function:

def process(text: str, config: Config, tracker: CostTracker) -> tuple[Result, CostTracker]:
    # 1. 路由模型
    model = select_model(len(text), estimated_items, config.force_model)

    # 2. 检查预算
    if tracker.over_budget:
        raise BudgetExceededError(tracker.total_cost, tracker.budget_limit)

    # 3. 带重试与缓存的调用
    response = call_with_retry(lambda: client.messages.create(
        model=model,
        messages=build_cached_messages(system_prompt, text),
    ))

    # 4. 跟踪成本 (不可变模式)
    record = CostRecord(model=model, input_tokens=..., output_tokens=..., cost_usd=...)
    tracker = tracker.add(record)

    return parse_result(response), tracker

Price Reference (2025-2026)

Model	Input ($/1M tokens)	Output ($/1M tokens)	Relative Cost
Haiku 4.5	$0.80	$4.00	1x
Sonnet 4.6	$3.00	$15.00	~4x
Opus 4.5	$15.00	$75.00	~19x

Best Practices

• Start with the cheapest model, and route to expensive models only when a complexity threshold is reached.
• Set clear budget limits before batch processing—prefer failing early over overspending.
• Log model selection decisionsso thresholds can be adjusted based on real data.
• Use prompt caching for system prompts exceeding 1024 tokens—this saves both cost and reduces latency.
• Never retry on authentication or validation errors—retry only transient failures (network, rate limits, server errors).

Anti-Patterns to Avoid

• Using the most expensive model for all requests regardless of complexity.
• Retrying all errors (wasting budget on permanent failures).
• Modifying cost tracking state (increases debugging and auditing difficulty).
• Hardcoding model names throughout the codebase (use constants or configuration instead).
• Ignoring prompt caching for repeated system prompts.

Use Cases

• Any application that calls Claude, OpenAI, or similar LLM APIs.
• Batch processing pipelines where costs accumulate rapidly.
• Multi-model architectures that require intelligent routing.
• Production systems that need budget safety Guardrails.