""" agent_pipeline.py — Pipeline idiomático de orquestación de agentes IA. CULTIVA IA · IA-Ingeniería-MLOps

Demuestra patrones pythónicos aplicados a un dominio real de agencia:

• Type hints modernos (Python 3.10+)
• Dataclasses con validación (post_init)
• EAFP en lugar de LBYL
• Jerarquía de excepciones propias
• Context managers con @contextmanager
• Generadores para procesamiento perezoso
• Decoradores (timer, retry)
• Concurrencia con ThreadPoolExecutor
• slots para modelos de datos frecuentes
• Organización de imports (stdlib → terceros → locales) """

──────────────────────────────────────────────

stdlib

──────────────────────────────────────────────

import json import logging import time import functools import concurrent.futures from contextlib import contextmanager from dataclasses import dataclass, field from datetime import datetime from enum import Enum, auto from io import StringIO from pathlib import Path from typing import Any, Callable, Iterator, NamedTuple, Protocol, TypeVar

──────────────────────────────────────────────

Configuración de logging (explícita, sin side-effects ocultos)

──────────────────────────────────────────────

logging.basicConfig( level=logging.INFO, format="%(asctime)s · %(name)s · %(levelname)s · %(message)s", datefmt="%H:%M:%S", ) logger = logging.getLogger("cultiva.pipeline")

──────────────────────────────────────────────

Type aliases

──────────────────────────────────────────────

JSON = dict[str, Any] | list[Any] | str | int | float | bool | None AgentResult = dict[str, str | float | list[str]] T = TypeVar("T")

──────────────────────────────────────────────

Jerarquía de excepciones propia

──────────────────────────────────────────────

class PipelineError(Exception): """Base de todos los errores del pipeline."""

class BriefValidationError(PipelineError): """El brief recibido no supera la validación."""

class AgentNotFoundError(PipelineError): """No existe un agente registrado para el tipo solicitado."""

class AgentTimeoutError(PipelineError): """El agente superó el tiempo máximo de ejecución."""

──────────────────────────────────────────────

Enums

──────────────────────────────────────────────

class TaskType(str, Enum): COPY = "copy" SEO = "seo" ADS = "ads" EMAIL = "email" AUTOMATION = "automation"

class BriefStatus(Enum): PENDING = auto() PROCESSING = auto() DONE = auto() FAILED = auto()

──────────────────────────────────────────────

Modelos de datos — dataclasses con validación

──────────────────────────────────────────────

@dataclass class ClientBrief: """Brief de cliente que entra al pipeline.""" task_type: TaskType description: str company: str budget_eur: float received_at: datetime = field(default_factory=datetime.now) status: BriefStatus = BriefStatus.PENDING

def __post_init__(self) -> None:
    if not self.description.strip():
        raise BriefValidationError("La descripción del brief no puede estar vacía.")
    if self.budget_eur < 0:
        raise BriefValidationError(
            f"El presupuesto no puede ser negativo: {self.budget_eur}"
        )
    if len(self.company) < 2:
        raise BriefValidationError(
            f"Nombre de empresa demasiado corto: '{self.company}'"
        )

@dataclass class PipelineConfig: """Configuración inmutable del pipeline.""" max_workers: int = 4 timeout_sec: float = 30.0 log_level: str = "INFO" output_dir: Path = field(default_factory=lambda: Path("./output"))

def __post_init__(self) -> None:
    if self.max_workers < 1:
        raise ValueError("max_workers debe ser >= 1")
    self.output_dir = Path(self.output_dir)

NamedTuple para resultados inmutables (ligeros, comparables)

class AgentReport(NamedTuple): agent_id: str task_type: TaskType tokens_used: int duration_sec: float output: str

──────────────────────────────────────────────

Protocol — duck typing para agentes

──────────────────────────────────────────────

class AgentProtocol(Protocol): agent_id: str

def run(self, brief: ClientBrief) -> AgentReport:
    """Ejecuta la tarea y devuelve un AgentReport."""
    ...

──────────────────────────────────────────────

Decoradores

──────────────────────────────────────────────

def timer(func: Callable[..., T]) -> Callable[..., T]: """Mide y loguea el tiempo de ejecución de cualquier función.""" @functools.wraps(func) def wrapper(*args: Any, **kwargs: Any) -> T: start = time.perf_counter() result = func(*args, **kwargs) elapsed = time.perf_counter() - start logger.info("%s completado en %.3fs", func.name, elapsed) return result return wrapper

