Skip to content

Repository files navigation

Atoms of Thought: Atomic Evaluation Engine

A deterministic, zero-leakage AI evaluation system built on the "Atoms of Thought" (AoT) philosophy using the Agno framework.

Overview

This system replaces probabilistic "vibes-based" AI evaluation with deterministic logic gates (Atoms) composed into workflows (Molecules) that produce structured, auditable outputs without narrative reasoning.

Core Concept: Break complex evaluation into three atomic operations:

  1. Decomposer - Extract atomic claims from generated text
  2. Verifier - Boolean check: is each claim supported by ground truth?
  3. Scorer - Deterministic calculation of final score with pass/fail status

Quick Start

Prerequisites

  • Python 3.11+
  • OpenRouter API key (free models available)

Installation

# Clone or navigate to repository
cd AtomsOfThought

# Install dependencies
uv sync --extra dev

# Configure API key
cp .env.example .env
# Edit .env and add your OPENROUTER_API_KEY
# Get free API key at: https://openrouter.ai/

Option 1: Run API Server (Recommended)

# Start verification API server
uv run python src/api/server.py

# In another terminal, run examples
uv run python examples/api_client_example.py

# Or use curl
curl -X POST http://localhost:8000/v1/verify \
  -H "Content-Type: application/json" \
  -H "X-API-Key: dev-key-12345" \
  -d '{
    "agent_output": "Patient is allergic to Aspirin.",
    "ground_truth": "Patient has allergy to Penicillin.",
    "criteria": "Medical accuracy"
  }'

# Interactive API docs at http://localhost:8000/docs

Option 2: Run Demo Script

uv run python examples/demo.py

This runs the medical hallucination detection example from the architecture notes.

Run Tests

# Schema validation tests (no API key needed)
uv run pytest tests/test_schemas.py -v

# Scorer unit tests (no API key needed)
uv run pytest tests/atoms/test_scorer.py -v

# All tests including integration (requires API key)
uv run pytest -v

# With coverage
uv run pytest --cov=src --cov-report=html

Usage

REST API (Production - Phase 1 ✅)

Use the verification API to add hallucination detection to any AI agent:

import requests

response = requests.post(
    "http://localhost:8000/v1/verify",
    headers={"X-API-Key": "dev-key-12345"},
    json={
        "agent_output": "Patient is allergic to Aspirin.",
        "ground_truth": "Patient has allergy to Penicillin.",
        "criteria": "Medical accuracy - zero tolerance for medication errors"
    }
)

result = response.json()
if result["status"] == "FAIL":
    print(f"⚠️  Hallucination detected! Don't use agent output.")
    for claim in result["failed_claims"]:
        print(f"  - {claim['claim_text']}")

See: docs/API_REFERENCE.md for complete API documentation

Python Library (Direct Integration)

from src.workflows.evaluator import evaluate

result = evaluate(
    generated_text="Patient is allergic to Aspirin.",
    ground_truth="Patient has allergy to Penicillin.",
    criteria="Medical accuracy - zero tolerance for medication errors",
    criticality="high"
)

print(f"Score: {result.score}/100")
print(f"Status: {result.status}")

for failed in result.failed_claims:
    print(f"Failed: {failed.claim.claim_text}")

Async/Parallel Verification (Phase 4 ✅)

from src.workflows.evaluator_async import evaluate_async_sync

# 2.7x faster for multi-claim verification
result = evaluate_async_sync(
    generated_text="Multiple claims to verify...",
    ground_truth="Source of truth...",
    criteria="Strict accuracy",
    max_concurrency=5  # Verify 5 claims in parallel
)

Architecture

The Atom Pattern

Each atom is a single-purpose cognitive unit:

  • One purpose: Single, well-defined transformation
  • Explicit inputs: Strict Pydantic input schema
  • One verifiable output: Strict Pydantic output schema
  • No narrative reasoning: Internal deliberation never exposed

Critical Parameters

All Agno agents use these parameters to prevent narrative leakage:

from agno.models.openrouter import OpenRouter

agent = Agent(
    model=OpenRouter(id="mistralai/mistral-large-2512", temperature=0),  # Determinism via OpenRouter
    response_model=MyPydanticSchema,                                      # Structured output
    show_full_reasoning=False,                                            # No leakage
    markdown=False,                                                       # No formatting
    instructions=[...]
)

Current Model: mistralai/mistral-large-2512 via OpenRouter

  • Cost: Free (promotional pricing, Dec 2025)
  • Context: 262K tokens
  • Constraint: Only models <$2/M output tokens used

Workflow Pipeline

InputContext
    ↓
Decomposer Atom → ClaimsList
    ↓
Verifier Atom → List[VerificationResult]
    ↓
Scorer Function → FinalScore

Project Structure

AtomsOfThought/
├── src/
│   ├── schemas.py           # Pydantic type system
│   ├── atoms/
│   │   ├── decomposer.py    # Atom 1: Claim extraction
│   │   ├── verifier.py      # Atom 2: Truth verification
│   │   ├── verifier_async.py  # Async parallel verification
│   │   └── scorer.py        # Atom 3: Score calculation
│   ├── workflows/
│   │   ├── evaluator.py     # Workflow orchestration
│   │   └── evaluator_async.py  # Async workflow (2.7x faster)
│   ├── api/                 # Phase 1: REST API ✅
│   │   ├── server.py        # FastAPI application
│   │   ├── models.py        # API request/response models
│   │   ├── routes/          # API endpoints
│   │   └── middleware/      # Auth & rate limiting
│   └── db/                  # Phase 3: Audit & Persistence
│       ├── client.py        # Database connection
│       ├── models.py        # SQLAlchemy ORM models
│       ├── trace_logger.py  # Trace logging service
│       └── queries.py       # Query interface
├── tests/
│   ├── test_schemas.py      # Schema validation
│   ├── atoms/               # Atom unit tests
│   ├── workflows/           # Integration tests
│   ├── api/                 # API integration tests ✅
│   ├── db/                  # Database integration tests
│   └── golden_set/          # Regression test cases
├── scripts/
│   ├── init_db.py           # Database initialization
│   └── check_openrouter_models.py  # Model pricing verification
├── examples/
│   ├── demo.py              # Medical hallucination demo
│   ├── benchmark_async.py   # Performance benchmarks
│   └── api_client_example.py  # API integration examples ✅
└── docs/
    ├── phase3_design.md     # Audit & persistence design
    ├── phase4_design.md     # Production features design
    └── API_REFERENCE.md     # Complete API documentation ✅

