Módulo 5: ReAct, Self-Consistency y Patrones Avanzados

8. Proyecto: Multi-Strategy Problem Solver

Siguiente →

Descripción

En este proyecto construirás un sistema completo que recibe cualquier tipo de problema, lo clasifica automáticamente, selecciona la técnica de prompt engineering más adecuada, ejecuta esa técnica, y opcionalmente compara los resultados de múltiples técnicas para validar la respuesta.

Este proyecto integra todo lo aprendido en el módulo: ReAct, Self-Consistency, Tree-of-Thought, Meta-prompting y Self-Refine.

Arquitectura del Sistema

                    MULTI-STRATEGY PROBLEM SOLVER
                    
    Input: Problema/Pregunta
           │
           ▼
    ┌─────────────────┐
    │  CLASIFICADOR   │  → tipo: math/factual/reasoning/creative/code/planning
    └────────┬────────┘
             │
             ▼
    ┌─────────────────┐
    │     ROUTER      │  → selecciona técnica óptima según tipo + constraints
    └────────┬────────┘
             │
    ┌────────┴──────────────────────────────────┐
    │                                           │
    ▼                                           ▼
 Técnica A                                  Técnica B (si comparación)
 (principal)                                (alternativa)
    │                                           │
    └────────────────────┬──────────────────────┘
                         ▼
                  ┌──────────────┐
                  │  EVALUADOR   │  → score calidad, confianza
                  └──────┬───────┘
                         │
                         ▼
                  Output estructurado
                  {respuesta, tecnica_usada, confianza, metricas}

Implementación Completa

from openai import OpenAI
from collections import Counter
from dataclasses import dataclass, field
from typing import Optional, Any
import json
import re
import time
import math

client = OpenAI()

# ============================================================
# MODELOS DE DATOS
# ============================================================

@dataclass
class ProblemaClasificado:
    texto: str
    tipo: str                     # math, factual, reasoning, creative, code, planning, qa_realtime
    complejidad: str              # simple, media, alta
    necesita_externos: bool
    accuracy_critica: bool
    tiene_respuesta_unica: bool   # False para creative

@dataclass
class ResultadoTecnica:
    tecnica: str
    respuesta: str
    tiempo_segundos: float
    n_llamadas: int
    confianza: float
    metadata: dict = field(default_factory=dict)

@dataclass
class ResultadoFinal:
    problema: str
    clasificacion: ProblemaClasificado
    resultado_principal: ResultadoTecnica
    resultado_alternativo: Optional[ResultadoTecnica]
    respuesta_consenso: str
    metricas: dict
    recomendacion_tecnica: str


# ============================================================
# MÓDULO 1: CLASIFICADOR
# ============================================================

def clasificar_problema(problema: str) -> ProblemaClasificado:
    """
    Analiza un problema y lo clasifica para el routing.
    
    Args:
        problema: El texto del problema a clasificar
    
    Returns:
        ProblemaClasificado con todos los atributos necesarios para el routing
    """
    prompt_clasificacion = f"""Analiza el siguiente problema y clasifícalo con precisión.

Problema: {problema}

Responde en JSON con este schema exacto:
{{
    "tipo": "math|factual|reasoning|creative|code|planning|qa_realtime",
    "complejidad": "simple|media|alta",
    "necesita_externos": true|false,
    "accuracy_critica": true|false,
    "tiene_respuesta_unica": true|false,
    "razon_tipo": "explicación en 10 palabras"
}}

Guía de tipos:
- math: operaciones numéricas, álgebra, cálculo, estadística
- factual: hechos que el modelo conoce del entrenamiento
- reasoning: deducciones lógicas, inferencias, multi-paso sin cálculo
- creative: escritura, generación, sin respuesta correcta única
- code: escribir, revisar, o depurar código
- planning: diseñar estrategias, arquitecturas, roadmaps
- qa_realtime: preguntas sobre datos actuales (precios, noticias, fechas)

Guía necesita_externos: true si la respuesta cambia con el tiempo o requiere APIs/BD
Guía accuracy_critica: true si errores tienen alto impacto (finanzas, medicina, legal)"""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt_clasificacion}],
        temperature=0,
        response_format={"type": "json_object"}
    )
    
    try:
        datos = json.loads(response.choices[0].message.content)
        return ProblemaClasificado(
            texto=problema,
            tipo=datos.get("tipo", "reasoning"),
            complejidad=datos.get("complejidad", "media"),
            necesita_externos=datos.get("necesita_externos", False),
            accuracy_critica=datos.get("accuracy_critica", False),
            tiene_respuesta_unica=datos.get("tiene_respuesta_unica", True)
        )
    except Exception as e:
        # Fallback conservador
        return ProblemaClasificado(
            texto=problema,
            tipo="reasoning",
            complejidad="media",
            necesita_externos=False,
            accuracy_critica=False,
            tiene_respuesta_unica=True
        )


