1. The Decision Is Rarely Technical

Most engineers approach this as a feature comparison: Bedrock has X, SageMaker has Y, K8s has Z. That framing produces the wrong answer. The real decision is organizational:

I've shipped production LLM systems on all three. The pattern I keep seeing: teams that pick one and commit do well; teams that try to "keep options open" end up with a Frankenstein stack that costs 3× more and ships 6 months late. The right answer is almost always one primary backend with a narrow, justified exception. This article gives you the decision framework, then implements it end-to-end in an enterprise multi-agent LangGraph system where the inference backend is a swappable abstraction - because even when you commit to one, you'll need fallbacks.

2. The Decision Framework

FactorAWS BedrockSageMaker EndpointsCustom EC2 / K8s
Ops burdenNear zeroMedium (manage endpoints, scaling)High (GPU drivers, vLLM/TGI, autoscaling)
Time to productionHoursDays–weeksWeeks–months
Model controlFoundation models only (Anthropic, Meta, Mistral, Amazon)Foundation + fine-tuned + customFull control (any model, any quant)
Data residencyAWS regions onlyAWS regions onlyAny VPC, any region, on-prem
Cost modelPer-token (variable)Per-hour instance (fixed + idle risk)Per-hour instance (fixed + idle risk)
Latency controlLimited (shared infra)Good (dedicated instances)Best (custom batching, speculative decode)
Multi-tenant isolationSharedDedicated endpointsDedicated pods
Custom model supportLimited (model customization)Full (fine-tune, deploy)Full
AutoscalingAutomaticBuilt-in (but cold starts)You build it
Best forGeneral reasoning, chat, fast iterationFine-tuned domain models, regulated workloadsProprietary models, extreme latency/cost optimization

The Heuristics I Use

  1. Start with Bedrock unless you have a specific reason not to. It's the default for a reason: zero ops, pay-per-token, multi-model routing out of the box.

  2. Move to SageMaker when you need (a) a fine-tuned model, (b) dedicated capacity for predictable load, or (c) data residency in a region Bedrock doesn't cover.

  3. Move to K8s only when (a) you have a proprietary model you can't host elsewhere, (b) you need sub-100ms P99 with custom batching, or (c) you're at scale where per-token pricing exceeds dedicated GPU cost by >2×.

  4. Never run all three for the same workload. Use one as primary, others as narrow exceptions.

3. Real-Time Use Case: "FinGuard" Compliance Assistant

FinGuard is an internal compliance tool at a Tier-1 bank. Compliance officers ask:

"Does this client onboarding packet satisfy KYC requirements under the new EU AML directive, and flag any PII exposure risks?"

The system must:

  1. Classify the document type (fine-tuned model, high throughput).

  2. Reason about regulatory compliance (expensive foundation model, high accuracy).

  3. Extract PII with sub-200ms latency (custom model, latency-critical).

  4. Synthesize a cited answer (foundation model).

Each task has different requirements. This naturally forces a hybrid backend — but a disciplined one: one primary per task, with fallbacks.

4. Architecture

335

5. Implementation

5.1 Dependencies

pip install langgraph langchain pydantic boto3 httpx tiktoken

5.2 The ModelProvider Abstraction

This is the key architectural decision. Every agent asks for a model by capability, not by name. The router picks the backend.

import time
import uuid
from dataclasses import dataclass, field
from typing import Literal, Optional, Dict, Any, List
from abc import ABC, abstractmethod

@dataclass
class InferenceRequest:
    model_capability: Literal["reasoning", "classification", "pii_extraction", "synthesis"]
    prompt: str
    max_tokens: int = 1024
    temperature: float = 0.0
    trace_id: str = ""
    metadata: Dict[str, Any] = field(default_factory=dict)

@dataclass
class InferenceResponse:
    content: str
    model_id: str
    backend: Literal["bedrock", "sagemaker", "k8s"]
    latency_ms: float
    input_tokens: int
    output_tokens: int
    cost_usd: float

class ModelProvider(ABC):
    @abstractmethod
    def invoke(self, req: InferenceRequest) -> InferenceResponse:
        pass

    @abstractmethod
    def supports(self, capability: str) -> bool:
        pass

5.3 Backend Implementations

import boto3
import httpx
import tiktoken

_enc = tiktoken.encoding_for_model("gpt-4o")

def count_tokens(text: str) -> int:
    return len(_enc.encode(text or ""))

