2. Clean Architecture for AI

Description

Clean architecture for AI apps is not full hexagonal architecture — it's a pragmatic 4-layer version adapted to the specific needs of LLM apps: prompt templates, business logic, LLM infrastructure, and output processing. This capsule defines each layer, its responsibilities, and the resulting directory structure.

The 4 layers: complete definition

┌─────────────────────────────────────────────────────────────────┐
│  LAYER 1: PROMPT TEMPLATES                                      │
│                                                                 │
│  What's inside?                                                 │
│    - YAML/JSON files with prompt templates                      │
│    - Metadata: version, author, date, description               │
│    - Variables: {text}, {max_words}, {language}                 │
│    - Loader: reads the template and enables rendering           │
│                                                                 │
│  What's NOT inside?                                             │
│    - Business logic                                             │
│    - API calls                                                  │
│    - Parsing code                                               │
│                                                                 │
│  Files: prompts/*.yaml, src/prompts/loader.py                   │
└─────────────────────────────────────────────────────────────────┘
                           ↓ depends on
┌─────────────────────────────────────────────────────────────────┐
│  LAYER 2: BUSINESS LOGIC (domain)                               │
│                                                                 │
│  What's inside?                                                 │
│    - Use cases: analyze_sentiment(), summarize_text()           │
│    - Business rules: "score < 0.5 is unreliable"                │
│    - Orchestration: calls prompts, calls provider, processes    │
│                                                                 │
│  What's NOT inside?                                             │
│    - Imports of `openai`, `anthropic`                           │
│    - JSON parsing code                                          │
│    - Temperature/model configuration                            │
│    - Logging or guardrails code                                 │
│                                                                 │
│  Files: src/domain/*.py                                         │
└─────────────────────────────────────────────────────────────────┘
              ↓ uses                             ↓ uses
┌──────────────────────────┐   ┌────────────────────────────────┐
│  LAYER 3: INFRASTRUCTURE │   │  LAYER 4: OUTPUT PROCESSING    │
│                          │   │                                │
│  What's inside?          │   │  What's inside?                │
│  - OpenAIProvider        │   │  - Parsers: str → dict         │
│  - AnthropicProvider     │   │  - Validators: Pydantic models │
│  - MockProvider          │   │  - Transformers: format output │
│  - LLMProvider Protocol  │   │                                │
│                          │   │  What's NOT inside?            │
│  What's NOT inside?      │   │  - API calls                   │
│  - Business logic        │   │  - Business logic              │
│  - Parsing               │   │                                │
│                          │   │  Files: src/processing/*.py    │
│  Files: src/infra/*.py   │   │                                │
└──────────────────────────┘   └────────────────────────────────┘

Complete directory structure

project/
├── prompts/                        # Layer 1: Prompt Templates
│   ├── sentiment/
│   │   ├── v1.yaml                 # Template v1
│   │   └── v2.yaml                 # Improved template
│   ├── summarization/
│   │   └── v1.yaml
│   └── common/
│       └── system_prompts.yaml     # Reusable system prompts
│
├── src/
│   ├── config.py                   # pydantic-settings (global)
│   │
│   ├── domain/                     # Layer 2: Business Logic
│   │   ├── __init__.py
│   │   ├── sentiment_service.py    # Use case: analyze sentiment
│   │   ├── models.py               # Domain models (no Pydantic I/O)
│   │   └── exceptions.py           # Domain exceptions
│   │
│   ├── infrastructure/             # Layer 3: LLM Infrastructure
│   │   ├── __init__.py
│   │   ├── llm_provider.py         # Protocol (interface)
│   │   ├── openai_provider.py      # OpenAI implementation
│   │   ├── anthropic_provider.py   # Anthropic implementation (optional)
│   │   ├── mock_provider.py        # Mock for tests
│   │   └── fallback_provider.py    # Fallback between providers
│   │
│   ├── processing/                 # Layer 4: Output Processing
│   │   ├── __init__.py
│   │   ├── sentiment_parser.py     # Sentiment-specific parser
│   │   └── extractors.py           # Extractors for JSON, text, etc.
│   │
│   ├── prompts/                    # Prompt loader
│   │   ├── __init__.py
│   │   └── loader.py               # load_prompt() function
│   │
│   ├── guardrails/                 # From Module 4 (infrastructure)
│   │   └── pipeline.py
│   │
│   ├── logging_config.py           # From Module 5 (infrastructure)
│   ├── tracing.py
│   └── middleware.py
│
├── src/app/                        # Entry point (FastAPI)
│   ├── __init__.py
│   ├── main.py                     # App factory, register middleware
│   ├── dependencies.py             # FastAPI Depends() providers
│   └── routers/
│       └── sentiment.py            # Sentiment endpoints
│
├── tests/
│   ├── unit/
│   │   ├── test_sentiment_service.py   # Tests with MockProvider
│   │   ├── test_sentiment_parser.py    # Parser tests
│   │   └── test_config.py              # Config tests
│   └── integration/
│       └── test_e2e.py
│
├── .env.example
├── .env.development
├── requirements.txt
└── README.md

The dependency rule: what can import what

# ✅ ALLOWED: dependencies point inward

# Domain can import from:
from src.infrastructure.llm_provider import LLMProvider  # Protocol (interface)
from src.processing.sentiment_parser import parse_sentiment  # Processing
from src.prompts.loader import load_prompt  # Prompts

# Processing can import from:
from pydantic import BaseModel  # External libs
# Nothing from domain or infrastructure

# Infrastructure can import from:
from src.infrastructure.llm_provider import LLMProvider  # Own layer
from openai import OpenAI  # External libs
# Nothing from domain or processing

# ❌ FORBIDDEN: outward dependencies

# Domain must NOT import:
# from openai import OpenAI  ← Infrastructure detail
# from src.logging_config import log  ← Infrastructure

# Infrastructure must NOT import:
# from src.domain.sentiment_service import analyze_sentiment  ← Domain

# Processing must NOT import:
# from src.infrastructure.openai_provider import OpenAIProvider  ← Infrastructure

When to use 4 layers vs when to simplify

# RULE: the architecture must serve the code, not the other way around

# ✅ Full 4 layers makes sense when:
# - The app has multiple endpoints with different logic
# - You're considering switching LLM provider
# - You have A/B testing of prompts
# - The team has more than 1 person
# - The app has more than 500 lines of code

# ⚠️  Reasonable simplification for small apps (< 200 lines):
# - domain/ and infrastructure/ as 2 modules, not 4 folders
# - Prompts as constants in domain (if there's only 1 and it doesn't change)
# - Processing as functions in domain (if it has only 1 parser)

# Example of a minimal app with good structure:
src/
├── config.py           # Settings
├── provider.py         # LLMProvider protocol + OpenAIProvider
├── service.py          # Business logic (imports provider as protocol)
└── app.py              # FastAPI with dependency injection

How the layers connect in a real call

# src/app/routers/sentiment.py
# Entry point: joins all the layers

from fastapi import APIRouter, Depends
from src.domain.sentiment_service import analyze_sentiment
from src.app.dependencies import get_provider
from src.config import get_settings

router = APIRouter()

@router.post("/analyze")
async def analyze_endpoint(
    body: AnalyzeRequest,
    provider = Depends(get_provider)   # Infrastructure injected by Depends
):
    # Business logic: knows nothing about FastAPI, OpenAI, or Pydantic I/O
    result = analyze_sentiment(
        text=body.text,
        provider=provider
    )
    return AnalyzeResponse(**result)


# src/app/dependencies.py
# Here you configure which implementation to use

from src.config import get_settings
from src.infrastructure.openai_provider import OpenAIProvider
from src.infrastructure.mock_provider import MockProvider

def get_provider():
    settings = get_settings()
    if settings.use_mock:
        return MockProvider()
    return OpenAIProvider(
        client=settings.create_openai_client(),
        model=settings.model,
        temperature=settings.temperature,
        max_tokens=settings.max_tokens
    )


# src/domain/sentiment_service.py
# Pure business logic

from src.infrastructure.llm_provider import LLMProvider
from src.processing.sentiment_parser import parse_sentiment_output
from src.prompts.loader import load_prompt

def analyze_sentiment(text: str, provider: LLMProvider) -> dict:
    """
    Use case: analyze the sentiment of a text.
    
    Only knows:
    - LLMProvider (interface, not implementation)
    - parse_sentiment_output (processing function)
    - load_prompt (prompt loader)
    
    Does NOT know:
    - OpenAI, Anthropic, or any specific provider
    - The JSON format of the response (delegates it to the parser)
    - The temperature or the model (they're in the provider)
    """
    prompt_template = load_prompt("sentiment/v1")
    prompt = prompt_template.render(text=text)
    
    messages = [
        {"role": "system", "content": prompt_template.system},
        {"role": "user", "content": prompt}
    ]
    
    raw_response = provider.complete(messages)
    return parse_sentiment_output(raw_response)

Where guardrails and logging go in this architecture

# Guardrails: they're middleware — they wrap the business logic without mixing in

# Option A: FastAPI middleware (for all endpoints)
app.add_middleware(GuardrailsMiddleware, config=PUBLIC_API_CONFIG)

# Option B: Wrapper in the endpoint (for a specific endpoint)
@router.post("/analyze")
async def analyze_endpoint(body: AnalyzeRequest, provider = Depends(get_provider)):
    guardrail_result = guardrails_pipeline.process(body.text)
    if guardrail_result.blocked:
        raise HTTPException(400, guardrail_result.block_reason)
    
    result = analyze_sentiment(guardrail_result.processed_input, provider)
    return result

# Logging: provider wrapper (infrastructure)
# The OpenAIProvider can be decorated with a LoggingProvider wrapper

class LoggingProviderWrapper:
    """Wrapper that adds logging to any LLMProvider."""
    
    def __init__(self, inner: LLMProvider):
        self._inner = inner
    
    def complete(self, messages: list) -> str:
        import structlog, time
        log = structlog.get_logger()
        start = time.time()
        
        try:
            result = self._inner.complete(messages)
            log.info("llm_completed", duration_ms=(time.time()-start)*1000)
            return result
        except Exception as e:
            log.error("llm_failed", error=str(e))
            raise

# In dependencies.py:
def get_provider():
    base_provider = OpenAIProvider(...)
    return LoggingProviderWrapper(base_provider)  # Adds logging automatically

Tests with clean architecture

# With clean architecture, tests are simpler and more robust

# tests/unit/test_sentiment_service.py

from src.domain.sentiment_service import analyze_sentiment
from src.infrastructure.mock_provider import MockProvider

# ✅ We don't need patch() for the test
# ✅ The test is completely deterministic
# ✅ The test works without an API key or internet connection

class TestAnalyzeSentiment:
    def test_positive_sentiment(self):
        mock = MockProvider(response='{"sentiment": "positive", "score": 0.8}')
        result = analyze_sentiment("This is great!", mock)
        assert result["sentiment"] == "positive"
        assert result["score"] == 0.8
    
    def test_invalid_json_returns_unknown(self):
        mock = MockProvider(response="not json")
        result = analyze_sentiment("test", mock)
        assert result["sentiment"] == "unknown"
    
    def test_negative_sentiment(self):
        mock = MockProvider(response='{"sentiment": "negative", "score": -0.7}')
        result = analyze_sentiment("This is terrible", mock)
        assert result["sentiment"] == "negative"


class MockProvider:
    """Configurable mock for tests."""
    def __init__(self, response: str):
        self._response = response
    
    def complete(self, messages: list) -> str:
        return self._response

Exercises

Exercise 1: Design the structure for a new endpoint

You need to add a /summarize endpoint that accepts a text and returns a summary. What files would you create in each layer?

prompts/summarization/v1.yaml  ← Summary prompt template

src/domain/summary_service.py  ← use case: summarize_text(text, provider) → dict
src/processing/summary_parser.py ← parse_summary_output(raw: str) → dict
src/app/routers/summary.py     ← Endpoint /summarize

# src/infrastructure/ doesn't change: the providers already exist and are reused
# src/config.py may need new parameters if summarize has different config

Exercise 2: Identify the architecture violation

# What violates the dependency rule here?
# src/domain/sentiment_service.py
from openai import OpenAI
from src.processing.sentiment_parser import parse_output

def analyze(text: str) -> dict:
    client = OpenAI()
    response = client.chat.completions.create(...)
    return parse_output(response.choices[0].message.content)

Summary

• 4 pragmatic layers: prompts (config), domain (business logic), infrastructure (providers), processing (parsers)
• The dependency rule: dependencies only point inward — the domain doesn't know OpenAI
• Guardrails and logging: they're middleware or wrappers — they don't contaminate the domain
• Simpler tests: with DI, the domain's unit tests use MockProvider without patch()
• Pragmatism: for apps < 200 lines, you can simplify the structure without violating the principles

Additional resources

1. Clean Architecture (Robert C. Martin) — The original article
2. The Dependency Rule — Why dependencies point inward
3. Domain-Driven Design (Eric Evans) — The domain layer concept
4. Python Project Structure (Hitchhiker's Guide) — Practical guide for Python