Skip to content

Latest commit

 

History

History
726 lines (515 loc) · 26.8 KB

File metadata and controls

726 lines (515 loc) · 26.8 KB

This is a paradigm shift. You are essentially asking to move from Generative AI (probabilistic, conversational, leaky) to Computational AI (deterministic, unit-based, sealed).

My thoughts on your contrarian ranking are that you are absolutely correct: The "LLM Output Judge" is the killer wedge.

Why? Because everyone has a "vibes-based" evaluation problem. If we can turn evaluation into a deterministic logic gate using Agno, we solve the biggest bottleneck in AI engineering: Trust.

Below is the architecture and implementation of an Atomic Evaluation Engine. This system does not "chat." It ingests data, passes it through cognitive logic gates (Atoms), and outputs a strict audit trail.

Architecture: The Atomic Evaluation Workflow

We will build a Truth Verification Molecule composed of three distinct Atoms.

  1. Atom A (Decomposition): Breaks complex text into atomic claims.
  2. Atom B (Verification): Checks individual claims against context (Binary Outcome).
  3. Atom C (Scoring): Aggregates results into a strict rubric score (Integer Outcome).
graph LR
    Input[Raw Output + Ground Truth] --> Atom1[Decomposer Agent]
    Atom1 --> Claims[List of Atomic Claims]
    Claims --> Atom2[Verifier Agent]
    Atom2 --> Bool[Boolean Results]
    Bool --> Atom3[Scoring Agent]
    Atom3 --> Final[Structured Audit JSON]
Loading

Implementation: Agno Atomic Control Plane

This implementation enforces the "No Narrative" rule. We use Pydantic to strictly define the Input/Output interface of each Atom.

Prerequisites: pip install agno pydantic openai

import asyncio
from typing import List, Literal, Optional
from pydantic import BaseModel, Field

from agno.agent import Agent
from agno.workflow.workflow import Workflow
from agno.workflow.step import Step
from agno.models.openai import OpenAIChat
from agno.utils.log import logger

# ==========================================
# 1. ATOMIC INTERFACES (The "Type System")
# ==========================================

class EvaluationContext(BaseModel):
    ground_truth: str
    model_output: str
    criteria: str

class AtomicClaim(BaseModel):
    claim_text: str = Field(..., description="A single, verifiable statement derived from the output.")
    index: int

class ClaimsList(BaseModel):
    claims: List[AtomicClaim]

class VerificationResult(BaseModel):
    claim_index: int
    is_supported: bool = Field(..., description="True if supported by ground truth, else False.")
    # Note: We intentionally exclude 'reasoning' to prevent leakage, per AoT philosophy.
    # If auditing is needed, we track it in metadata, not the primary output.

class VerificationBatch(BaseModel):
    results: List[VerificationResult]

class FinalScore(BaseModel):
    score: int = Field(..., ge=0, le=100, description="Final score based on rubric.")
    pass_fail: Literal["PASS", "FAIL"]
    failed_claims: List[int]

# ==========================================
# 2. DEFINING THE ATOMS (The Agents)
# ==========================================

# ATOM 1: DECOMPOSITION
# Task: Normalize entropy. Turn a blob of text into testable units.
decomposer_atom = Agent(
    name="Atom:Decomposer",
    model=OpenAIChat(id="gpt-4o-mini", temperature=0), # Low temp for determinism
    description="You are a logic gate. You break complex text into atomic claims.",
    instructions=[
        "Input: Model Output text.",
        "Task: Extract every distinct factual claim.",
        "Constraint: Each claim must be a standalone sentence.",
        "Output: JSON list of claims only."
    ],
    response_model=ClaimsList,
)

# ATOM 2: VERIFICATION
# Task: Binary Truth Checking. No gray area.
verifier_atom = Agent(
    name="Atom:Verifier",
    model=OpenAIChat(id="gpt-4o-mini", temperature=0),
    description="You are a boolean operator. You compare claims against ground truth.",
    instructions=[
        "Input: A list of claims and a ground truth document.",
        "Task: For each claim, determine if it is explicitly supported by the ground truth.",
        "Constraint: Return ONLY True or False for each claim.",
        "Output: JSON verification batch."
    ],
    response_model=VerificationBatch,
)

