5.1 Objective

Chapters 1–4 established the initial ACAI pipeline:

User
 ↓
API
 ↓
Orchestrator
 ├── Planner
 ├── Retrieval
 └── Memory
 ↓
Model
 ↓
Response

The next problem is model selection.

A single model does not necessarily provide the best combination of:

• reasoning capability

• coding capability

• speed

• cost

• context length

• availability

• privacy requirements

Chapter 5 introduces the Adaptive Model Router.

The router evaluates the task and selects an appropriate model configuration.

5.2 Core Principle

The router should not simply do:

Every request
     ↓
Same model

Instead:

User Request
     ↓
Task Analysis
     ↓
Capability Requirements
     ↓
Candidate Models
     ↓
Scoring
     ↓
Selected Model
     ↓
Generation

The important engineering principle is:

Model selection should be based on measurable requirements rather than assuming that one model is optimal for every task.

5.3 Chapter 5 Architecture

                         USER
                           │
                           ▼
                      FastAPI API
                           │
                           ▼
                   ACAI Orchestrator
                           │
              ┌────────────┼────────────┐
              ▼            ▼            ▼
           Planner       Memory       Retrieval
              │            │            │
              └────────────┼────────────┘
                           ▼
                    Model Router
                           │
                  ┌────────┼────────┐
                  ▼        ▼        ▼
                Fast     Reasoning  Coding
                Model      Model     Model
                  │        │        │
                  └────────┼────────┘
                           ▼
                        Response

5.4 Model Metadata

Create:

app/services/model_router.py

Start with a model description.

from dataclasses import dataclass


@dataclass
class ModelProfile:
    name: str
    capabilities: set[str]
    latency_score: float
    cost_score: float
    reasoning_score: float
    coding_score: float

A model profile describes what the router knows about a model.

For example:

Model
 ├── capabilities
 ├── latency
 ├── cost
 ├── reasoning
 └── coding

These values should eventually come from actual benchmarks or provider metadata rather than invented performance claims.

5.5 Candidate Models

Add:

DEFAULT_MODELS = [
    ModelProfile(
        name="fast-model",
        capabilities={
            "general",
            "classification",
        },
        latency_score=0.95,
        cost_score=0.90,
        reasoning_score=0.55,
        coding_score=0.60,
    ),

    ModelProfile(
        name="reasoning-model",
        capabilities={
            "general",
            "research",
            "reasoning",
        },
        latency_score=0.55,
        cost_score=0.50,
        reasoning_score=0.95,
        coding_score=0.80,
    ),

    ModelProfile(
        name="coding-model",
        capabilities={
            "general",
            "coding",
        },
        latency_score=0.70,
        cost_score=0.65,
        reasoning_score=0.80,
        coding_score=0.95,
    ),
]

These are example profiles for the routing prototype, not claims that these fictional model names have those real-world capabilities.

5.6 Task Requirements

The router needs to convert the Planner result into requirements.

@dataclass
class TaskRequirements:
    task_type: str
    complexity: str
    required_capabilities: set[str]

Add:

def create_requirements(
    task_type: str,
    complexity: str,
) -> TaskRequirements:

    capabilities = {
        "general"
    }

    if task_type == "research":
        capabilities.add("research")

    if task_type == "coding":
        capabilities.add("coding")

    if complexity == "complex":
        capabilities.add("reasoning")

    return TaskRequirements(
        task_type=task_type,
        complexity=complexity,
        required_capabilities=capabilities,
    )

5.7 Model Scoring

A simple scoring function can evaluate candidate models.

def score_model(
    model: ModelProfile,
    requirements: TaskRequirements,
) -> float:

    capability_score = sum(
        capability in model.capabilities
        for capability
        in requirements.required_capabilities
    )

    score = capability_score * 2.0

    if requirements.complexity == "complex":
        score += model.reasoning_score

    if requirements.task_type == "coding":
        score += model.coding_score

    score += model.latency_score * 0.25
    score += model.cost_score * 0.25

    return score

This is intentionally simple.

A production router should use measured performance, actual pricing, latency observations, availability, and policy constraints.

5.8 Model Router

Complete app/services/model_router.py:

from dataclasses import dataclass


@dataclass
class ModelProfile:
    name: str
    capabilities: set[str]
    latency_score: float
    cost_score: float
    reasoning_score: float
    coding_score: float


