ADR-011: Cost Tracking and Prediction System

Status: Proposed Date: 2024-12-12 Deciders: LLM Council Technical Story: Design comprehensive cost tracking, prediction, and budget controls for the council system

Context and Problem Statement

The council system uses multiple LLM calls per query (3-10 models × 3 stages), making costs unpredictable. Users need:

1. Pre-submission estimates: Know costs before running a query
2. Post-submission breakdowns: Understand where money went
3. Budget controls: Prevent surprise charges
4. Optimization guidance: Reduce costs without sacrificing quality

Current State

• Token counts tracked per stage (stage1, stage1.5, stage2, stage3)
• OpenRouter provides usage: {prompt_tokens, completion_tokens, total_tokens}
• No cost calculation or prediction
• No budget enforcement

Key Challenge: Peer Review Cost Growth

Stage 2 (peer review) input size grows as O(N × M) where: - N = number of models - M = average Stage 1 response length

With 5 models generating 500 tokens each, each reviewer sees ~2500+ input tokens.

Decision Drivers

• Accuracy: Cost predictions should be within 20% of actuals
• Reliability: Never fail a query due to pricing lookup issues
• User Safety: Prevent runaway costs before they happen
• Transparency: Users understand exactly how costs are calculated
• Extensibility: Support future providers beyond OpenRouter

Design Questions & Decisions

1. Pricing Data Source

Question: How to get and cache model pricing data?

Options Considered: | Approach | Pros | Cons | |----------|------|------| | Hardcoded only | Simple, no dependencies | Stale quickly, maintenance burden | | Dynamic API only | Always current | API failures break pricing | | Hybrid (chosen) | Resilient, current when available | Slightly more complex |

Decision: Hybrid approach with cached dynamic fetching + hardcoded fallback

class PricingService:
    def __init__(self):
        self.cache_ttl = 3600  # 1 hour
        self.fallback_prices = {
            # Per million tokens
            "openai/gpt-4o": {"input": 2.50, "output": 10.00},
            "anthropic/claude-sonnet-4-20250514": {"input": 3.00, "output": 15.00},
            "google/gemini-1.5-pro": {"input": 1.25, "output": 5.00},
        }
        self._cache = {}

    async def get_price(self, model_id: str) -> dict:
        if self._is_cache_valid(model_id):
            return self._cache[model_id]

        try:
            price = await self._fetch_from_openrouter(model_id)
            self._update_cache(model_id, price)
            return price
        except APIError:
            return self.fallback_prices.get(model_id)

Rationale: - OpenRouter prices change frequently (new models, price drops) - API failures shouldn't break cost tracking - Cached prices provide sub-millisecond lookups during query execution - Fallback ensures graceful degradation

2. Token Estimation for Cost Prediction

Question: How to estimate completion tokens before a query runs?

Options Considered: | Approach | Pros | Cons | |----------|------|------| | Simple multiplier | Easy | Ignores model/stage variance | | Trained ML predictor | Accurate | Complex, data-hungry | | Historical percentiles (chosen) | Accurate, simple | Needs bootstrap data |

Decision: Model-specific historical percentiles with stage multipliers

class CostPredictor:
    def __init__(self):
        # Per (model, stage) completion token statistics
        self.completion_stats = {
            "openai/gpt-4o": {
                "initial_response": {"p50": 450, "p75": 680, "p95": 1200},
                "peer_review": {"p50": 580, "p75": 850, "p95": 1400},
                "synthesis": {"p50": 400, "p75": 600, "p95": 1000},
            },
            # ... other models
        }

    def estimate_query_cost(
        self,
        prompt_tokens: int,
        models: list[str],
        confidence: str = "p75"  # p50, p75, p95
    ) -> CostEstimate:
        """
        Estimate total cost for a council query.

        Returns low/expected/high range based on historical data.
        """
        estimates = []

        for stage in ["initial_response", "peer_review", "synthesis"]:
            stage_models = self._models_for_stage(stage, models)
            stage_prompt = self._estimate_stage_prompt(stage, prompt_tokens)

            for model in stage_models:
                completion = self.completion_stats[model][stage][confidence]
                price = self.pricing.get_price(model)

                cost = (
                    (stage_prompt * price["input"]) +
                    (completion * price["output"])
                ) / 1_000_000

                estimates.append(StageCostEstimate(stage, model, cost))

        total = sum(e.cost for e in estimates)
        return CostEstimate(
            low=total * 0.6,      # p25 equivalent
            expected=total,       # chosen confidence
            high=total * 1.5,     # p95 buffer
            breakdown=estimates
        )

    def update_stats(self, model: str, stage: str, actual_tokens: int):
        """Update statistics after each completion (exponential moving average)."""
        alpha = 0.1
        current = self.completion_stats[model][stage]["p50"]
        self.completion_stats[model][stage]["p50"] = (
            alpha * actual_tokens + (1 - alpha) * current
        )