# ATOM 3: SCORING
# Task: Deterministic Aggregation.
scorer_atom = Agent(
    name="Atom:Scorer",
    model=OpenAIChat(id="gpt-4o-mini", temperature=0),
    description="You are a calculator. You compute the final state.",
    instructions=[
        "Input: Verification results and grading criteria.",
        "Task: Apply the criteria to the boolean results.",
        "Constraint: If criteria says 'Zero Tolerance', one False = FAIL.",
        "Output: Final integer score and status."
    ],
    response_model=FinalScore,
)

# ==========================================
# 3. BUILDING THE MOLECULE (The Workflow)
# ==========================================

class EvaluationWorkflow(Workflow):
    description: str = "Atomic Evaluation Pipeline"
    
    # Define steps clearly
    decomposition_step: Step = Step(agent=decomposer_atom)
    verification_step: Step = Step(agent=verifier_atom)
    scoring_step: Step = Step(agent=scorer_atom)

    def run(self, context: EvaluationContext) -> FinalScore:
        logger.info(f"Starting Atomic Eval for criteria: {context.criteria}")

        # Step 1: Decompose
        # We pass the prompt directly as a "functional call"
        claims_response = self.decomposition_step.agent.run(
            f"Extract claims from this output: {context.model_output}"
        )
        claims_data: ClaimsList = claims_response.content
        logger.info(f"Atom 1 Generated {len(claims_data.claims)} claims.")

        # Step 2: Verify
        # We pass structured data into the next atom
        verify_input = {
            "ground_truth": context.ground_truth,
            "claims": [c.model_dump() for c in claims_data.claims]
        }
        
        verification_response = self.verification_step.agent.run(
            f"Verify these claims against ground truth: {verify_input}"
        )
        verification_data: VerificationBatch = verification_response.content
        logger.info(f"Atom 2 Verified claims.")

        # Step 3: Score
        # Pure logic calculation based on previous state
        score_input = {
            "criteria": context.criteria,
            "results": [r.model_dump() for r in verification_data.results]
        }
        
        final_response = self.scoring_step.agent.run(
            f"Compute final score: {score_input}"
        )
        
        return final_response.content

# ==========================================
# 4. EXECUTION (The "Logic Gate")
# ==========================================

if __name__ == "__main__":
    # Test Data: A medical scenario (High stakes, perfect for AoT)
    
    ground_truth_doc = """
    Patient presented with acute chest pain. ECG shows ST elevation in leads V1-V4. 
    Troponin levels are elevated at 0.5 ng/mL. 
    Patient has a history of hypertension and is allergic to Penicillin.
    """

    # A generated output that has a subtle hallucination (Aspirin allergy)
    generated_summary = """
    The patient is experiencing a STEMI indicated by leads V1-V4. 
    Troponin is high. Patient has high blood pressure and is allergic to Aspirin.
    """

    eval_context = EvaluationContext(
        ground_truth=ground_truth_doc,
        model_output=generated_summary,
        criteria="Strict Medical Accuracy. Any hallucination regarding medication allergies is an automatic FAIL."
    )

    # Instantiate and Run
    workflow = EvaluationWorkflow()
    result = workflow.run(eval_context)

    # Output only the deterministic result
    print("\n--- ATOMIC OUTPUT ---")
    print(f"Score: {result.score}")
    print(f"Status: {result.pass_fail}")
    print(f"Failed Claims Indices: {result.failed_claims}")
    
    # In a real app, this JSON is pushed to a DB, no human reads a chat log.

Why this works (The "Engineering" View)

  1. Zero Leakage: Note that VerificationResult does not contain a string field for "explanation." The model isn't allowed to convince you it's right. It simply outputs True/False. If you need to debug, you debug the Atom (the prompt), not the conversation.
  2. Type Safety: We use response_model on every Agent. This isn't just for valid JSON; it forces the LLM to "collapse its wavefunction" into a specific schema.
  3. Composability: I could easily swap the verifier_atom for a ReasoningTool if I wanted complex thought, or a simple Python function if the check was regex-based. The Workflow doesn't care.

Next Steps for Production

