7. Proyecto: Refactored AI App

Descripción

Este es el proyecto integrador del Módulo 6. Tomas la app de análisis de sentimiento que has venido construyendo (con tests, guardrails, y logging) y la refactorizas aplicando todos los patterns del módulo: clean architecture en 4 capas, prompts externalizados, LLMProvider con DI, pydantic-settings, y environment management. El resultado es un codebase que un equipo puede mantener, extender, y debuggear.

---

La app antes del refactoring

# src/app/main.py — ANTES (estado típico después de M1-M5)
import os, json, time, structlog
from openai import OpenAI
from fastapi import FastAPI
from pydantic import BaseModel
from src.guardrails.pipeline import GuardrailsPipeline, PUBLIC_API_CONFIG

# Config dispersa
API_KEY = os.getenv("OPENAI_API_KEY")
MODEL = os.getenv("MODEL", "gpt-4o-mini")
TEMP = float(os.getenv("TEMPERATURE", "0.7"))

# Prompt hardcodeado en el módulo
SYSTEM = "You are a sentiment analysis expert."
PROMPT = """Analyze the sentiment of: {text}
Return JSON: {{"sentiment": "positive|negative|neutral|mixed", "score": float}}"""

log = structlog.get_logger()

app = FastAPI()
client = OpenAI(api_key=API_KEY)
pipeline = GuardrailsPipeline(config=PUBLIC_API_CONFIG, openai_client=client)

class AnalyzeRequest(BaseModel):
    text: str

@app.post("/analyze")
async def analyze(req: AnalyzeRequest):
    # God function: hace todo
    guardrail = pipeline.process(req.text)
    if guardrail.blocked:
        return {"error": guardrail.block_reason}
    
    messages = [
        {"role": "system", "content": SYSTEM},
        {"role": "user", "content": PROMPT.format(text=guardrail.processed_input)}
    ]
    
    start = time.time()
    response = client.chat.completions.create(
        model=MODEL, messages=messages, temperature=TEMP, max_tokens=500
    )
    
    raw = response.choices[0].message.content
    try:
        result = json.loads(raw)
    except:
        result = {"sentiment": "unknown", "score": 0.0}
    
    log.info("done", duration_ms=(time.time()-start)*1000,
             tokens=response.usage.total_tokens)
    
    return result

---

La estructura del proyecto después del refactoring

proyecto/
├── prompts/
│   ├── sentiment/
│   │   └── v1.yaml
│   └── README.md
│
├── src/
│   ├── config.py                    ← pydantic-settings centralizado
│   ├── startup.py                   ← Startup checks
│   │
│   ├── prompts/
│   │   └── loader.py                ← Cargador de prompts YAML
│   │
│   ├── domain/
│   │   ├── __init__.py
│   │   ├── sentiment_service.py     ← Use case: analyze_sentiment()
│   │   └── exceptions.py            ← LowConfidenceError, etc.
│   │
│   ├── infrastructure/
│   │   ├── __init__.py
│   │   ├── llm_provider.py          ← Protocol LLMProvider
│   │   ├── openai_provider.py       ← Implementación OpenAI
│   │   ├── mock_provider.py         ← Mock para tests
│   │   └── fallback_provider.py     ← Fallback entre providers
│   │
│   ├── processing/
│   │   ├── __init__.py
│   │   └── sentiment_parser.py      ← Parser y SentimentOutput
│   │
│   ├── guardrails/                  ← Del Módulo 4
│   │   └── pipeline.py
│   │
│   ├── logging_config.py            ← Del Módulo 5
│   ├── tracing.py
│   └── middleware.py
│
├── src/app/
│   ├── __init__.py
│   ├── main.py                      ← App factory
│   ├── dependencies.py              ← FastAPI Depends() providers
│   └── routers/
│       └── sentiment.py             ← Endpoint /analyze
│
├── tests/
│   ├── conftest.py
│   ├── unit/
│   │   ├── test_sentiment_service.py
│   │   ├── test_sentiment_parser.py
│   │   ├── test_config.py
│   │   └── test_mock_provider.py
│   └── integration/
│       └── test_e2e.py
│
├── .env
├── .env.development
├── .env.example
├── requirements.txt
└── README.md

---

Paso 1: Crear el archivo de prompt

