Dynamic complexity and cost adaptation for analyzer tasks

[rbpotter-aws]

Description of the Need

As a BADGERS user, I want the system to intelligently manage model selection and token usage based on task complexity and cost constraints, So that I can control per-page analysis costs while maintaining quality where it matters.

Context

Different documents have wildly different complexity levels. A simple typed form doesn't need the same model or token budget as a dense handwritten ledger with marginalia. Currently every page gets the same model and prompt treatment regardless of complexity, leading to unnecessary cost on simple pages and potentially insufficient analysis on complex ones. Customers want cost predictability (e.g. ~$0.60/page average) with quality that scales to the task.

Feature Areas

1. Automatic model routing based on task complexity

• Implement a lightweight pre-classification step that assesses page complexity (e.g. simple/medium/complex)
• Route simple pages to cheaper/faster models, complex pages to more capable models
• Classification could factor in: image resolution, detected text density, document type, handwriting presence
• Routing rules configurable per analyzer

2. Token Budgeting per Analyzer

• Define a target cost-per-page for each analyzer (e.g. $0.60 average)
• System tracks running average cost and adjusts model/prompt selection to stay within budget
• Support hard caps (never exceed $X per page) and soft targets (average over N pages)
• Surface alerts when an analyzer is trending over budget

3. Prompt Cost Analysis and Optimization Recommendations

• Break down cost per invocation: input tokens, image tokens, output tokens
• Identify which prompt components contribute most to cost
• Recommend optimizations: prompt trimming, fewer examples, output format changes, model downsizing
• Provide a cost analysis report per analyzer (on-demand or periodic)

Acceptance Criteria

• A complexity classifier can categorize pages before full analysis
• Model routing rules are configurable per analyzer with sensible defaults
• Per-analyzer cost targets can be defined in configuration
• The system tracks and reports actual cost per page against targets
• A cost breakdown is available per invocation showing input/image/output token costs
• The system can generate optimization recommendations for a given analyzer's prompt configuration
• Existing deployments without cost config continue to work unchanged

Technical Considerations

• Complexity classification should be cheap itself — could use a small model, heuristics, or image analysis
• Token counting needs to happen pre- and post-invocation for accurate tracking
• Bedrock pricing varies by model and region; need a pricing reference table (or API)
• Image token costs follow different pricing models than text tokens
• This feature depends on the pluggable LLM backends feature for model routing to work across providers

Out of Scope (future consideration):

• Real-time cost dashboards
• Automatic prompt rewriting/compression
• Cross-analyzer budget pooling

[rbpotter-aws]

Response to Issue #12: Dynamic Complexity and Cost Adaptation

TL;DR - Immediate Value Path

The cost optimization features described in issue #12 can be implemented immediately using the existing Bedrock integration.

---

Architecture: What Already Exists

BADGERS analyzer manifests already contain LLM configuration:

{
  "analyzer_name": "clinical_form_extraction",
  "inference": {
    "provider": "bedrock",
    "model_id": "us.anthropic.claude-3-5-sonnet-20241022-v2:0",
    "temperature": 0.1,
    "max_tokens": 8192
  }
}

The Bedrock Converse API already returns token counts in responses:

response = bedrock_client.converse(...)
# response['usage'] includes inputTokens, outputTokens

We have everything we need to implement complexity-based routing and cost governance today.

---

Proposed Implementation: Minimal Complexity-Based Routing

1. Extend Analyzer Manifest with Routing Rules

Add optional complexity-aware routing to existing configs:

{
  "analyzer_name": "clinical_form_extraction",
  "inference": {
    "provider": "bedrock",
    "model_id": "us.anthropic.claude-3-5-sonnet-20241022-v2:0",
    "temperature": 0.1,
    "max_tokens": 8192
  },
  "complexity_routing": {
    "enabled": false,
    "classifier": {
      "strategy": "hybrid",
      "lightweight_model_id": "us.amazon.nova-micro-v1:0"
    },
    "rules": [
      {
        "complexity": "simple",
        "model_id": "us.amazon.nova-lite-v1:0",
        "max_tokens": 2048
      },
      {
        "complexity": "moderate",
        "model_id": "us.anthropic.claude-3-5-haiku-20241022-v1:0",
        "max_tokens": 4096
      },
      {
        "complexity": "complex",
        "model_id": "us.anthropic.claude-3-5-sonnet-20241022-v2:0",
        "max_tokens": 8192
      }
    ]
  },
  "budget": {
    "target_cost_per_page_usd": 0.60,
    "hard_cap_per_page_usd": 1.50,
    "tracking_window_pages": 100,
    "breach_action": "alert"
  }
}

Backward compatibility: When complexity_routing.enabled is false or omitted, the system uses the default inference config exactly as before.

2. Implementation Flow (No Abstraction Layers Required)

async def process_document(image: bytes, analyzer_config: dict):
    # Step 1: Determine which model config to use
    if analyzer_config.get("complexity_routing", {}).get("enabled"):
        # Assess complexity
        complexity = await assess_complexity(
            image=image,
            strategy=analyzer_config["complexity_routing"]["classifier"]["strategy"],
            lightweight_model=analyzer_config["complexity_routing"]["classifier"]["lightweight_model_id"]
        )

        # Select routing rule
        model_config = next(
            rule for rule in analyzer_config["complexity_routing"]["rules"]
            if rule["complexity"] == complexity.level
        )

        # Check budget (pre-flight)
        estimated_cost = estimate_cost(complexity.estimated_tokens, model_config["model_id"])
        if estimated_cost > analyzer_config["budget"]["hard_cap_per_page_usd"]:
            # Apply breach action: downgrade, alert, or block
            model_config = handle_budget_breach(analyzer_config, model_config, estimated_cost)
    else:
        # Use default inference config
        model_config = analyzer_config["inference"]
        complexity = None

    # Step 2: Execute inference using existing Bedrock client
    response = bedrock_client.converse(
        modelId=model_config["model_id"],
        messages=[...],
        inferenceConfig={
            "temperature": model_config.get("temperature", 0.1),
            "maxTokens": model_config["max_tokens"]
        }
    )

    # Step 3: Track actual cost
    actual_cost = calculate_cost(
        model_id=model_config["model_id"],
        input_tokens=response['usage']['inputTokens'],
        output_tokens=response['usage']['outputTokens']
    )

    await cost_tracker.record_invocation(
        analyzer_name=analyzer_config["analyzer_name"],
        complexity_level=complexity.level if complexity else "default",
        model_id=model_config["model_id"],
        cost_usd=actual_cost,
        tokens=response['usage']
    )

    # Step 4: Check budget (post-flight) and emit alerts if needed
    budget_status = await cost_tracker.get_budget_status(
        analyzer_name=analyzer_config["analyzer_name"],
        window_pages=analyzer_config["budget"]["tracking_window_pages"]
    )

    if budget_status.actual_avg_cost > analyzer_config["budget"]["target_cost_per_page_usd"]:
        emit_budget_alert(analyzer_config["analyzer_name"], budget_status)

    return extract_structured_output(response)

Key insight: This is just conditional logic selecting which model_id to pass to the existing Bedrock client. No abstraction layers, no provider interfaces—just practical routing.

---

Complexity Classification Strategies

Strategy 1: Heuristic-Based (Cheapest, Fastest)

Pure image analysis without LLM inference:

@dataclass
class ComplexitySignals:
    image_size_kb: int
    estimated_text_density: float  # From preliminary OCR or pixel analysis
    image_quality_score: float      # Based on resolution, contrast
    element_count_estimate: int     # Heuristic edge detection for tables/forms

async def assess_complexity_heuristic(image: bytes) -> ComplexityAssessment:
    signals = analyze_image_characteristics(image)

    # Simple decision tree
    if signals.image_size_kb < 100 and signals.image_quality_score > 0.8:
        level = ComplexityLevel.SIMPLE
    elif signals.estimated_text_density > 500 or signals.image_quality_score < 0.5:
        level = ComplexityLevel.COMPLEX
    else:
        level = ComplexityLevel.MODERATE

    return ComplexityAssessment(
        level=level,
        confidence=calculate_confidence(signals),
        signals=signals,
        estimated_tokens=estimate_tokens_from_image(image),
        reasoning=f"Image {signals.image_size_kb}KB, quality {signals.image_quality_score:.2f}"
    )