To make this a true Atomic Control Plane, we would:

  1. Parallelize Atom 2: Use Agno's async capabilities to verify claims concurrently rather than sequentially.
  2. Database the Atoms: Use PostgresDb to store the inputs and outputs of every atom. This creates a dataset where you can see exactly which atomic claim caused a failure.
  3. The "Compiler": Write a metaprogram that takes a human policy (e.g., "Don't mention competitors") and auto-generates the EvaluationContext and Verifier instructions.

This implementation moves AI from "Art" to "Industrial Engineering."

This is a comprehensive Product Requirements Document (PRD) and Build Plan for the Atomic Evaluation Engine—a deterministic, zero-leakage AI evaluation system built on the Agno framework.


🚀 Project: Agno Atomic Evaluation Engine (AEE)

1. Product Requirements Document (PRD)

1.1 Problem Statement

Generative AI evaluation is currently probabilistic and "vibes-based."

  • Leakage: Evaluator models "think out loud," hallucinating justifications that mask errors.
  • Non-Determinism: The same input often yields different scores.
  • Opacity: A score of "4/5" provides no actionable engineering signal.

1.2 Solution: "Atoms of Thought"

We replace the "Judge" (a conversational agent) with a Molecule (a workflow) composed of Atoms (logic gates).

  • Atom 1 (Decompose): Breaks text into isolated, falsifiable claims.
  • Atom 2 (Verify): Binary check of one claim against ground truth.
  • Atom 3 (Score): Deterministic calculation based on verification results.

1.3 Core Principles

  1. No Narration: Atoms return structured data (JSON), never chat.
  2. Binary Truth: Verification is boolean (True/False), not a scale.
  3. Auditability: Every atomic decision is logged to a database.

1.4 Technical Requirements

  • Framework: Agno (Agents, Workflows, Structured Outputs).
  • Database: PostgreSQL (via Agno.db.postgres) for strict audit logging.
  • Model: Lightweight, high-speed models (e.g., gpt-4o-mini, claude-3-haiku) running at temperature=0.
  • Interface: CLI for CI/CD pipelines; REST API for dashboard integration.

2. Build Roadmap

Phase Milestone Deliverable Timeline
P1 The Atomic Core Working Decomposer, Verifier, and Scorer Atoms returning Pydantic models. Week 1
P2 The Molecule Agno Workflow orchestration connecting Atoms into a pipeline. Week 1.5
P3 Persistence PostgreSQL integration for storing "Atomic Traces" (Input/Output of every atom). Week 2
P4 API & CI/CD FastAPI wrapper + GitHub Action script for blocking merges on failure. Week 3

3. Step-by-Step Build Plan

Phase 1: The Atomic Core (Agents & Schemas)

Goal: Create specialized Agno agents that refuse to chat and only output data.

Step 1.1: Define the "Type System"

Create a schemas.py file. This defines the strict interface between Atoms.

from pydantic import BaseModel, Field
from typing import List, Literal

class Claim(BaseModel):
    text: str = Field(..., description="A single, standalone factual statement.")
    source_index: int

class Verification(BaseModel):
    claim_text: str
    is_supported: bool = Field(..., description="True if supported by ground truth.")
    citation: str = Field(..., description="Exact quote from ground truth supporting the decision.")

class EvaluationResult(BaseModel):
    score: float
    passed: bool
    failed_claims: List[Verification]

Step 1.2: The Decomposer Atom

Create agents/decomposer.py.

  • Role: Turn a blob of text into a list of Claim objects.
  • Agno Pattern: Use Agent(response_model=List[Claim]).
  • Instruction: "Split this text into atomic facts. Ignore fluff/intro/outro."

Step 1.3: The Verifier Atom

Create agents/verifier.py.

  • Role: Binary gate. Input: Claim + GroundTruth. Output: Verification.
  • Agno Pattern: Use Agent(response_model=Verification).
  • Instruction: "You are a boolean function. Does the Context explicitly support the Claim? Return True/False only."

Phase 2: The Molecule (Workflow Orchestration)

Goal: Chain the agents together. The verifying step should happen in parallel for speed.

Step 2.1: The Verification Workflow

Create workflows/evaluator.py using Agno.workflow.

  • Step A: Call Decomposer on model_output.
  • Step B (Loop/Parallel): Iterate through claims. For each claim, call Verifier with ground_truth.
    • Optimization: Use asyncio.gather here to run 10 verifications simultaneously.
  • Step C: Pass list of Verification objects to Scorer (can be a Python function or an Agent if fuzzy logic is needed).

