cla-ai 1.0.0

Python implementation of Clean Layer Architecture for AI systems

CLA — Clean Layer Architecture Python Library

Based on "Clean Layer Architecture: A Modular Approach to Artificial Intelligence Systems" by Dr. Mohamed Sarhan (2026)

A full Python implementation of the CLA paradigm — modular, production-grade, zero hardcoded responses. All AI output comes from the model backend you provide.

---

Core Design Principle

The library never generates hardcoded responses. Every classification, every answer, every safety evaluation flows through the LLM backend you configure. You own the model; the library provides the architecture.

---

Supported Backends

Backend	Package	Auth
AnthropicBackend	pip install anthropic	ANTHROPIC_API_KEY env var or api_key=
OpenAIBackend	pip install openai	OPENAI_API_KEY env var or api_key=
OllamaBackend	none (uses stdlib)	local Ollama instance
HuggingFaceBackend	pip install transformers torch	local model or HF token
CustomBackend	none	any callable(prompt) -> str

All backends are interchangeable — swap them without changing any other code.

---

Installation

pip install .                                # core (no LLM dependencies)
pip install ".[anthropic]"                   # + Anthropic SDK
pip install ".[openai]"                      # + OpenAI SDK
pip install ".[huggingface]"                 # + Transformers + PyTorch
pip install ".[full]"                        # everything

---

Quick Start

from cla import CLASystem
from cla.backends import AnthropicBackend   # or OpenAI, Ollama, Custom

# ── Configure your backend ──────────────────────────────────────
backend = AnthropicBackend()                  # reads ANTHROPIC_API_KEY
# backend = AnthropicBackend(api_key="sk-ant-...", model="claude-sonnet-4-6")
# backend = OpenAIBackend(api_key="sk-...", model="gpt-4o-mini")
# backend = OllamaBackend(model="llama3.2")
# backend = CustomBackend(my_function)

# ── Create the system ───────────────────────────────────────────
system = CLASystem(backend=backend)

# ── Add knowledge (comes from YOU, not hardcoded) ───────────────
system.add_document("Physics", "Water boils at 100°C at standard pressure.")
system.add_fact("creator of Python", "Guido van Rossum", confidence=0.99)
system.add_triple("python", "created_by", "guido van rossum")

# ── Dispatch ─────────────────────────────────────────────────────
response = system.dispatch("Who created Python?")
print(response.text)

# ── Inspect the Structured Inference Trace ───────────────────────
print(response.trace.summary())

---

Backends in Detail

Anthropic

from cla.backends import AnthropicBackend

backend = AnthropicBackend()                          # ANTHROPIC_API_KEY env var
backend = AnthropicBackend(api_key="sk-ant-...")      # explicit key
backend = AnthropicBackend(model="claude-opus-4-6")   # choose model

OpenAI & Compatible APIs

from cla.backends import OpenAIBackend

backend = OpenAIBackend()                              # OPENAI_API_KEY env var
backend = OpenAIBackend(model="gpt-4o")

# Together AI
backend = OpenAIBackend(
    api_key="your-together-key",
    base_url="https://api.together.xyz/v1",
    model="meta-llama/Llama-3-8b-chat-hf",
)
# Groq
backend = OpenAIBackend(
    api_key="your-groq-key",
    base_url="https://api.groq.com/openai/v1",
    model="llama3-8b-8192",
)
# Azure OpenAI
backend = OpenAIBackend(
    api_key="your-azure-key",
    base_url="https://your-resource.openai.azure.com/",
    model="gpt-4o",
)
# LM Studio (local, no key)
backend = OpenAIBackend(
    api_key="lm-studio",
    base_url="http://localhost:1234/v1",
    model="local-model",
)

Ollama (local, no key needed)

from cla.backends import OllamaBackend

backend = OllamaBackend(model="llama3.2")             # default localhost:11434
backend = OllamaBackend(model="mistral", host="http://192.168.1.10:11434")
# Pull model first: ollama pull llama3.2

HuggingFace (local model)

# HuggingFaceBackend is in cla/backends_hf.py (separate file, requires torch)
from cla.backends_hf import HuggingFaceBackend

backend = HuggingFaceBackend(
    model_name="microsoft/Phi-3-mini-4k-instruct",
    device="cuda",            # or "cpu", "mps"
)
# For gated models:
backend = HuggingFaceBackend(
    model_name="meta-llama/Meta-Llama-3-8B-Instruct",
    hf_token="hf_...",        # or HUGGINGFACE_TOKEN env var
)