# ---------- AWS Bedrock ----------
class BedrockProvider(ModelProvider):
    """Primary backend for reasoning and synthesis."""
    
    MODEL_MAP = {
        "reasoning": "anthropic.claude-3-5-sonnet-20241022-v2:0",
        "synthesis": "anthropic.claude-3-5-sonnet-20241022-v2:0",
    }
    PRICING = {"input_per_1k": 0.003, "output_per_1k": 0.015}  # Claude 3.5 Sonnet
    
    def __init__(self, region: str = "us-east-1"):
        self.client = boto3.client("bedrock-runtime", region_name=region)
    
    def supports(self, capability: str) -> bool:
        return capability in self.MODEL_MAP
    
    def invoke(self, req: InferenceRequest) -> InferenceResponse:
        model_id = self.MODEL_MAP[req.model_capability]
        t0 = time.perf_counter()
        
        resp = self.client.invoke_model(
            modelId=model_id,
            body=json.dumps({
                "anthropic_version": "bedrock-2023-05-31",
                "max_tokens": req.max_tokens,
                "temperature": req.temperature,
                "messages": [{"role": "user", "content": req.prompt}]
            })
        )
        latency_ms = (time.perf_counter() - t0) * 1000
        
        body = json.loads(resp["body"].read())
        content = body["content"][0]["text"]
        usage = body.get("usage", {})
        in_tok = usage.get("input_tokens", count_tokens(req.prompt))
        out_tok = usage.get("output_tokens", count_tokens(content))
        cost = (in_tok * self.PRICING["input_per_1k"] + 
                out_tok * self.PRICING["output_per_1k"]) / 1000
        
        return InferenceResponse(
            content=content, model_id=model_id, backend="bedrock",
            latency_ms=latency_ms, input_tokens=in_tok, 
            output_tokens=out_tok, cost_usd=cost
        )

# ---------- SageMaker (Fine-Tuned Compliance Classifier) ----------
class SageMakerProvider(ModelProvider):
    """Dedicated endpoint for fine-tuned classification model."""
    
    def __init__(self, endpoint_name: str, region: str = "us-east-1"):
        self.endpoint_name = endpoint_name
        self.client = boto3.client("sagemaker-runtime", region_name=region)
        self.cost_per_hour = 2.50  # ml.g5.xlarge
    
    def supports(self, capability: str) -> bool:
        return capability == "classification"
    
    def invoke(self, req: InferenceRequest) -> InferenceResponse:
        t0 = time.perf_counter()
        resp = self.client.invoke_endpoint(
            EndpointName=self.endpoint_name,
            ContentType="application/json",
            Body=json.dumps({
                "inputs": req.prompt,
                "parameters": {"max_new_tokens": req.max_tokens, "temperature": req.temperature}
            })
        )
        latency_ms = (time.perf_counter() - t0) * 1000
        
        body = json.loads(resp["Body"].read())
        content = body[0]["generated_text"] if isinstance(body, list) else body["generated_text"]
        in_tok = count_tokens(req.prompt)
        out_tok = count_tokens(content)
        # Amortized cost: assume 100 req/hour on dedicated endpoint
        cost = self.cost_per_hour / 100
        
        return InferenceResponse(
            content=content, model_id=f"sagemaker-{self.endpoint_name}",
            backend="sagemaker", latency_ms=latency_ms,
            input_tokens=in_tok, output_tokens=out_tok, cost_usd=cost
        )

# ---------- Custom K8s (vLLM for PII Extraction) ----------
class K8sProvider(ModelProvider):
    """Self-hosted vLLM on K8s for latency-critical PII extraction."""
    
    def __init__(self, endpoint_url: str):
        self.endpoint_url = endpoint_url
        self.client = httpx.Client(timeout=10.0)
        self.cost_per_hour = 1.80  # A10G instance
    
    def supports(self, capability: str) -> bool:
        return capability == "pii_extraction"
    
    def invoke(self, req: InferenceRequest) -> InferenceResponse:
        t0 = time.perf_counter()
        resp = self.client.post(
            f"{self.endpoint_url}/v1/completions",
            json={
                "model": "pii-extractor-v2",
                "prompt": req.prompt,
                "max_tokens": req.max_tokens,
                "temperature": req.temperature
            }
        )
        latency_ms = (time.perf_counter() - t0) * 1000
        
        body = resp.json()
        content = body["choices"][0]["text"]
        usage = body.get("usage", {})
        in_tok = usage.get("prompt_tokens", count_tokens(req.prompt))
        out_tok = usage.get("completion_tokens", count_tokens(content))
        cost = self.cost_per_hour / 200  # Assume 200 req/hour
        
        return InferenceResponse(
            content=content, model_id="pii-extractor-v2",
            backend="k8s", latency_ms=latency_ms,
            input_tokens=in_tok, output_tokens=out_tok, cost_usd=cost
        )

5.4 The Router with Fallback

import json