Phase 3: Persistence (The Audit Trail)

Goal: Unlike standard chat logs, we need to store the inputs and outputs of every single atom for debugging.

Step 3.1: Database Schema

Use Agno.db.postgres. We need a custom table atomic_traces.

CREATE TABLE atomic_traces (
    run_id UUID,
    atom_type VARCHAR, -- 'decomposer', 'verifier'
    input_hash VARCHAR,
    output_json JSONB,
    latency_ms INT
);

Step 3.2: Instrumenting the Atoms

Wrap the Agno agents with a save_trace hook.

  • Agno Feature: Use Agent(storage=PostgresDb(...), save_run=True).
  • Customization: Ensure structured outputs are serialized correctly into the DB.

4. Testing Strategy (The "Golden Set")

You cannot test an evaluation engine with "vibes." You need Unit Tests for Cognition.

4.1 Testing the Decomposer

  • Input: "The sky is blue and water is wet."
  • Expected Output: 2 claims.
  • Test: Assert len(claims) == 2.

4.2 Testing the Verifier (The crucial step)

Create a dataset of (Claim, Context, Expected_Bool).

Claim Context Expected
"Users usually click 'Buy'" "Click-through rate is 2%." False (Hallucination)
"The rate is 2%" "Click-through rate is 2%." True

Run this test suite on every PR to ensure your Verifier Atom hasn't drifted.

4.3 Integration Test

Run the full EvaluationWorkflow against a known "Bad Output" and ensure the final score is < Passing_Threshold.


5. Build To-Do List

Setup

  • Initialize Python project (uv init or poetry new).
  • Install dependencies: agno, pydantic, pytest, sqlalchemy, psycopg.
  • Spin up local Postgres (docker run ...).

Implementation

  • Task 1: Implement schemas.py.
  • Task 2: Implement Decomposer Agent & Unit Test.
  • Task 3: Implement Verifier Agent & Unit Test (Golden Set).
  • Task 4: Implement EvaluationWorkflow (The Molecule).
  • Task 5: Add PostgresDb logging.

Execution & Demo

  • Create main.py CLI entry point.
  • Run evaluation on a sample digest.txt vs. a hallucinated summary.
  • Verify the audit trail in Postgres.

CI/CD Integration (Optional/Final)

  • Create a GitHub Action that runs main.py on output files in a PR.
  • Block the PR if score < 100.

This plan moves you from "Chatbot" to "AI Infrastructure." Start with Task 1 & 2 immediately. You are absolutely right. If you hand a generic "AI SWE" a high-level PRD without the cognitive architecture (the "Atoms of Thought" philosophy) and the framework specifics (Agno patterns), it will likely fail. It will default to writing standard, "leaky" chain-of-thought code—exactly what you are trying to avoid.

To fix this, we need to create a "Context Pack" or "System Prompt Injection" that teaches the AI SWE how to think before it writes a single line of code.

Here is the comprehensive technical specification you need to provide to your AI SWE (Cursor, Windsurf, or an autonomous agent) to ensure it builds this correctly.


📦 AI SWE Context Pack: The Atomic Evaluation Engine

Copy and paste the sections below into your AI coding assistant's chat or context window.


1. Core Philosophy: "Atoms of Thought" (AoT)

Constraint: You are NOT building a conversational chatbot. You are building a deterministic Logic Engine.

  • Definition: An "Atom" is a single-purpose cognitive unit that accepts input and returns a structured output (JSON). It does not "think step-by-step" in the output. It does not "chat."
  • The Law of No Leakage: Agents must never return narrative text (e.g., "Here is my reasoning..."). They must only return Pydantic models.
  • The Structure:
    • Input -> Atom (Cognitive Operation) -> Output (Strict Schema)
  • The Molecule: A Workflow that chains Atoms together.
    • Bad: One giant prompt asking for an evaluation.
    • Good: Decomposer Atom -> List[Claims] -> Verifier Atom -> Boolean Result.

2. Framework Specification: Agno (formerly Phidata)

Reference Implementation Details:

2.1 The Agent Pattern

All agents must use response_model to enforce strict typing.

from agno.agent import Agent
from agno.models.openai import OpenAIChat