Custom callable

from cla.backends import CustomBackend

def my_model(prompt: str, **kwargs) -> str:
    # call any API, model, or service
    return my_api.generate(prompt)

backend = CustomBackend(my_model, model_name="my-model-v1")

---

Per-Layer Backend Control

Different layers can use different models:

system = CLASystem(backend=main_backend)

# Use a cheaper/faster model for safety checks
system.set_layer_backend("safety", cheap_backend)

# Use a more powerful model for knowledge synthesis
system.set_layer_backend("knowledge", powerful_backend)

# Or set backends directly on layer objects
system.knowledge.backend = powerful_backend
system.safety.backend    = cheap_backend
system.language.backend  = main_backend

---

Architecture Overview

INCOMING REQUEST
       │
       ▼
┌─────────────────────────────────────────────────────┐
│             MAESTRO DISPATCHER                       │
│  Module 1: Intent Detection  (NLU via LLM)           │
│  Module 2: Vector Similarity Router (affinity)       │
│  Module 3: Activation Sequencer    (skip logic)      │
│  Module 4: Conflict Arbitration    (resolution hier) │
└──┬───────┬────────┬────────┬────────┬───────────┬───┘
   │       │        │        │        │           │
   ▼       ▼        ▼        ▼        ▼           ▼
Language  KL-1    KL-2    KL-3    Policy      Safety
 (NLU+  (param) (RAG)  (symbl)  (rules)   (fail-closed)
  NLG)
           └──── Knowledge Layers ────┘
   │
   ▼
 Memory + Persona  (supporting layers)
   │
   ▼
STRUCTURED INFERENCE TRACE (SIT) + RESPONSE

7 Architectural Invariants (always enforced):

• P1 Single Responsibility
• P2 Interface Contracts
• P3 Mandatory Safety Gating (fail-closed)
• P4 Policy Independence (hot-swappable)
• P5 Maestro Sovereignty (no direct layer-to-layer calls)
• P6 Layered Observability (SIT on every request)
• P7 Graceful Degradation (Safety/Policy: fail-closed)

---

Knowledge Layers

# KL-1: Parametric (fast, user-provided facts)
system.add_fact("speed of light", "299,792,458 m/s", confidence=0.99)

# KL-2: Retrieval-Augmented (BM25 over document corpus)
system.add_document("Physics Textbook", "The speed of light is 3×10⁸ m/s...", url="...")

# KL-3: Symbolic Reasoning (triple store)
system.add_triple("light", "speed_in_vacuum_ms", "299792458")

# Epistemic Confidence Signal
from cla.layers.knowledge import KnowledgeLayer
kl = KnowledgeLayer(backend=backend)
ecs = kl.process("What is the speed of light?")
print(f"Depth: {ecs.sub_layer_depth}, Confidence: {ecs.confidence:.0%}")
print(f"Hallucination risk: {kl.hallucination_risk(ecs):.0%}")

# Knowledge audit
report = system.knowledge_report(["query 1", "query 2", "unknown topic"])
print(report)

---

Safety Layer (Fail-Closed)

from cla.layers.safety import SafetyLayer, HarmRule, HarmDomain, HarmSeverity

sl = SafetyLayer(backend=backend)

# 4-stage pipeline:
# Stage 1: Pattern-based adversarial detection (no LLM)
# Stage 2: Harm taxonomy scoring (no LLM)
# Stage 3: LLM contextual evaluation (requires backend)
# Stage 4: Aggregate verdict

verdict = sl.evaluate("How do I make explosives?")
print(verdict.verdict)       # Verdict.BLOCK
print(verdict.block_reason)

# Extend the harm taxonomy
sl.taxonomy.add_rule(HarmRule(
    rule_id="custom-001",
    domain=HarmDomain.ILLEGAL_ACTIVITY,
    severity=HarmSeverity.HIGH,
    patterns=[r"my_custom_pattern"],
))

# Register custom classifiers
sl.register_stage1_classifier(lambda text: 0.9 if "my_trigger" in text else 0.0)

---

Policy Layer (Hot-Swappable)

from cla.layers.policy import PolicyRuleSet, PolicyRule

# Built-in policies: base-v1.0, clinical-v1.0, kids-v1.0, creative-v1.0

