Patrones que se ven bien en teoría
Los artículos anteriores explicaron cinco patrones agénticos por separado: Tool Use, Parallelization, Routing, Prompt Chaining y Reflection. Cada uno funciona bien de forma aislada. Pero la pregunta que nadie contestaba era: ¿Cómo se integran en un sistema coherente sin convertirse en un spaghetti de callbacks?
La respuesta está en la disciplina de diseño y en mantener cada patrón en su lugar. Cada patrón necesita una responsabilidad clara, un punto de entrada definido y un punto de salida predecible. Si no, los patrones se solapan, las dependencias se entrelazan y el sistema se vuelve imposible de mantener.
Construimos un analizador financiero trivial para demostrarlo. No es un sistema de análisis financiero real. Es un sistema donde cada patrón es visible, separado y funcional. El dominio financiero es simplemente un vehículo, tiene datos estructurados, reglas claras y un resultado fácil de entender.
Arquitectura general
El flujo sigue cinco etapas:
- Tool Use: Obtener datos de fuentes externas (financials, balance, news).
- Parallelization: Ejecutar las tres herramientas concurrentemente.
- Routing: Clasificar la empresa según métricas (growth, risk, stable).
- Prompt Chaining: Generar el informe paso a paso (resumen → riesgos → perspectivas → informe).
- Reflection: Revisar y corregir el informe con un loop Generator-Critic.
Lo que esto no hace es intentar ser inteligente. Lo que intenta es que puedas señalar cualquier bloque del código y decir «esto es Tool Use» o «esto es Reflection». Ese es el objetivo.
El código está disponible en github.
Contexto central: La decisión que simplifica todo
El primer problema que aparece al integrar múltiples patrones es la gestión de datos. ¿Cómo pasan los datos entre patrones sin convertir el código en un mapa de metro?
La solución: un contexto central compartido.
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class AnalysisContext:
"""Contexto compartido entre todos los patrones del pipeline."""
# Entrada
company: str
# Datos crudos (Tool Use + Parallelization)
financials: dict = field(default_factory=dict)
balance: dict = field(default_factory=dict)
news: list = field(default_factory=list)
# Clasificación (Routing)
company_type: Optional[str] = None
# Generación (Prompt Chaining)
summary: str = ""
risks: str = ""
outlook: str = ""
report_draft: str = ""
# Revisión (Reflection)
final_report: str = ""
reflection_iterations: int = 0
critic_feedback: str = ""Cada patrón lee y escribe en el mismo objeto. No hay que pasar 15 variables entre funciones. El código se mantiene limpio y el flujo es visible.
Esta decisión tiene un tradeoff: El contexto central es mutable, lo que introduce el riesgo de que un patrón modifique datos que otro patrón aún necesita. En este sistema no es problema porque el flujo es secuencial y cada patrón escribe en campos distintos. En sistemas más complejos, considerar un patrón de eventos o un estado inmutable.
Módulo 1: Tool Use y Parallelization
Las herramientas son funciones que obtienen datos. Tres herramientas, tres fuentes independientes:
get_financials(): ingresos, crecimiento, margen, EPS.get_balance(): efectivo, deuda, activos, pasivos.get_news(): noticias recientes sobre la empresa.
La independencia es clave. Ninguna herramienta depende de la otra. Eso las convierte en candidatas ideales para Parallelization.
async def fetch_data(ctx: AnalysisContext) -> None:
"""Tool Use + Parallelization.
Tres herramientas independientes se ejecutan concurrentemente.
"""
financials, balance, news = await asyncio.gather(
asyncio.to_thread(get_financials, ctx.company),
asyncio.to_thread(get_balance, ctx.company),
asyncio.to_thread(get_news, ctx.company),
)
ctx.financials = financials
ctx.balance = balance
ctx.news = newsasyncio.gather() ejecuta las tres herramientas concurrentemente. asyncio.to_thread() las mueve a un thread separado porque son llamadas bloqueantes. El tiempo total es el de la herramienta más lenta, no la suma de las tres.
Nótese que las herramientas son funciones sincrónicas normales. asyncio.to_thread() se encarga de ejecutarlas en un thread pool. Esto permite usar cualquier función existente sin tener que reescribirla como async.
Módulo 2: Routing
El router clasifica la empresa según métricas financieras. No necesita LLM: las reglas son claras y reproducibles.
def route_company(ctx: AnalysisContext) -> str:
"""Clasifica la empresa en un tipo basado en métricas financieras."""
financials = ctx.financials
balance = ctx.balance
revenue_growth = financials.get("revenue_growth", 0)
debt = balance.get("debt", 0)
equity = balance.get("equity", 1)
debt_ratio = debt / equity if equity > 0 else 0
if revenue_growth > 0.20:
ctx.company_type = "growth"
elif debt_ratio > 0.6:
ctx.company_type = "risk"
else:
ctx.company_type = "stable"
return ctx.company_typeTres reglas. Tres tipos de empresa. El router determinista es más rápido y reproducible que un router basado en LLM. Cuando las categorías son claras, no necesitas un modelo para decidir.
El tipo de empresa afecta los prompts en Prompt Chaining: una empresa «growth» recibe prompts enfocados en escalabilidad y valoración. Una empresa «risk» recibe prompts enfocados en solvencia y liquidez. El routing personaliza la generación sin añadir complejidad.
Módulo 3: Prompt Chaining
Aquí ocurre la generación real. El patrón Prompt Chaining descompone una tarea compleja (generar un informe financiero) en etapas secuenciales más simples y controlables.
Cuatro pasos. Cada paso recibe el output del paso anterior como input:
- Resumen ejecutivo: financials + balance →
ctx.summary - Análisis de riesgos: summary + news →
ctx.risks - Perspectivas: summary + risks + company_type →
ctx.outlook - Informe completo: summary + risks + outlook →
ctx.report_draft
async def run_chaining(ctx: AnalysisContext) -> str:
"""Ejecuta la cadena completa de prompts."""
await generate_summary(ctx)
await generate_risks(ctx)
await generate_outlook(ctx)
await generate_report(ctx)
return ctx.report_draftCada paso llama al modelo local. La llamada es simple: prompt + temperatura + max tokens.
async def call_llm(prompt: str, system: str = "", temperature: float = 0.3) -> str:
"""Llama al modelo local vía API OpenAI-compatible."""
headers = {"Content-Type": "application/json"}
messages = []
if system:
messages.append({"role": "system", "content": system})
messages.append({"role": "user", "content": prompt})
payload = {
"model": LLM_MODEL,
"messages": messages,
"temperature": temperature,
"max_tokens": 2048,
}
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{LLM_BASE_URL}/chat/completions",
json=payload,
headers=headers,
)
response.raise_for_status()
data = response.json()
return data["choices"][0]["message"]["content"]La temperatura varía según el paso: 0.3 para resumen y riesgos (necesitan creatividad controlada), 0.2 para el informe final (necesita consistencia). El timeout de 60 segundos cubre la mayoría de casos con modelos locales de 12B.
Módulo 4: Reflection
El Generator ya produjo un draft del informe. El Critic lo revisa contra los datos originales y proporciona feedback. Si hay problemas, el Generator corrige. Loop hasta OUTPUT_ACCEPTABLE o máximo 3 iteraciones.
async def reflect(ctx: AnalysisContext) -> str:
"""Ejecuta el loop de Reflection."""
for iteration in range(MAX_ITERATIONS):
ctx.reflection_iterations = iteration + 1
is_acceptable, feedback = await critic_review(ctx)
if is_acceptable:
ctx.final_report = ctx.report_draft
return ctx.final_report
# Feedback no aceptable → corregir
ctx.critic_feedback = feedback
await correct_report(ctx)
# Si llegamos aquí, se agotaron las iteraciones
ctx.final_report = ctx.report_draft
return ctx.final_reportEl Critic evalúa cuatro cosas: coherencia financiera, riesgos omitidos, afirmaciones sin soporte y contradicciones. Usa temperatura 0.1 para consistencia en la evaluación.
Por qué 3 iteraciones y no 5. Lo que vimos fue que después de la tercera iteración, la mejora marginal cae por debajo del 5% en la mayoría de casos. El coste crece linealmente pero el retorno no. Truncamos el historial después de la tercera iteración y el informe final suele ser aceptable.
Orquestador: El corazón del sistema
Todo converge en analyze_company(). Cinco pasos, cinco patrones, un contexto compartido.
async def analyze_company(company: str) -> str:
"""Orquestador principal."""
# Paso 1: Crear contexto compartido
ctx = AnalysisContext(company=company)
# Paso 2: Tool Use + Parallelization
await fetch_data(ctx)
# Paso 3: Routing
company_type = route_company(ctx)
# Paso 4: Prompt Chaining
await run_chaining(ctx)
# Paso 5: Reflection
final_report = await reflect(ctx)
# Paso 6: Guardar informe
save_report(ctx, final_report)
return final_reportEl código es limpio porque cada patrón tiene una responsabilidad clara. No hay solapamiento. No hay dependencias cruzadas. El flujo es secuencial y predecible.
Estructura del proyecto
| Archivo | Patrón | Responsabilidad |
|---|---|---|
main.py | Orquestador | Integra todos los patrones en un flujo coherente |
models.py | Contexto | Objeto compartido entre todos los patrones |
tools.py | Tool Use | Herramientas de datos (financials, balance, news) |
routing.py | Routing | Clasificación de empresa según métricas |
chaining.py | Prompt Chaining | Generación paso a paso del informe |
reflection.py | Reflection | Revisión y corrección con Generator-Critic |
prompts.py | Prompts | Templates de prompts para cada etapa |
Lo que aprendimos construyéndolo
El primer bug que encontramos fue silencioso.
El código estaba configurado con gemma-3-12b como modelo, pero el modelo en el servidor era google/gemma3-12b. El endpoint redirigía automáticamente a qwen3.6-27b como fallback, un modelo de razonamiento que no exponía la salida en el campo content.
Nuestro pipeline leía únicamente content. El resultado era sistemáticamente vacío en cada etapa del flujo. El sistema no fallaba ni lanzaba excepciones: Producía informes con estructura válida en Markdown, pero sin contenido real, solo plantillas.
Este es uno de los fallos más peligrosos en sistemas agénticos: La ejecución es correcta, pero la semántica está vacía.
Segundo problema: Prompts con placeholders interpretados literalmente
En el prompt del informe utilizábamos instrucciones del tipo:
[Desarrolla el resumen con análisis de métricas clave]
El modelo interpretaba estos bloques como texto literal, no como instrucciones. El resultado era un draft compuesto por instrucciones sin ejecutar.
El problema se amplifica en pipelines multi-etapa: El Critic detecta inconsistencias, pero al reinyectar el estado, el sistema sigue operando sobre una base vacía. El bucle se vuelve estéril.
Tercer problema: Portabilidad del entorno
El modelo y el endpoint estaban hardcodeados. En entornos donde el modelo no existía, la API devolvía 400 Bad Request.
La solución fue introducir una capa de auto-detección:
- Consulta
/modelsal iniciar - Validación de disponibilidad del modelo
- Fallback automático a un modelo compatible
Esto elimina una clase completa de errores de despliegue silencioso.
Lecciones concretas
La simplicidad en configuración es deuda técnica diferida. Variables de entorno y auto-detección reducen fallos de producción.
Un modelo de razonamiento no es equivalente a un modelo instructivo. La diferencia no es conceptual, es de interfaz: la salida puede estar en campos distintos.
Los placeholders en prompts ([haz esto]) degradan la ejecución del modelo en lugar de guiarla. Deben separarse claramente instrucciones y datos.
En pipelines agénticos, los errores silenciosos se propagan. Si el primer nodo produce vacío, el sistema entero ejecuta correctamente sobre entrada inválida.
Los errores de entorno pueden simular errores de modelo. Siempre hay que validar capa de inferencia antes de depurar prompts.
Lo importante no son los cinco patrones
El objetivo del proyecto no era construir un analista financiero. El objetivo era demostrar que los patrones agénticos pueden componerse de forma limpia cuando cada uno tiene una responsabilidad bien definida.
El mismo diseño podría reutilizarse para:
- Análisis legal.
- Investigación científica.
- Auditoría de código.
- Análisis de contratos.
- Revisión documental.
- Inteligencia competitiva.
Cambiando las herramientas y los prompts, la arquitectura permanece prácticamente igual. No se trata de enseñar un analista financiero. Se trata de enseñar una forma de diseñar agentes. Y esa diferencia es enorme.
Conclusión
Cinco patrones. Un contexto compartido. Un flujo coherente.
Lo que este proyecto demuestra no es que los patrones agenticos sean complejos. Demuestra que la complejidad no está en los patrones individuales, está en integrarlos sin que se solapen. Cuando cada patrón tiene una responsabilidad clara, el sistema se mantiene limpio incluso con cinco componentes.
El objetivo nunca fue construir el mejor analista financiero posible. El objetivo era demostrar que los patrones agenticos pueden convivir sin convertirse en un sistema imposible de mantener. Si después de leer el código puedes identificar dónde empieza y termina cada patrón, el experimento ha funcionado.