# ============================================================
# MÓDULO 2: IMPLEMENTACIONES DE TÉCNICAS
# ============================================================

def _extraer_numero(texto: str) -> str:
    """Extrae el número final de un texto de razonamiento."""
    patrones = [
        r'(?:respuesta|answer|resultado)[:\s=]+(-?\d+\.?\d*)',
        r'\*\*(-?\d+\.?\d*)\*\*',
        r'=\s*(-?\d+\.?\d*)\s*$'
    ]
    for p in patrones:
        m = re.search(p, texto, re.IGNORECASE | re.MULTILINE)
        if m:
            return m.group(1)
    nums = re.findall(r'-?\d+\.?\d*', texto)
    return nums[-1] if nums else texto.strip()[-20:]

def ejecutar_zero_shot(problema: str) -> ResultadoTecnica:
    """Resolución directa sin técnica especial."""
    t0 = time.time()
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": problema}],
        temperature=0.3,
        max_tokens=400
    ).choices[0].message.content
    return ResultadoTecnica(
        tecnica="zero_shot",
        respuesta=resp,
        tiempo_segundos=time.time() - t0,
        n_llamadas=1,
        confianza=0.7
    )

def ejecutar_cot(problema: str) -> ResultadoTecnica:
    """Chain-of-Thought: razonamiento paso a paso."""
    t0 = time.time()
    prompt = f"""{problema}

Piensa paso a paso:
1. Identifica qué sabes y qué necesitas encontrar
2. Aplica el proceso correcto paso a paso
3. Verifica si tu respuesta tiene sentido
4. Escribe: "Respuesta final: [respuesta]"
"""
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
        max_tokens=500
    ).choices[0].message.content
    return ResultadoTecnica(
        tecnica="chain_of_thought",
        respuesta=resp,
        tiempo_segundos=time.time() - t0,
        n_llamadas=1,
        confianza=0.82,
        metadata={"tiene_razonamiento": True}
    )

def ejecutar_self_consistency(problema: str, n: int = 5) -> ResultadoTecnica:
    """Self-Consistency: N respuestas con majority vote."""
    t0 = time.time()
    respuestas = []
    
    for _ in range(n):
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"{problema}\nPiensa paso a paso. Respuesta: [valor]"}],
            temperature=0.7,
            max_tokens=400
        ).choices[0].message.content
        respuestas.append(_extraer_numero(resp))
    
    conteo = Counter(respuestas)
    ganadora, votos = conteo.most_common(1)[0]
    confianza = votos / n
    
    return ResultadoTecnica(
        tecnica="self_consistency",
        respuesta=ganadora,
        tiempo_segundos=time.time() - t0,
        n_llamadas=n,
        confianza=confianza,
        metadata={"n_muestras": n, "distribucion": dict(conteo), "votos_ganador": votos}
    )

def ejecutar_react(problema: str) -> ResultadoTecnica:
    """
    ReAct simplificado: razonamiento con herramientas de cálculo y búsqueda.
    """
    t0 = time.time()
    
    TOOLS = [
        {
            "type": "function",
            "function": {
                "name": "calcular",
                "description": "Evalúa expresiones matemáticas. Usa para cálculos numéricos.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "expresion": {"type": "string", "description": "Expresión a calcular"}
                    },
                    "required": ["expresion"]
                }
            }
        },
        {
            "type": "function",
            "function": {
                "name": "obtener_dato",
                "description": "Simula obtener un dato factual actual. En producción: llama API real.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "consulta": {"type": "string"}
                    },
                    "required": ["consulta"]
                }
            }
        }
    ]
    
    messages = [
        {
            "role": "system",
            "content": "Usa herramientas para obtener datos y calcular. No inventes números. Cuando tengas la respuesta, termina con 'Respuesta: [resultado]'"
        },
        {"role": "user", "content": problema}
    ]
    
    n_llamadas = 1
    herramientas_usadas = []
    
    for _ in range(6):
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
            temperature=0
        )
        msg = response.choices[0].message
        n_llamadas += 1
        
        if not msg.tool_calls:
            respuesta_final = msg.content or "Sin respuesta"
            break
        
        messages.append({
            "role": "assistant",
            "content": msg.content or "",
            "tool_calls": [tc.model_dump() for tc in msg.tool_calls]
        })
        
        for tc in msg.tool_calls:
            tool_name = tc.function.name
            args = json.loads(tc.function.arguments)
            herramientas_usadas.append(tool_name)
            
            if tool_name == "calcular":
                expr = args.get("expresion", "0")
                try:
                    # Safe eval solo operaciones matemáticas
                    allowed = set("0123456789+-*/.() ")
                    if all(c in allowed for c in expr):
                        resultado = str(round(eval(expr, {"__builtins__": {}}, {"math": math}), 4))
                    else:
                        resultado = "Expresión no permitida"
                except Exception as e:
                    resultado = f"Error: {e}"
            else:
                resultado = f"[Dato simulado para: {args.get('consulta', '')}]"
            
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": resultado
            })
    else:
        respuesta_final = "Límite de pasos alcanzado"
    
    return ResultadoTecnica(
        tecnica="react",
        respuesta=respuesta_final,
        tiempo_segundos=time.time() - t0,
        n_llamadas=n_llamadas,
        confianza=0.85 if herramientas_usadas else 0.7,
        metadata={"herramientas_usadas": herramientas_usadas}
    )