def retry(max_attempts: int = 3, delay_sec: float = 1.0): """Reintenta la función hasta max_attempts veces ante excepciones transitorias.""" def decorator(func: Callable[..., T]) -> Callable[..., T]: @functools.wraps(func) def wrapper(*args: Any, **kwargs: Any) -> T: last_exc: Exception | None = None for attempt in range(1, max_attempts + 1): try: return func(*args, **kwargs) except PipelineError: raise # no reintentar errores de negocio except Exception as exc: last_exc = exc logger.warning( "Intento %d/%d fallido en %s: %s", attempt, max_attempts, func.name, exc, ) time.sleep(delay_sec * attempt) raise RuntimeError( f"{func.name} falló tras {max_attempts} intentos" ) from last_exc return wrapper return decorator

──────────────────────────────────────────────

Context managers

──────────────────────────────────────────────

@contextmanager def pipeline_span(name: str) -> Iterator[None]: """Traza el inicio/fin de una fase del pipeline con tiempo.""" logger.info("[INICIO] %s", name) start = time.perf_counter() try: yield finally: elapsed = time.perf_counter() - start logger.info("[FIN] %s — %.3fs", name, elapsed)

──────────────────────────────────────────────

Agentes concretos (simulados con slots)

──────────────────────────────────────────────

class _BaseAgent: """Agente base ligero con slots para optimizar memoria.""" slots = ("agent_id",)

def __init__(self, agent_id: str) -> None:
    self.agent_id = agent_id

def run(self, brief: ClientBrief) -> AgentReport:
    raise NotImplementedError

class CopywritingAgent(_BaseAgent): slots = ("agent_id", "tone")

def __init__(self, tone: str = "persuasivo") -> None:
    super().__init__("copy-v2")
    self.tone = tone

def run(self, brief: ClientBrief) -> AgentReport:
    time.sleep(0.05)  # simula LLM call
    output = (
        f"[{self.tone.upper()}] Texto para {brief.company}: "
        f"{brief.description[:60]}... — CTA: ¡Descúbrelo hoy!"
    )
    return AgentReport(
        agent_id=self.agent_id,
        task_type=brief.task_type,
        tokens_used=312,
        duration_sec=0.05,
        output=output,
    )

class SEOAgent(_BaseAgent): slots = ("agent_id",)

def __init__(self) -> None:
    super().__init__("seo-v1")

def run(self, brief: ClientBrief) -> AgentReport:
    time.sleep(0.08)
    keywords = ["IA para empresas", f"{brief.company} automatización", "ROI IA"]
    output = f"Keywords objetivo: {', '.join(keywords)}. H1 optimizado generado."
    return AgentReport(
        agent_id=self.agent_id,
        task_type=brief.task_type,
        tokens_used=198,
        duration_sec=0.08,
        output=output,
    )

class AdsAgent(_BaseAgent): slots = ("agent_id",)

def __init__(self) -> None:
    super().__init__("ads-v1")

def run(self, brief: ClientBrief) -> AgentReport:
    time.sleep(0.06)
    output = (
        f"Titular A: '¿Tu empresa sin IA en 2025?' | "
        f"Titular B: 'Automatiza {brief.company} en 30 días' | "
        f"Presupuesto sugerido: {brief.budget_eur * 0.15:.0f} €/mes"
    )
    return AgentReport(
        agent_id=self.agent_id,
        task_type=brief.task_type,
        tokens_used=224,
        duration_sec=0.06,
        output=output,
    )

──────────────────────────────────────────────

Registro de agentes

──────────────────────────────────────────────

AGENT_REGISTRY: dict[TaskType, AgentProtocol] = { TaskType.COPY: CopywritingAgent(tone="persuasivo"), TaskType.SEO: SEOAgent(), TaskType.ADS: AdsAgent(), }

def get_agent(task_type: TaskType) -> AgentProtocol: """EAFP: intenta obtener el agente, falla explícitamente si no existe.""" try: return AGENT_REGISTRY[task_type] except KeyError: raise AgentNotFoundError( f"No hay agente registrado para '{task_type.value}'. " f"Disponibles: {[t.value for t in AGENT_REGISTRY]}" ) from None

──────────────────────────────────────────────

Parser de briefs — generador (lazy)

──────────────────────────────────────────────

