A Coding Guide to Building a Brain-Inspired Hierarchical Reasoning AI Agent with Hugging Face Models

!pip -q set up -U transformers speed up bitsandbytes wealthy


import os, re, json, textwrap, traceback
from typing import Dict, Any, Checklist
from wealthy 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.float32

tok = AutoTokenizer.from_pretrained(MODEL_NAME, use_fast=True)
mannequin = AutoModelForCausalLM.from_pretrained(
   MODEL_NAME,
   device_map="auto",
   torch_dtype=DTYPE,
   load_in_4bit=True
)
gen = pipeline(
   "text-generation",
   mannequin=mannequin,
   tokenizer=tok,
   return_full_text=False
)

def chat(immediate: str, system: str = "", max_new_tokens: int = 512, temperature: float = 0.3) -> str:
   msgs = []
   if system:
       msgs.append({"position":"system","content material":system})
   msgs.append({"position":"consumer","content material":immediate})
   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"{[sS]*}$", txt.strip())
   if not m:
       m = re.search(r"{[sS]*?}", txt)
   strive:
       return json.hundreds(m.group(0)) if m else {}
   besides Exception:
       # fallback: strip code fences
       s = re.sub(r"^```.*?n|n```$", "", txt, flags=re.S)
       strive:
           return json.hundreds(s)
       besides Exception:
           return {}

def extract_code(txt: str) -> str:
   m = re.search(r"```(?:python)?s*([sS]*?)```", 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.replace(env)
   buf = io.StringIO()
   strive:
       with contextlib.redirect_stdout(buf):
           exec(code, g, l)
       out = l.get("RESULT", g.get("RESULT"))
       return {"okay": True, "end result": out, "stdout": buf.getvalue()}
   besides Exception as e:
       return {"okay": False, "error": str(e), "hint": traceback.format_exc(), "stdout": buf.getvalue()}


PLANNER_SYS = """You're the HRM Planner.
Decompose the TASK into 2–4 atomic, code-solvable subgoals.
Return compact JSON solely: {"subgoals":[...], "final_format":"<one-line reply format>"}."""


SOLVER_SYS = """You're the HRM Solver.
Given SUBGOAL and CONTEXT vars, output a single Python snippet.
Guidelines:
- Compute deterministically.
- Set a variable RESULT to the reply.
- Hold code quick; stdlib solely.
Return solely a Python code block."""


CRITIC_SYS = """You're the HRM Critic.
Given TASK and LOGS (subgoal outcomes), resolve if remaining reply is prepared.
Return JSON solely: "revise","critique":"...", "fix_hint":"<if revise>"."""


SYNTH_SYS = """You're the HRM Synthesizer.
Given TASK, LOGS, and final_format, output solely the ultimate reply (no steps).
Observe final_format precisely."""

def plan(process: str) -> Dict[str, Any]:
   p = f"TASK:n{process}nReturn JSON solely."
   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]:
   immediate = f"SUBGOAL:n{subgoal}nCONTEXT vars: {listing(context.keys())}nReturn Python code solely."
   code = extract_code(chat(immediate, SOLVER_SYS, temperature=0.2, max_new_tokens=400))
   res = run_python(code, env=context)
   return {"subgoal": subgoal, "code": code, "run": res}


def critic(process: str, logs: Checklist[Dict[str, Any]]) -> Dict[str, Any]:
   pl = [{"subgoal": L["subgoal"], "end result": L["run"].get("end result"), "okay": L["run"]["ok"]} for L in logs]
   out = chat("TASK:n"+process+"nLOGS:n"+json.dumps(pl, ensure_ascii=False, indent=2)+"nReturn JSON solely.",
              CRITIC_SYS, temperature=0.1, max_new_tokens=250)
   return extract_json(out)


def refine(process: str, logs: Checklist[Dict[str, Any]]) -> Dict[str, Any]:
   sys = "Refine subgoals minimally to repair points. Return identical JSON schema as planner."
   out = chat("TASK:n"+process+"nLOGS:n"+json.dumps(logs, ensure_ascii=False)+"nReturn JSON solely.",
              sys, temperature=0.2, max_new_tokens=250)
   j = extract_json(out)
   return j if j.get("subgoals") else {}


def synthesize(process: str, logs: Checklist[Dict[str, Any]], final_format: str) -> str:
   packed = [{"subgoal": L["subgoal"], "end result": L["run"].get("end result")} for L in logs]
   return chat("TASK:n"+process+"nLOGS:n"+json.dumps(packed, ensure_ascii=False)+
               f"nfinal_format: {final_format}nOnly the ultimate reply.",
               SYNTH_SYS, temperature=0.0, max_new_tokens=120).strip()


def hrm_agent(process: str, context: Dict[str, Any] | None = None, funds: int = 2) -> Dict[str, Any]:
   ctx = dict(context or {})
   hint, plan_json = [], plan(process)
   for round_id in vary(1, funds+1):
       logs = [solve_subgoal(sg, ctx) for sg in plan_json.get("subgoals", [])]
       for L in logs:
           ctx_key = f"g{len(hint)}_{abs(hash(L['subgoal']))%9999}"
           ctx[ctx_key] = L["run"].get("end result")
       verdict = critic(process, logs)
       hint.append({"spherical": round_id, "plan": plan_json, "logs": logs, "verdict": verdict})
       if verdict.get("motion") == "submit": break
       plan_json = refine(process, logs) or plan_json
   remaining = synthesize(process, hint[-1]["logs"], plan_json.get("final_format", "Reply: <worth>"))
   return {"remaining": remaining, "hint": hint}

ARC_TASK = textwrap.dedent("""
Infer the transformation rule from prepare examples and apply to check.
Return precisely: "Reply: <grid>", the place <grid> is a Python listing of lists of ints.
""").strip()
ARC_DATA = {
   "prepare": [
       {"inp": [[0,0],[1,0]], "out": [[1,1],[0,1]]},
       {"inp": [[0,1],[0,0]], "out": [[1,0],[1,1]]}
   ],
   "check": [[0,0],[0,1]]
}
res1 = hrm_agent(ARC_TASK, context={"TRAIN": ARC_DATA["train"], "TEST": ARC_DATA["test"]}, funds=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 precisely: 'Reply: <liters>'."
res2 = hrm_agent(WM_TASK, context={}, funds=2)
rprint("n[bold]Demo 2 — Phrase Math[/bold]")
rprint(res2["final"])


rprint("n[dim]Rounds executed (Demo 1):[/dim]", len(res1["trace"]))