# ACAI — Chapter 6: Verification and Evaluation Layer ![Post cover](https://dev-to-uploads.s3.us-east-2.amazonaws.com/uploads/articles/youwfq7w63h0t8op4ei3.png align="center") ## 6.1 Objective Chapters 1–5 established the main ACAI pipeline: ```text 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: * incomplete * inconsistent * unsupported by retrieved evidence * incorrectly formatted * technically invalid * based on irrelevant context Chapter 6 therefore introduces a **Verification and Evaluation Layer**. The basic pipeline becomes: ```text 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: ```text 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: ```text Accuracy Relevance Completeness Latency Cost Failure rate ``` Therefore: ```text Verification → Individual response Evaluation → System performance ``` * * * # 6.3 Chapter 6 Architecture ```text 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: ```text app/services/verification.py ``` Start with: ```python 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: ```text passed score reasons needs_revision ``` * * * # 6.5 Basic Output Validation The first verifier should perform deterministic checks. ```python 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: ```python 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. ```python 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: ```text Response length ``` does not prove correctness. Similarly: ```text Keyword overlap ``` does not prove factual accuracy. Therefore ACAI should use multiple independent signals. A future verification system can contain: ```text Format Check + Evidence Check + Consistency Check + Task Completion Check + Domain Validation ``` * * * # 6.9 Integrating Verification into the Orchestrator Update the Orchestrator: ```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 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: ```text 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: ```text Generate ↓ Verify ↓ Passed? ┌─┴─┐ YES NO │ │ ▼ ▼ Done Revise │ ▼ Generate ``` However, retries must be bounded. Otherwise the system could enter: ```text Generate ↓ Fail ↓ Generate ↓ Fail ↓ Generate ↓ Fail ... ``` Therefore use a maximum retry count. * * * # 6.12 Bounded Revision Add: ```python MAX_REVISIONS = 2 ``` Then conceptually: ```python 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: ```text Original request: Previous response: Verification problems: Produce a corrected response. ``` This creates: ```text Generation ↓ Verification ↓ Feedback ↓ Revision ↓ Verification ``` * * * # 6.14 Safe Revision Implementation Update the Orchestrator generation section: ```python 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: ```text Maximum attempts = 3 ``` * * * # 6.15 Verification Result The API can now expose: ```json { "verification": { "passed": true, "score": 1.0, "reasons": [], "needs_revision": false } } ``` Or, when problems occur: ```json { "verification": { "passed": false, "score": 0.5, "reasons": [ "Response is too short." ], "needs_revision": true } } ``` * * * # 6.16 API Schema Update Add: ```python class VerificationResponse(BaseModel): passed: bool score: float reasons: list[str] needs_revision: bool ``` Then include it in: ```python 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: ```text tests/test_verification.py ``` Add: ```python 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: ```powershell pytest ``` At this stage the project should test: ```text API Planner Retrieval Memory Router Verification ``` Conceptually: ```text 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: ```text data/evaluation/ benchmark.json ``` Example structure: ```json [ { "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: ```text tests/evaluation_runner.py ``` Basic structure: ```python 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: ```powershell 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 ```text Model only ``` ### Configuration B ```text Model + Planner ``` ### Configuration C ```text Model + Planner + Retrieval ``` ### Configuration D ```text Model + Planner + Retrieval + Memory ``` ### Configuration E ```text 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: ```text 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: ```python 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: ```text 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: ```text Request ↓ Generation ↓ Verification ↓ Metrics ↓ Logging ↓ Monitoring Dashboard ``` Useful operational metrics: ```text 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: ```text Code ↓ Syntax Check ↓ Unit Tests ↓ Static Analysis ``` For structured JSON: ```text JSON ↓ Schema Validation ``` For retrieved research: ```text Answer ↓ Citation/Evidence Check ↓ Source Validation ``` For calculations: ```text Generated Result ↓ Independent Calculator ↓ Compare ``` This is much stronger than asking another model: ```text "Are you correct?" ``` * * * # 6.28 Independent Verification Whenever possible, verification should be independent from generation. For example: ```text Generator ↓ Result ↓ Deterministic Validator ``` rather than: ```text Generator ↓ Same Generator ↓ "Is your answer correct?" ``` Independent checks reduce the risk of simply reproducing the same mistake. * * * # 6.29 Chapter 6 Complete Architecture ```text 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: ```text [✓] 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: ```text 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: ```text 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: ```text User Goal ↓ Planner ↓ Task Graph ↓ ┌───────────────┐ │ Task A │ └──────┬────────┘ │ ┌────┴────┐ ▼ ▼ Task B Task C │ │ └────┬────┘ ▼ Task D │ ▼ Verification │ ▼ Result ``` **End of Chapter 6**