The Model CommitteeWhy Modern AI Is Not One Brain — It's Eight Specialists and a Router

# ──────────────────────────────────────────────────────────────
#  Rule-based router — no training data, no latency overhead,
#  transparent logic. Best for narrow products with predictable
#  query distributions.
# ──────────────────────────────────────────────────────────────
from dataclasses import dataclass
from typing import Literal

ModelFamily = Literal["slm", "llm", "moe", "vlm", "lam", "sam", "lcm", "mlm"]


@dataclass
class Query:
    text: str
    has_image: bool = False
    wants_action: bool = False         # set by upstream intent classifier
    needs_pixel_mask: bool = False     # set when request is "cut this out"
    is_generative_image: bool = False  # set for "draw me a ..." prompts


def route(q: Query) -> ModelFamily:
    # Hard rules first — these short-circuit everything else.
    if q.needs_pixel_mask:
        return "sam"                   # pixel-perfect isolation
    if q.is_generative_image:
        return "lcm"                   # fast image generation
    if q.has_image:
        return "vlm"                   # any other image input

    # Action before reasoning: LAM beats LLM when the user wants
    # something DONE, not explained.
    if q.wants_action:
        return "lam"

    # Text path: short factual → SLM, long complex → LLM/MoE.
    token_estimate = len(q.text.split())
    if token_estimate < 20 and _looks_factual(q.text):
        return "slm"                   # on-device, <200ms

    # Embedding/classification jobs ride the MLM family.
    if _is_classification_task(q.text):
        return "mlm"

    # Default escalation: MoE if available, otherwise dense LLM.
    return "moe" if _moe_endpoint_healthy() else "llm"


def _looks_factual(text: str) -> bool:
    return text.strip().endswith("?") and not any(
        marker in text.lower()
        for marker in ("explain", "analyze", "why", "how does", "compare")
    )


def _is_classification_task(text: str) -> bool:
    return text.startswith(("classify:", "sentiment:", "intent:"))


def _moe_endpoint_healthy() -> bool:
    # In production this checks a circuit breaker; stubbed here.
    return True

# ──────────────────────────────────────────────────────────────
#  Classifier router — one small MLM decides which specialist
#  handles the query. ~10ms overhead; handles the long tail
#  better than hand-coded rules.
# ──────────────────────────────────────────────────────────────
import torch
from transformers import AutoTokenizer, AutoModelForSequenceClassification

ROUTER_MODEL = "distilbert-base-uncased"          # fine-tuned offline
LABELS = ["slm", "llm", "moe", "vlm", "lam", "sam", "lcm", "mlm"]

# Load once at process start; hold in memory.
_tokenizer = AutoTokenizer.from_pretrained(ROUTER_MODEL)
_model = AutoModelForSequenceClassification.from_pretrained(
    ROUTER_MODEL,
    num_labels=len(LABELS),
)
_model.eval()


def classify_route(query: str, confidence_floor: float = 0.65) -> str:
    """Return the predicted target model family, or 'llm' as safe default."""
    with torch.inference_mode():
        inputs = _tokenizer(
            query,
            return_tensors="pt",
            truncation=True,
            max_length=128,
        )
        logits = _model(**inputs).logits
        probs = torch.softmax(logits, dim=-1)[0]
        best_idx = int(torch.argmax(probs).item())
        best_conf = float(probs[best_idx].item())

    # Below the confidence floor we escalate to the generalist LLM
    # rather than risk a wrong specialist pick on an ambiguous query.
    if best_conf < confidence_floor:
        return "llm"

    return LABELS[best_idx]


# Example usage in a request handler
def handle_request(query: str) -> str:
    target = classify_route(query)
    # dispatch_to_family is the thin adapter that talks to each model
    # provider — Anthropic, OpenAI, Mistral, on-device runtime, etc.
    return dispatch_to_family(target, query)

# ──────────────────────────────────────────────────────────────
#  Cascading router — escalate through a cost-ascending cascade,
#  stopping at the first model whose answer clears the gate.
# ──────────────────────────────────────────────────────────────
from dataclasses import dataclass
from anthropic import Anthropic

client = Anthropic()

@dataclass
class CascadeStep:
    model: str
    confidence_gate: float   # min confidence to accept this answer
    cost_per_1k_in: float    # illustrative $/1k input tokens
    cost_per_1k_out: float   # illustrative $/1k output tokens


CASCADE: list[CascadeStep] = [
    CascadeStep("claude-haiku-4-5",  confidence_gate=0.85,
                cost_per_1k_in=0.0008, cost_per_1k_out=0.004),
    CascadeStep("claude-sonnet-4-6", confidence_gate=0.90,
                cost_per_1k_in=0.003,  cost_per_1k_out=0.015),
    CascadeStep("claude-opus-4-6",   confidence_gate=1.00,  # final
                cost_per_1k_in=0.015,  cost_per_1k_out=0.075),
]


