2. Clean Architecture para AI

Descripción

Clean architecture para AI apps no es hexagonal architecture completa — es una versión pragmática de 4 capas adaptada a las necesidades específicas de LLM apps: prompt templates, business logic, LLM infrastructure, y output processing. Esta cápsula define cada capa, sus responsabilidades, y la estructura de directorios resultante.

Las 4 capas: definición completa

┌─────────────────────────────────────────────────────────────────┐
│  CAPA 1: PROMPT TEMPLATES                                       │
│                                                                 │
│  ¿Qué contiene?                                                 │
│    - Archivos YAML/JSON con templates de prompts                │
│    - Metadata: versión, autor, fecha, descripción              │
│    - Variables: {text}, {max_words}, {language}                 │
│    - Cargador: función que lee el template y permite renderizar │
│                                                                 │
│  ¿Qué NO contiene?                                              │
│    - Lógica de negocio                                          │
│    - Llamadas a APIs                                            │
│    - Código de parsing                                          │
│                                                                 │
│  Archivos: prompts/*.yaml, src/prompts/loader.py               │
└─────────────────────────────────────────────────────────────────┘
                           ↓ depende de
┌─────────────────────────────────────────────────────────────────┐
│  CAPA 2: BUSINESS LOGIC (domain)                                │
│                                                                 │
│  ¿Qué contiene?                                                 │
│    - Use cases: analyze_sentiment(), summarize_text()           │
│    - Reglas de negocio: "score < 0.5 es unreliable"            │
│    - Orquestación: llama a prompts, llama a provider, procesa  │
│                                                                 │
│  ¿Qué NO contiene?                                              │
│    - Importaciones de `openai`, `anthropic`                     │
│    - Código de parsing de JSON                                  │
│    - Configuración de temperatura/modelo                        │
│    - Código de logging o guardrails                             │
│                                                                 │
│  Archivos: src/domain/*.py                                      │
└─────────────────────────────────────────────────────────────────┘
              ↓ usa                              ↓ usa
┌──────────────────────────┐   ┌────────────────────────────────┐
│  CAPA 3: INFRASTRUCTURE  │   │  CAPA 4: OUTPUT PROCESSING     │
│                          │   │                                │
│  ¿Qué contiene?          │   │  ¿Qué contiene?                │
│  - OpenAIProvider        │   │  - Parsers: str → dict         │
│  - AnthropicProvider     │   │  - Validators: Pydantic models │
│  - MockProvider          │   │  - Transformers: format output │
│  - LLMProvider Protocol  │   │                                │
│                          │   │  ¿Qué NO contiene?             │
│  ¿Qué NO contiene?       │   │  - Llamadas a APIs             │
│  - Lógica de negocio     │   │  - Lógica de negocio           │
│  - Parsing               │   │                                │
│                          │   │  Archivos: src/processing/*.py │
│  Archivos: src/infra/*.py│   │                                │
└──────────────────────────┘   └────────────────────────────────┘

Estructura de directorios completa

proyecto/
├── prompts/                        # Capa 1: Prompt Templates
│   ├── sentiment/
│   │   ├── v1.yaml                 # Template v1
│   │   └── v2.yaml                 # Template mejorado
│   ├── summarization/
│   │   └── v1.yaml
│   └── common/
│       └── system_prompts.yaml     # System prompts reutilizables
│
├── src/
│   ├── config.py                   # pydantic-settings (global)
│   │
│   ├── domain/                     # Capa 2: Business Logic
│   │   ├── __init__.py
│   │   ├── sentiment_service.py    # Use case: analizar sentimiento
│   │   ├── models.py               # Domain models (no Pydantic I/O)
│   │   └── exceptions.py           # Domain exceptions
│   │
│   ├── infrastructure/             # Capa 3: LLM Infrastructure
│   │   ├── __init__.py
│   │   ├── llm_provider.py         # Protocol (interface)
│   │   ├── openai_provider.py      # OpenAI implementation
│   │   ├── anthropic_provider.py   # Anthropic implementation (opcional)
│   │   ├── mock_provider.py        # Mock para tests
│   │   └── fallback_provider.py    # Fallback entre providers
│   │
│   ├── processing/                 # Capa 4: Output Processing
│   │   ├── __init__.py
│   │   ├── sentiment_parser.py     # Parser específico de sentimiento
│   │   └── extractors.py           # Extractors de JSON, texto, etc.
│   │
│   ├── prompts/                    # Cargador de prompts
│   │   ├── __init__.py
│   │   └── loader.py               # Función load_prompt()
│   │
│   ├── guardrails/                 # Del Módulo 4 (infrastructure)
│   │   └── pipeline.py
│   │
│   ├── logging_config.py           # Del Módulo 5 (infrastructure)
│   ├── tracing.py
│   └── middleware.py
│
├── src/app/                        # Entry point (FastAPI)
│   ├── __init__.py
│   ├── main.py                     # App factory, registrar middleware
│   ├── dependencies.py             # FastAPI Depends() providers
│   └── routers/
│       └── sentiment.py            # Endpoints de sentimiento
│
├── tests/
│   ├── unit/
│   │   ├── test_sentiment_service.py   # Tests con MockProvider
│   │   ├── test_sentiment_parser.py    # Tests del parser
│   │   └── test_config.py              # Tests de la config
│   └── integration/
│       └── test_e2e.py
│
├── .env.example
├── .env.development
├── requirements.txt
└── README.md

La regla de dependencia: qué puede importar a qué

# ✅ PERMITIDO: dependencias apuntan hacia adentro

# Domain puede importar de:
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 puede importar de:
from pydantic import BaseModel  # External libs
# Nada de domain ni de infrastructure

# Infrastructure puede importar de:
from src.infrastructure.llm_provider import LLMProvider  # Propia capa
from openai import OpenAI  # External libs
# Nada de domain ni de processing

# ❌ PROHIBIDO: dependencias hacia afuera

# Domain NO debe importar:
# from openai import OpenAI  ← Infrastructure detail
# from src.logging_config import log  ← Infraestructura

# Infrastructure NO debe importar:
# from src.domain.sentiment_service import analyze_sentiment  ← Domain

# Processing NO debe importar:
# from src.infrastructure.openai_provider import OpenAIProvider  ← Infrastructure

Cuándo usar 4 capas vs cuándo simplificar

# REGLA: la arquitectura debe servir al código, no al revés

# ✅ 4 capas completas tiene sentido cuando:
# - La app tiene múltiples endpoints con lógica diferente
# - Estás considerando cambiar de proveedor LLM
# - Tienes A/B testing de prompts
# - El equipo tiene más de 1 persona
# - La app tiene más de 500 líneas de código

# ⚠️  Simplificación razonable para apps pequeñas (< 200 líneas):
# - domain/ e infrastructure/ como 2 módulos, no 4 carpetas
# - Prompts como constantes en domain (si solo hay 1 y no cambia)
# - Processing como funciones en domain (si solo tiene 1 parser)

# Ejemplo de app mínima con buena estructura:
src/
├── config.py           # Settings
├── provider.py         # LLMProvider protocol + OpenAIProvider
├── service.py          # Business logic (importa provider como protocol)
└── app.py              # FastAPI con dependency injection

Cómo conectan las capas en una llamada real

# src/app/routers/sentiment.py
# Entry point: une todas las capas

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 inyectada por Depends
):
    # Business logic: no sabe nada de FastAPI, OpenAI, ni Pydantic I/O
    result = analyze_sentiment(
        text=body.text,
        provider=provider
    )
    return AnalyzeResponse(**result)


# src/app/dependencies.py
# Aquí se configura qué implementación usar

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
# Business logic pura

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: analizar sentimiento de un texto.
    
    Solo conoce:
    - LLMProvider (interface, no implementación)
    - parse_sentiment_output (función de processing)
    - load_prompt (cargador de prompts)
    
    NO conoce:
    - OpenAI, Anthropic, ni ningún provider específico
    - El formato JSON de la respuesta (lo delega a parser)
    - La temperatura ni el modelo (están en el 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)

Dónde van guardrails y logging en esta arquitectura

# Guardrails: son middleware — envuelven el business logic sin mezclarse

# Opción A: FastAPI middleware (para todos los endpoints)
app.add_middleware(GuardrailsMiddleware, config=PUBLIC_API_CONFIG)

# Opción B: Wrapper en el endpoint (para un endpoint específico)
@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: wrapper del provider (infrastructure)
# El OpenAIProvider puede ser decorado con un LoggingProvider wrapper

class LoggingProviderWrapper:
    """Wrapper que añade logging a cualquier 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

# En dependencies.py:
def get_provider():
    base_provider = OpenAIProvider(...)
    return LoggingProviderWrapper(base_provider)  # Añade logging automáticamente

Tests con clean architecture

# Con clean architecture, los tests son más simples y más robustos

# tests/unit/test_sentiment_service.py

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

# ✅ No necesitamos patch() para el test
# ✅ El test es completamente determinístico
# ✅ El test funciona sin API key ni conexión a internet

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:
    """Mock configurable para tests."""
    def __init__(self, response: str):
        self._response = response
    
    def complete(self, messages: list) -> str:
        return self._response

Ejercicios

Ejercicio 1: Diseñar la estructura para un nuevo endpoint

Tienes que añadir un endpoint /summarize que acepta un texto y devuelve un resumen. ¿Qué archivos crearías en cada capa?

prompts/summarization/v1.yaml  ← Template del prompt de resumen

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/ no cambia: los providers ya existen y se reutilizan
# src/config.py puede necesitar nuevos parámetros si summarize tiene config diferente

Ejercicio 2: Identificar la violación de arquitectura

# ¿Qué viola la regla de dependencia aquí?
# 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)

Resumen

• 4 capas pragmáticas: prompts (config), domain (business logic), infrastructure (providers), processing (parsers)
• La regla de dependencia: las dependencias solo apuntan hacia adentro — domain no conoce OpenAI
• Guardrails y logging: son middleware o wrappers — no contaminan el domain
• Tests más simples: con DI, los unit tests del domain usan MockProvider sin patch()
• Pragmatismo: para apps < 200 líneas, puedes simplificar la estructura sin violar los principios

Recursos adicionales

1. Clean Architecture (Robert C. Martin) — El artículo original
2. The Dependency Rule — Por qué las dependencias apuntan hacia adentro
3. Domain-Driven Design (Eric Evans) — El concepto de domain layer
4. Python Project Structure (Hitchhiker's Guide) — Guía práctica para Python