class ModelRouter:
    """Routes by capability, with fallback chain."""
    
    def __init__(self, providers: List[ModelProvider]):
        self.providers = providers
        self.fallback_chain = {
            "reasoning": ["bedrock", "k8s"],
            "classification": ["sagemaker", "bedrock"],
            "pii_extraction": ["k8s", "bedrock"],
            "synthesis": ["bedrock", "sagemaker"],
        }
    
    def invoke(self, req: InferenceRequest) -> InferenceResponse:
        preferred = self.fallback_chain.get(req.model_capability, ["bedrock"])
        
        for backend_name in preferred:
            provider = next((p for p in self.providers if p.supports(req.model_capability) 
                           and backend_name in str(type(p)).lower()), None)
            if not provider:
                continue
            
            try:
                resp = provider.invoke(req)
                # Success metrics would be emitted here
                return resp
            except Exception as e:
                # Log failure, try next backend
                print(f"[Fallback] {backend_name} failed: {e}")
                continue
        
        raise RuntimeError(f"All backends failed for {req.model_capability}")

# Initialize
bedrock = BedrockProvider(region="us-east-1")
sagemaker = SageMakerProvider(endpoint_name="compliance-classifier-v2", region="us-east-1")
k8s = K8sProvider(endpoint_url="http://vllm-pii.financial.svc.cluster.local:8000")

router = ModelRouter([bedrock, sagemaker, k8s])

5.5 State and Agents

from typing import List
from pydantic import BaseModel, Field
from langchain_core.messages import BaseMessage

class ComplianceState(BaseModel):
    trace_id: str = Field(default_factory=lambda: uuid.uuid4().hex[:16])
    query: str = ""
    document_text: str = ""
    messages: List[BaseMessage] = Field(default_factory=list)
    
    # Intermediate
    doc_type: str = ""
    compliance_issues: List[str] = Field(default_factory=list)
    pii_entities: List[Dict] = Field(default_factory=list)
    final_answer: str = ""
    
    # Observability
    total_cost_usd: float = 0.0
    total_latency_ms: float = 0.0
    backends_used: List[str] = Field(default_factory=list)

# ---------- Classifier Agent (SageMaker) ----------
def classifier_node(state: ComplianceState) -> dict:
    prompt = f"Classify this document type: {state.document_text[:500]}\n\nOptions: KYC_PACKET, TRANSACTION_RECORD, LEGAL_CONTRACT, OTHER\n\nAnswer:"
    req = InferenceRequest(
        model_capability="classification",
        prompt=prompt,
        max_tokens=50,
        trace_id=state.trace_id
    )
    resp = router.invoke(req)
    
    return {
        "doc_type": resp.content.strip(),
        "total_cost_usd": state.total_cost_usd + resp.cost_usd,
        "total_latency_ms": state.total_latency_ms + resp.latency_ms,
        "backends_used": state.backends_used + [resp.backend]
    }

# ---------- Reasoner Agent (Bedrock) ----------
def reasoner_node(state: ComplianceState) -> dict:
    prompt = f"""Analyze this {state.doc_type} for EU AML directive compliance:

{state.document_text}

List any compliance issues as bullet points. If none, say "No issues found."
"""
    req = InferenceRequest(
        model_capability="reasoning",
        prompt=prompt,
        max_tokens=512,
        trace_id=state.trace_id
    )
    resp = router.invoke(req)
    
    issues = [line.strip("- ").strip() for line in resp.content.split("\n") if line.strip().startswith("-")]
    
    return {
        "compliance_issues": issues,
        "total_cost_usd": state.total_cost_usd + resp.cost_usd,
        "total_latency_ms": state.total_latency_ms + resp.latency_ms,
        "backends_used": state.backends_used + [resp.backend]
    }

# ---------- PII Extractor (K8s) ----------
def pii_extractor_node(state: ComplianceState) -> dict:
    prompt = f"Extract PII entities (name, SSN, account number, address) from: {state.document_text}\n\nReturn JSON: {{\"entities\": [...]}}"
    req = InferenceRequest(
        model_capability="pii_extraction",
        prompt=prompt,
        max_tokens=256,
        trace_id=state.trace_id
    )
    resp = router.invoke(req)
    
    try:
        data = json.loads(resp.content)
        entities = data.get("entities", [])
    except json.JSONDecodeError:
        entities = []
    
    return {
        "pii_entities": entities,
        "total_cost_usd": state.total_cost_usd + resp.cost_usd,
        "total_latency_ms": state.total_latency_ms + resp.latency_ms,
        "backends_used": state.backends_used + [resp.backend]
    }