# prompts/sentiment/v1.yaml
version: "v1"
author: "team"
created: "2024-01-15"
description: >
  Sentiment analysis prompt. Classifies text as positive, negative,
  neutral, or mixed. Returns JSON with sentiment label and score.

system: >
  You are a sentiment analysis expert. You always respond with
  valid JSON and nothing else.

template: |
  Analyze the sentiment of the following text.
  Return a JSON object with exactly these fields:
  - "sentiment": one of "positive", "negative", "neutral", "mixed"
  - "score": float from -1.0 (very negative) to 1.0 (very positive)
  - "confidence": float from 0.0 to 1.0

  Text to analyze:
  {text}

---

Paso 2: El cargador de prompts

# src/prompts/loader.py
import yaml
from pathlib import Path
from dataclasses import dataclass
from functools import lru_cache

PROMPTS_DIR = Path(__file__).parent.parent.parent / "prompts"

@dataclass
class PromptTemplate:
    version: str
    system: str
    template: str
    description: str = ""
    
    def render(self, **kwargs) -> str:
        try:
            return self.template.format(**kwargs)
        except KeyError as e:
            raise ValueError(f"Missing variable in prompt: {e}") from e

@lru_cache(maxsize=32)
def load_prompt(name: str) -> PromptTemplate:
    """
    Carga un prompt desde YAML con caché.
    
    Args:
        name: Ruta relativa al directorio prompts/ sin extensión.
              Ejemplo: "sentiment/v1"
    """
    path = PROMPTS_DIR / f"{name}.yaml"
    if not path.exists():
        raise FileNotFoundError(f"Prompt not found: {path}")
    
    with open(path, encoding="utf-8") as f:
        data = yaml.safe_load(f)
    
    return PromptTemplate(
        version=str(data.get("version", "v1")),
        system=str(data.get("system", "")).strip(),
        template=str(data.get("template", "")),
        description=str(data.get("description", ""))
    )

---

Paso 3: El domain service limpio

# src/domain/exceptions.py
class SentimentAnalysisError(Exception):
    """Error base del dominio de análisis de sentimiento."""

class LowConfidenceError(SentimentAnalysisError):
    """El LLM no pudo analizar con suficiente confianza."""

class ProviderUnavailableError(SentimentAnalysisError):
    """El LLM provider no está disponible."""

# src/domain/sentiment_service.py
from src.infrastructure.llm_provider import LLMProvider, LLMProviderError
from src.processing.sentiment_parser import parse_sentiment_output
from src.prompts.loader import load_prompt
from src.domain.exceptions import LowConfidenceError, ProviderUnavailableError

MINIMUM_CONFIDENCE = 0.3

def analyze_sentiment(text: str, provider: LLMProvider) -> dict:
    """
    Use case: analizar el sentimiento de un texto.
    
    Args:
        text: Texto ya sanitizado por guardrails
        provider: LLM provider inyectado
    
    Returns:
        dict con: sentiment (str), score (float), confidence (float)
    
    Raises:
        LowConfidenceError: confidence < MINIMUM_CONFIDENCE
        ProviderUnavailableError: El provider falló
    """
    template = load_prompt("sentiment/v1")
    
    messages = [
        {"role": "system", "content": template.system},
        {"role": "user", "content": template.render(text=text)}
    ]
    
    try:
        raw_response = provider.complete(messages)
    except LLMProviderError as e:
        raise ProviderUnavailableError(
            f"LLM provider unavailable: {e}"
        ) from e
    
    result = parse_sentiment_output(raw_response)
    
    if result["confidence"] < MINIMUM_CONFIDENCE:
        raise LowConfidenceError(
            f"Confidence too low: {result['confidence']:.2f} "
            f"(minimum: {MINIMUM_CONFIDENCE})"
        )
    
    return result

---

Paso 4: El parser en processing

# src/processing/sentiment_parser.py
import json
import re
from pydantic import BaseModel, field_validator
from typing import Optional

class SentimentOutput(BaseModel):
    sentiment: str
    score: float
    confidence: float = 1.0
    
    @field_validator("sentiment")
    @classmethod
    def normalize_sentiment(cls, v: str) -> str:
        VALID = {"positive", "negative", "neutral", "mixed"}
        normalized = v.lower().strip()
        return normalized if normalized in VALID else "unknown"
    
    @field_validator("score")
    @classmethod
    def clamp_score(cls, v: float) -> float:
        return max(-1.0, min(1.0, float(v)))
    
    @field_validator("confidence")
    @classmethod
    def clamp_confidence(cls, v: float) -> float:
        return max(0.0, min(1.0, float(v)))

