Módulo 5: ReAct, Self-Consistency y Patrones Avanzados
2. ReAct: Reasoning + Acting
Descripción
ReAct (Reasoning + Acting) es una técnica propuesta por Yao et al. en 2022 que combina el razonamiento verbal (Thought) con la acción sobre entornos externos (Action). La idea central es simple pero poderosa: los modelos de lenguaje no deberían solo "pensar" — también deberían poder actuar sobre el mundo y observar los resultados para continuar razonando.
En la implementación práctica con OpenAI, esto se hace mediante function calling: el modelo decide qué herramienta llamar, con qué parámetros, y luego recibe el resultado de esa herramienta como contexto adicional.
El Problema que Resuelve ReAct
Considera este escenario:
Pregunta: "Si invierto $10,000 en Tesla hoy, y el precio sube
un 8% en el próximo trimestre, ¿cuánto tendré?"
CoT puro:
Thought: Necesito el precio actual de Tesla
Thought: No sé el precio actual → asumiré $200 (inventado/desactualizado)
Thought: 10000 / 200 = 50 acciones
Thought: 50 * 200 * 1.08 = $10,800
Answer: $10,800
PROBLEMA: El precio asumido puede ser completamente incorrecto.
ReAct:
Thought: Necesito buscar el precio actual de Tesla
Action: search("Tesla stock price today")
Observation: Tesla (TSLA): $348.50 (8 marzo 2026)
Thought: Tengo el precio real. 10000 / 348.50 ≈ 28.7 acciones
Action: calculator("10000 / 348.50 * 348.50 * 1.08")
Observation: 10800.0
Answer: Con $10,000 tendrías aproximadamente $10,800 después del 8% de subida.
El Patrón ReAct Completo
El ciclo ReAct se compone de:
Thought: [Razonamiento sobre qué hacer a continuación]
Action: [nombre_herramienta(parámetros)]
Observation: [Resultado de la herramienta]
... (repetir hasta tener respuesta)
Answer: [Respuesta final]
Reglas del patrón
- Thought siempre precede a Action: El modelo debe razonar antes de actuar
- Una acción a la vez: No ejecutar múltiples acciones sin procesar las observaciones
- Observation informa el siguiente Thought: El resultado de la herramienta debe influir en el razonamiento
- Terminar cuando hay suficiente información: No llamar herramientas innecesariamente
Implementación Completa con OpenAI
from openai import OpenAI
import json
import math
client = OpenAI()
# --- Definición de herramientas ---
def calculadora(expresion: str) -> str:
"""Evalúa expresiones matemáticas de forma segura."""
# En producción: usar una librería de evaluación segura
# Aquí permitimos solo operaciones matemáticas básicas
allowed_chars = set("0123456789+-*/.() ")
if not all(c in allowed_chars for c in expresion):
return "Error: Expresión no permitida"
try:
resultado = eval(expresion, {"__builtins__": {}}, {"math": math})
return str(round(float(resultado), 4))
except Exception as e:
return f"Error: {e}"
def buscar_info(consulta: str) -> str:
"""Simula búsqueda de información."""
# En producción: llamar a una API real (Google, Bing, etc.)
base_datos = {
"tesla stock price": "Tesla (TSLA): $348.50 (8 marzo 2026)",
"apple stock": "Apple (AAPL): $218.30 (8 marzo 2026)",
"pib españa 2024": "PIB España 2024: ~1.47 billones EUR (Banco de España)",
"poblacion mexico": "Población México 2024: ~129.9 millones (INEGI)",
"tipo cambio eur usd": "EUR/USD: 1.0847 (8 marzo 2026)",
}
consulta_lower = consulta.lower()
for clave, valor in base_datos.items():
if any(word in consulta_lower for word in clave.split()):
return valor
return f"No se encontró información específica sobre: {consulta}"
def convertir_unidades(valor: float, de: str, a: str) -> str:
"""Convierte entre unidades comunes."""
conversiones = {
("km", "miles"): 0.621371,
("miles", "km"): 1.60934,
("kg", "lb"): 2.20462,
("lb", "kg"): 0.453592,
("celsius", "fahrenheit"): lambda x: x * 9/5 + 32,
("fahrenheit", "celsius"): lambda x: (x - 32) * 5/9,
}
key = (de.lower(), a.lower())
if key in conversiones:
conv = conversiones[key]
resultado = conv(valor) if callable(conv) else valor * conv
return f"{valor} {de} = {round(resultado, 4)} {a}"
return f"Conversión {de} → {a} no disponible"
# --- Definición del schema de herramientas para OpenAI ---
TOOLS = [
{
"type": "function",
"function": {
"name": "calculadora",
"description": "Evalúa expresiones matemáticas. Usa cuando necesites calcular números, porcentajes, operaciones aritméticas.",
"parameters": {
"type": "object",
"properties": {
"expresion": {
"type": "string",
"description": "La expresión matemática a evaluar. Ejemplo: '100 * 1.08' o '(500 - 200) / 300'"
}
},
"required": ["expresion"]
}
}
},
{
"type": "function",
"function": {
"name": "buscar_info",
"description": "Busca información factual actual sobre precios, datos económicos, estadísticas. Usa cuando necesites datos del mundo real.",
"parameters": {
"type": "object",
"properties": {
"consulta": {
"type": "string",
"description": "La consulta de búsqueda. Sé específico. Ejemplo: 'Tesla stock price today' o 'PIB España 2024'"
}
},
"required": ["consulta"]
}
}
},
{
"type": "function",
"function": {
"name": "convertir_unidades",
"description": "Convierte entre unidades de medida: distancia, peso, temperatura.",
"parameters": {
"type": "object",
"properties": {
"valor": {"type": "number", "description": "El valor a convertir"},
"de": {"type": "string", "description": "Unidad de origen (km, miles, kg, lb, celsius, fahrenheit)"},
"a": {"type": "string", "description": "Unidad de destino"}
},
"required": ["valor", "de", "a"]
}
}
}
]
# --- Dispatcher de herramientas ---
TOOL_FUNCTIONS = {
"calculadora": lambda args: calculadora(args["expresion"]),
"buscar_info": lambda args: buscar_info(args["consulta"]),
"convertir_unidades": lambda args: convertir_unidades(
args["valor"], args["de"], args["a"]
)
}
# --- Motor ReAct principal ---
def run_react(
problema: str,
max_steps: int = 8,
verbose: bool = True
) -> dict:
"""
Ejecuta el ciclo ReAct para resolver un problema.
Args:
problema: La pregunta o problema a resolver
max_steps: Máximo número de ciclos Thought→Action→Observation
verbose: Si True, imprime el proceso paso a paso
Returns:
dict con 'respuesta', 'pasos', 'herramientas_usadas'
"""
system_prompt = """Eres un asistente que resuelve problemas usando herramientas.
Para cada problema:
1. Piensa (Thought) qué información necesitas o qué calcular
2. Usa las herramientas disponibles (Action) cuando necesites datos o cálculos
3. Observa el resultado (lo recibirás como Observation)
4. Repite hasta tener suficiente información
5. Da una respuesta final clara y completa
IMPORTANTE:
- Usa la herramienta calculadora para TODOS los cálculos numéricos
- Usa buscar_info cuando necesites datos actuales o hechos específicos
- No inventes datos que podrías buscar con las herramientas
"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": problema}
]
pasos = []
herramientas_usadas = []
for step in range(max_steps):
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=TOOLS,
tool_choice="auto",
temperature=0
)
msg = response.choices[0].message
# Registrar el pensamiento
if msg.content:
pasos.append({"tipo": "thought", "contenido": msg.content})
if verbose:
print(f"\n[Thought {step+1}]: {msg.content}")
# Si no hay tool calls, el modelo tiene la respuesta
if not msg.tool_calls:
respuesta = msg.content or "Sin respuesta"
if verbose:
print(f"\n[Answer]: {respuesta}")
return {
"respuesta": respuesta,
"pasos": pasos,
"herramientas_usadas": herramientas_usadas,
"steps_used": step + 1
}
# Procesar tool calls
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)
if verbose:
print(f"\n[Action]: {tool_name}({args})")
# Ejecutar herramienta
if tool_name in TOOL_FUNCTIONS:
resultado = TOOL_FUNCTIONS[tool_name](args)
else:
resultado = f"Error: Herramienta '{tool_name}' no disponible"
herramientas_usadas.append(tool_name)
pasos.append({
"tipo": "action",
"herramienta": tool_name,
"args": args,
"resultado": resultado
})
if verbose:
print(f"[Observation]: {resultado}")
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": resultado
})
# Si llegamos al límite de pasos
return {
"respuesta": "Alcanzado límite de pasos. Último resultado: " + str(pasos[-1] if pasos else "ninguno"),
"pasos": pasos,
"herramientas_usadas": herramientas_usadas,
"steps_used": max_steps
}
# --- Ejemplos de uso ---
if __name__ == "__main__":
print("=" * 60)
print("Ejemplo 1: Cálculo con datos reales")
print("=" * 60)
resultado = run_react(
"Si invierto $5,000 en acciones de Apple hoy y el precio sube 12%, "
"¿cuánto tendré? ¿Cuántas acciones compraría?",
verbose=True
)
print(f"\nRespuesta final: {resultado['respuesta']}")
print(f"Herramientas usadas: {resultado['herramientas_usadas']}")
print("\n" + "=" * 60)
print("Ejemplo 2: Conversión con cálculo")
print("=" * 60)
resultado2 = run_react(
"Un corredor corre un maratón (42.195 km). "
"¿Cuántas millas son? ¿Y cuántos metros?",
verbose=True
)
print(f"\nRespuesta final: {resultado2['respuesta']}")
ReAct vs CoT Puro: Comparación Detallada
| Aspecto | CoT Puro | ReAct |
|---|---|---|
| Datos externos | No (usa conocimiento del modelo) | Sí (APIs, bases de datos) |
| Verificación | Solo mental | Con herramientas reales |
| Actualidad | Limitado por corte de entrenamiento | Tiempo real (si tienes las tools) |
| Número de llamadas API | 1 | 1 + N (por cada action) |
| Costo | 1x | 3-10x dependiendo de pasos |
| Latencia | Baja | Media-Alta |
| Precisión en math | Media | Alta (con calculadora real) |
| Cuándo preferirlo | Math/lógica pura, sin datos externos | Siempre que necesites datos reales |
Patrones Avanzados de ReAct
Pattern 1: ReAct con Validación
def react_con_validacion(problema: str) -> dict:
"""ReAct que valida el resultado antes de retornarlo."""
resultado = run_react(problema, verbose=False)
# Paso de validación adicional
validacion_prompt = f"""
Problema original: {problema}
Respuesta propuesta: {resultado['respuesta']}
Pasos tomados: {json.dumps(resultado['pasos'], indent=2)}
¿La respuesta es correcta y completa? Si hay errores, corrígelos.
Si es correcta, di "VÁLIDO: " seguido de la respuesta final.
"""
validacion = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": validacion_prompt}],
temperature=0
)
respuesta_validada = validacion.choices[0].message.content
return {
**resultado,
"respuesta_validada": respuesta_validada
}
Pattern 2: ReAct con Cache de Observaciones
from functools import lru_cache
@lru_cache(maxsize=100)
def buscar_con_cache(consulta: str) -> str:
"""Cachear resultados de búsqueda para evitar llamadas repetidas."""
return buscar_info(consulta)
def react_con_cache(problema: str) -> dict:
"""ReAct que evita repetir búsquedas idénticas."""
# Reemplazar la función de búsqueda con la versión cacheada
TOOL_FUNCTIONS["buscar_info"] = lambda args: buscar_con_cache(args["consulta"])
return run_react(problema, verbose=False)
Pattern 3: ReAct Multi-Agente
def react_multi_agente(problema: str) -> dict:
"""
Divide el problema en sub-problemas, resuelve cada uno con ReAct,
luego sintetiza los resultados.
"""
# Paso 1: Descomponer el problema
descomposicion = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"Descompón este problema en 2-3 sub-preguntas independientes: {problema}\nResponde en formato: 1. ... 2. ... 3. ..."
}],
temperature=0
).choices[0].message.content
# Extraer sub-preguntas (simplificado)
sub_preguntas = [line.strip() for line in descomposicion.split("\n") if line.strip() and line[0].isdigit()]
# Paso 2: Resolver cada sub-pregunta con ReAct
resultados_parciales = []
for sub_pregunta in sub_preguntas[:3]: # Máximo 3
resultado = run_react(sub_pregunta, verbose=False)
resultados_parciales.append({
"pregunta": sub_pregunta,
"respuesta": resultado["respuesta"]
})
# Paso 3: Sintetizar
sintesis_prompt = f"""
Problema original: {problema}
Sub-respuestas encontradas:
{json.dumps(resultados_parciales, ensure_ascii=False, indent=2)}
Sintetiza una respuesta final completa al problema original.
"""
sintesis = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": sintesis_prompt}],
temperature=0
).choices[0].message.content
return {
"respuesta_final": sintesis,
"sub_resultados": resultados_parciales
}
Integración con Anthropic
import anthropic
client_anthropic = anthropic.Anthropic()
tools_anthropic = [
{
"name": "calculadora",
"description": "Evalúa expresiones matemáticas. Usa para calcular números.",
"input_schema": {
"type": "object",
"properties": {
"expresion": {"type": "string", "description": "Expresión matemática"}
},
"required": ["expresion"]
}
},
{
"name": "buscar_info",
"description": "Busca información factual actual.",
"input_schema": {
"type": "object",
"properties": {
"consulta": {"type": "string", "description": "Consulta de búsqueda"}
},
"required": ["consulta"]
}
}
]
def run_react_anthropic(problema: str, max_steps: int = 6) -> str:
"""
Implementación de ReAct usando Claude (Anthropic).
La API de Anthropic maneja tool_use de forma diferente a OpenAI.
"""
messages = [{"role": "user", "content": problema}]
for step in range(max_steps):
response = client_anthropic.messages.create(
model="claude-3-5-haiku-20241022",
max_tokens=1000,
tools=tools_anthropic,
messages=messages,
system="Resuelve problemas usando herramientas. Piensa antes de actuar. Usa las herramientas cuando necesites datos reales o cálculos."
)
# Anthropic usa stop_reason para indicar tool_use
if response.stop_reason == "end_turn":
# Respuesta final
for block in response.content:
if hasattr(block, "text"):
return block.text
# Procesar tool_use
tool_results = []
messages.append({"role": "assistant", "content": response.content})
for block in response.content:
if block.type == "tool_use":
tool_name = block.name
args = block.input
if tool_name in TOOL_FUNCTIONS:
resultado = TOOL_FUNCTIONS[tool_name](args)
else:
resultado = f"Herramienta no disponible: {tool_name}"
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": resultado
})
if tool_results:
messages.append({"role": "user", "content": tool_results})
return "Límite de pasos alcanzado"
Casos de Uso Reales
Caso 1: Asistente de Finanzas Personales
problemas_financieros = [
"Tengo €15,000 ahorros. Si los pongo en un depósito al 2.5% anual durante 3 años, ¿cuánto tendré?",
"¿Cuánto es mi salario neto si gano €35,000 brutos al año en España?",
"Si envío 500 USD a México hoy, ¿cuántos pesos mexicanos recibirá mi familia?"
]
for problema in problemas_financieros:
print(f"\nPROBLEMA: {problema}")
resultado = run_react(problema, verbose=False)
print(f"RESPUESTA: {resultado['respuesta']}")
print(f"HERRAMIENTAS: {resultado['herramientas_usadas']}")
Caso 2: Research Automatizado
def research_react(tema: str) -> dict:
"""Investigar un tema usando múltiples búsquedas encadenadas."""
prompt = f"""
Investiga sobre: {tema}
Usa las herramientas de búsqueda para recopilar:
1. Datos estadísticos actuales
2. Contexto histórico si es relevante
3. Tendencias actuales
Al final, genera un resumen estructurado con los datos encontrados.
"""
return run_react(prompt, max_steps=10, verbose=True)
resultado = research_react("estado actual de la industria de semiconductores en 2026")
Troubleshooting
Problema 1: El modelo entra en un loop
Síntoma: El modelo llama la misma herramienta repetidamente sin avanzar.
Causas:
- La observación no satisface lo que el modelo esperaba
- El modelo no procesa bien el resultado de la herramienta
Solución:
def detect_loop(pasos: list, window: int = 3) -> bool:
"""Detectar si el modelo está en loop."""
if len(pasos) < window * 2:
return False
ultimas_acciones = [p.get("herramienta") for p in pasos[-window*2:] if p.get("tipo") == "action"]
# Si las últimas window*2 acciones son todas iguales, es un loop
if len(set(ultimas_acciones)) == 1 and len(ultimas_acciones) >= window:
return True
return False
# En run_react, agregar:
if detect_loop(pasos):
messages.append({
"role": "user",
"content": "Parece que estás repitiendo las mismas acciones. ¿Tienes suficiente información para responder? Si es así, da tu mejor respuesta con lo que tienes."
})
Problema 2: Tool call con argumentos incorrectos
Síntoma: json.loads() falla o los argumentos no tienen el formato esperado.
Solución:
def ejecutar_herramienta_seguro(tool_name: str, args_str: str) -> str:
"""Ejecutar herramienta con manejo de errores robusto."""
try:
args = json.loads(args_str)
except json.JSONDecodeError as e:
return f"Error al parsear argumentos: {e}. Argumentos recibidos: {args_str}"
if tool_name not in TOOL_FUNCTIONS:
return f"Herramienta '{tool_name}' no existe. Herramientas disponibles: {list(TOOL_FUNCTIONS.keys())}"
try:
return TOOL_FUNCTIONS[tool_name](args)
except KeyError as e:
return f"Argumento faltante: {e}"
except Exception as e:
return f"Error ejecutando {tool_name}: {e}"
Problema 3: Observation no usada en el siguiente Thought
Síntoma: El modelo responde como si no hubiera recibido la observación.
Solución: Ser más explícito en el system prompt:
system_prompt_mejorado = """Eres un asistente que DEBE usar herramientas.
REGLAS ESTRICTAS:
1. NUNCA inventes datos numéricos — usa calculadora
2. NUNCA asumas precios o datos actuales — busca con buscar_info
3. SIEMPRE menciona en tu Thought el dato que acabas de recibir de la Observation
4. Si la Observation tiene un error, inténtalo con una consulta diferente
Ejemplo de uso correcto:
Thought: Necesito el precio de Apple para calcular.
Action: buscar_info("Apple AAPL stock price")
Observation: Apple (AAPL): $218.30
Thought: El precio actual de Apple es $218.30. Ahora calculo cuántas acciones puedo comprar con $5000.
Action: calculadora("5000 / 218.30")
...
"""
Ejercicios
Ejercicio 1: Agregar una nueva herramienta
Implementa una herramienta obtener_fecha(zona_horaria: str) -> str que retorne la fecha y hora actual para una zona horaria dada. Integra esta herramienta al sistema ReAct y prueba con la pregunta: "¿Qué día de la semana es en Tokio ahora mismo?"
Ver solución
from datetime import datetime
import pytz # pip install pytz
def obtener_fecha(zona_horaria: str) -> str:
"""Obtiene la fecha y hora actual para una zona horaria."""
zona_map = {
"tokio": "Asia/Tokyo",
"nueva_york": "America/New_York",
"madrid": "Europe/Madrid",
"mexico": "America/Mexico_City",
"london": "Europe/London"
}
tz_name = zona_map.get(zona_horaria.lower(), zona_horaria)
try:
tz = pytz.timezone(tz_name)
now = datetime.now(tz)
return f"Fecha/hora en {zona_horaria}: {now.strftime('%A %d/%m/%Y %H:%M:%S')} ({tz_name})"
except Exception:
return f"Zona horaria no reconocida: {zona_horaria}"
# Agregar al schema de tools:
nueva_tool = {
"type": "function",
"function": {
"name": "obtener_fecha",
"description": "Obtiene la fecha y hora actual en una zona horaria específica.",
"parameters": {
"type": "object",
"properties": {
"zona_horaria": {
"type": "string",
"description": "Zona horaria (tokio, nueva_york, madrid, mexico, london)"
}
},
"required": ["zona_horaria"]
}
}
}
TOOLS.append(nueva_tool)
TOOL_FUNCTIONS["obtener_fecha"] = lambda args: obtener_fecha(args["zona_horaria"])
# Probar:
resultado = run_react("¿Qué día de la semana es en Tokio ahora mismo?")
Ejercicio 2: ReAct con historial
Implementa una versión de ReAct que recuerde conversaciones previas. El usuario debe poder hacer preguntas de seguimiento.
Usuario: "¿Cuántas acciones de Tesla puedo comprar con $10,000?"
Bot: [responde con ReAct]
Usuario: "¿Y si el precio baja 20%?" ← Debe recordar la conversación anterior
Bot: [responde considerando el contexto previo]
Ver solución
class ReActConversacion:
def __init__(self):
self.historial = []
self.system_prompt = """Eres un asistente financiero.
Recuerdas la conversación previa y usas herramientas para datos actuales."""
def preguntar(self, pregunta: str) -> str:
# Añadir pregunta al historial
self.historial.append({"role": "user", "content": pregunta})
# Construir messages con historial
messages = [{"role": "system", "content": self.system_prompt}] + self.historial.copy()
for _ in range(8):
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=TOOLS,
tool_choice="auto",
temperature=0
)
msg = response.choices[0].message
if not msg.tool_calls:
respuesta = msg.content
self.historial.append({"role": "assistant", "content": respuesta})
return respuesta
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:
args = json.loads(tc.function.arguments)
resultado = TOOL_FUNCTIONS.get(tc.function.name, lambda _: "Error")(args)
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": resultado
})
return "Límite de pasos alcanzado"
# Uso:
bot = ReActConversacion()
print(bot.preguntar("¿Cuántas acciones de Tesla puedo comprar con $10,000?"))
print(bot.preguntar("¿Y si el precio baja 20%?"))
Ejercicio 3: Comparar ReAct vs CoT
Diseña un experimento que compare la precisión de ReAct vs CoT puro para 5 preguntas que requieren cálculos matemáticos con datos reales. ¿En cuáles gana ReAct? ¿En cuáles son equivalentes?
Ver solución
from collections import namedtuple
Pregunta = namedtuple("Pregunta", ["texto", "respuesta_correcta", "tipo"])
preguntas = [
Pregunta(
"¿Cuánto es 15% de 2,840?",
"426",
"math_puro" # CoT y ReAct deberían empatar
),
Pregunta(
"Si Apple está a $218.30, ¿cuántas acciones compro con $5,000?",
"22", # 5000/218.30 ≈ 22.9
"datos_externos" # ReAct debe ganar
),
Pregunta(
"¿Cuántos kilómetros son 100 millas?",
"160.93",
"conversion" # CoT puede saberlo del entrenamiento; ReAct lo confirma
),
]
def cot_puro(pregunta: str) -> str:
"""CoT sin herramientas."""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"{pregunta}\n\nPiensa paso a paso y da la respuesta numérica."
}],
temperature=0
)
return response.choices[0].message.content
resultados = []
for p in preguntas:
cot_resp = cot_puro(p.texto)
react_resp = run_react(p.texto, verbose=False)["respuesta"]
resultados.append({
"pregunta": p.texto[:50],
"tipo": p.tipo,
"cot": cot_resp[:100],
"react": react_resp[:100],
"correcta": p.respuesta_correcta
})
print(f"\n{p.texto[:50]}...")
print(f" CoT: {cot_resp[:80]}...")
print(f" ReAct: {react_resp[:80]}...")
Resultado esperado: Para math_puro, ambos son similares. Para datos_externos, ReAct es más preciso porque usa el precio real.
Ejercicio 4: ReAct con timeout
Implementa un decorador que limite el tiempo total de ejecución de un ciclo ReAct a N segundos. Si el tiempo se agota, retorna la mejor respuesta parcial.
Ver solución
import signal
import time
class TimeoutError(Exception):
pass
def con_timeout(segundos: int):
"""Decorador que limita el tiempo de ejecución."""
def decorator(func):
def handler(signum, frame):
raise TimeoutError(f"Timeout: {segundos}s excedidos")
def wrapper(*args, **kwargs):
signal.signal(signal.SIGALRM, handler)
signal.alarm(segundos)
try:
resultado = func(*args, **kwargs)
signal.alarm(0) # Cancelar alarm
return resultado
except TimeoutError:
return {"respuesta": "Timeout: respuesta parcial no disponible",
"pasos": [], "error": "timeout"}
return wrapper
return decorator
@con_timeout(15) # 15 segundos máximo
def react_con_timeout(problema: str) -> dict:
return run_react(problema, verbose=False)
# Uso:
resultado = react_con_timeout("Problema muy complejo que podría tardar mucho...")
Nota: signal.SIGALRM solo funciona en Unix/macOS. En Windows, se necesita threading.Timer.
Resumen
- ReAct = Thought → Action → Observation → repeat hasta tener respuesta
- Cuándo usar: Siempre que necesites datos externos (precios, fechas, búsquedas, cálculos precisos)
- Cuándo NO usar: Math puro sin datos externos (CoT es más rápido y barato)
- Implementación: OpenAI function calling + dispatcher de herramientas
- Precauciones: Limitar max_steps, detectar loops, manejar errores por herramienta
- Costo: 3-10x más que CoT simple, justificado cuando los datos son críticos