def parse_briefs(raw_data: list[JSON]) -> Iterator[ClientBrief]: """ Parsea una lista de dicts JSON como ClientBrief. Usa generador: no carga todos en memoria a la vez. """ for item in raw_data: if not isinstance(item, dict): logger.warning("Elemento ignorado (no es dict): %r", item) continue try: yield ClientBrief( task_type=TaskType(item["task_type"]), description=item["description"], company=item["company"], budget_eur=float(item.get("budget_eur", 0)), ) except (KeyError, ValueError, BriefValidationError) as exc: logger.error("Brief inválido ignorado: %s — %r", exc, item)

──────────────────────────────────────────────

Formateo de resultados — join en lugar de += en bucle

──────────────────────────────────────────────

def format_report(reports: list[AgentReport]) -> str: """Construye el informe de resultados. Usa join (O(n)) en lugar de += (O(n²)).""" buffer = StringIO() buffer.write("=" * 60 + "\n") buffer.write(" INFORME DE EJECUCIÓN · CULTIVA IA PIPELINE\n") buffer.write("=" * 60 + "\n\n") for r in reports: buffer.write(f" Agente : {r.agent_id}\n") buffer.write(f" Tarea : {r.task_type.value}\n") buffer.write(f" Tokens usados: {r.tokens_used}\n") buffer.write(f" Duración : {r.duration_sec:.3f}s\n") buffer.write(f" Output : {r.output}\n") buffer.write("-" * 60 + "\n") total_tokens = sum(r.tokens_used for r in reports) buffer.write(f"\n TOTAL tokens: {total_tokens}\n") return buffer.getvalue()

──────────────────────────────────────────────

Pipeline principal

──────────────────────────────────────────────

@timer def run_pipeline( raw_briefs: list[JSON], config: PipelineConfig | None = None, ) -> list[AgentReport]: """ Orquesta la ejecución concurrente de agentes sobre los briefs recibidos. Devuelve la lista de AgentReport completados. """ if config is None: config = PipelineConfig()

reports: list[AgentReport] = []

with pipeline_span("parse_briefs"):
    briefs = list(parse_briefs(raw_briefs))
    logger.info("%d briefs válidos para procesar.", len(briefs))

with pipeline_span("dispatch_agents"), \
     concurrent.futures.ThreadPoolExecutor(max_workers=config.max_workers) as pool:

    future_map: dict[concurrent.futures.Future[AgentReport], ClientBrief] = {}

    for brief in briefs:
        try:
            agent = get_agent(brief.task_type)
        except AgentNotFoundError as exc:
            logger.warning("Skipping brief: %s", exc)
            continue

        brief.status = BriefStatus.PROCESSING
        future_map[pool.submit(agent.run, brief)] = brief

    for future in concurrent.futures.as_completed(
        future_map, timeout=config.timeout_sec
    ):
        brief = future_map[future]
        try:
            report = future.result()
            brief.status = BriefStatus.DONE
            reports.append(report)
            logger.info(
                "OK · %s · %s · %d tokens",
                brief.company, brief.task_type.value, report.tokens_used,
            )
        except Exception as exc:
            brief.status = BriefStatus.FAILED
            logger.error("FAIL · %s · %s", brief.company, exc)

return reports

──────────────────────────────────────────────

Entrypoint de demostración

──────────────────────────────────────────────

if name == "main": # Datos de ejemplo: briefs de clientes ficticios de CULTIVA IA sample_briefs: list[JSON] = [ { "task_type": "copy", "description": "Landing page para lanzamiento de herramienta de análisis de datos con IA para PYMEs industriales.", "company": "DataForge SL", "budget_eur": 2400, }, { "task_type": "seo", "description": "Posicionamiento de blog técnico sobre automatización con n8n y Make para e-commerce.", "company": "AutoFlow Agency", "budget_eur": 800, }, { "task_type": "ads", "description": "Campaña de captación para consultora de transformación digital en el sector salud.", "company": "HealthTech Partners", "budget_eur": 5000, }, { "task_type": "email", # sin agente registrado → warning controlado "description": "Secuencia de nurturing para leads de webinar sobre IA generativa.", "company": "LearnAI Corp", "budget_eur": 600, }, { # brief inválido: descripción vacía → BriefValidationError capturada "task_type": "copy", "description": " ", "company": "BrokenClient", "budget_eur": 100, }, ]

config = PipelineConfig(max_workers=3, timeout_sec=15.0)

print("\n" + "=" * 60)
print("  CULTIVA IA · Pipeline de Agentes IA (demo)")
print("=" * 60 + "\n")

results = run_pipeline(sample_briefs, config)
print("\n" + format_report(results))