Cost: ~$0.001-0.005/page (if using Textract for OCR preview)
Latency: 100-500ms
Accuracy: 70-80%

Strategy 2: Lightweight Model (More Accurate)

Use Nova Micro or Haiku to classify complexity:

async def assess_complexity_model(image: bytes, model_id: str) -> ComplexityAssessment:
    prompt = """Analyze this document image and assess its extraction complexity.

Consider:
- Text clarity (typed vs handwritten)
- Layout complexity (simple form vs dense multi-column)
- Image quality (resolution, contrast, noise)
- Element density (tables, signatures, stamps)

Respond with JSON:
{
  "complexity": "simple|moderate|complex",
  "confidence": 0.0-1.0,
  "reasoning": "brief explanation"
}"""

    response = bedrock_client.converse(
        modelId=model_id,  # e.g., "us.amazon.nova-micro-v1:0"
        messages=[{
            "role": "user",
            "content": [
                {"image": {"format": "png", "source": {"bytes": image}}},
                {"text": prompt}
            ]
        }],
        inferenceConfig={"maxTokens": 200, "temperature": 0.0}
    )

    result = json.loads(response['output']['message']['content'][0]['text'])

    return ComplexityAssessment(
        level=ComplexityLevel(result["complexity"]),
        confidence=result["confidence"],
        signals=None,  # Model-based, no explicit signals
        estimated_tokens=estimate_tokens_from_image(image),
        reasoning=result["reasoning"]
    )

Cost: ~$0.01-0.03/page
Latency: 500-1500ms
Accuracy: 85-95%

Strategy 3: Hybrid (Recommended Default)

Use heuristics first, fall back to model when uncertain:

async def assess_complexity_hybrid(
    image: bytes,
    lightweight_model_id: str
) -> ComplexityAssessment:
    # Try heuristics first
    heuristic_result = await assess_complexity_heuristic(image)

    # If confident, use heuristic result
    if heuristic_result.confidence > 0.75:
        return heuristic_result

    # Otherwise, use lightweight model
    return await assess_complexity_model(image, lightweight_model_id)

Cost: ~$0.002-0.01/page (amortized)
Latency: 100-1500ms (variable)
Accuracy: 80-90%

---

Cost Tracking Implementation

Pricing Table (Static JSON)

Start with a simple static pricing file for Bedrock models:

{
  "pricing": {
    "us.anthropic.claude-3-5-sonnet-20241022-v2:0": {
      "input_per_1k": 0.003,
      "output_per_1k": 0.015
    },
    "us.anthropic.claude-3-5-haiku-20241022-v1:0": {
      "input_per_1k": 0.001,
      "output_per_1k": 0.005
    },
    "us.amazon.nova-lite-v1:0": {
      "input_per_1k": 0.00006,
      "output_per_1k": 0.00024
    },
    "us.amazon.nova-micro-v1:0": {
      "input_per_1k": 0.000035,
      "output_per_1k": 0.00014
    }
  }
}

Future enhancement: Fetch from AWS Pricing API, but static is fine to start.

Cost Calculation

class CostCalculator:
    def __init__(self, pricing_table: dict):
        self.pricing = pricing_table["pricing"]

    def calculate_cost(
        self,
        model_id: str,
        input_tokens: int,
        output_tokens: int
    ) -> float:
        """Calculate cost in USD for a single invocation."""
        model_pricing = self.pricing[model_id]

        input_cost = (input_tokens / 1000) * model_pricing["input_per_1k"]
        output_cost = (output_tokens / 1000) * model_pricing["output_per_1k"]

        return input_cost + output_cost

    def estimate_cost(
        self,
        model_id: str,
        estimated_input_tokens: int,
        estimated_output_tokens: int
    ) -> float:
        """Estimate cost before invocation (for budget pre-flight checks)."""
        return self.calculate_cost(model_id, estimated_input_tokens, estimated_output_tokens)