def ejecutar_tot_simplificado(problema: str, breadth: int = 3) -> ResultadoTecnica:
    """Tree-of-Thought simplificado: generar enfoques, evaluar, continuar con el mejor."""
    t0 = time.time()
    n_llamadas = 0
    
    # Generar enfoques
    resp_enfoques = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""Problema: {problema}

Genera {breadth} enfoques DISTINTOS para resolver este problema.
Cada uno debe ser una estrategia diferente.
Formato: "1. [enfoque]" por línea."""}],
        temperature=0.8,
        max_tokens=300
    )
    n_llamadas += 1
    
    lineas = [l.lstrip('0123456789.-) ').strip() 
              for l in resp_enfoques.choices[0].message.content.split('\n')
              if l.strip() and (l.strip()[0].isdigit() or l.strip()[0] == '-')]
    enfoques = lineas[:breadth]
    
    if not enfoques:
        enfoques = ["Enfoque directo paso a paso"]
    
    # Evaluar enfoques
    mejores = []
    for enfoque in enfoques:
        eval_resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"Problema: {problema}\nEnfoque: {enfoque}\n¿Prometedor? Score 0.0-1.0. Solo el número."}],
            temperature=0,
            max_tokens=10
        )
        n_llamadas += 1
        try:
            score = float(eval_resp.choices[0].message.content.strip()[:4])
            score = max(0.0, min(1.0, score))
        except ValueError:
            score = 0.5
        mejores.append((score, enfoque))
    
    mejor_enfoque = max(mejores, key=lambda x: x[0])[1]
    
    # Resolver con el mejor enfoque
    solucion = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""Problema: {problema}

Empezando con: {mejor_enfoque}

Continúa paso a paso hasta la solución completa.
Al final: "Respuesta: [respuesta]"
"""}],
        temperature=0,
        max_tokens=500
    )
    n_llamadas += 1
    
    return ResultadoTecnica(
        tecnica="tree_of_thought",
        respuesta=solucion.choices[0].message.content,
        tiempo_segundos=time.time() - t0,
        n_llamadas=n_llamadas,
        confianza=0.78,
        metadata={"mejor_enfoque": mejor_enfoque, "n_enfoques_evaluados": len(enfoques)}
    )

def ejecutar_self_refine(problema: str, max_iter: int = 2) -> ResultadoTecnica:
    """Self-Refine: generar, criticar, mejorar."""
    t0 = time.time()
    n_llamadas = 0
    
    # Generación inicial
    resp_inicial = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": problema}],
        temperature=0.3,
        max_tokens=500
    ).choices[0].message.content
    n_llamadas += 1
    
    respuesta = resp_inicial
    score_actual = 0.5
    
    for _ in range(max_iter):
        # Crítica
        critica_resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"""Problema: {problema}
Respuesta propuesta: {respuesta}

Evalúa: ¿Hay errores? ¿Falta algo? ¿Se puede mejorar?
Responde en JSON: {{"score": 0.0-1.0, "mejoras": ["mejora1", "mejora2"], "puede_mejorar": true/false}}"""}],
            temperature=0,
            response_format={"type": "json_object"}
        ).choices[0].message.content
        n_llamadas += 1
        
        try:
            critica = json.loads(critica_resp)
            score = float(critica.get("score", 0.5))
            mejoras = critica.get("mejoras", [])
            puede_mejorar = critica.get("puede_mejorar", True)
        except Exception:
            break
        
        if not puede_mejorar or score >= 0.88:
            score_actual = score
            break
        
        # Refinamiento
        mejoras_str = "\n".join([f"- {m}" for m in mejoras[:3]])
        refinada = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"""Problema: {problema}
Respuesta anterior: {respuesta}
Mejoras requeridas:
{mejoras_str}

Genera respuesta mejorada incorporando todas las mejoras."""}],
            temperature=0.2,
            max_tokens=500
        ).choices[0].message.content
        n_llamadas += 1
        
        respuesta = refinada
        score_actual = score
    
    return ResultadoTecnica(
        tecnica="self_refine",
        respuesta=respuesta,
        tiempo_segundos=time.time() - t0,
        n_llamadas=n_llamadas,
        confianza=min(score_actual, 0.95),
        metadata={"iteraciones": max_iter}
    )