def extract_json(raw: str) -> Optional[str]:
    """Extrae JSON de un string que puede contener texto extra."""
    stripped = raw.strip()
    if stripped.startswith("{"):
        return stripped
    
    # JSON en markdown code block
    match = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", raw, re.DOTALL)
    if match:
        return match.group(1)
    
    # JSON en el texto
    match = re.search(r"\{[^{}]+\}", raw, re.DOTALL)
    if match:
        return match.group()
    
    return None

def parse_sentiment_output(raw: str) -> dict:
    """Parsea el output del LLM y retorna un dict validado."""
    json_str = extract_json(raw)
    if json_str is None:
        return SentimentOutput(
            sentiment="unknown", score=0.0, confidence=0.0
        ).model_dump()
    
    try:
        data = json.loads(json_str)
        return SentimentOutput(**data).model_dump()
    except (json.JSONDecodeError, ValueError):
        return SentimentOutput(
            sentiment="unknown", score=0.0, confidence=0.0
        ).model_dump()

---

Paso 5: Infrastructure (providers)

# src/infrastructure/llm_provider.py
from typing import Protocol, runtime_checkable

@runtime_checkable
class LLMProvider(Protocol):
    def complete(self, messages: list[dict], **kwargs) -> str: ...

class LLMProviderError(Exception):
    def __init__(self, message: str, original_error: Exception = None):
        super().__init__(message)
        self.original_error = original_error

# src/infrastructure/openai_provider.py
import time, structlog
from src.infrastructure.llm_provider import LLMProvider, LLMProviderError
from src.logging_config import calculate_cost

log = structlog.get_logger()

class OpenAIProvider:
    def __init__(self, client, model: str, temperature: float,
                 max_tokens: int, seed: int = None):
        self._client = client
        self._model = model
        self._temperature = temperature
        self._max_tokens = max_tokens
        self._seed = seed
    
    def complete(self, messages: list[dict], **kwargs) -> str:
        start = time.time()
        params = dict(
            model=self._model, messages=messages,
            temperature=self._temperature, max_tokens=self._max_tokens
        )
        if self._seed is not None:
            params["seed"] = self._seed
        params.update(kwargs)
        
        try:
            r = self._client.chat.completions.create(**params)
            log.info("llm_call_completed", model=self._model,
                    input_tokens=r.usage.prompt_tokens,
                    output_tokens=r.usage.completion_tokens,
                    cost_usd=calculate_cost(self._model,
                        r.usage.prompt_tokens, r.usage.completion_tokens),
                    duration_ms=round((time.time()-start)*1000, 1))
            return r.choices[0].message.content
        except Exception as e:
            raise LLMProviderError(str(e), original_error=e)
    
    @classmethod
    def from_settings(cls, settings) -> "OpenAIProvider":
        return cls(
            client=settings.create_openai_client(),
            model=settings.model,
            temperature=settings.temperature,
            max_tokens=settings.max_tokens,
            seed=settings.seed
        )

# src/infrastructure/mock_provider.py
from typing import Union, Callable, Optional

class MockProvider:
    def __init__(self, response: Union[str, Callable] = None,
                 responses: list = None, raise_error: Exception = None):
        self._response = response or '{"sentiment":"positive","score":0.8,"confidence":0.9}'
        self._responses = responses
        self._raise_error = raise_error
        self._call_count = 0
        self.calls = []
    
    def complete(self, messages: list[dict], **kwargs) -> str:
        self.calls.append(messages)
        self._call_count += 1
        if self._raise_error:
            raise self._raise_error
        if self._responses:
            return self._responses[(self._call_count-1) % len(self._responses)]
        if callable(self._response):
            return self._response(messages)
        return self._response
    
    @property
    def call_count(self) -> int:
        return self._call_count
    
    def get_last_user_message(self) -> Optional[str]:
        if not self.calls:
            return None
        for msg in reversed(self.calls[-1]):
            if msg.get("role") == "user":
                return msg.get("content")
        return None

---

Paso 6: App FastAPI refactorizada