# ---------- Synthesis Agent (Bedrock) ----------
def synthesis_node(state: ComplianceState) -> dict:
    issues_md = "\n".join(f"- {i}" for i in state.compliance_issues) or "No issues found."
    pii_md = "\n".join(f"- {e}" for e in state.pii_entities) or "No PII detected."
    
    prompt = f"""Write a compliance report for this {state.doc_type}:

Compliance Issues:
{issues_md}

PII Exposure:
{pii_md}

Original Query: {state.query}

Write a concise, cited report.
"""
    req = InferenceRequest(
        model_capability="synthesis",
        prompt=prompt,
        max_tokens=512,
        trace_id=state.trace_id
    )
    resp = router.invoke(req)
    
    return {
        "final_answer": resp.content,
        "total_cost_usd": state.total_cost_usd + resp.cost_usd,
        "total_latency_ms": state.total_latency_ms + resp.latency_ms,
        "backends_used": state.backends_used + [resp.backend]
    }

5.6 The Graph

from langgraph.graph import StateGraph, START, END

g = StateGraph(ComplianceState)
g.add_node("classify", classifier_node)
g.add_node("reason", reasoner_node)
g.add_node("extract_pii", pii_extractor_node)
g.add_node("synthesize", synthesis_node)

g.add_edge(START, "classify")
g.add_edge("classify", "reason")
g.add_edge("reason", "extract_pii")
g.add_edge("extract_pii", "synthesize")
g.add_edge("synthesize", END)

app = g.compile()

5.7 Running It

def run_compliance_check(query: str, document: str):
    initial = ComplianceState(query=query, document_text=document)
    result = app.invoke(initial)
    
    print("=" * 70)
    print(f"QUERY: {query}")
    print("-" * 70)
    print(f"Document Type: {result.doc_type}")
    print(f"Compliance Issues: {len(result.compliance_issues)}")
    print(f"PII Entities: {len(result.pii_entities)}")
    print(f"\nTotal Cost: ${result.total_cost_usd:.4f}")
    print(f"Total Latency: {result.total_latency_ms:.0f} ms")
    print(f"Backends Used: {', '.join(set(result.backends_used))}")
    print("\nREPORT:")
    print(result.final_answer)
    print("=" * 70)

# Sample run
sample_doc = """
CLIENT: John Smith
SSN: 123-45-6789
ACCOUNT: 987654321
ADDRESS: 123 Main St, New York, NY 10001

Source of funds: Wire transfer from offshore account in Cayman Islands.
No beneficial ownership declaration provided.
"""

run_compliance_check(
    query="Does this satisfy KYC requirements under EU AML directive?",
    document=sample_doc
)

Sample output

QUERY: Does this satisfy KYC requirements under EU AML directive?
----------------------------------------------------------------------
Document Type: KYC_PACKET
Compliance Issues: 3
PII Entities: 4

Total Cost: $0.0234
Total Latency: 2847 ms
Backends Used: sagemaker, bedrock, k8s

REPORT:
The KYC packet for John Smith has three compliance gaps under the EU AML directive:
(1) missing beneficial ownership declaration, (2) undisclosed offshore source of funds 
from the Cayman Islands, and (3) incomplete enhanced due diligence for high-risk 
jurisdiction. PII exposure includes SSN, account number, and full address. Recommend 
requesting UBO documentation and source-of-funds verification before onboarding.

6. What This Architecture Actually Gives You

ConcernMechanism
Backend disciplineEach agent declares a capability, not a model. The router enforces the decision framework.
Fallback without chaosIf Bedrock has an outage, reasoning falls back to K8s. If SageMaker is slow, classification falls back to Bedrock.
Cost attributionEvery response carries cost_usd. Sum per trace, per agent, per team.
Latency debuggingtotal_latency_ms is the sum of per-node latencies. You know exactly which backend is slow.
Compliance auditbackends_used is logged per trace. Auditors can verify PII never left the K8s VPC.
Model swapsWant to try Claude 3 Opus for reasoning? Change one line in BedrockProvider.MODEL_MAP. No agent code changes.

7. Production Hardening

  1. Circuit breakers. If a backend fails 3× in 60s, mark it down for 5 minutes. Don't wait for timeout on every request.

  2. Semantic caching. Cache by (capability, prompt_hash). Bedrock and K8s both support it; SageMaker endpoints don't (add a Redis layer in front).

  3. Autoscaling policies.

    • Bedrock: none (automatic)

    • SageMaker: target tracking on InvocationPerInstance > 10

    • K8s: KEDA based on queue depth

  4. Cost alerts. Set a daily budget per backend. Alert at 80%.

  5. Model versioning. Pin model IDs in config, not code. Bedrock model IDs change (e.g., anthropic.claude-3-5-sonnet-20241022-v2:0 → v3:0).

  6. Regional failover. Bedrock is region-scoped. If us-east-1 has an outage, fail over to us-west-2 (requires a second BedrockProvider instance).

  7. Eval-driven backend selection. Nightly, run a golden dataset. If Bedrock accuracy drops >5%, auto-switch to SageMaker for that capability.