def cascade_answer(query: str, system_prompt: str) -> dict:
    """Walk the cascade until an answer clears its step's confidence gate."""
    last_answer = None
    last_model = None

    for step in CASCADE:
        response = client.messages.create(
            model=step.model,
            max_tokens=1024,
            system=system_prompt,
            messages=[{"role": "user", "content": query}],
        )
        answer = response.content[0].text
        confidence = _judge_quality(query, answer, step.model)

        last_answer, last_model = answer, step.model

        # Clear the gate → return immediately, don't pay for escalation.
        if confidence >= step.confidence_gate:
            return {
                "model": step.model,
                "answer": answer,
                "confidence": confidence,
                "escalations_used": CASCADE.index(step),
            }

    # Exhausted the cascade — return the last (most expensive) answer anyway.
    return {
        "model": last_model,
        "answer": last_answer,
        "confidence": confidence,
        "escalations_used": len(CASCADE) - 1,
    }


def _judge_quality(query: str, answer: str, model: str) -> float:
    """Lightweight self-evaluation signal. In production this is either
    a small judge model or a rule-based post-hoc validator."""
    if not answer or len(answer) < 20:
        return 0.0
    if any(refusal in answer.lower() for refusal in ("i can't", "i don't know")):
        return 0.3
    return 0.92  # placeholder — real systems use an embedding-based judge

# ──────────────────────────────────────────────────────────────
#  LLM Council — parallel adversarial review pattern.
#  Claude writes. Gemini and Codex critique in parallel.
#  Writer revises until both reviewers approve, or max rounds hit.
# ──────────────────────────────────────────────────────────────
import os
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from anthropic import Anthropic
from openai import OpenAI
import google.generativeai as genai

anthropic_client = Anthropic()
openai_client = OpenAI()
genai.configure(api_key=os.environ["GEMINI_API_KEY"])
gemini_model = genai.GenerativeModel("gemini-2.5-pro")


@dataclass
class Review:
    reviewer: str
    approved: bool
    feedback: str


def llm_council_write(
    prompt: str,
    system: str = "You are a rigorous technical writer.",
    max_rounds: int = 3,
) -> str:
    """Iterate until both reviewers approve or max_rounds reached."""
    draft = _claude_write(prompt, system)

    for round_num in range(max_rounds):
        # Parallel adversarial review — Gemini and Codex run concurrently.
        with ThreadPoolExecutor(max_workers=2) as pool:
            gemini_future = pool.submit(_gemini_critique, prompt, draft)
            codex_future  = pool.submit(_codex_critique, prompt, draft)
            gemini_review = gemini_future.result()
            codex_review  = codex_future.result()

        if gemini_review.approved and codex_review.approved:
            return draft  # consensus reached

        # Writer revises against both reviewers' feedback simultaneously.
        draft = _claude_revise(draft, gemini_review, codex_review)

    return draft  # ship with residual warnings after max_rounds


def _claude_write(prompt: str, system: str) -> str:
    response = anthropic_client.messages.create(
        model="claude-opus-4-6",
        max_tokens=4096,
        system=system,
        messages=[{"role": "user", "content": prompt}],
    )
    return response.content[0].text


def _gemini_critique(prompt: str, draft: str) -> Review:
    critique_prompt = (
        f"Original request:\n{prompt}\n\nDraft:\n{draft}\n\n"
        "Find every factual error, logical gap, or missing requirement. "
        "End with exactly one line: 'VERDICT: APPROVE' or 'VERDICT: REJECT'."
    )
    response = gemini_model.generate_content(critique_prompt)
    text = response.text
    return Review(
        reviewer="gemini",
        approved="VERDICT: APPROVE" in text,
        feedback=text,
    )


def _codex_critique(prompt: str, draft: str) -> Review:
    critique_prompt = (
        f"Original request:\n{prompt}\n\nDraft:\n{draft}\n\n"
        "Review as an adversarial technical reviewer. "
        "End with exactly one line: 'VERDICT: APPROVE' or 'VERDICT: REJECT'."
    )
    response = openai_client.chat.completions.create(
        model="gpt-5-codex",
        messages=[{"role": "user", "content": critique_prompt}],
    )
    text = response.choices[0].message.content or ""
    return Review(
        reviewer="codex",
        approved="VERDICT: APPROVE" in text,
        feedback=text,
    )


def _claude_revise(draft: str, g: Review, c: Review) -> str:
    response = anthropic_client.messages.create(
        model="claude-opus-4-6",
        max_tokens=4096,
        messages=[{
            "role": "user",
            "content": (
                f"Current draft:\n{draft}\n\n"
                f"Gemini said:\n{g.feedback}\n\n"
                f"Codex said:\n{c.feedback}\n\n"
                "Revise the draft to address both reviewers' concerns. "
                "Return only the revised draft."
            ),
        }],
    )
    return response.content[0].text