Módulo 6: Code Quality Patterns para AI
3. Separation of Concerns
Descripción
Separation of concerns es el principio que elimina las god functions — esas funciones de 60 líneas que hacen prompt construction, API call, JSON parsing, validation, logging y business logic, todo en un solo bloque. Esta cápsula muestra cómo identificar las responsabilidades en código AI, cómo extraerlas, y cómo decidir cuándo NO separar más.
La god function: anatomía del problema
# Esta función real de una app AI antes del módulo 6.
# Cuenta cuántas responsabilidades tiene:
def analyze_text(text: str) -> dict:
# Responsabilidad 1: VALIDACIÓN DE INPUT
if not text or len(text) < 5:
raise ValueError("Text too short")
if len(text) > 10000:
text = text[:10000] # Truncar silenciosamente (ocultar el hecho)
# Responsabilidad 2: CONSTRUCCIÓN DEL PROMPT
system_prompt = "You are a sentiment analysis expert."
user_prompt = f"""
Analyze the sentiment of the following text.
Return a JSON with these exact fields:
- sentiment: "positive", "negative", "neutral", or "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}
"""
# Responsabilidad 3: LLAMADA AL LLM (infrastructure)
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
response = client.chat.completions.create(
model=os.getenv("MODEL", "gpt-4o-mini"),
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=float(os.getenv("TEMPERATURE", "0.7")),
max_tokens=int(os.getenv("MAX_TOKENS", "500"))
)
# Responsabilidad 4: PARSEO DE RESPUESTA
raw = response.choices[0].message.content
try:
result = json.loads(raw)
except json.JSONDecodeError:
# Intentar extraer JSON con regex
match = re.search(r'\{.*\}', raw, re.DOTALL)
if match:
result = json.loads(match.group())
else:
result = {"sentiment": "unknown", "score": 0.0, "confidence": 0.0}
# Responsabilidad 5: VALIDACIÓN DE NEGOCIO
if "sentiment" not in result:
result["sentiment"] = "unknown"
if result.get("score", 0) > 1.0:
result["score"] = 1.0
if result.get("score", 0) < -1.0:
result["score"] = -1.0
# Responsabilidad 6: LOGGING
print(f"[{datetime.now()}] Analyzed text ({len(text)} chars): {result['sentiment']}")
return result
# Cuenta: 6 responsabilidades en ~55 líneas
# Problemas:
# - Para testear el parser, necesitas la llamada al LLM
# - Para testear el LLM call, necesitas el prompt hardcodeado
# - Para cambiar el prompt, editas esta función
# - Para cambiar de OpenAI a Anthropic, editas esta función
# - Para testear la validación, necesitas mockear todo lo anterior
# - Para reutilizar el parser en otro endpoint, copias códigoEl refactoring paso a paso
Paso 1: Extraer el prompt a configuración
# prompts/sentiment/v1.yaml
---
version: "v1"
author: "team"
created: "2024-01-15"
description: "Sentiment analysis prompt - first version"
system: "You are a sentiment analysis expert. Return only valid JSON."
template: |
Analyze the sentiment of the following text.
Return a JSON object with these exact 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}
# Variables requeridas: [text]
# Variables opcionales: [language, max_words]# src/prompts/loader.py
import yaml
from pathlib import Path
from dataclasses import dataclass
from typing import Optional
PROMPTS_DIR = Path(__file__).parent.parent.parent / "prompts"
@dataclass
class PromptTemplate:
version: str
system: str
template: str
description: Optional[str] = None
def render(self, **kwargs) -> str:
"""Renderiza el template con las variables dadas."""
try:
return self.template.format(**kwargs)
except KeyError as e:
raise ValueError(f"Missing variable in prompt template: {e}")
def load_prompt(name: str) -> PromptTemplate:
"""
Carga un prompt desde un archivo YAML.
Args:
name: nombre relativo al directorio prompts/, sin extensión
Ejemplo: "sentiment/v1" carga prompts/sentiment/v1.yaml
Returns:
PromptTemplate con el template cargado
Raises:
FileNotFoundError: si el archivo no existe
"""
path = PROMPTS_DIR / f"{name}.yaml"
if not path.exists():
raise FileNotFoundError(f"Prompt not found: {path}")
with open(path) as f:
data = yaml.safe_load(f)
return PromptTemplate(
version=data.get("version", "v1"),
system=data.get("system", ""),
template=data.get("template", ""),
description=data.get("description")
)Paso 2: Extraer el parser a output processing
# src/processing/sentiment_parser.py
import json
import re
from typing import Optional
from pydantic import BaseModel, field_validator
class SentimentOutput(BaseModel):
"""Schema de output del análisis de sentimiento."""
sentiment: str
score: float
confidence: float = 1.0
@field_validator("sentiment")
@classmethod
def valid_sentiment(cls, v: str) -> str:
allowed = {"positive", "negative", "neutral", "mixed", "unknown"}
v_lower = v.lower().strip()
if v_lower not in allowed:
return "unknown"
return v_lower
@field_validator("score")
@classmethod
def clamp_score(cls, v: float) -> float:
return max(-1.0, min(1.0, v))
@field_validator("confidence")
@classmethod
def clamp_confidence(cls, v: float) -> float:
return max(0.0, min(1.0, v))
def extract_json(raw: str) -> Optional[str]:
"""Intenta extraer un objeto JSON de un string que puede contener texto extra."""
# Intentar parsear directamente
stripped = raw.strip()
if stripped.startswith("{"):
return stripped
# Buscar JSON embebido en markdown code block
code_block = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", raw, re.DOTALL)
if code_block:
return code_block.group(1)
# Buscar JSON en cualquier parte del string
json_match = re.search(r"\{[^{}]*\}", raw, re.DOTALL)
if json_match:
return json_match.group()
return None
def parse_sentiment_output(raw: str) -> dict:
"""
Parsea el output del LLM y retorna un dict validado.
Estrategia:
1. Intentar parsear JSON directamente
2. Si falla, intentar extraer JSON con regex
3. Si falla, retornar output de fallback
"""
json_str = extract_json(raw)
if json_str is None:
# Fallback: no pudimos extraer JSON
return SentimentOutput(
sentiment="unknown",
score=0.0,
confidence=0.0
).model_dump()
try:
data = json.loads(json_str)
output = SentimentOutput(**data)
return output.model_dump()
except (json.JSONDecodeError, ValueError):
return SentimentOutput(
sentiment="unknown",
score=0.0,
confidence=0.0
).model_dump()Paso 3: Extraer la llamada al LLM a infrastructure
# src/infrastructure/llm_provider.py
from typing import Protocol, runtime_checkable
@runtime_checkable
class LLMProvider(Protocol):
"""
Interface para LLM providers.
Cualquier clase que implemente este Protocol puede usarse
como provider sin herencia explícita.
"""
def complete(self, messages: list[dict], **kwargs) -> str:
"""
Envía mensajes al LLM y retorna el contenido de la respuesta.
Args:
messages: Lista de mensajes en formato OpenAI
[{"role": "system", "content": "..."}, ...]
Returns:
El contenido del mensaje de respuesta del LLM
Raises:
LLMProviderError: Si el provider no puede completar la request
"""
...
class LLMProviderError(Exception):
"""Error genérico del LLM provider."""
pass
# src/infrastructure/openai_provider.py
import structlog
import time
from src.infrastructure.llm_provider import LLMProvider, LLMProviderError
log = structlog.get_logger()
class OpenAIProvider:
"""
Implementación de LLMProvider para OpenAI.
Encapsula toda la lógica específica de OpenAI.
"""
def __init__(self, client, model: str, temperature: float, max_tokens: int):
self._client = client
self._model = model
self._temperature = temperature
self._max_tokens = max_tokens
def complete(self, messages: list[dict], **kwargs) -> str:
"""Realiza una llamada al API de OpenAI."""
start = time.time()
try:
response = self._client.chat.completions.create(
model=self._model,
messages=messages,
temperature=self._temperature,
max_tokens=self._max_tokens,
**kwargs
)
duration_ms = (time.time() - start) * 1000
log.info(
"llm_call_completed",
model=self._model,
input_tokens=response.usage.prompt_tokens,
output_tokens=response.usage.completion_tokens,
duration_ms=round(duration_ms, 1)
)
return response.choices[0].message.content
except Exception as e:
log.error(
"llm_call_failed",
error_type=type(e).__name__,
model=self._model
)
raise LLMProviderError(f"OpenAI call failed: {e}") from ePaso 4: Business logic limpia
# src/domain/sentiment_service.py
"""
Business logic para análisis de sentimiento.
Este módulo NO importa:
- openai, anthropic, ni ningún provider específico
- json, re, ni lógica de parsing
- logging, structlog
Solo orquesta el flujo de alto nivel.
"""
from typing import Optional
from src.infrastructure.llm_provider import LLMProvider
from src.processing.sentiment_parser import parse_sentiment_output
from src.prompts.loader import load_prompt
# Threshold de negocio: si confidence < MINIMUM_CONFIDENCE, marcar como low_confidence
MINIMUM_CONFIDENCE = 0.3
class LowConfidenceError(Exception):
"""El LLM no pudo analizar con suficiente confianza."""
pass
def analyze_sentiment(text: str, provider: LLMProvider) -> dict:
"""
Analiza el sentimiento de un texto.
Business rules:
- Si confidence < 0.3, lanza LowConfidenceError
- El texto debe haber pasado los guardrails antes de llegar aquí
Args:
text: Texto a analizar (ya sanitizado por guardrails)
provider: LLM provider inyectado
Returns:
dict con: sentiment, score, confidence
"""
prompt_template = load_prompt("sentiment/v1")
messages = [
{"role": "system", "content": prompt_template.system},
{"role": "user", "content": prompt_template.render(text=text)}
]
raw_response = provider.complete(messages)
result = parse_sentiment_output(raw_response)
# Regla de negocio: confidence mínima
if result["confidence"] < MINIMUM_CONFIDENCE:
raise LowConfidenceError(
f"Analysis confidence too low: {result['confidence']:.2f} "
f"(minimum: {MINIMUM_CONFIDENCE})"
)
return resultComparación: antes vs después
ANTES (god function de 55 líneas):
┌─────────────────────────────────────────────────────────────┐
│ analyze_text(text: str) -> dict │
│ ├── Validar input (debería ser guardrail) │
│ ├── Construir prompt (debería ser prompts/) │
│ ├── Llamar OpenAI directamente (debería ser infra) │
│ ├── Parsear JSON con try/except + regex (debería ser │
│ │ processing) │
│ ├── Validar y normalizar valores (debería ser Pydantic) │
│ └── Loguear con print() (debería ser wrapper de infra) │
└─────────────────────────────────────────────────────────────┘
DESPUÉS (4 módulos, cada uno con una responsabilidad):
┌─────────────────────────────────────────────────────────────┐
│ prompts/sentiment/v1.yaml (8 líneas) │
│ └── El template del prompt, versionado, editable │
├─────────────────────────────────────────────────────────────┤
│ src/processing/sentiment_parser.py (45 líneas) │
│ └── extract_json() + parse_sentiment_output() │
│ Testeable de forma totalmente independiente │
├─────────────────────────────────────────────────────────────┤
│ src/infrastructure/openai_provider.py (35 líneas) │
│ └── complete() → llamada al API + logging │
│ Intercambiable con MockProvider │
├─────────────────────────────────────────────────────────────┤
│ src/domain/sentiment_service.py (20 líneas) │
│ └── analyze_sentiment() — orquesta los 3 anteriores │
│ Completamente testeable con MockProvider │
└─────────────────────────────────────────────────────────────┘
Total: similar número de líneas, pero 4 responsabilidades claras
Cuándo NO separar más
# La separación tiene un límite. Separar demasiado crea "over-engineering"
# ❌ Over-engineering: crear abstracciones sin valor real
class PromptRenderer:
class TemplateLoader:
class FileSystemAdapter:
class PathResolver:
def resolve_path(self, name: str) -> Path: ...
# Para una función que solo hace:
PROMPTS_DIR / f"{name}.yaml"
# Esto es 4 capas de abstracción para 1 línea de código
# ✅ Regla práctica: separar si...
# 1. Quieres testear cada parte por separado
# → extract_json() tiene tests, parse_sentiment_output() tiene tests
# 2. Quieres reutilizar en otro lugar
# → parse_sentiment_output() se usa en /analyze Y en /batch
# 3. El cambio afecta a una cosa, no a las demás
# → Cambiar el parser no debe afectar al LLM call
# 4. La función hace más de una cosa
# → "hace esto Y hace aquello" = separar
# ❌ NO separar si...
# - La abstracción solo existe para existir (YAGNI)
# - Navegas 5 archivos para entender una operación de 3 líneas
# - La "separación" duplica código sin reducir acoplamientoTests después de la separación
# Ahora cada parte tiene sus propios tests independientes
# tests/unit/test_sentiment_parser.py
import pytest
from src.processing.sentiment_parser import parse_sentiment_output, extract_json
class TestExtractJson:
def test_plain_json(self):
assert extract_json('{"a": 1}') == '{"a": 1}'
def test_json_in_markdown(self):
raw = '```json\n{"a": 1}\n```'
assert extract_json(raw) == '{"a": 1}'
def test_json_with_surrounding_text(self):
raw = 'Sure! Here is the result: {"sentiment": "positive"} Hope this helps!'
result = extract_json(raw)
assert '{"sentiment": "positive"}' in result
def test_no_json_returns_none(self):
assert extract_json("no json here") is None
class TestParseSentimentOutput:
def test_valid_output(self):
raw = '{"sentiment": "positive", "score": 0.8, "confidence": 0.9}'
result = parse_sentiment_output(raw)
assert result["sentiment"] == "positive"
assert result["score"] == 0.8
def test_score_clamped(self):
raw = '{"sentiment": "positive", "score": 1.5}'
result = parse_sentiment_output(raw)
assert result["score"] == 1.0 # Clamped
def test_invalid_json_returns_unknown(self):
result = parse_sentiment_output("not valid json")
assert result["sentiment"] == "unknown"
assert result["score"] == 0.0
def test_invalid_sentiment_becomes_unknown(self):
raw = '{"sentiment": "very_positive", "score": 0.9}'
result = parse_sentiment_output(raw)
assert result["sentiment"] == "unknown" # Normalizado
# tests/unit/test_sentiment_service.py
import pytest
from src.domain.sentiment_service import analyze_sentiment, LowConfidenceError
class MockProvider:
def __init__(self, response: str):
self._response = response
def complete(self, messages: list) -> str:
return self._response
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"
def test_low_confidence_raises(self):
mock = MockProvider('{"sentiment": "mixed", "score": 0.1, "confidence": 0.1}')
with pytest.raises(LowConfidenceError):
analyze_sentiment("ambiguous text", mock)
def test_parser_called_with_provider_output(self):
"""Domain delega el parsing al parser, no lo hace inline."""
mock = MockProvider("```json\n{\"sentiment\": \"negative\", \"score\": -0.5, \"confidence\": 0.8}\n```")
result = analyze_sentiment("This is bad", mock)
# El parser debe manejar el markdown code block
assert result["sentiment"] == "negative"Ejercicios
Ejercicio 1: Identificar responsabilidades
Lee esta función y lista cada responsabilidad (mínimo 4):
def generate_summary(article: str, max_words: int = 200) -> dict:
if len(article) > 20000:
article = article[:20000]
prompt = f"Summarize in {max_words} words: {article}"
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
resp = client.chat.completions.create(model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}])
raw = resp.choices[0].message.content
try:
result = json.loads(raw)
except:
result = {"summary": raw, "word_count": len(raw.split())}
if "summary" not in result:
result["summary"] = raw
print(f"Summary generated: {len(result['summary'])} chars")
return resultVer solución
- Validación/sanitización de input:
if len(article) > 20000→ debería ser guardrail o en entrypoint - Construcción del prompt:
f"Summarize in {max_words} words..."→ debería ser prompts/summarization/v1.yaml - Llamada al LLM:
client.chat.completions.create(...)→ debería serOpenAIProvider.complete() - Parseo del output:
json.loads(raw)→ debería serparse_summary_output()en processing/ - Validación del output:
if "summary" not in result→ debería serSummaryOutputPydantic model - Logging:
print(...)→ debería ser structlog en el provider wrapper
Ejercicio 2: Refactoring incremental
Propón el orden de los pasos para refactorizar la función anterior sin romper nada:
Ver guía
Orden seguro (red-green-refactor):
- Añadir tests de la función original (si no existen) → red de seguridad
- Extraer el parser a
processing/summary_parser.py+ tests del parser - Extraer el prompt a
prompts/summarization/v1.yaml+ loader - Crear MockProvider para tests
- Extraer la llamada al LLM a
OpenAIProvider.complete() - Limpiar la función principal para que solo orqueste
- Correr todos los tests → deben pasar
En cada paso: correr tests antes y después.
Resumen
- God functions son el enemigo: una función que hace 6 cosas no puede testearse, ni cambiarse, ni reutilizarse
- El refactoring incremental: extraer una responsabilidad a la vez, verificando tests en cada paso
- El límite de la separación: separar cuando facilita testing, reutilización, o mantenimiento — no por dogma
- Tests como demostración del valor: con el parser separado, sus tests son triviales y dan alta confianza
Recursos adicionales
- Single Responsibility Principle (SRP) — El principio fundacional
- Refactoring: Improving the Design of Existing Code (Fowler) — El libro de referencia
- Working Effectively with Legacy Code (Feathers) — Para refactorizar código sin tests
- YAGNI — You Ain't Gonna Need It — El principio de no over-engineer