Cost Tracker (DynamoDB or In-Memory)

@dataclass
class CostRecord:
    analyzer_name: str
    document_id: str
    timestamp: datetime
    complexity_level: Optional[str]
    model_id: str
    input_tokens: int
    output_tokens: int
    cost_usd: float

class CostTracker:
    """Tracks per-analyzer costs with rolling window support."""

    async def record_invocation(
        self,
        analyzer_name: str,
        complexity_level: Optional[str],
        model_id: str,
        cost_usd: float,
        tokens: dict
    ):
        """Record a single invocation."""
        record = CostRecord(
            analyzer_name=analyzer_name,
            document_id=generate_document_id(),
            timestamp=datetime.utcnow(),
            complexity_level=complexity_level,
            model_id=model_id,
            input_tokens=tokens['inputTokens'],
            output_tokens=tokens['outputTokens'],
            cost_usd=cost_usd
        )

        # Store in DynamoDB for persistent tracking
        await self.storage.put_item(record)

        # Emit CloudWatch metrics
        cloudwatch.put_metric_data(
            Namespace='BADGERS/Cost',
            MetricData=[
                {
                    'MetricName': 'CostPerPage',
                    'Value': cost_usd,
                    'Unit': 'None',
                    'Dimensions': [
                        {'Name': 'AnalyzerName', 'Value': analyzer_name},
                        {'Name': 'ModelId', 'Value': model_id}
                    ]
                },
                {
                    'MetricName': 'TokensUsed',
                    'Value': tokens['inputTokens'] + tokens['outputTokens'],
                    'Unit': 'Count',
                    'Dimensions': [
                        {'Name': 'AnalyzerName', 'Value': analyzer_name}
                    ]
                }
            ]
        )

    async def get_budget_status(
        self,
        analyzer_name: str,
        window_pages: int
    ) -> BudgetStatus:
        """Get rolling average cost for an analyzer."""
        recent_records = await self.storage.query_recent(
            analyzer_name=analyzer_name,
            limit=window_pages
        )

        if not recent_records:
            return BudgetStatus(
                analyzer_name=analyzer_name,
                actual_avg_cost_per_page=0.0,
                pages_processed=0,
                total_cost_usd=0.0
            )

        total_cost = sum(r.cost_usd for r in recent_records)
        avg_cost = total_cost / len(recent_records)

        return BudgetStatus(
            analyzer_name=analyzer_name,
            actual_avg_cost_per_page=avg_cost,
            pages_processed=len(recent_records),
            total_cost_usd=total_cost
        )

---

Budget Governance

Pre-Flight Check (Before Inference)

async def check_budget_before_inference(
    analyzer_config: dict,
    estimated_cost: float,
    complexity: ComplexityAssessment
) -> dict:
    """
    Check if estimated cost exceeds hard cap.
    Apply breach action if needed.
    """
    hard_cap = analyzer_config["budget"]["hard_cap_per_page_usd"]

    if estimated_cost <= hard_cap:
        return None  # No action needed

    breach_action = analyzer_config["budget"]["breach_action"]

    if breach_action == "alert":
        logger.warning(
            f"Estimated cost ${estimated_cost:.4f} exceeds hard cap ${hard_cap:.4f} "
            f"for {analyzer_config['analyzer_name']}, but proceeding per breach_action=alert"
        )
        return None

    elif breach_action == "downgrade":
        # Find cheapest model in routing rules
        cheapest_rule = min(
            analyzer_config["complexity_routing"]["rules"],
            key=lambda r: estimate_model_cost(r["model_id"], complexity.estimated_tokens)
        )
        logger.warning(
            f"Downgrading to {cheapest_rule['model_id']} due to budget cap"
        )
        return cheapest_rule

    elif breach_action == "block":
        raise BudgetExceededError(
            f"Estimated cost ${estimated_cost:.4f} exceeds hard cap ${hard_cap:.4f}"
        )

    return None

Post-Flight Check (After Inference)

