6.1 Objective
Chapters 1–5 established the main ACAI pipeline:
User
↓
API
↓
Orchestrator
↓
Planner
↓
Memory + Retrieval
↓
Model Router
↓
Model
↓
Response
There is now a critical question:
How does ACAI determine whether the generated response is good enough to return?
Generation alone is not sufficient.
A model can produce an answer that is:
Chapter 6 therefore introduces a Verification and Evaluation Layer.
The basic pipeline becomes:
Request
↓
Plan
↓
Retrieve
↓
Remember
↓
Route
↓
Generate
↓
Verify
↓
Accept / Revise / Reject
↓
Final Response
6.2 Verification vs Evaluation
These concepts should be separated.
Verification
Verification asks:
"Does this particular output satisfy the required conditions?"
For example:
Is the response empty?
Does it contain required information?
Does it contradict retrieved evidence?
Does it follow the requested format?
Evaluation
Evaluation asks:
"How well did the system perform?"
For example:
Accuracy
Relevance
Completeness
Latency
Cost
Failure rate
Therefore:
Verification
→ Individual response
Evaluation
→ System performance
6.3 Chapter 6 Architecture
USER
│
▼
FastAPI API
│
▼
ACAI Orchestrator
│
┌───────────────────┼───────────────────┐
▼ ▼ ▼
Planner Memory Retrieval
│ │ │
└───────────────────┼───────────────────┘
▼
Model Router
│
▼
Model Service
│
▼
Generation
│
▼
Verification
│
┌─────────────┼─────────────┐
▼ ▼ ▼
Validity Evidence Quality
Check Check Check
│ │ │
└─────────────┼─────────────┘
▼
Decision Engine
│
┌─────────┼─────────┐
▼ ▼ ▼
Accept Revise Reject
6.4 Verification Data Model
Create:
app/services/verification.py
Start with:
from dataclasses import dataclass
@dataclass
class VerificationResult:
passed: bool
score: float
reasons: list[str]
needs_revision: bool
The verifier returns four important pieces of information:
passed
score
reasons
needs_revision
6.5 Basic Output Validation
The first verifier should perform deterministic checks.
class BasicVerifier:
MIN_RESPONSE_LENGTH = 10
def verify(
self,
response: str,
) -> VerificationResult:
reasons = []
if not response.strip():
return VerificationResult(
passed=False,
score=0.0,
reasons=[
"Response is empty."
],
needs_revision=True,
)
if len(response.strip()) < (
self.MIN_RESPONSE_LENGTH
):
reasons.append(
"Response is too short."
)
score = 1.0
if reasons:
score = 0.5
return VerificationResult(
passed=not reasons,
score=score,
reasons=reasons,
needs_revision=bool(reasons),
)
This is deliberately simple.
It provides a deterministic baseline before introducing more sophisticated evaluation.
6.6 Context Consistency
If ACAI used retrieved evidence, it should be possible to check whether the answer has at least some relationship to that evidence.
Add:
class ContextVerifier:
def verify(
self,
response: str,
context: str,
) -> VerificationResult:
if not context.strip():
return VerificationResult(
passed=True,
score=1.0,
reasons=[
"No retrieval context was provided."
],
needs_revision=False,
)
response_words = {
word.lower().strip(".,!?;:")
for word in response.split()
if word.strip()
}
context_words = {
word.lower().strip(".,!?;:")
for word in context.split()
if word.strip()
}
overlap = (
response_words & context_words
)
if not overlap:
return VerificationResult(
passed=False,
score=0.0,
reasons=[
"Response has no lexical overlap "
"with retrieved context."
],
needs_revision=True,
)
return VerificationResult(
passed=True,
score=1.0,
reasons=[
"Response has overlap with "
"retrieved context."
],
needs_revision=False,
)
This is not a factuality proof.
Lexical overlap does not establish truth.
It is simply an inexpensive consistency signal.
6.7 Composite Verification
Now combine multiple checks.
class VerificationService:
def __init__(self) -> None:
self.basic = BasicVerifier()
self.context = ContextVerifier()
def verify(
self,
response: str,
context: str = "",
) -> VerificationResult:
basic_result = self.basic.verify(
response
)
context_result = self.context.verify(
response=response,
context=context,
)
scores = [
basic_result.score,
context_result.score,
]
score = sum(scores) / len(scores)
reasons = (
basic_result.reasons
+ context_result.reasons
)
passed = (
basic_result.passed
and context_result.passed
)
return VerificationResult(
passed=passed,
score=score,
reasons=reasons,
needs_revision=not passed,
)
verification_service = VerificationService()
6.8 Why Multiple Checks Matter
A single check can be misleading.
For example:
Response length
does not prove correctness.
Similarly:
Keyword overlap
does not prove factual accuracy.
Therefore ACAI should use multiple independent signals.
A future verification system can contain:
Format Check
+
Evidence Check
+
Consistency Check
+
Task Completion Check
+
Domain Validation
6.9 Integrating Verification into the Orchestrator
Update the Orchestrator:
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
from app.services.verification import verification_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,
model_name=(
routing_decision.model_name
),
)
verification = (
verification_service.verify(
response=response,
context=retrieval_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,
},
"verification": {
"passed":
verification.passed,
"score":
verification.score,
"reasons":
verification.reasons,
"needs_revision":
verification.needs_revision,
},
}
orchestrator = ACAIOrchestrator()
6.10 Verification in the Complete Pipeline
The system now operates like:
USER
│
▼
API
│
▼
ORCHESTRATOR
│
┌───────────┼───────────┐
▼ ▼ ▼
PLANNER MEMORY RETRIEVAL
│ │ │
└───────────┼───────────┘
▼
MODEL ROUTER
│
▼
MODEL SERVICE
│
▼
GENERATION
│
▼
VERIFICATION
│
┌────┴────┐
▼ ▼
PASS FAIL
│ │
▼ ▼
Response Revision
6.11 Revision Loop
A powerful improvement is allowing ACAI to retry when verification fails.
The basic loop is:
Generate
↓
Verify
↓
Passed?
┌─┴─┐
YES NO
│ │
▼ ▼
Done Revise
│
▼
Generate
However, retries must be bounded.
Otherwise the system could enter:
Generate
↓
Fail
↓
Generate
↓
Fail
↓
Generate
↓
Fail
...
Therefore use a maximum retry count.
6.12 Bounded Revision
Add:
MAX_REVISIONS = 2
Then conceptually:
for attempt in range(MAX_REVISIONS + 1):
response = await model_service.generate(...)
verification = verification_service.verify(
response=response,
context=retrieval_context,
)
if verification.passed:
break
A bounded loop prevents runaway generation.
6.13 Revision Prompt
If verification fails, the model can receive structured feedback.
For example:
Original request:
<user request>
Previous response:
<generated response>
Verification problems:
<verification reasons>
Produce a corrected response.
This creates:
Generation
↓
Verification
↓
Feedback
↓
Revision
↓
Verification
6.14 Safe Revision Implementation
Update the Orchestrator generation section:
MAX_REVISIONS = 2
verification = None
response = ""
for attempt in range(
MAX_REVISIONS + 1
):
if attempt == 0:
generation_prompt = (
cleaned_message
)
else:
generation_prompt = (
f"Original request:\n"
f"{cleaned_message}\n\n"
f"Previous response:\n"
f"{response}\n\n"
f"Verification feedback:\n"
f"{'; '.join(verification.reasons)}\n\n"
"Produce an improved response."
)
response = await model_service.generate(
prompt=generation_prompt,
context=retrieval_context,
memory=memory_context,
model_name=(
routing_decision.model_name
),
)
verification = (
verification_service.verify(
response=response,
context=retrieval_context,
)
)
if verification.passed:
break
The important property is the bounded maximum:
Maximum attempts = 3
6.15 Verification Result
The API can now expose:
{
"verification": {
"passed": true,
"score": 1.0,
"reasons": [],
"needs_revision": false
}
}
Or, when problems occur:
{
"verification": {
"passed": false,
"score": 0.5,
"reasons": [
"Response is too short."
],
"needs_revision": true
}
}
6.16 API Schema Update
Add:
class VerificationResponse(BaseModel):
passed: bool
score: float
reasons: list[str]
needs_revision: bool
Then include it in:
class ChatResponse(BaseModel):
success: bool
response: str
model: str
mode: str
plan: PlanResponse
retrieval: RetrievalResponse
memory: MemoryResponse
routing: RoutingResponse
verification: VerificationResponse
6.17 Testing Verification
Create:
tests/test_verification.py
Add:
from app.services.verification import (
VerificationService,
)
def test_valid_response():
service = VerificationService()
result = service.verify(
response=(
"ACAI is a modular architecture "
"for coordinating AI components."
)
)
assert result.passed is True
def test_empty_response():
service = VerificationService()
result = service.verify(
response=""
)
assert result.passed is False
def test_short_response():
service = VerificationService()
result = service.verify(
response="Hi"
)
assert result.passed is False
def test_context_overlap():
service = VerificationService()
result = service.verify(
response=(
"ACAI uses a planner "
"for task analysis."
),
context=(
"The planner analyzes "
"incoming ACAI requests."
),
)
assert result.passed is True
def test_context_without_overlap():
service = VerificationService()
result = service.verify(
response=(
"Completely unrelated information."
),
context=(
"ACAI planner analyzes requests."
),
)
assert result.passed is False
6.18 Run the Full Test Suite
Run:
pytest
At this stage the project should test:
API
Planner
Retrieval
Memory
Router
Verification
Conceptually:
ACAI TEST SUITE
│
┌────────────────┼────────────────┐
▼ ▼ ▼
API Planner Retrieval
│ │ │
├────────────────┼────────────────┤
▼ ▼ ▼
Memory Router Verification
6.19 Evaluation Dataset
Verification checks individual outputs.
For system evaluation, ACAI needs a benchmark dataset.
Create:
data/evaluation/
benchmark.json
Example structure:
[
{
"id": "task-001",
"category": "general",
"prompt": "Explain HTTP in simple terms.",
"expected_properties": [
"definition",
"simple explanation"
]
},
{
"id": "task-002",
"category": "coding",
"prompt": "Explain how a Python API works.",
"expected_properties": [
"Python",
"API",
"request",
"response"
]
}
]
This is a benchmark structure, not a claim about actual system performance.
6.20 Evaluation Runner
Create:
tests/evaluation_runner.py
Basic structure:
import json
from pathlib import Path
def load_benchmark():
path = Path(
"data/evaluation/benchmark.json"
)
return json.loads(
path.read_text(
encoding="utf-8"
)
)
def main():
benchmark = load_benchmark()
print(
f"Loaded {len(benchmark)} evaluation tasks."
)
if __name__ == "__main__":
main()
Run:
python tests/evaluation_runner.py
6.21 Metrics
The evaluation framework should measure at least:
Task Success
Did the system satisfy the requested task?
Relevance
Was the response related to the request?
Completeness
Did it contain the required components?
Verification Pass Rate
How many generated responses passed verification?
Revision Rate
How frequently did the system need a retry?
Latency
How long did a request take?
Cost
For paid models, how much did each request cost?
6.22 Example Evaluation Table
| Metric | Baseline | ACAI |
|---|
| Task Success | Measure | Measure |
| Relevance | Measure | Measure |
| Verification Pass Rate | Measure | Measure |
| Revision Rate | Measure | Measure |
| Latency | Measure | Measure |
| Cost | Measure | Measure |
Do not fill this table with invented numbers.
The purpose of the framework is to produce real measurements.
6.23 Ablation Study
The system should be tested incrementally.
Configuration A
Model only
Configuration B
Model
+
Planner
Configuration C
Model
+
Planner
+
Retrieval
Configuration D
Model
+
Planner
+
Retrieval
+
Memory
Configuration E
Model
+
Planner
+
Retrieval
+
Memory
+
Router
+
Verification
Compare all configurations.
This is important because adding more components does not automatically mean better performance.
6.24 Verification Failure Analysis
When verification fails, ACAI should record why.
For example:
Failure Type
────────────
Empty response
Too short
Missing context
Context mismatch
Task incomplete
Invalid format
Model error
Timeout
This allows developers to discover where the system actually fails.
6.25 Logging
Create a structured event:
verification_log = {
"passed": verification.passed,
"score": verification.score,
"reasons": verification.reasons,
"revision_count": attempt,
}
In production, logs should avoid unnecessarily storing sensitive user content.
A better design is:
Request ID
Timestamp
Model
Latency
Verification Score
Failure Category
Revision Count
rather than automatically storing complete conversations.
6.26 Monitoring Architecture
The future monitoring pipeline can look like:
Request
↓
Generation
↓
Verification
↓
Metrics
↓
Logging
↓
Monitoring Dashboard
Useful operational metrics:
Requests/minute
Error rate
Average latency
P95 latency
Verification failure rate
Revision rate
Model selection distribution
Token usage
Cost
6.27 Production Verification
A mature verification system can eventually include domain-specific validators.
For example:
Code
↓
Syntax Check
↓
Unit Tests
↓
Static Analysis
For structured JSON:
JSON
↓
Schema Validation
For retrieved research:
Answer
↓
Citation/Evidence Check
↓
Source Validation
For calculations:
Generated Result
↓
Independent Calculator
↓
Compare
This is much stronger than asking another model:
"Are you correct?"
6.28 Independent Verification
Whenever possible, verification should be independent from generation.
For example:
Generator
↓
Result
↓
Deterministic Validator
rather than:
Generator
↓
Same Generator
↓
"Is your answer correct?"
Independent checks reduce the risk of simply reproducing the same mistake.
6.29 Chapter 6 Complete Architecture
USER
│
▼
FastAPI API
│
▼
ACAI ORCHESTRATOR
│
┌────────────────────────┼────────────────────────┐
│ │ │
▼ ▼ ▼
PLANNER MEMORY RETRIEVAL
│ │ │
└────────────────────────┼────────────────────────┘
│
▼
MODEL ROUTER
│
▼
MODEL SERVICE
│
▼
GENERATE
│
▼
VERIFICATION
│
┌────────────┼────────────┐
▼ ▼ ▼
FORMAT CONTEXT QUALITY
CHECK CHECK CHECK
│ │ │
└────────────┼────────────┘
▼
DECISION ENGINE
│
┌───────────┼───────────┐
▼ ▼ ▼
ACCEPT REVISE REJECT
│ │
│ ▼
│ GENERATE
│ │
│ ▼
│ VERIFY AGAIN
│
▼
RESPONSE
6.30 Chapter 6 Success Criteria
Chapter 6 is complete when:
[✓] Verification service exists
[✓] Basic output validation works
[✓] Context consistency checking exists
[✓] Composite verification exists
[✓] Verification is integrated into the orchestrator
[✓] Failed generations can be revised
[✓] Revision attempts are bounded
[✓] Verification results are exposed through the API
[✓] Verification tests exist
[✓] Evaluation benchmark structure exists
[✓] Metrics are defined
[✓] Ablation methodology is defined
6.31 Current ACAI System
After Chapter 6:
ACAI
│
┌───────────┼───────────┐
▼ ▼ ▼
Planner Retrieval Memory
│ │ │
└───────────┼───────────┘
▼
Model Router
│
▼
Model Service
│
▼
Generation
│
▼
Verification
│
┌─────┴─────┐
▼ ▼
Accept Revise
│ │
└─────┬─────┘
▼
Response
6.32 What Comes Next?
The system can now plan, retrieve, remember, route, generate, and verify.
The next major engineering problem is orchestration at a larger scale.
ACAI needs to handle:
Multiple steps
Parallel tasks
Dependencies
Retries
Timeouts
Failures
Fallbacks
Long-running jobs
That leads to:
Chapter 7 — Workflow Orchestration and Agent Execution
The next layer will transform ACAI from a simple request-response pipeline into a system capable of executing multi-step workflows:
User Goal
↓
Planner
↓
Task Graph
↓
┌───────────────┐
│ Task A │
└──────┬────────┘
│
┌────┴────┐
▼ ▼
Task B Task C
│ │
└────┬────┘
▼
Task D
│
▼
Verification
│
▼
Result
End of Chapter 6
Comments
Post a Comment