# ============================================================
# MÓDULO 3: ROUTER
# ============================================================

def seleccionar_y_ejecutar(
    clasificacion: ProblemaClasificado,
    modo_comparacion: bool = False,
    n_sc: int = 5
) -> tuple[ResultadoTecnica, Optional[ResultadoTecnica]]:
    """
    Selecciona y ejecuta la técnica según la clasificación del problema.
    
    Args:
        clasificacion: Resultado del clasificador
        modo_comparacion: Si True, también ejecuta una técnica alternativa para comparar
        n_sc: Número de muestras para Self-Consistency
    
    Returns:
        Tupla (resultado_principal, resultado_alternativo)
    """
    tipo = clasificacion.tipo
    accuracy = clasificacion.accuracy_critica
    externos = clasificacion.necesita_externos
    complejidad = clasificacion.complejidad
    tiene_unica = clasificacion.tiene_respuesta_unica
    
    # Selección de técnica principal
    if externos:
        principal = ejecutar_react(clasificacion.texto)
    elif tipo == "math" and accuracy and complejidad in ["media", "alta"]:
        principal = ejecutar_self_consistency(clasificacion.texto, n=n_sc)
    elif tipo == "math":
        principal = ejecutar_cot(clasificacion.texto)
    elif tipo in ["reasoning", "logico"] and complejidad == "alta":
        principal = ejecutar_self_consistency(clasificacion.texto, n=3)
    elif tipo in ["reasoning", "logico"]:
        principal = ejecutar_cot(clasificacion.texto)
    elif tipo == "planning" and complejidad == "alta":
        principal = ejecutar_tot_simplificado(clasificacion.texto)
    elif tipo == "code":
        principal = ejecutar_self_refine(clasificacion.texto, max_iter=2)
    elif tipo == "creative" or not tiene_unica:
        principal = ejecutar_zero_shot(clasificacion.texto)
    else:
        # factual, qa_realtime, default
        principal = ejecutar_cot(clasificacion.texto) if externos else ejecutar_zero_shot(clasificacion.texto)
    
    # Técnica alternativa para comparación
    alternativa = None
    if modo_comparacion:
        if principal.tecnica == "chain_of_thought":
            alternativa = ejecutar_self_consistency(clasificacion.texto, n=3)
        elif principal.tecnica == "self_consistency":
            alternativa = ejecutar_cot(clasificacion.texto)
        elif principal.tecnica == "tree_of_thought":
            alternativa = ejecutar_cot(clasificacion.texto)
        elif principal.tecnica == "self_refine":
            alternativa = ejecutar_cot(clasificacion.texto)
    
    return principal, alternativa


# ============================================================
# MÓDULO 4: EVALUADOR Y CONSENSO
# ============================================================

def evaluar_calidad_respuesta(problema: str, respuesta: str) -> dict:
    """
    Evalúa la calidad de una respuesta sin conocer la respuesta correcta.
    """
    prompt = f"""Evalúa esta respuesta:

Problema: {problema}
Respuesta: {respuesta}

Responde en JSON:
{{
    "score_completitud": 0.0-1.0,
    "score_precision": 0.0-1.0,
    "score_claridad": 0.0-1.0,
    "tiene_respuesta": true/false,
    "posibles_errores": ["error si existe"]
}}"""
    
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
        response_format={"type": "json_object"}
    ).choices[0].message.content
    
    try:
        datos = json.loads(resp)
        score_promedio = (
            datos.get("score_completitud", 0.5) * 0.4 +
            datos.get("score_precision", 0.5) * 0.4 +
            datos.get("score_claridad", 0.5) * 0.2
        )
        return {**datos, "score_promedio": score_promedio}
    except Exception:
        return {"score_promedio": 0.5, "tiene_respuesta": True, "posibles_errores": []}