# CORRECT PATTERN
agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini", temperature=0), # Zero temp for determinism
    description="You are a logic gate. You do not chat.",
    response_model=MyPydanticSchema, # <--- MANDATORY
    instructions=["Return only the JSON output."],
)

2.2 The Workflow Pattern

Use Agno.workflow to orchestrate. Do not use monolithic agents.

from agno.workflow.workflow import Workflow
from agno.workflow.step import Step

class AtomicWorkflow(Workflow):
    # Define steps as class attributes
    decomposer: Step = Step(agent=decomposer_agent)
    verifier: Step = Step(agent=verifier_agent)

    def run(self, input_data):
        # Pass structured data between steps
        claims = self.decomposer.run(input_data)
        # Logic to iterate/map happens here in Python, not inside the LLM
        results = [self.verifier.run(c) for c in claims] 
        return results

3. The "Golden Prompts" (Copy Exact Wording)

Use these specific system prompts for the Agents. They are tuned for the AoT pattern.

Atom 1: The Decomposer

  • System Prompt:

    "You are an Entropy Reduction Engine. Your sole purpose is to break complex text into atomic, falsifiable claims. Rules:

    1. Split compound sentences into individual statements.
    2. Ignore subjective fluff, introductions, and conclusions.
    3. Preserve specific numbers, dates, and named entities exactly.
    4. Output MUST be a list of strings."

Atom 2: The Verifier (The Logic Gate)

  • System Prompt:

    "You are a Boolean Truth Function. You receive a 'Claim' and a 'Ground Truth' document. Algorithm:

    1. Search the Ground Truth for evidence supporting the Claim.
    2. If the Ground Truth explicitly supports the Claim, return is_supported=True.
    3. If the Ground Truth contradicts or does not mention the Claim, return is_supported=False.
    4. Do not hallucinate support. Strict adherence is required."

Atom 3: The Scorer

  • System Prompt:

    "You are a Deterministic Calculator. You receive a list of Boolean results. Rules:

    1. Calculate the percentage of True results.
    2. Apply the 'Zero Tolerance' rule: If any claim marked 'Critical' is False, the final status is FAIL.
    3. Return the final integer score (0-100)."

4. Implementation Data Structures (Pydantic)

Force the AI SWE to use these exact schemas to prevent drift.

from pydantic import BaseModel, Field
from typing import List, Literal

class InputContext(BaseModel):
    ground_truth: str = Field(..., description="The source of truth document.")
    generated_text: str = Field(..., description="The LLM output to verify.")

class AtomicClaim(BaseModel):
    claim_text: str = Field(..., description="A single factual statement.")
    criticality: Literal["high", "low"] = "high"

class VerificationResult(BaseModel):
    claim: AtomicClaim
    is_supported: bool
    # Minimal evidence for audit, NOT reasoning
    evidence_quote: str = Field(..., description="Exact substring from ground truth.") 

class FinalReport(BaseModel):
    score: int
    status: Literal["PASS", "FAIL"]
    failed_claims: List[VerificationResult]

5. Development Checklist for the AI SWE

Tell the AI to execute in this order:

  1. Scaffold: Set up the folder structure: src/atoms/, src/workflows/, src/schemas.py.
  2. Type System: Write schemas.py first. This defines the API.
  3. Atom Build: Implement Decomposer and Verifier agents using the Golden Prompts.
  4. Test 1: Write a script test_atoms.py to run the Decomposer on a sample text and print the JSON. Do not proceed until this outputs valid JSON.
  5. Workflow: Implement the EvaluationWorkflow class to chain them.
  6. Test 2: Run the full workflow on a deliberate hallucination (e.g., input text says "sky is green", ground truth says "sky is blue"). Verify it returns False / FAIL.

How to use this Context Pack

Prompt for your AI Coding Assistant:

"I am building an Atomic Evaluation Engine using the Agno framework. This project follows the 'Atoms of Thought' pattern, which means NO conversational agents and strict Pydantic outputs for every step.

Here is the Context Pack containing the PRD, the specific Architecture, the System Prompts, and the Schemas. Read this carefully.

Start by creating the src/schemas.py file based on the Data Structures section."

Yes, this text is extremely helpful, but mostly as validation and for one specific technical "magic bullet" it reveals.