async def check_budget_after_inference(
    analyzer_config: dict,
    cost_tracker: CostTracker
):
    """
    Check rolling average cost against target.
    Emit alerts if trending over budget.
    """
    budget_status = await cost_tracker.get_budget_status(
        analyzer_name=analyzer_config["analyzer_name"],
        window_pages=analyzer_config["budget"]["tracking_window_pages"]
    )

    target = analyzer_config["budget"]["target_cost_per_page_usd"]
    actual = budget_status.actual_avg_cost_per_page

    if actual > target:
        margin_percent = ((actual - target) / target) * 100

        # Emit CloudWatch alarm
        sns.publish(
            TopicArn=os.environ['BUDGET_ALERT_TOPIC'],
            Subject=f"BADGERS Budget Alert: {analyzer_config['analyzer_name']}",
            Message=f"""
Analyzer: {analyzer_config['analyzer_name']}
Rolling Average Cost: ${actual:.4f}/page (over last {budget_status.pages_processed} pages)
Target Cost: ${target:.4f}/page
Over Budget: {margin_percent:.1f}%

Consider:
- Reviewing complexity classification accuracy
- Adjusting routing rules to use cheaper models
- Analyzing prompt cost breakdown
            """.strip()
        )

        logger.warning(
            f"{analyzer_config['analyzer_name']} over budget: "
            f"${actual:.4f}/page vs target ${target:.4f}/page "
            f"({margin_percent:.1f}% over)"
        )

---

Cost Analysis Tooling

CLI Tool: Analyze Analyzer Costs

async def analyze_analyzer_cost(
    analyzer_name: str,
    sample_documents: list[str],  # S3 paths
    time_range: Optional[tuple[datetime, datetime]] = None
):
    """
    Generate cost breakdown report for an analyzer.

    Usage:
        python -m badgers.tools.cost_analyzer \\
            --analyzer clinical_form_extraction \\
            --samples s3://bucket/samples/*.png \\
            --days 7
    """
    # Fetch historical cost data
    cost_records = await cost_tracker.get_records(
        analyzer_name=analyzer_name,
        time_range=time_range
    )

    # Calculate statistics
    total_cost = sum(r.cost_usd for r in cost_records)
    avg_cost = total_cost / len(cost_records)

    cost_by_complexity = {}
    for record in cost_records:
        level = record.complexity_level or "default"
        cost_by_complexity.setdefault(level, []).append(record.cost_usd)

    cost_by_model = {}
    for record in cost_records:
        cost_by_model.setdefault(record.model_id, []).append(record.cost_usd)

    # Run sample documents to get token breakdown
    sample_results = []
    for doc_path in sample_documents[:10]:  # Analyze up to 10 samples
        image = await load_image_from_s3(doc_path)

        # Run inference with token tracking
        response = await process_document(image, analyzer_config)

        sample_results.append({
            "document": doc_path,
            "input_tokens": response.tokens['inputTokens'],
            "output_tokens": response.tokens['outputTokens'],
            "cost": response.cost_usd,
            "model": response.model_id
        })

    # Generate report
    report = f"""
# Cost Analysis Report: {analyzer_name}

## Historical Performance (Last {len(cost_records)} Pages)

- **Total Cost:** ${total_cost:.2f}
- **Average Cost/Page:** ${avg_cost:.4f}
- **Target Cost/Page:** ${analyzer_config['budget']['target_cost_per_page_usd']:.4f}
- **Status:** {"✓ Under Budget" if avg_cost <= analyzer_config['budget']['target_cost_per_page_usd'] else "⚠ Over Budget"}

## Cost by Complexity Level

{format_cost_breakdown(cost_by_complexity)}

## Cost by Model

{format_cost_breakdown(cost_by_model)}

## Sample Document Analysis

{format_sample_results(sample_results)}

## Recommendations

{generate_recommendations(analyzer_config, cost_records, sample_results)}
    """

    print(report)

    # Optionally save to S3 or CloudWatch Logs
    await save_report(report, analyzer_name)

Example Recommendations