Testing Strategy

Unit Tests

  • Schemas: Pydantic validation (18 tests, no API key needed)
  • Scorer: Pure Python logic (13 tests, no API key needed)
  • Database: Trace logging and queries (10 tests, no API key needed)

Integration Tests

  • Decomposer: Claim extraction (requires API key)
  • Verifier: Golden set regression (requires API key)
  • Workflow: End-to-end pipeline (requires API key)

Total Tests: 108 (99% passing - Phase 1 complete)

Golden Set Testing

Regression tests prevent atom drift:

  • hallucination_cases.json - Claims that should fail verification
  • supported_cases.json - Claims that should pass verification
  • edge_cases.json - Boundary conditions

Target: 100% pass rate on golden set (zero tolerance for drift)

Development

Running Specific Tests

# Schema tests only
uv run pytest tests/test_schemas.py

# Scorer tests only
uv run pytest tests/atoms/test_scorer.py

# Golden set regression
uv run pytest tests/golden_set/ -v

# Integration tests
uv run pytest tests/workflows/ -v

# Quiet mode (for CI)
uv run pytest -q

Adding to Golden Set

  1. Add test case to appropriate JSON file:

    • tests/golden_set/hallucination_cases.json
    • tests/golden_set/supported_cases.json
    • tests/golden_set/edge_cases.json
  2. Run tests to verify:

    uv run pytest tests/golden_set/test_verifier_golden_set.py -v

Commit Guidelines

Per AGENTS.md:

  • Format: Imperative present tense with scope (atoms: add verifier schema)
  • Run uv run pytest -q before committing
  • Include before/after artifacts for agent behavior changes

Core Principles

  1. No Leakage: Agents return structured data only, never narrative
  2. Binary Truth: Verification is boolean (True/False), not probabilistic
  3. Composability: Complex tasks decompose into single-purpose Atoms
  4. Auditability: Every atomic decision logged with input/output pairs

Phase 3: Audit & Persistence (✅ Complete)

The system now includes persistent audit trails for all atomic operations.

Database Setup

# Initialize database (creates tables)
uv run python scripts/init_db.py

# Database URL is configurable in .env
# Default: sqlite:///./atomic_traces.db (zero setup)
# Production: postgresql://user:pass@host/db

Querying Traces

from src.db.queries import TraceQueries, RunQueries

# Get all traces for a specific run
traces = TraceQueries.get_run_traces(run_id)

# Get recent evaluation runs
runs = RunQueries.get_evaluation_runs(limit=10)

# Get failed traces for debugging
failed = TraceQueries.get_failed_traces(limit=10)

# Calculate pass rate over last 7 days
pass_rate = RunQueries.get_pass_rate(days=7)

Features

  • Atomic trace logging: Every atom execution logged with input/output
  • Run tracking: High-level metadata for complete workflows
  • Query interface: Rich API for debugging and analysis
  • Deduplication: Input hashing prevents redundant processing
  • Flexible database: SQLite (dev) or PostgreSQL (prod)

See PHASE3_COMPLETE.md for complete documentation.

Phase Completion Status

✅ Phase 1: Core API (Complete)

  • FastAPI REST endpoints
  • API key authentication
  • Rate limiting (100 req/min)
  • Complete API documentation
  • 16 integration tests passing

See: PHASE1_COMPLETE.md

✅ Phase 3: Audit & Persistence (Complete)

  • PostgreSQL/SQLite audit trail
  • Atomic trace logging
  • Query interface
  • Deduplication
  • 10 database tests passing

See: PHASE3_COMPLETE.md

✅ Phase 4.1: Async Verification (Complete)

  • Parallel claim verification
  • 2.7x-5x performance speedup
  • Async workflow with tracing
  • 10 async tests passing

See: docs/phase4_design.md

⏸️ Phase 2: Async & Batch API (Planned)

  • Job queue for async verification
  • Batch endpoint (100 verifications)
  • Job status polling
  • Redis integration

⏸️ Phase 4.2: Framework Adapters (Planned)

  • LangChain adapter
  • CrewAI adapter
  • AutoGPT adapter
  • Python SDK library

⏸️ Phase 5: Production Hardening (Planned)

  • Security audit
  • Load testing (10K req/min)
  • Monitoring & alerting
  • CI/CD pipeline

Known Limitations

  • LLM non-determinism: Even at temp=0, outputs may vary slightly
  • Golden set coverage: 18 test cases may not cover all edge cases
  • Framework adapters: Not yet built (manual integration required)

References

  • Architecture: See notes1.md for original AoT design
  • Agno Integration: See notes1a.md for framework alignment
  • Guidelines: See AGENTS.md for development rules
  • Claude Code Guide: See CLAUDE.md for AI assistant instructions

License

See project root for license information.

About

verification as a service

Topics

Resources

Stars

0 stars

Watchers

0 watching

Forks

Releases

Packages

Contributors

Languages