def determinar_consenso(
    resultado_principal: ResultadoTecnica,
    resultado_alternativo: Optional[ResultadoTecnica],
    clasificacion: ProblemaClasificado
) -> str:
    """
    Determina la respuesta de consenso cuando hay dos técnicas.
    """
    if resultado_alternativo is None:
        return resultado_principal.respuesta
    
    # Para math: comparar respuestas numéricas
    if clasificacion.tipo == "math":
        num_p = _extraer_numero(resultado_principal.respuesta)
        num_a = _extraer_numero(resultado_alternativo.respuesta)
        if num_p == num_a:
            return resultado_principal.respuesta  # Consenso
        # Diferencia: elegir la de mayor confianza
        if resultado_principal.confianza >= resultado_alternativo.confianza:
            return resultado_principal.respuesta
        return resultado_alternativo.respuesta
    
    # Para otras: elegir la de mayor score de calidad
    eval_p = evaluar_calidad_respuesta(clasificacion.texto, resultado_principal.respuesta)
    eval_a = evaluar_calidad_respuesta(clasificacion.texto, resultado_alternativo.respuesta)
    
    if eval_p["score_promedio"] >= eval_a["score_promedio"]:
        return resultado_principal.respuesta
    return resultado_alternativo.respuesta


# ============================================================
# SISTEMA PRINCIPAL
# ============================================================

def resolver_problema(
    problema: str,
    modo_comparacion: bool = False,
    n_sc: int = 5,
    verbose: bool = True
) -> ResultadoFinal:
    """
    Sistema principal del Multi-Strategy Problem Solver.
    
    Args:
        problema: El problema o pregunta a resolver
        modo_comparacion: Si True, ejecuta dos técnicas y compara
        n_sc: N para Self-Consistency si aplica
        verbose: Si True, imprime el proceso
    
    Returns:
        ResultadoFinal con respuesta, métricas y metadata completa
    """
    t_inicio = time.time()
    
    if verbose:
        print(f"\n{'='*60}")
        print(f"MULTI-STRATEGY PROBLEM SOLVER")
        print(f"{'='*60}")
        print(f"Problema: {problema[:80]}...")
    
    # Paso 1: Clasificar
    if verbose:
        print("\n[1/4] Clasificando problema...")
    clasificacion = clasificar_problema(problema)
    
    if verbose:
        print(f"  Tipo: {clasificacion.tipo}")
        print(f"  Complejidad: {clasificacion.complejidad}")
        print(f"  Datos externos: {clasificacion.necesita_externos}")
        print(f"  Accuracy crítica: {clasificacion.accuracy_critica}")
    
    # Paso 2: Seleccionar y ejecutar técnica
    if verbose:
        print("\n[2/4] Seleccionando y ejecutando técnica...")
    
    resultado_principal, resultado_alternativo = seleccionar_y_ejecutar(
        clasificacion, modo_comparacion, n_sc
    )
    
    if verbose:
        print(f"  Técnica principal: {resultado_principal.tecnica}")
        print(f"  Llamadas API: {resultado_principal.n_llamadas}")
        print(f"  Tiempo: {resultado_principal.tiempo_segundos:.2f}s")
        if resultado_alternativo:
            print(f"  Técnica alternativa: {resultado_alternativo.tecnica}")
    
    # Paso 3: Evaluar calidad
    if verbose:
        print("\n[3/4] Evaluando calidad...")
    evaluacion = evaluar_calidad_respuesta(problema, resultado_principal.respuesta)
    
    if verbose:
        print(f"  Score calidad: {evaluacion.get('score_promedio', 0):.2f}")
    
    # Paso 4: Determinar respuesta de consenso
    if verbose:
        print("\n[4/4] Determinando respuesta final...")
    respuesta_consenso = determinar_consenso(
        resultado_principal, resultado_alternativo, clasificacion
    )
    
    # Calcular métricas
    n_llamadas_total = resultado_principal.n_llamadas
    if resultado_alternativo:
        n_llamadas_total += resultado_alternativo.n_llamadas
    # Incluir llamadas de clasificación y evaluación
    n_llamadas_total += 2
    
    metricas = {
        "tiempo_total_segundos": time.time() - t_inicio,
        "n_llamadas_total": n_llamadas_total,
        "costo_estimado_usd": n_llamadas_total * 0.0002,  # Estimación gpt-4o-mini
        "score_calidad": evaluacion.get("score_promedio", 0),
        "confianza_tecnica": resultado_principal.confianza,
        "hay_comparacion": resultado_alternativo is not None
    }
    
    if verbose:
        print(f"\n{'='*60}")
        print(f"RESULTADO FINAL:")
        print(f"Técnica usada: {resultado_principal.tecnica}")
        print(f"Respuesta: {respuesta_consenso[:200]}...")
        print(f"Tiempo total: {metricas['tiempo_total_segundos']:.2f}s")
        print(f"Llamadas API: {n_llamadas_total}")
        print(f"Costo estimado: ${metricas['costo_estimado_usd']:.4f}")
        print(f"{'='*60}")
    
    return ResultadoFinal(
        problema=problema,
        clasificacion=clasificacion,
        resultado_principal=resultado_principal,
        resultado_alternativo=resultado_alternativo,
        respuesta_consenso=respuesta_consenso,
        metricas=metricas,
        recomendacion_tecnica=resultado_principal.tecnica
    )