Rationale: - Different models have different verbosity (Claude verbose, GPT concise) - Different stages have different output patterns (reviews longer than synthesis) - Percentiles let users choose risk tolerance - EMA updates improve accuracy over time without complex ML

3. Budget Enforcement Strategy

Question: How to handle spending limits?

Options Considered: | Approach | Pros | Cons | |----------|------|------| | Hard reject | Safe | Frustrating UX, estimates uncertain | | Warn only | User-friendly | Can exceed budget | | Abort mid-query | Real-time control | Wastes spent tokens | | Tiered (chosen) | Flexible, safe | More complex |

Decision: Tiered enforcement with configurable strictness

class BudgetEnforcer:
    class Mode(Enum):
        STRICT = "strict"      # Reject if p75 estimate exceeds
        BALANCED = "balanced"  # Reject if p50 exceeds, warn if p75 exceeds
        PERMISSIVE = "permissive"  # Warn only, never reject upfront

    def pre_query_check(
        self,
        estimate: CostEstimate,
        budget_remaining: float,
        mode: Mode = Mode.BALANCED
    ) -> BudgetDecision:
        if mode == Mode.STRICT:
            if estimate.high > budget_remaining:
                return BudgetDecision.REJECT, self._suggest_cheaper(estimate)

        elif mode == Mode.BALANCED:
            if estimate.expected > budget_remaining:
                return BudgetDecision.REJECT, "Likely to exceed budget"
            elif estimate.high > budget_remaining:
                return BudgetDecision.WARN, f"May exceed (${estimate.expected:.2f} expected, up to ${estimate.high:.2f})"

        elif mode == Mode.PERMISSIVE:
            if estimate.expected > budget_remaining:
                return BudgetDecision.WARN, "Likely to exceed budget"

        return BudgetDecision.ALLOW, None

    def mid_query_check(
        self,
        spent_so_far: float,
        remaining_stages: list[str],
        budget_remaining: float
    ) -> BudgetDecision:
        """Check between stages - abort gracefully if over budget."""
        if spent_so_far > budget_remaining:
            return BudgetDecision.ABORT_GRACEFULLY, "Budget exceeded. Returning partial results."
        return BudgetDecision.CONTINUE, None

Critical UX considerations: - Never abort mid-completion: Wastes tokens, creates broken responses - Check between stages: Can return partial results gracefully - Suggest alternatives: "Removing GPT-4 reduces estimate to $0.28"

4. Cost Attribution in Peer Review

Question: In Stage 2, each model reviews all others. How to attribute costs?

Options Considered: | Approach | Pros | Cons | |----------|------|------| | Split among reviewed | Intuitive | Masks reviewer verbosity | | Reviewer only (chosen) | Actionable, causal | Less intuitive | | Both views | Complete | Complex |

Decision: Primary attribution to reviewer, with cross-reference tagging

@dataclass
class CostAttribution:
    stage: str
    model: str
    tokens_in: int
    tokens_out: int
    cost: float
    reviewing_models: list[str] | None = None  # For peer review stage

    @property
    def cost_per_review_target(self) -> float | None:
        """Secondary view: split cost among reviewed responses."""
        if self.reviewing_models:
            return self.cost / len(self.reviewing_models)
        return None

Rationale: - Causality: The reviewer's verbosity determines cost - Actionability: "Claude's reviews cost 2x GPT's" → adjust reviewer selection - Simplicity: Primary attribution is straightforward - Flexibility: Can still compute "cost to be reviewed" analytically

Example output:

{
  "model_costs": {
    "openai/gpt-4o": {
      "direct_cost": 0.0234,
      "cost_as_review_target": 0.0156,
      "breakdown": {
        "initial_response": 0.0089,
        "peer_review": 0.0102,
        "synthesis": 0.0043
      }
    }
  },
  "stage_costs": {
    "initial_response": 0.0312,
    "peer_review": 0.0847,
    "synthesis": 0.0098
  }
}

