En este tutorial, nos propusimos recrear el espíritu del Modelo de razonamiento jerárquico (HRM) utilizando un modelo de cara de abrazo libre que se ejecuta localmente. Caminamos por el diseño de un agente de razonamiento ligero pero estructurado, donde actuamos como arquitectos y experimentadores. Al dividir los problemas en subggoals, resolverlos con Python, criticar los resultados y sintetizar una respuesta final, podemos experimentar cómo la planificación y ejecución jerárquica pueden mejorar el rendimiento del razonamiento. Este proceso nos permite ver, en tiempo real, cómo se puede implementar un flujo de trabajo inspirado en el cerebro sin requerir tamaños de modelo masivos o API costosas. Mira el Papel y Códigos completos.
!pip -q install -U transformers accelerate bitsandbytes rich
import os, re, json, textwrap, traceback
from typing import Dict, Any, List
from rich import print as rprint
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
MODEL_NAME = "Qwen/Qwen2.5-1.5B-Instruct"
DTYPE = torch.bfloat16 if torch.cuda.is_available() else torch.float32Comenzamos instalando las bibliotecas requeridas y cargando el modelo QWEN2.5-1.5B-Instructo desde la cara abrazada. Establecimos el tipo de datos en función de la disponibilidad de GPU para garantizar una ejecución eficiente del modelo en Colab.
tok = AutoTokenizer.from_pretrained(MODEL_NAME, use_fast=True)
model = AutoModelForCausalLM.from_pretrained(
MODEL_NAME,
device_map="auto",
torch_dtype=DTYPE,
load_in_4bit=True
)
gen = pipeline(
"text-generation",
model=model,
tokenizer=tok,
return_full_text=False
)
Cargamos el tokenizador y el modelo, lo configuramos para que se ejecute en 4 bits para obtener eficiencia y envolvemos todo en una tubería de generación de texto para que podamos interactuar con el modelo fácilmente en Colab. Mira el Papel y Códigos completos.
def chat(prompt: str, system: str = "", max_new_tokens: int = 512, temperature: float = 0.3) -> str:
msgs = []
if system:
msgs.append({"role":"system","content":system})
msgs.append({"role":"user","content":prompt})
inputs = tok.apply_chat_template(msgs, tokenize=False, add_generation_prompt=True)
out = gen(inputs, max_new_tokens=max_new_tokens, do_sample=(temperature>0), temperature=temperature, top_p=0.9)
return out[0]["generated_text"].strip()
def extract_json(txt: str) -> Dict[str, Any]:
m = re.search(r"\{[\s\S]*\}$", txt.strip())
if not m:
m = re.search(r"\{[\s\S]*?\}", txt)
try:
return json.loads(m.group(0)) if m else {}
except Exception:
# fallback: strip code fences
s = re.sub(r"^```.*?\n|\n```$", "", txt, flags=re.S)
try:
return json.loads(s)
except Exception:
return {}Definimos las funciones auxiliares: la función de chat nos permite enviar indicaciones al modelo con instrucciones del sistema opcionales y controles de muestreo, mientras que Extract_json nos ayuda a analizar las salidas JSON estructuradas desde el modelo de manera confiable, incluso si la respuesta incluye cercas de código o texto adicional. Mira el Papel y Códigos completos.
def extract_code(txt: str) -> str:
m = re.search(r"```(?:python)?\s*([\s\S]*?)```", txt, flags=re.I)
return (m.group(1) if m else txt).strip()
def run_python(code: str, env: Dict[str, Any] | None = None) -> Dict[str, Any]:
import io, contextlib
g = {"__name__": "__main__"}; l = {}
if env: g.update(env)
buf = io.StringIO()
try:
with contextlib.redirect_stdout(buf):
exec(code, g, l)
out = l.get("RESULT", g.get("RESULT"))
return {"ok": True, "result": out, "stdout": buf.getvalue()}
except Exception as e:
return {"ok": False, "error": str(e), "trace": traceback.format_exc(), "stdout": buf.getvalue()}
PLANNER_SYS = """You are the HRM Planner.
Decompose the TASK into 2–4 atomic, code-solvable subgoals.
Return compact JSON only: {"subgoals":[...], "final_format":"<one-line answer format>"}."""
SOLVER_SYS = """You are the HRM Solver.
Given SUBGOAL and CONTEXT vars, output a single Python snippet.
Rules:
- Compute deterministically.
- Set a variable RESULT to the answer.
- Keep code short; stdlib only.
Return only a Python code block."""
CRITIC_SYS = """You are the HRM Critic.
Given TASK and LOGS (subgoal results), decide if final answer is ready.
Return JSON only: {"action":"submit"|"revise","critique":"...", "fix_hint":"<if revise>"}."""
SYNTH_SYS = """You are the HRM Synthesizer.
Given TASK, LOGS, and final_format, output only the final answer (no steps).
Follow final_format exactly."""
Agregamos dos piezas importantes: funciones de utilidad y indicaciones del sistema. La función Extract_code extrae los fragmentos de Python de la salida del modelo, mientras que Run_python ejecuta de forma segura esos fragmentos y captura sus resultados. Junto, definimos cuatro indicaciones de roles, planificador, solucionador, crítico y sintetizador, que guía el modelo para dividir las tareas en subggoal, resolverlas con código, verificar la corrección y finalmente producir una respuesta limpia. Mira el Papel y Códigos completos.
def plan(task: str) -> Dict[str, Any]:
p = f"TASK:\n{task}\nReturn JSON only."
return extract_json(chat(p, PLANNER_SYS, temperature=0.2, max_new_tokens=300))
def solve_subgoal(subgoal: str, context: Dict[str, Any]) -> Dict[str, Any]:
prompt = f"SUBGOAL:\n{subgoal}\nCONTEXT vars: {list(context.keys())}\nReturn Python code only."
code = extract_code(chat(prompt, SOLVER_SYS, temperature=0.2, max_new_tokens=400))
res = run_python(code, env=context)
return {"subgoal": subgoal, "code": code, "run": res}
def critic(task: str, logs: List[Dict[str, Any]]) -> Dict[str, Any]:
pl = [{"subgoal": L["subgoal"], "result": L["run"].get("result"), "ok": L["run"]["ok"]} for L in logs]
out = chat("TASK:\n"+task+"\nLOGS:\n"+json.dumps(pl, ensure_ascii=False, indent=2)+"\nReturn JSON only.",
CRITIC_SYS, temperature=0.1, max_new_tokens=250)
return extract_json(out)
def refine(task: str, logs: List[Dict[str, Any]]) -> Dict[str, Any]:
sys = "Refine subgoals minimally to fix issues. Return same JSON schema as planner."
out = chat("TASK:\n"+task+"\nLOGS:\n"+json.dumps(logs, ensure_ascii=False)+"\nReturn JSON only.",
sys, temperature=0.2, max_new_tokens=250)
j = extract_json(out)
return j if j.get("subgoals") else {}
def synthesize(task: str, logs: List[Dict[str, Any]], final_format: str) -> str:
packed = [{"subgoal": L["subgoal"], "result": L["run"].get("result")} for L in logs]
return chat("TASK:\n"+task+"\nLOGS:\n"+json.dumps(packed, ensure_ascii=False)+
f"\nfinal_format: {final_format}\nOnly the final answer.",
SYNTH_SYS, temperature=0.0, max_new_tokens=120).strip()
def hrm_agent(task: str, context: Dict[str, Any] | None = None, budget: int = 2) -> Dict[str, Any]:
ctx = dict(context or {})
trace, plan_json = [], plan(task)
for round_id in range(1, budget+1):
logs = [solve_subgoal(sg, ctx) for sg in plan_json.get("subgoals", [])]
for L in logs:
ctx_key = f"g{len(trace)}_{abs(hash(L['subgoal']))%9999}"
ctx[ctx_key] = L["run"].get("result")
verdict = critic(task, logs)
trace.append({"round": round_id, "plan": plan_json, "logs": logs, "verdict": verdict})
if verdict.get("action") == "submit": break
plan_json = refine(task, logs) or plan_json
final = synthesize(task, trace[-1]["logs"], plan_json.get("final_format", "Answer: <value>"))
return {"final": final, "trace": trace}Implementamos el bucle completo de HRM: planificamos subggoals, resolvemos cada una generando y ejecutando Python (capturando el resultado), luego criticamos, opcionalmente refinamos el plan y sintetizamos una respuesta final limpia. Orquestamos estas rondas en HRM_AGENT, llevando los resultados intermedios como contexto, por lo que mejoramos y nos detenemos iterativamente una vez que el crítico dice “enviar”. Mira el Papel y Códigos completos.
ARC_TASK = textwrap.dedent("""
Infer the transformation rule from train examples and apply to test.
Return exactly: "Answer: <grid>", where <grid> is a Python list of lists of ints.
""").strip()
ARC_DATA = {
"train": [
{"inp": [[0,0],[1,0]], "out": [[1,1],[0,1]]},
{"inp": [[0,1],[0,0]], "out": [[1,0],[1,1]]}
],
"test": [[0,0],[0,1]]
}
res1 = hrm_agent(ARC_TASK, context={"TRAIN": ARC_DATA["train"], "TEST": ARC_DATA["test"]}, budget=2)
rprint("\n[bold]Demo 1 — ARC-like Toy[/bold]")
rprint(res1["final"])
WM_TASK = "A tank holds 1200 L. It leaks 2% per hour for 3 hours, then is refilled by 150 L. Return exactly: 'Answer: <liters>'."
res2 = hrm_agent(WM_TASK, context={}, budget=2)
rprint("\n[bold]Demo 2 — Word Math[/bold]")
rprint(res2["final"])
rprint("\n[dim]Rounds executed (Demo 1):[/dim]", len(res1["trace"]))Ejecutamos dos demostraciones para validar al agente: una tarea de estilo ARC donde inferimos una transformación de los pares de trenes y la aplicamos a una cuadrícula de prueba, y un problema de madera que verifica el razonamiento numérico. Llamamos HRM_AGENT con cada tarea, imprimimos las respuestas finales y también mostramos el número de rondas de razonamiento que toma la ejecución de ARC.
En conclusión, reconocemos que lo que hemos construido es más que una simple demostración; Es una ventana de cómo el razonamiento jerárquico puede hacer que los modelos más pequeños golpeen por encima de su peso. Al planificar, resolver y criticar en capas, empoderamos a un modelo de abrazadera gratuita para realizar tareas con sorprendente robustez. Dejamos una apreciación más profunda de cómo las estructuras inspiradas en el cerebro, cuando se combinan con herramientas prácticas de código abierto, nos permiten explorar puntos de referencia de razonamiento y experimentar creativamente sin incurrir en altos costos. Este viaje práctico nos muestra que los flujos de trabajo cognitivos avanzados son accesibles para cualquier persona dispuesta a jugar, iterar y aprender.
Mira el Papel y Códigos completos. No dude en ver nuestro Página de Github para tutoriales, códigos y cuadernos. Además, siéntete libre de seguirnos Gorjeo Y no olvides unirte a nuestro Subreddit de 100k+ ml y suscribirse a Nuestro boletín.
Asif Razzaq es el CEO de MarktechPost Media Inc .. Como empresario e ingeniero visionario, ASIF se compromete a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, MarktechPost, que se destaca por su cobertura profunda de noticias de aprendizaje automático y de aprendizaje profundo que es técnicamente sólido y fácilmente comprensible por una audiencia amplia. La plataforma cuenta con más de 2 millones de vistas mensuales, ilustrando su popularidad entre el público.