@dataclass
class TaskRequirements:
    task_type: str
    complexity: str
    required_capabilities: set[str]


@dataclass
class RoutingDecision:
    model_name: str
    score: float
    requirements: TaskRequirements


DEFAULT_MODELS = [

    ModelProfile(
        name="fast-model",
        capabilities={
            "general",
            "classification",
        },
        latency_score=0.95,
        cost_score=0.90,
        reasoning_score=0.55,
        coding_score=0.60,
    ),

    ModelProfile(
        name="reasoning-model",
        capabilities={
            "general",
            "research",
            "reasoning",
        },
        latency_score=0.55,
        cost_score=0.50,
        reasoning_score=0.95,
        coding_score=0.80,
    ),

    ModelProfile(
        name="coding-model",
        capabilities={
            "general",
            "coding",
        },
        latency_score=0.70,
        cost_score=0.65,
        reasoning_score=0.80,
        coding_score=0.95,
    ),
]


def create_requirements(
    task_type: str,
    complexity: str,
) -> TaskRequirements:

    capabilities = {
        "general"
    }

    if task_type == "research":
        capabilities.add("research")

    if task_type == "coding":
        capabilities.add("coding")

    if complexity == "complex":
        capabilities.add("reasoning")

    return TaskRequirements(
        task_type=task_type,
        complexity=complexity,
        required_capabilities=capabilities,
    )


def score_model(
    model: ModelProfile,
    requirements: TaskRequirements,
) -> float:

    capability_score = sum(
        capability in model.capabilities
        for capability
        in requirements.required_capabilities
    )

    score = capability_score * 2.0

    if requirements.complexity == "complex":
        score += model.reasoning_score

    if requirements.task_type == "coding":
        score += model.coding_score

    score += model.latency_score * 0.25
    score += model.cost_score * 0.25

    return score


class ModelRouter:

    def __init__(
        self,
        models: list[ModelProfile] | None = None,
    ) -> None:

        self.models = (
            models
            if models is not None
            else DEFAULT_MODELS
        )

    def route(
        self,
        task_type: str,
        complexity: str,
    ) -> RoutingDecision:

        requirements = create_requirements(
            task_type=task_type,
            complexity=complexity,
        )

        if not self.models:
            raise RuntimeError(
                "No models are available."
            )

        scored_models = [
            (
                score_model(
                    model,
                    requirements,
                ),
                model,
            )
            for model in self.models
        ]

        scored_models.sort(
            key=lambda item: item[0],
            reverse=True,
        )

        best_score, best_model = (
            scored_models[0]
        )

        return RoutingDecision(
            model_name=best_model.name,
            score=best_score,
            requirements=requirements,
        )


model_router = ModelRouter()

5.9 Integrating the Router

The Orchestrator now has to call the Planner first.

Then the router uses the Planner output.

Update app/orchestrator.py:

from app.services.memory import memory_service
from app.services.model_router import model_router
from app.services.model_service import model_service
from app.services.planner import planner
from app.services.retrieval import retrieval_service


class ACAIOrchestrator:

    async def process(
        self,
        message: str,
    ) -> dict:

        cleaned_message = message.strip()

        if not cleaned_message:
            raise ValueError(
                "Message cannot be empty."
            )

        plan = planner.create_plan(
            cleaned_message
        )

        routing_decision = model_router.route(
            task_type=plan.task_type,
            complexity=plan.complexity,
        )

        retrieval_context = (
            retrieval_service.build_context(
                query=cleaned_message,
                top_k=3,
            )
        )

        memory_context = (
            memory_service.build_context(
                query=cleaned_message,
                top_k=5,
            )
        )

        response = await model_service.generate(
            prompt=cleaned_message,
            context=retrieval_context,
            memory=memory_context,
        )

        memory_service.remember_if_useful(
            content=cleaned_message,
            memory_type="conversation",
        )

        return {
            "response": response,

            "plan": {
                "task_type": plan.task_type,
                "complexity": plan.complexity,
                "steps": plan.steps,
            },

            "routing": {
                "model_name":
                    routing_decision.model_name,
                "score":
                    routing_decision.score,
                "required_capabilities":
                    list(
                        routing_decision
                        .requirements
                        .required_capabilities
                    ),
            },

            "retrieval": {
                "used": bool(
                    retrieval_context
                ),
                "context": retrieval_context,
            },

            "memory": {
                "used": bool(
                    memory_context
                ),
                "context": memory_context,
            },
        }