5. Where Cost Tracking Lives

Question: Core library, optional module, or cloud-only?

Decision: Core library with pluggable interfaces

┌─────────────────────────────────────────────────────────────────┐
│                     OPEN SOURCE CORE                            │
├─────────────────────────────────────────────────────────────────┤
│  llm_council/                                                   │
│  └── cost_tracking/                                             │
│      ├── __init__.py                                            │
│      ├── types.py          # CostEstimate, Attribution DTOs     │
│      ├── calculator.py     # Pure token→cost math               │
│      ├── predictor.py      # Estimation algorithms              │
│      ├── enforcer.py       # Budget enforcement                 │
│      └── interfaces.py     # Abstract PricingProvider           │
│                                                                 │
│  llm_council/cost_tracking/backends/                            │
│  ├── memory_storage.py     # In-memory (default)                │
│  ├── sqlite_storage.py     # Local persistence                  │
│  └── static_pricing.py     # Bundled fallback prices            │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│                    PAID CLOUD TIER                              │
├─────────────────────────────────────────────────────────────────┤
│  • PostgreSQL storage backend                                   │
│  • Real-time OpenRouter pricing sync                            │
│  • Cross-organization prediction models                         │
│  • Budget alerts, dashboards, admin controls                    │
│  • Cost anomaly detection                                       │
│  • Multi-tenant quotas (org/team/user)                          │
└─────────────────────────────────────────────────────────────────┘

Feature Matrix: | Feature | Open Source | Cloud | |---------|-------------|-------| | Per-query cost calculation | Yes | Yes | | Cost estimation (local history) | Yes | Yes (global models) | | Budget warnings | Yes | Yes | | Budget enforcement | Yes (local) | Yes (org-wide) | | Historical dashboards | No | Yes | | Cross-user prediction models | No | Yes | | Cost anomaly alerts | No | Yes |

Rationale: - Safety default: OSS users need visibility into spend - Transparency: Users can see exactly how costs are calculated - Extensibility: Cloud tier adds value without restricting core functionality - Community contribution: Open cost logic means community can improve it

Implementation

API Integration

# In run_full_council()
async def run_full_council(
    user_query: str,
    cost_tracker: CostTracker | None = None,
    budget_limit: float | None = None,
) -> CouncilResult:
    cost_tracker = cost_tracker or DefaultCostTracker()

    # Pre-flight cost estimation
    estimate = cost_tracker.estimate(user_query, COUNCIL_MODELS)

    # Budget check
    if budget_limit:
        decision, msg = cost_tracker.enforcer.pre_query_check(
            estimate, budget_limit
        )
        if decision == BudgetDecision.REJECT:
            raise BudgetExceededError(msg, estimate=estimate)

    # ... run stages, tracking actual costs ...

    return CouncilResult(
        answer=synthesis,
        cost_summary=cost_tracker.get_summary()
    )

Configuration

# config.py additions
DEFAULT_BUDGET_MODE = "balanced"  # strict, balanced, permissive
DEFAULT_COST_TRACKING = True
DEFAULT_PRICING_CACHE_TTL = 3600  # seconds

Response Schema

@dataclass
class CostSummary:
    total_cost: float
    currency: str = "USD"
    by_stage: dict[str, float]
    by_model: dict[str, float]
    estimate_accuracy: float  # actual / estimated
    tokens: TokenUsage

@dataclass
class TokenUsage:
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    by_stage: dict[str, dict]

Consequences

Positive

• Users can predict costs before running queries
• Budget controls prevent surprise charges
• Cost breakdowns enable optimization
• OSS users get full cost visibility

Negative

• Adds complexity to core library
• Pricing data maintenance required
• Estimates are inherently uncertain

Risks

• Stale prices: Mitigated by dynamic fetching + short TTL
• Inaccurate estimates: Mitigated by percentile ranges
• Budget too strict: Mitigated by configurable modes

Migration Path

1. Phase 1: Add cost calculation (post-query only)
2. Phase 2: Add cost estimation (pre-query)
3. Phase 3: Add budget warnings
4. Phase 4: Add budget enforcement (opt-in)
5. Phase 5: Add cost dashboards in cloud tier

References

• ADR-009: HTTP API and Open Core Boundary
• OpenRouter Pricing API
• Token Estimation Best Practices