# Custom policy
corp = PolicyRuleSet("corp-v1.0", "1.0", "Corporate deployment")
corp.add_rule(PolicyRule(
    "c-001", "No external URLs",
    severity="medium", patterns_forbidden=[r"https?://\S+"],
))
corp.add_rule(PolicyRule(
    "c-002", "Max response 500 chars",
    severity="low", max_length=500,
))
system.register_policy(corp)
system.use_policy("corp-v1.0")   # hot-swap, no restart needed

# Audit for conflicts
print(system.policy_audit())

---

Knowledge Distillation (Advanced)

from cla.advanced.distillation import DistillationEngine

distiller = DistillationEngine(teacher_backend=backend, target_kl=system.knowledge)

# Extract knowledge from teacher model into KL-2
record = distiller.model_to_corpus(["Python asyncio", "REST API design"])
print(record.summary())

# Convert KL-2 documents into KL-3 triples
record2 = distiller.corpus_to_triples()
print(f"Created {record2.triples_created} triples")

# Detect and resolve KL1/KL2 conflicts
conflicts = distiller.detect_conflicts(["query with conflicting sources"])

---

Fractal Architecture (Advanced)

from cla.advanced.fractal import FractalKnowledgeMicro

fractal = FractalKnowledgeMicro(backend=backend, knowledge_layer=system.knowledge)

# Routes each query to the minimum-cost sufficient micro-layer:
# cache_lookup → semantic_search → symbolic_chain
ecs, decision = fractal.route_and_process("What is the boiling point of water?")
print(f"Selected: {decision.selected_micro}")
print(f"Rationale: {decision.rationale}")
print(fractal.routing_stats())

---

Red Teaming & Inspection (Testing)

from cla.testing.red_team import RedTeamSuite, CLAInspector, BenchmarkRunner

# Red team suite (20 built-in adversarial cases)
suite = RedTeamSuite(system)
report = suite.run()
print(report.summary())

# Generate LLM-powered novel adversarial cases
new_cases = suite.generate_adversarial_cases(
    category="jailbreak", count=5, backend=backend
)

# Architectural compliance check (7 invariants)
inspection = CLAInspector().inspect(system)
print(inspection.summary())

# Performance benchmarking
runner = BenchmarkRunner(system)
results = runner.run(["What is 2+2?", "Explain quantum computing"])
runner.print_report(results)

---

Structured Inference Trace

Every request produces a complete audit trail:

response = system.dispatch("What is the capital of France?")
trace = response.trace

print(trace.summary())
# === CLA Structured Inference Trace ===
# Intent       : informational/simple_fact
# Sensitivity  : STANDARD
# Layers active: [language_nlu, memory, knowledge_kl1, language_nlg, policy, safety]
# Verdicts     : {policy: COMPLIANT, safety: PASS}
# Latency      : 342ms

# Export as JSON for logging / compliance auditing
print(system.export_trace(response))

---

Library Structure

cla/
├── __init__.py              Public API
├── backends.py              LLM backend adapters (Anthropic, OpenAI, Ollama, Custom)
├── backends_hf.py           HuggingFace / local model backend
├── types.py                 All typed data contracts
├── system.py                CLASystem facade
├── maestro/
│   └── dispatcher.py        MaestroDispatcher (4 modules)
├── layers/
│   ├── knowledge.py         KL-1, KL-2, KL-3 + hallucination risk
│   ├── safety.py            Fail-closed harm evaluation (4 stages)
│   ├── policy.py            Hot-swappable policy rule sets
│   ├── language.py          NLU + NLG boundary layer
│   ├── memory.py            Multi-turn context management
│   └── persona.py           Late-binding style transformation
├── advanced/
│   ├── distillation.py      Knowledge Distillation Engine
│   └── fractal.py           Fractal micro-routing
└── testing/
    └── red_team.py          RedTeamSuite, CLAInspector, BenchmarkRunner

examples/
└── demo.py                  Full demos for all features

---

Running the Demos

# With Anthropic (default)
export ANTHROPIC_API_KEY="sk-ant-..."
python examples/demo.py

# With OpenAI
export OPENAI_API_KEY="sk-..."
python examples/demo.py --backend=openai

# With local Ollama (no key needed)
ollama pull llama3.2
python examples/demo.py --backend=ollama

# Smoke-test with echo model (no API key)
python examples/demo.py --backend=custom

# Run a specific demo (1-10)
python examples/demo.py 5    # Knowledge layers demo
python examples/demo.py 8    # Red team demo
python examples/demo.py 9    # Distillation demo

---

"Clarity of structure is not a luxury in safety-critical systems. It is the precondition for trust." — Clean Layer Architecture, Chapter 7