# ACAI — Chapter 5: Adaptive Model Router

![Post cover](https://dev-to-uploads.s3.us-east-2.amazonaws.com/uploads/articles/ojgzddn71vwz9p6crx54.png align="center")

  

## 5.1 Objective

Chapters 1–4 established the initial ACAI pipeline:

```text
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:

```text
Every request
     ↓
Same model
```

Instead:

```text
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

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

* * *

# 5.4 Model Metadata

Create:

```text
app/services/model_router.py
```

Start with a model description.

```python
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:

```text
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:

```python
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.

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

Add:

```python
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.

```python
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`:

```python
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`:

```python
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.

```python
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`:

```python
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:

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

* * *

# 5.12 API Schema Update

Update `app/schemas.py`:

```python
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:

```text
tests/test_model_router.py
```

Add:

```python
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:

```powershell
pytest
```

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

* * *

# 5.15 Example: Research Request

Input:

```text
Compare several research papers
and analyze their conclusions.
```

Planner:

```text
Task Type:
research

Complexity:
complex
```

Router requirements:

```text
general
research
reasoning
```

Candidate selection:

```text
fast-model
    ↓
Weak research capability

reasoning-model
    ↓
Research + reasoning
    ↓
Selected
```

The final route becomes:

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

* * *

# 5.16 Example: Coding Request

Input:

```text
Build and debug a Python API.
```

Planner:

```text
Task Type:
coding

Complexity:
complex
```

Router:

```text
coding
+
reasoning
```

The coding-capable model becomes the preferred candidate.

* * *

# 5.17 Example: Simple Request

Input:

```text
What is HTTP?
```

Planner:

```text
general
simple
```

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

This demonstrates an important idea:

```text
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:

```text
Task
Expected Capability
Selected Model
Task Success
Latency
Cost
```

For example:

| Task | Expected Capability | Selected Model | Success | Latency |
| --- | --- | --- | --- | --- |
| Classification | General | Fast | Measure | Measure |
| Research | Research/Reasoning | Reasoning | Measure | Measure |
| Coding | Coding | Coding | Measure | Measure |

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

* * *

# 5.19 Routing Baseline

Create two experimental systems.

### Baseline A

```text
Every task
 ↓
One fixed model
```

### System B

```text
Task
 ↓
Planner
 ↓
Adaptive Router
 ↓
Selected model
```

Then compare:

```text
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:

```text
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:

```text
                    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:

```text
[✓] 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:

```text
                         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:

```text
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:

```text
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**