Suite de Pruebas

# ============================================================
# TESTS DEL SISTEMA
# ============================================================

CASOS_PRUEBA = [
    {
        "categoria": "Math simple",
        "problema": "¿Cuánto es 15% de 2,400?",
        "respuesta_esperada": "360",
        "tecnica_esperada": "chain_of_thought"
    },
    {
        "categoria": "Math crítico",
        "problema": "Si invierto €10,000 con interés compuesto al 6% anual durante 5 años, ¿cuánto tendré?",
        "respuesta_esperada": "13382",
        "tecnica_esperada": "self_consistency"
    },
    {
        "categoria": "Razonamiento lógico",
        "problema": "Todos los managers trabajan los lunes. Alicia trabaja los lunes. ¿Alicia es manager?",
        "respuesta_esperada": "no necesariamente",
        "tecnica_esperada": "chain_of_thought"
    },
    {
        "categoria": "Planificación",
        "problema": "Diseña un plan de 3 meses para aprender Machine Learning partiendo de Python básico. Incluye recursos específicos.",
        "respuesta_esperada": None,  # Evaluación cualitativa
        "tecnica_esperada": "tree_of_thought"
    },
    {
        "categoria": "Código",
        "problema": "Escribe una función Python que encuentre todos los números primos hasta N usando el método Sieve of Eratosthenes.",
        "respuesta_esperada": None,
        "tecnica_esperada": "self_refine"
    },
    {
        "categoria": "Creativo",
        "problema": "Escribe un haiku sobre la programación.",
        "respuesta_esperada": None,
        "tecnica_esperada": "zero_shot"
    }
]

def ejecutar_suite_pruebas(verbose: bool = True) -> dict:
    """
    Ejecuta todos los casos de prueba y genera un reporte.
    """
    print("\n" + "="*70)
    print("SUITE DE PRUEBAS: MULTI-STRATEGY PROBLEM SOLVER")
    print("="*70)
    
    resultados_suite = []
    
    for i, caso in enumerate(CASOS_PRUEBA, 1):
        print(f"\n[Test {i}/{len(CASOS_PRUEBA)}] {caso['categoria']}")
        print(f"Problema: {caso['problema'][:60]}...")
        
        # Ejecutar
        resultado = resolver_problema(
            caso["problema"],
            modo_comparacion=False,
            verbose=False  # Silencioso en la suite
        )
        
        # Evaluar
        tecnica_correcta = resultado.recomendacion_tecnica == caso.get("tecnica_esperada")
        
        respuesta_correcta = None
        if caso.get("respuesta_esperada"):
            resp_num = _extraer_numero(resultado.respuesta_consenso)
            try:
                correcta_num = case.get("respuesta_esperada", "")
                respuesta_correcta = abs(float(resp_num) - float(correcta_num)) < 1.0
            except ValueError:
                respuesta_correcta = caso["respuesta_esperada"].lower() in resultado.respuesta_consenso.lower()
        
        resultado_test = {
            "categoria": caso["categoria"],
            "tecnica_usada": resultado.recomendacion_tecnica,
            "tecnica_esperada": caso.get("tecnica_esperada"),
            "tecnica_correcta": tecnica_correcta,
            "respuesta_correcta": respuesta_correcta,
            "confianza": resultado.metricas["confianza_tecnica"],
            "llamadas": resultado.metricas["n_llamadas_total"],
            "tiempo": resultado.metricas["tiempo_total_segundos"]
        }
        resultados_suite.append(resultado_test)
        
        if verbose:
            print(f"  Técnica: {resultado.recomendacion_tecnica} ({'✓' if tecnica_correcta else '✗'})")
            if respuesta_correcta is not None:
                print(f"  Respuesta: {'✓ Correcta' if respuesta_correcta else '✗ Incorrecta'}")
            print(f"  Llamadas: {resultado.metricas['n_llamadas_total']}, Tiempo: {resultado.metricas['tiempo_total_segundos']:.2f}s")
    
    # Resumen
    tecnicas_correctas = sum(1 for r in resultados_suite if r["tecnica_correcta"])
    respuestas_evaluables = [r for r in resultados_suite if r["respuesta_correcta"] is not None]
    respuestas_correctas = sum(1 for r in respuestas_evaluables if r["respuesta_correcta"])
    
    total_llamadas = sum(r["llamadas"] for r in resultados_suite)
    tiempo_promedio = sum(r["tiempo"] for r in resultados_suite) / len(resultados_suite)
    
    print(f"\n{'='*70}")
    print("RESUMEN DE PRUEBAS")
    print(f"{'='*70}")
    print(f"Tests ejecutados: {len(CASOS_PRUEBA)}")
    print(f"Técnica correcta: {tecnicas_correctas}/{len(CASOS_PRUEBA)} ({tecnicas_correctas/len(CASOS_PRUEBA):.0%})")
    if respuestas_evaluables:
        print(f"Respuesta correcta: {respuestas_correctas}/{len(respuestas_evaluables)} ({respuestas_correctas/len(respuestas_evaluables):.0%})")
    print(f"Total llamadas API: {total_llamadas}")
    print(f"Tiempo promedio: {tiempo_promedio:.2f}s")
    print(f"Costo total estimado: ${total_llamadas * 0.0002:.4f}")
    
    return {
        "detalle": resultados_suite,
        "accuracy_tecnica": tecnicas_correctas / len(CASOS_PRUEBA),
        "total_llamadas": total_llamadas,
        "tiempo_promedio": tiempo_promedio
    }