orchestrator = ACAIOrchestrator()

5.10 Model Service Integration

At this stage, the router selects a model name, but the prototype ModelService still uses the mock implementation.

Update it so the selected model is visible.

from app.config import settings


class ModelService:

    def __init__(self) -> None:

        self.provider = (
            settings.model_provider
        )

        self.model_name = (
            settings.model_name
        )

    async def generate(
        self,
        prompt: str,
        context: str = "",
        memory: str = "",
        model_name: str | None = None,
    ) -> str:

        selected_model = (
            model_name
            or self.model_name
        )

        if self.provider == "mock":

            return self._mock_generate(
                prompt=prompt,
                context=context,
                memory=memory,
                model_name=selected_model,
            )

        raise RuntimeError(
            f"Unsupported model provider: "
            f"{self.provider}"
        )

    def _mock_generate(
        self,
        prompt: str,
        context: str,
        memory: str,
        model_name: str,
    ) -> str:

        sections = [
            "ACAI Demo Model Response",
            "",
            f"Selected Model: {model_name}",
            "",
            f"Question:\n{prompt}",
        ]

        if context:

            sections.extend([
                "",
                f"Retrieved Context:\n{context}",
            ])

        if memory:

            sections.extend([
                "",
                f"Relevant Memory:\n{memory}",
            ])

        sections.extend([
            "",
            "The ACAI pipeline completed "
            "model routing successfully.",
        ])

        return "\n".join(sections)


model_service = ModelService()

5.11 Pass the Routing Decision to the Model

Change the call in orchestrator.py:

response = await model_service.generate(
    prompt=cleaned_message,
    context=retrieval_context,
    memory=memory_context,
    model_name=routing_decision.model_name,
)

Now the entire pipeline is connected:

User
 ↓
Planner
 ↓
Task Type + Complexity
 ↓
Model Router
 ↓
Selected Model
 ↓
Retrieval + Memory
 ↓
Model Service
 ↓
Response

5.12 API Schema Update

Update app/schemas.py:

from pydantic import BaseModel, Field


class ChatRequest(BaseModel):

    message: str = Field(
        ...,
        min_length=1,
        max_length=10000,
    )


class PlanResponse(BaseModel):

    task_type: str
    complexity: str
    steps: list[str]


class RetrievalResponse(BaseModel):

    used: bool
    context: str


class MemoryResponse(BaseModel):

    used: bool
    context: str


class RoutingResponse(BaseModel):

    model_name: str
    score: float
    required_capabilities: list[str]


class ChatResponse(BaseModel):

    success: bool
    response: str
    model: str
    mode: str

    plan: PlanResponse

    retrieval: RetrievalResponse

    memory: MemoryResponse

    routing: RoutingResponse

5.13 Router Tests

Create:

tests/test_model_router.py

Add:

from app.services.model_router import (
    ModelRouter,
)


def test_research_task():

    router = ModelRouter()

    decision = router.route(
        task_type="research",
        complexity="complex",
    )

    assert decision.model_name == (
        "reasoning-model"
    )


def test_coding_task():

    router = ModelRouter()

    decision = router.route(
        task_type="coding",
        complexity="complex",
    )

    assert decision.model_name == (
        "coding-model"
    )


def test_general_simple_task():

    router = ModelRouter()

    decision = router.route(
        task_type="general",
        complexity="simple",
    )

    assert decision.model_name in {
        "fast-model",
        "reasoning-model",
        "coding-model",
    }


def test_no_models():

    router = ModelRouter(models=[])

    try:

        router.route(
            task_type="general",
            complexity="simple",
        )

        assert False

    except RuntimeError as exc:

        assert "No models" in str(exc)

5.14 Test the Router

Run:

pytest

The new router tests should pass along with the previous tests.

5.15 Example: Research Request

Input:

Compare several research papers
and analyze their conclusions.

Planner:

Task Type:
research

Complexity:
complex

Router requirements:

general
research
reasoning

Candidate selection:

fast-model
    ↓
Weak research capability