def generate_recommendations(
    analyzer_config: dict,
    cost_records: list[CostRecord],
    sample_results: list[dict]
) -> str:
    recommendations = []

    avg_cost = sum(r.cost_usd for r in cost_records) / len(cost_records)
    target = analyzer_config["budget"]["target_cost_per_page_usd"]

    if avg_cost > target:
        # Analyze where costs are high
        avg_input_tokens = sum(r["input_tokens"] for r in sample_results) / len(sample_results)
        avg_output_tokens = sum(r["output_tokens"] for r in sample_results) / len(sample_results)

        if avg_input_tokens > 10000:
            recommendations.append(
                "**HIGH IMPACT:** Input tokens average {avg_input_tokens:.0f}. "
                "Consider reducing few-shot examples or simplifying system prompt."
            )

        if not analyzer_config.get("complexity_routing", {}).get("enabled"):
            recommendations.append(
                "**HIGH IMPACT:** Enable complexity-based routing to route simple documents "
                "to cheaper models. Estimated savings: 30-50% on simple pages."
            )

        if avg_output_tokens > 4000:
            max_tokens = analyzer_config["inference"]["max_tokens"]
            recommended_max = int(avg_output_tokens * 1.2)  # 20% buffer
            recommendations.append(
                f"**MEDIUM IMPACT:** Output tokens average {avg_output_tokens:.0f} but "
                f"max_tokens is {max_tokens}. Reduce max_tokens to {recommended_max} "
                f"to prevent unnecessary allocation."
            )

    if not recommendations:
        recommendations.append("✓ Cost performance is within target. No immediate optimizations needed.")

    return "\n".join(f"{i+1}. {rec}" for i, rec in enumerate(recommendations))

---

Implementation Phases (Revised for Immediate Value)

Phase 1: Cost Tracking (1-2 weeks)

Goal: Instrument existing pipeline with cost tracking. No functional changes to extraction.

• Implement CostCalculator with static Bedrock pricing table
• Implement CostTracker with DynamoDB backend
• Add token/cost recording to existing inference flow
• Add CloudWatch metrics for per-analyzer cost
• Deliverable: Cost dashboard showing actual cost per page for all analyzers

Phase 2: Complexity Classification (1-2 weeks)

Goal: Add optional complexity assessment without changing routing yet.

• Implement all three complexity classifier strategies (heuristic, model, hybrid)
• Add complexity_routing.classifier config to analyzer manifests (opt-in)
• Run classifier and record complexity assessment alongside extraction results
• Collect data for validation (complexity vs actual cost)
• Deliverable: Complexity assessments logged for sample analyzers; validation data

Phase 3: Model Routing (1 week)

Goal: Enable dynamic model selection based on complexity.

• Implement routing logic: select model config from routing rules based on complexity
• Add complexity_routing.rules to analyzer manifest format
• Update analyzer execution to use selected model config
• Log routing decisions for observability
• Deliverable: Analyzers can route to different Bedrock models based on page complexity

Phase 4: Budget Governance (1 week)

Goal: Enforce cost targets and caps.

• Implement pre-flight budget check (estimated cost vs hard cap)
• Implement post-flight budget check (rolling average vs target)
• Add SNS alerts for budget breaches
• Implement breach actions (alert, downgrade, block)
• Deliverable: Budget enforcement with automated alerts

Phase 5: Cost Analysis Tooling (1 week)

Goal: Provide cost optimization recommendations.

• Implement CLI cost analyzer tool
• Build cost breakdown reports
• Generate recommendations based on actual usage patterns
• Document cost optimization best practices
• Deliverable: On-demand cost analysis for any analyzer

Total implementation time: 6-7 weeks for complete feature set

---

Migration Path: Pluggable Backends as Future Enhancement