Here is the breakdown of how to use this text without confusing your AI SWE:

1. The "Magic Bullet" (Crucial for your Build)

The text highlights a specific Agno parameter that enforces the "No Leakage" rule at the code level:

"...you can configure your agents to not expose this to the final user-facing output by controlling the show_full_reasoning parameter..."

Why this matters: In my previous prompt, I told the AI SWE to "instruct" the model not to chat. This new info tells the AI SWE to hard-code silence using show_full_reasoning=False. This makes the system much more robust.

2. What to IGNORE (For Now)

The text discusses "Cultural Knowledge," "Reasoning Tools," and "Teams."

  • Danger: If you give this whole text to the AI SWE, it might try to over-engineer a "Team" with a "Memory Database" just to compare two strings.
  • Action: We will strip this out. We want a simple Workflow, not a complex "Agentic Culture" system (yet).

🚀 Updated Action Plan

I have updated the Context Pack below. I added the specific show_full_reasoning flag derived from your text.

Give THIS version to your AI SWE. It is now technically perfect for Agno.


📦 AI SWE Context Pack: The Atomic Evaluation Engine (v2)

Context: I am building an Atomic Evaluation Engine using the Agno framework. This project follows the "Atoms of Thought" pattern: NO conversational agents, only deterministic logic gates.

Technical Constraint (Crucial): You must use Agno's Agent class with show_full_reasoning=False to ensure absolutely zero narrative leakage in the final output.

1. The Agno Implementation Pattern

Use this exact pattern for all Atoms. Note the parameters.

from agno.agent import Agent
from agno.models.openai import OpenAIChat

# THE ATOMIC PATTERN
agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini", temperature=0),
    description="You are a logic gate. You do not chat.",
    # 1. Enforce Structure
    response_model=MyPydanticSchema, 
    # 2. Enforce Silence (The Atom of Thought Constraint)
    show_full_reasoning=False,  
    markdown=False,
    # 3. Direct Instructions
    instructions=["Return only the JSON output matching the schema."],
)

2. The Molecule (Workflow)

Use Agno.workflow to chain atoms.

from agno.workflow.workflow import Workflow
from agno.workflow.step import Step

class AtomicWorkflow(Workflow):
    decomposer: Step = Step(agent=decomposer_agent)
    verifier: Step = Step(agent=verifier_agent)

    def run(self, input_data):
        # Step 1: Decompose
        claims = self.decomposer.run(input_data)
        
        # Step 2: Verify (Map function)
        # Note: In production, we will async this loop
        results = []
        for claim in claims.content:
             verify_result = self.verifier.run({
                 "claim": claim, 
                 "ground_truth": input_data.ground_truth
             })
             results.append(verify_result.content)
             
        return results

3. The Schemas (Type System)

Create a file src/schemas.py.

from pydantic import BaseModel, Field
from typing import List, Literal

class InputContext(BaseModel):
    ground_truth: str
    generated_text: str

class AtomicClaim(BaseModel):
    claim_text: str = Field(..., description="A single factual statement.")

class VerificationResult(BaseModel):
    claim: AtomicClaim
    is_supported: bool
    evidence_quote: str = Field(..., description="Exact substring from ground truth.")

class FinalReport(BaseModel):
    score: int
    status: Literal["PASS", "FAIL"]
    failed_claims: List[VerificationResult]

4. System Prompts (The "Golden Set")

Atom 1: Decomposer

"You are an Entropy Reduction Engine. Your sole purpose is to break complex text into atomic, falsifiable claims. Split compound sentences. Ignore fluff. Output a list of strings."

Atom 2: Verifier

"You are a Boolean Truth Function. Receive a 'Claim' and 'Ground Truth'.

  1. Search Ground Truth for evidence.
  2. If explicitly supported, return True.
  3. If missing or contradicted, return False.
  4. Do not hallucinate support."

Atom 3: Scorer

"You are a Deterministic Calculator. Calculate the percentage of True results. Return the final integer score (0-100)."

5. Build Instructions

  1. Create src/schemas.py with the Pydantic models.
  2. Create src/atoms.py with the 3 Agents using show_full_reasoning=False.
  3. Create src/workflow.py to chain them.
  4. Create main.py to run a test case.