# src/app/dependencies.py
from functools import lru_cache
from src.config import get_settings
from src.infrastructure.llm_provider import LLMProvider
from src.infrastructure.openai_provider import OpenAIProvider
from src.infrastructure.mock_provider import MockProvider

def get_llm_provider() -> LLMProvider:
    """El único lugar donde se decide qué provider usar."""
    settings = get_settings()
    if settings.use_mock_llm:
        return MockProvider(response=settings.mock_response)
    return OpenAIProvider.from_settings(settings)

# src/app/routers/sentiment.py
import structlog
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from src.domain.sentiment_service import analyze_sentiment
from src.domain.exceptions import LowConfidenceError, ProviderUnavailableError
from src.infrastructure.llm_provider import LLMProvider
from src.app.dependencies import get_llm_provider
from src.tracing import get_request_id

router = APIRouter()
log = structlog.get_logger()

class AnalyzeRequest(BaseModel):
    text: str

class AnalyzeResponse(BaseModel):
    sentiment: str
    score: float
    confidence: float
    request_id: str

@router.post("/analyze", response_model=AnalyzeResponse)
async def analyze_sentiment_endpoint(
    body: AnalyzeRequest,
    provider: LLMProvider = Depends(get_llm_provider)
):
    log.info("analyze_requested", text_length=len(body.text))
    
    try:
        result = analyze_sentiment(body.text, provider)
    except LowConfidenceError as e:
        log.warning("low_confidence_result", error=str(e))
        raise HTTPException(422, detail=str(e))
    except ProviderUnavailableError as e:
        log.error("provider_unavailable", error=str(e))
        raise HTTPException(503, detail="LLM service temporarily unavailable")
    
    return AnalyzeResponse(**result, request_id=get_request_id() or "unknown")

# src/app/main.py
import os
from fastapi import FastAPI
from src.config import get_settings
from src.logging_config import configure_logging
from src.middleware import RequestTracingMiddleware
from src.startup import run_startup_checks
from src.app.routers import sentiment

def create_app() -> FastAPI:
    """
    App factory — crea y configura la app FastAPI.
    Separar la creación del app de la ejecución facilita los tests.
    """
    settings = get_settings()
    
    # Configurar logging
    configure_logging(
        env=settings.environment,
        log_level=getattr(__import__("logging"), settings.log_level),
        log_file=settings.log_file_path if settings.log_to_file else None
    )
    
    # Startup checks
    run_startup_checks()
    
    # Crear la app
    app = FastAPI(
        title="AI Sentiment Analyzer",
        description="Sentiment analysis with clean architecture",
        version=settings.app_version
    )
    
    # Middleware
    app.add_middleware(RequestTracingMiddleware)
    
    # Routers
    app.include_router(sentiment.router, prefix="/api/v1", tags=["sentiment"])
    
    @app.get("/health")
    async def health():
        return {"status": "ok", "env": settings.environment}
    
    return app

app = create_app()

---

Paso 7: Tests del proyecto refactorizado

# tests/conftest.py
import pytest
from src.config import Settings, get_settings
from unittest.mock import patch

@pytest.fixture(autouse=True)
def reset_settings_cache():
    """Limpiar cache de settings antes/después de cada test."""
    get_settings.cache_clear()
    yield
    get_settings.cache_clear()

@pytest.fixture
def mock_settings():
    return Settings(
        environment="testing",
        use_mock_llm=True,
        openai_api_key="sk-test",
        log_level="WARNING",
        temperature=0.0
    )

@pytest.fixture
def override_settings(mock_settings):
    with patch("src.config.get_settings", return_value=mock_settings):
        yield mock_settings

# tests/unit/test_sentiment_service.py
import pytest
from src.domain.sentiment_service import analyze_sentiment
from src.domain.exceptions import LowConfidenceError, ProviderUnavailableError
from src.infrastructure.mock_provider import MockProvider
from src.infrastructure.llm_provider import LLMProviderError