When pluggable LLM backends (issue #11) are eventually implemented, the migration is straightforward:

Current implementation:

# Direct Bedrock client usage
response = bedrock_client.converse(
    modelId=selected_model_id,
    messages=messages,
    inferenceConfig=inference_config
)

Future with pluggable backends:

# Abstract provider interface
provider = get_provider(selected_config.provider)  # "bedrock", "sagemaker", "http"
response = await provider.invoke(
    model_id=selected_model_id,
    messages=messages,
    inference_config=inference_config
)

The routing logic, complexity classification, cost tracking, and budget governance remain identical. Only the execution layer changes.

Benefits of this approach:

1. Deliver cost optimization value in 6-7 weeks instead of waiting months
2. Validate complexity classification and routing strategies with real usage
3. Gather data on which models work best for which complexity levels
4. Build cost tracking infrastructure that will work with any future provider

When pluggable backends are needed:

• Multi-provider support (SageMaker, custom endpoints)
• Provider-specific features (streaming, batch inference)
• Testing and local development environments

But for Bedrock-only deployments focused on cost optimization, the direct integration is simpler and faster.

---

Open Questions for Discussion

1. Default Complexity Classifier Strategy

Should the default be:

• Heuristic-only (cheapest, fastest)
• Lightweight model (most accurate)
• Hybrid (balanced, my recommendation)

2. Default Breach Action

When hard cap is exceeded, default behavior should be:

• alert (log and continue - lowest friction, recommended for initial rollout)
• downgrade (automatically use cheaper model)
• block (refuse to process)

3. Cost Tracking Storage

For per-analyzer cost tracking:

• DynamoDB (persistent, cross-invocation continuity)
• In-memory (simpler, works for batch Lambda invocations)
• CloudWatch Logs + Logs Insights (no dedicated storage, post-hoc analysis)

Recommendation: DynamoDB for production, in-memory for development/testing.

4. Token Estimation Accuracy

For pre-flight budget checks, how accurate should token estimates be?

• ±10% (requires precise tokenization - expensive)
• ±25% (heuristic estimation - my recommendation)
• ±50% (coarse bucketing - too imprecise)

5. Pricing Table Updates

How to maintain Bedrock pricing data?

• Static JSON in repo (simple, requires manual updates)
• AWS Pricing API (automated, requires integration)
• Configurable per analyzer (most flexible)

Recommendation: Start with static JSON, add API integration in Phase 2.

---

Summary: Dependency Resolution

Original proposal: "Depends on issue #11 (Pluggable LLM Backends)"

Revised stance: "Pluggable backends are a complementary enhancement, not a blocker."

The cost optimization features can be implemented immediately using the existing Bedrock integration:

• Complexity classification: No provider abstraction needed
• Model routing: Just conditional model_id selection
• Cost tracking: Bedrock API already returns token counts
• Budget governance: Decision logic on top of cost data

Pluggable backends become valuable when:

• Supporting multiple providers (SageMaker, HTTP endpoints)
• Building testing/mocking infrastructure
• Standardizing provider-specific behavior

But for the core cost optimization use case (route simple docs to cheaper Bedrock models, track costs, enforce budgets), we can ship value now without architectural dependencies.

---

Recommendation: Implement Phases 1-5 as described above using direct Bedrock integration. Revisit pluggable backends as a separate refactoring effort when multi-provider support is actually needed.

[hshahrokni2]

The complexity classification step you're describing — pre-assessing a page as simple/medium/complex before committing to a full model call — is the right architecture. The key insight from similar implementations: the classifier itself needs to be cheap enough that misclassification overhead doesn't eat the savings.

A pattern that works well in practice:

• Heuristic-first: image DPI, text density estimation, handwriting detection can cover 60-70% of routing decisions without any LLM call
• Lightweight LLM fallback: for ambiguous cases, a fast small model (100-200 token prompt) confirms complex vs. simple
• Cost telemetry in the response: track actual cost per invocation and feed that back into the routing thresholds over time — static rules drift as model pricing changes

One thing worth noting on the dynamic model selection: for OpenAI-compatible inference endpoints, routing across different model tiers without changing your application code is solved by proxies that accept the standard format and swap the underlying model based on complexity. Komilion does this (400+ models, auto-routing to cheapest capable model per request). Whether you use that or build it yourself, the architecture of 'classify first, route second, verify cost against target' is sound and the acceptance criteria you've outlined maps well to that separation.