reasoning-model
    ↓
Research + reasoning
    ↓
Selected

The final route becomes:

User
 ↓
Planner
 ↓
research + complex
 ↓
Router
 ↓
reasoning-model

5.16 Example: Coding Request

Input:

Build and debug a Python API.

Planner:

Task Type:
coding

Complexity:
complex

Router:

coding
+
reasoning

The coding-capable model becomes the preferred candidate.

5.17 Example: Simple Request

Input:

What is HTTP?

Planner:

general
simple

The router can prefer a fast/low-cost candidate.

This demonstrates an important idea:

Simple Task
→ Prefer efficiency

Complex Task
→ Prefer capability

5.18 Routing Should Be Measurable

The router should eventually be evaluated using real data.

A benchmark can contain:

Task
Expected Capability
Selected Model
Task Success
Latency
Cost

For example:

The values in the final benchmark must come from actual experiments.

5.19 Routing Baseline

Create two experimental systems.

Baseline A

Every task
 ↓
One fixed model

System B

Task
 ↓
Planner
 ↓
Adaptive Router
 ↓
Selected model

Then compare:

Accuracy
Latency
Cost
Failure Rate
Task-Specific Performance

This provides evidence for whether routing actually improves the system.

5.20 Important Limitation

The Chapter 5 router is a prototype routing algorithm.

It does not yet include:

Real-time latency measurements
Real provider pricing
Model availability
Rate limits
Load balancing
Retries
Fallback models
Provider outages
Dynamic benchmark scores
User-specific constraints
Privacy policies

Those should be added before calling the router production-ready.

5.21 Future Production Router

A more advanced architecture could become:

                    Request
                       │
                       ▼
                    Planner
                       │
                       ▼
              Capability Analysis
                       │
                       ▼
               Candidate Models
                       │
        ┌──────────────┼──────────────┐
        ▼              ▼              ▼
     Quality         Cost          Latency
        │              │              │
        └──────────────┼──────────────┘
                       ▼
                 Policy Filter
                       │
                       ▼
                Model Selection
                       │
                       ▼
                  Generation
                       │
                 ┌─────┴─────┐
                 ▼           ▼
              Success      Failure
                 │           │
                 │           ▼
                 │        Fallback
                 │           │
                 └─────┬─────┘
                       ▼
                    Result

This is the direction for later chapters.

5.22 Chapter 5 Success Criteria

Chapter 5 is complete when:

[✓] Model profiles exist
[✓] Task requirements are generated
[✓] Candidate models can be scored
[✓] Router selects a candidate
[✓] Planner output controls routing
[✓] ModelService receives selected model
[✓] API exposes routing information
[✓] Router tests pass
[✓] Baseline comparison can be performed

5.23 Current ACAI Architecture

After Chapter 5:

                         USER
                           │
                           ▼
                      FastAPI API
                           │
                           ▼
                   ACAI Orchestrator
                           │
        ┌──────────────────┼──────────────────┐
        │                  │                  │
        ▼                  ▼                  ▼
     Planner             Memory           Retrieval
        │                  │                  │
        └──────────────────┼──────────────────┘
                           │
                           ▼
                     Model Router
                           │
              ┌────────────┼────────────┐
              ▼            ▼            ▼
            Fast       Reasoning      Coding
           Model         Model         Model
              │            │            │
              └────────────┼────────────┘
                           ▼
                     Model Service
                           │
                           ▼
                        Response

5.24 What Comes Next?

ACAI now has:

Chapter 1 → Core
Chapter 2 → Planner
Chapter 3 → Retrieval
Chapter 4 → Memory
Chapter 5 → Model Router

The next problem is critical:

How does ACAI know whether its generated answer is actually good enough?

A model can produce an answer that is:

• incomplete

• unsupported

• inconsistent

• incorrectly retrieved

• poorly structured

• technically invalid

Therefore, the next layer is:

Chapter 6 — Verification and Evaluation Layer

The next chapter will build a practical verification pipeline:

Generate
   ↓
Check
   ↓
Evidence Validation
   ↓
Consistency Check
   ↓
Quality Score
   ↓
Accept / Revise / Reject

The objective is to make the architecture measurable and testable, rather than simply claiming that a larger or more complicated system is automatically better.

End of Chapter 5