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:
| 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
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
Comments
Post a Comment