class TestAnalyzeSentiment:
    def test_positive_result(self):
        mock = MockProvider('{"sentiment":"positive","score":0.9,"confidence":0.95}')
        result = analyze_sentiment("Great product!", mock)
        assert result["sentiment"] == "positive"
        assert result["score"] == 0.9
    
    def test_raises_on_low_confidence(self):
        mock = MockProvider('{"sentiment":"mixed","score":0.0,"confidence":0.05}')
        with pytest.raises(LowConfidenceError):
            analyze_sentiment("ambiguous", mock)
    
    def test_raises_on_provider_failure(self):
        from src.infrastructure.llm_provider import LLMProviderError
        mock = MockProvider(raise_error=LLMProviderError("API down"))
        with pytest.raises(ProviderUnavailableError):
            analyze_sentiment("test", mock)
    
    def test_text_in_prompt(self):
        mock = MockProvider('{"sentiment":"positive","score":0.8,"confidence":0.9}')
        unique_text = "UNIQUETOKEN_12345_UNIQUETOKEN"
        analyze_sentiment(unique_text, mock)
        assert unique_text in mock.get_last_user_message()


# tests/unit/test_sentiment_parser.py
from src.processing.sentiment_parser import parse_sentiment_output, extract_json

class TestExtractJson:
    def test_plain_json(self):
        assert '{"a": 1}' in extract_json('{"a": 1}')
    
    def test_json_in_markdown(self):
        result = extract_json('```json\n{"a":1}\n```')
        assert result == '{"a":1}'
    
    def test_no_json_returns_none(self):
        assert extract_json("no json here") is None

class TestParseSentimentOutput:
    def test_valid(self):
        raw = '{"sentiment":"positive","score":0.8,"confidence":0.9}'
        r = parse_sentiment_output(raw)
        assert r["sentiment"] == "positive"
    
    def test_score_clamped_above(self):
        raw = '{"sentiment":"positive","score":2.5}'
        assert parse_sentiment_output(raw)["score"] == 1.0
    
    def test_invalid_json_fallback(self):
        r = parse_sentiment_output("sorry, i cannot analyze")
        assert r["sentiment"] == "unknown"
        assert r["confidence"] == 0.0


# tests/unit/test_config.py
import pytest
from src.config import Settings

def test_production_rejects_mock():
    with pytest.raises(ValueError, match="PRODUCCIÓN"):
        Settings(environment="production", use_mock_llm=True,
                 openai_api_key="sk-key", log_level="INFO")

def test_production_requires_api_key():
    with pytest.raises(ValueError, match="openai_api_key"):
        Settings(environment="production", use_mock_llm=False,
                 openai_api_key="", log_level="INFO")

def test_development_allows_mock():
    s = Settings(environment="development", use_mock_llm=True)
    assert s.use_mock_llm is True

---

Verificación: los tests de Phase 1 siguen pasando

# El objetivo del refactoring es que el comportamiento externo no cambie

# Antes del refactoring:
pytest tests/ -v → 47 passed

# Después de cada paso de refactoring:
pytest tests/ -v → 47 passed (+ nuevos tests del módulo 6)

# Si algún test falla después de un paso:
# 1. NO hacer commit
# 2. Revisar qué cambió
# 3. Arreglar el código (o actualizar el test si el comportamiento
#    cambió intencionalmente)
# 4. Volver a correr hasta que todos pasen

# El ciclo:
# EXTRAER CÓDIGO → CORRER TESTS → SI PASAN: CONTINUAR → SI FALLAN: REVERTIR

---

Checklist del proyecto

[ ] prompts/sentiment/v1.yaml creado con sistema y template
[ ] src/prompts/loader.py con load_prompt() y lru_cache
[ ] src/domain/exceptions.py con LowConfidenceError, ProviderUnavailableError
[ ] src/domain/sentiment_service.py limpio — no importa openai
[ ] src/processing/sentiment_parser.py con SentimentOutput y parse_sentiment_output()
[ ] src/infrastructure/llm_provider.py con Protocol y LLMProviderError
[ ] src/infrastructure/openai_provider.py implementa LLMProvider
[ ] src/infrastructure/mock_provider.py con call tracking
[ ] src/app/dependencies.py con get_llm_provider()
[ ] src/app/routers/sentiment.py con endpoint que usa Depends()
[ ] src/app/main.py como app factory
[ ] src/config.py con pydantic-settings y validadores por entorno
[ ] .env.example documentado
[ ] tests/unit/test_sentiment_service.py — usa MockProvider, sin patch()
[ ] tests/unit/test_sentiment_parser.py — testa parser de forma independiente
[ ] tests/unit/test_config.py — verifica validaciones de producción

[ ] VERIFICACIÓN FINAL: pytest tests/ → todos pasan