Extensiones Opcionales

Extensión 1: Interfaz de línea de comandos

import sys

def cli_interactive():
    """
    Interfaz interactiva para usar el solver desde terminal.
    Escribe 'salir' para terminar.
    """
    print("\n=== MULTI-STRATEGY PROBLEM SOLVER ===")
    print("Escribe cualquier problema o pregunta.")
    print("Comandos: 'comparar' (activa modo comparación), 'salir'\n")
    
    modo_comparacion = False
    
    while True:
        problema = input("Tu pregunta: ").strip()
        
        if problema.lower() == "salir":
            print("¡Hasta luego!")
            break
        
        if problema.lower() == "comparar":
            modo_comparacion = not modo_comparacion
            print(f"Modo comparación: {'ON' if modo_comparacion else 'OFF'}")
            continue
        
        if not problema:
            continue
        
        resultado = resolver_problema(problema, modo_comparacion=modo_comparacion, verbose=True)
        print(f"\nRESPUESTA FINAL:\n{resultado.respuesta_consenso}\n")

Extensión 2: Logging y métricas persistentes

import json
from datetime import datetime
from pathlib import Path

class SolverLogger:
    def __init__(self, log_file: str = "solver_logs.jsonl"):
        self.log_file = Path(log_file)
    
    def log(self, resultado: ResultadoFinal):
        """Guarda cada resolución en un archivo JSONL para análisis posterior."""
        entrada = {
            "timestamp": datetime.now().isoformat(),
            "problema": resultado.problema[:200],
            "tipo": resultado.clasificacion.tipo,
            "tecnica": resultado.recomendacion_tecnica,
            "n_llamadas": resultado.metricas["n_llamadas_total"],
            "tiempo_seg": resultado.metricas["tiempo_total_segundos"],
            "score_calidad": resultado.metricas["score_calidad"],
            "confianza": resultado.metricas["confianza_tecnica"]
        }
        with self.log_file.open("a") as f:
            f.write(json.dumps(entrada, ensure_ascii=False) + "\n")
    
    def estadisticas(self) -> dict:
        """Calcula estadísticas del uso del solver."""
        if not self.log_file.exists():
            return {}
        
        logs = [json.loads(l) for l in self.log_file.read_text().strip().split('\n') if l]
        
        from collections import Counter
        tecnicas_conteo = Counter(l["tecnica"] for l in logs)
        tipos_conteo = Counter(l["tipo"] for l in logs)
        
        return {
            "total_resoluciones": len(logs),
            "tecnicas_mas_usadas": tecnicas_conteo.most_common(5),
            "tipos_mas_comunes": tipos_conteo.most_common(5),
            "tiempo_promedio": sum(l["tiempo_seg"] for l in logs) / len(logs),
            "score_promedio": sum(l["score_calidad"] for l in logs) / len(logs),
            "llamadas_promedio": sum(l["n_llamadas"] for l in logs) / len(logs)
        }

# Uso con logging:
logger = SolverLogger()
resultado = resolver_problema("¿Cuánto es la raíz cuadrada de 2025?")
logger.log(resultado)
print(logger.estadisticas())

Extensión 3: Cache de respuestas similares

from hashlib import md5
import shelve

class CachedSolver:
    def __init__(self, cache_file: str = "solver_cache"):
        self.cache_file = cache_file
    
    def _hash_problema(self, problema: str) -> str:
        """Crea un hash único para el problema."""
        return md5(problema.strip().lower().encode()).hexdigest()
    
    def resolver_con_cache(self, problema: str, ttl_hours: int = 24) -> ResultadoFinal:
        """
        Resolver con cache: si el mismo problema fue resuelto recientemente,
        retorna el resultado cacheado.
        """
        clave = self._hash_problema(problema)
        
        with shelve.open(self.cache_file) as cache:
            if clave in cache:
                entrada = cache[clave]
                # Verificar TTL
                edad_horas = (time.time() - entrada["timestamp"]) / 3600
                if edad_horas < ttl_hours:
                    print(f"[Cache hit] Resultado encontrado ({edad_horas:.1f}h de antigüedad)")
                    return entrada["resultado"]
            
            # No en cache: resolver y guardar
            resultado = resolver_problema(problema, verbose=False)
            cache[clave] = {
                "resultado": resultado,
                "timestamp": time.time()
            }
            return resultado

Criterios de Éxito del Proyecto

Verifica que tu implementación cumple estos criterios:

  • El clasificador identifica correctamente el tipo de problema (math/factual/reasoning/creative/code/planning) en >80% de los casos
  • El router selecciona la técnica apropiada según el tipo y la complejidad
  • Cada técnica produce resultados de calidad consistente
  • El sistema de evaluación asigna scores de calidad razonables
  • El modo comparación ejecuta dos técnicas y determina la mejor respuesta
  • El sistema maneja errores graciosamente (sin crash en casos edge)
  • Los logs y métricas se registran correctamente (extensión)
  • El cache funciona para evitar llamadas repetidas (extensión)

Punto de Entrada Principal

if __name__ == "__main__":
    # Demo básico
    problemas_demo = [
        "¿Cuánto es el 30% de 1,750?",
        "Explica la diferencia entre una lista y una tupla en Python",
        "Diseña la arquitectura de una API REST para un sistema de inventario",
        "Escribe una función que invierta una cadena de texto sin usar reversed()",
    ]
    
    print("=== DEMO: MULTI-STRATEGY PROBLEM SOLVER ===\n")
    
    for problema in problemas_demo:
        print(f"\n{'─'*50}")
        resultado = resolver_problema(problema, verbose=True)
    
    # Suite de pruebas
    print("\n\n=== EJECUTANDO SUITE DE PRUEBAS ===")
    reporte = ejecutar_suite_pruebas(verbose=True)
    
    print(f"\nAccuracy en selección de técnica: {reporte['accuracy_tecnica']:.0%}")

Resumen

En este proyecto construiste un Multi-Strategy Problem Solver completo que:

  1. Clasifica automáticamente el tipo de problema (math, reasoning, código, planificación, etc.)
  2. Selecciona la técnica óptima basándose en el tipo, complejidad, y si requiere datos externos
  3. Ejecuta la técnica seleccionada con parámetros apropiados
  4. Evalúa la calidad de la respuesta
  5. Determina consenso cuando hay comparación entre técnicas
  6. Registra métricas de uso, costo y calidad

Las técnicas implementadas son: Zero-shot, CoT, Self-Consistency, ReAct, Tree-of-Thought, y Self-Refine.

Este sistema es extensible: puedes agregar nuevos tipos de problemas, nuevas técnicas, y conectar herramientas externas reales en el módulo ReAct.

Recursos adicionales

  1. ReAct: Synergizing Reasoning and Acting (Yao et al., 2022)
  2. Self-Consistency (Wang et al., 2022)
  3. Tree of Thoughts (Yao et al., 2023)
  4. Self-Refine (Madaan et al., 2023)
  5. OpenAI Function Calling
  6. LangChain Agent framework
  7. Evaluating LLM outputs - OpenAI evals
← Anterior7. Decision Framework: Qué Técnica UsarSiguiente →1. Introducción: Un Prompt No es Suficiente