{"id":485,"date":"2026-02-28T10:18:20","date_gmt":"2026-02-28T02:18:20","guid":{"rendered":"https:\/\/connectword.dpdns.org\/?p=485"},"modified":"2026-02-28T10:18:20","modified_gmt":"2026-02-28T02:18:20","slug":"a-coding-implementation-to-build-a-hierarchical-planner-ai-agent-using-open-source-llms-with-tool-execution-and-structured-multi-agent-reasoning","status":"publish","type":"post","link":"https:\/\/connectword.dpdns.org\/?p=485","title":{"rendered":"A Coding Implementation to Build a Hierarchical Planner AI Agent Using Open-Source LLMs with Tool Execution and Structured Multi-Agent Reasoning"},"content":{"rendered":"

In this tutorial, we build a hierarchical planner agent using an open-source instruct model. We design a structured multi-agent architecture comprising a planner agent, an executor agent, and an aggregator agent, where each component plays a specialized role in solving complex tasks. We use the planner agent to decompose high-level goals into actionable steps, the executor agent to execute those steps using reasoning or Python tool execution, and the aggregator agent to synthesize results into a coherent final response. By integrating tool usage, structured planning, and iterative execution, we create a fully autonomous agent system that demonstrates how modern AI agents reason, plan, and act in a scalable and modular manner.<\/p>\n

\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
!pip -q install -U transformers accelerate bitsandbytes sentencepiece\n\n\nimport json\nimport re\nimport io\nimport contextlib\nfrom dataclasses import dataclass\nfrom typing import Any, Dict, List, Optional\n\n\nimport torch\nfrom transformers import AutoTokenizer, AutoModelForCausalLM\n\n\n\n\nMODEL_ID = \"Qwen\/Qwen2.5-1.5B-Instruct\"\nDEVICE = \"cuda\" if torch.cuda.is_available() else \"cpu\"\n\n\nprint(\"Device:\", DEVICE)\n\n\ntokenizer = AutoTokenizer.from_pretrained(MODEL_ID, use_fast=True)\n\n\nmodel = None\nif DEVICE == \"cuda\":\n   try:\n       model = AutoModelForCausalLM.from_pretrained(\n           MODEL_ID,\n           device_map=\"auto\",\n           torch_dtype=\"auto\",\n           load_in_4bit=True,\n       )\n       print(\"Loaded model in 4-bit.\")\n   except Exception as e:\n       print(\"4-bit load failed, falling back to normal load:\", str(e)[:200])\n       model = AutoModelForCausalLM.from_pretrained(\n           MODEL_ID,\n           device_map=\"auto\",\n           torch_dtype=\"auto\",\n       )\nelse:\n   model = AutoModelForCausalLM.from_pretrained(\n       MODEL_ID,\n       torch_dtype=torch.float32,\n   ).to(DEVICE)\n\n\nmodel.eval()<\/code><\/pre>\n<\/div>\n<\/div>\n
We install the required libraries and import all the modules necessary to build our hierarchical agent system. We load the open-source Qwen instruct model and configure it to run efficiently on a GPU using 4-bit quantization when available. We initialize the tokenizer and model, ensuring that our agent has the language understanding and reasoning capabilities needed to plan and execute tasks.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
def llm_chat(system: str, user: str, max_new_tokens: int = 500, temperature: float = 0.3) -> str:\n   messages = [\n       {\"role\": \"system\", \"content\": system.strip()},\n       {\"role\": \"user\", \"content\": user.strip()},\n   ]\n   prompt = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)\n   inputs = tokenizer(prompt, return_tensors=\"pt\").to(model.device)\n\n\n   with torch.no_grad():\n       out = model.generate(\n           **inputs,\n           max_new_tokens=max_new_tokens,\n           do_sample=True if temperature > 0 else False,\n           temperature=temperature,\n           top_p=0.9,\n           repetition_penalty=1.05,\n           eos_token_id=tokenizer.eos_token_id,\n       )\n\n\n   text = tokenizer.decode(out[0], skip_special_tokens=True)\n   return text.split(user.strip())[-1].strip()\n\n\n\n\ndef run_python(code: str) -> Dict[str, Any]:\n   buf = io.StringIO()\n   env: Dict[str, Any] = {\"__builtins__\": __builtins__}\n   try:\n       with contextlib.redirect_stdout(buf):\n           exec(code, env, env)\n       return {\"ok\": True, \"stdout\": buf.getvalue(), \"error\": None}\n   except Exception as e:\n       return {\"ok\": False, \"stdout\": buf.getvalue(), \"error\": repr(e)}<\/code><\/pre>\n<\/div>\n<\/div>\n
We define the core interaction function that allows us to communicate with the language model using a structured system and user prompts. We generate responses from the model using controlled sampling parameters to ensure stable, coherent reasoning. We also implement a Python execution tool that allows our agent to dynamically execute generated code and capture its output safely.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
def extract_json_block(text: str) -> Optional[Any]:\n   fenced = re.search(r\"```jsons*(.*?)s*```\", text, flags=re.DOTALL | re.IGNORECASE)\n   if fenced:\n       cand = fenced.group(1).strip()\n       try:\n           return json.loads(cand)\n       except:\n           pass\n\n\n   start_obj = text.find(\"{\")\n   start_arr = text.find(\"[\")\n   starts = [i for i in [start_obj, start_arr] if i != -1]\n   if not starts:\n       return None\n   start = min(starts)\n\n\n   s = text[start:]\n   stack = []\n   end = None\n   for i, ch in enumerate(s):\n       if ch in \"{[\":\n           stack.append(ch)\n       elif ch in \"}]\":\n           if not stack:\n               continue\n           op = stack.pop()\n           if (op == \"{\" and ch != \"}\") or (op == \"[\" and ch != \"]\"):\n               return None\n           if not stack:\n               end = i + 1\n               break\n   if end is None:\n       return None\n\n\n   cand = s[:end].strip()\n   try:\n       return json.loads(cand)\n   except:\n       return None<\/code><\/pre>\n<\/div>\n<\/div>\n
We implement a robust JSON extraction mechanism that allows us to reliably parse structured plans generated by the planner agent. We handle multiple formats, including fenced JSON blocks and inline JSON, to ensure resilience against imperfect model outputs. We ensure that our agent can consistently convert raw language model outputs into structured data for downstream execution.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
PLANNER_SYSTEM = \"\"\"\nYou are a Hierarchical Planner Agent.\nYou break down the user's task into 3-8 clear steps.\nYou MUST output ONLY valid JSON (no extra text).\nSchema:\n{\n \"goal\": \"string\",\n \"assumptions\": [\"string\", ...],\n \"steps\": [\n   {\n     \"id\": 1,\n     \"title\": \"short title\",\n     \"instruction\": \"what to do\",\n     \"tool\": \"none|llm|python\",\n     \"expected_output\": \"what we should get\"\n   }\n ]\n}\nGuidelines:\n- Use tool=\"python\" only if calculation \/ data processing \/ simulation helps.\n- Keep steps independent and executable.\n\"\"\"\n\n\n\n\nEXECUTOR_SYSTEM = \"\"\"\nYou are an Executor Agent.\nGiven a step and the current context, you produce the result for that step.\nIf the step tool is \"python\", output ONLY Python code (no backticks).\nIf the step tool is \"llm\" or \"none\", output a concise result, and reference any prior step outputs when relevant.\n\"\"\"\n\n\n\n\nAGGREGATOR_SYSTEM = \"\"\"\nYou are an Aggregator Agent.\nYou combine step outputs into a final, polished response to the original task.\nBe structured, correct, and practical.\nIf the task asks for an actionable plan, include bullet points and clear next actions.\n\"\"\"\n\n\n\n\n@dataclass\nclass StepResult:\n   step_id: int\n   title: str\n   tool: str\n   output: str\n\n\n\n\ndef planner_agent(task: str) -> Dict[str, Any]:\n   raw = llm_chat(PLANNER_SYSTEM, f\"Task:n{task}nnReturn JSON only.\", max_new_tokens=650, temperature=0.2)\n   plan = extract_json_block(raw)\n   if plan is None or \"steps\" not in plan:\n       raw2 = llm_chat(\n           PLANNER_SYSTEM,\n           f\"Your last output was invalid. Task:n{task}nReturn ONLY valid JSON matching the schema.\",\n           max_new_tokens=650,\n           temperature=0.0,\n       )\n       plan = extract_json_block(raw2)\n   if plan is None:\n       plan = {\n           \"goal\": task,\n           \"assumptions\": [],\n           \"steps\": [\n               {\"id\": 1, \"title\": \"Analyze\", \"instruction\": \"Analyze the task and outline an approach.\", \"tool\": \"llm\", \"expected_output\": \"Approach\"},\n               {\"id\": 2, \"title\": \"Execute\", \"instruction\": \"Produce the main solution.\", \"tool\": \"llm\", \"expected_output\": \"Solution\"},\n               {\"id\": 3, \"title\": \"Refine\", \"instruction\": \"Improve clarity and add next actions.\", \"tool\": \"llm\", \"expected_output\": \"Polished final\"},\n           ],\n       }\n   return plan<\/code><\/pre>\n<\/div>\n<\/div>\n
We define the prompts for the planner, executor, and aggregator agent system that establishes the hierarchical reasoning architecture. We create a planner agent function that decomposes complex tasks into structured steps, using defined tools and expected outputs. We also define the StepResult structure to store execution outputs in a structured, reusable format for subsequent reasoning.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
def executor_agent(step: Dict[str, Any], context: Dict[str, Any]) -> StepResult:\n   step_id = int(step.get(\"id\", 0))\n   title = step.get(\"title\", f\"Step {step_id}\")\n   tool = step.get(\"tool\", \"llm\")\n\n\n   ctx_compact = {\n       \"goal\": context.get(\"goal\"),\n       \"assumptions\": context.get(\"assumptions\", []),\n       \"prior_results\": [\n           {\"step_id\": r.step_id, \"title\": r.title, \"tool\": r.tool, \"output\": r.output[:1500]}\n           for r in context.get(\"results\", [])\n       ],\n   }\n\n\n   if tool == \"python\":\n       code = llm_chat(\n           EXECUTOR_SYSTEM,\n           user=(\n               f\"Step:n{json.dumps(step, indent=2)}nn\"\n               f\"Context:n{json.dumps(ctx_compact, indent=2)}nn\"\n               f\"Write Python code that completes the step. Output ONLY code.\"\n           ),\n           max_new_tokens=700,\n           temperature=0.2,\n       )\n       py = run_python(code)\n       out = []\n       out.append(\"PYTHON_CODE:n\" + code)\n       out.append(\"nEXECUTION_OK: \" + str(py[\"ok\"]))\n       if py[\"stdout\"]:\n           out.append(\"nSTDOUT:n\" + py[\"stdout\"])\n       if py[\"error\"]:\n           out.append(\"nERROR:n\" + py[\"error\"])\n       return StepResult(step_id=step_id, title=title, tool=tool, output=\"n\".join(out))\n\n\n   result_text = llm_chat(\n       EXECUTOR_SYSTEM,\n       user=(\n           f\"Step:n{json.dumps(step, indent=2)}nn\"\n           f\"Context:n{json.dumps(ctx_compact, indent=2)}nn\"\n           f\"Return the step result.\"\n       ),\n       max_new_tokens=700,\n       temperature=0.3,\n   )\n   return StepResult(step_id=step_id, title=title, tool=tool, output=result_text)\n\n\n\n\ndef aggregator_agent(task: str, plan: Dict[str, Any], results: List[StepResult]) -> str:\n   payload = {\n       \"task\": task,\n       \"plan\": plan,\n       \"results\": [{\"step_id\": r.step_id, \"title\": r.title, \"tool\": r.tool, \"output\": r.output[:2500]} for r in results],\n   }\n   return llm_chat(\n       AGGREGATOR_SYSTEM,\n       user=f\"Combine everything into the final answer.nnINPUT:n{json.dumps(payload, indent=2)}\",\n       max_new_tokens=900,\n       temperature=0.2,\n   )\n\n\n\n\ndef run_hierarchical_agent(task: str, verbose: bool = True) -> Dict[str, Any]:\n   plan = planner_agent(task)\n\n\n   if verbose:\n       print(\"n====================\")\n       print(\"PLAN (from Planner)\")\n       print(\"====================\")\n       print(json.dumps(plan, indent=2))\n\n\n   context = {\n       \"goal\": plan.get(\"goal\", task),\n       \"assumptions\": plan.get(\"assumptions\", []),\n       \"results\": [],\n   }\n\n\n   results: List[StepResult] = []\n   for step in plan.get(\"steps\", []):\n       res = executor_agent(step, context)\n       results.append(res)\n       context[\"results\"].append(res)\n\n\n       if verbose:\n           print(\"n--------------------\")\n           print(f\"STEP {res.step_id}: {res.title}  [tool={res.tool}]\")\n           print(\"--------------------\")\n           print(res.output)\n\n\n   final = aggregator_agent(task, plan, results)\n   if verbose:\n       print(\"n====================\")\n       print(\"FINAL (from Aggregator)\")\n       print(\"====================\")\n       print(final)\n\n\n   return {\"task\": task, \"plan\": plan, \"results\": results, \"final\": final}\n\n\n\n\ndemo_task = \"\"\"\nCreate a practical checklist to launch a small multi-agent system in Python for coordinating logistics:\n- One planner agent that decomposes tasks\n- Two executor agents (routing + inventory)\n- A simple memory store for past decisions\nKeep it lightweight and runnable in Colab.\n\"\"\"\n\n\n_ = run_hierarchical_agent(demo_task, verbose=True)\n\n\nprint(\"nnType your own task (or press Enter to skip):\")\nuser_task = input().strip()\nif user_task:\n   _ = run_hierarchical_agent(user_task, verbose=True)<\/code><\/pre>\n<\/div>\n<\/div>\n
We implement the executor agent that executes each planned step using either language reasoning or dynamic Python code execution. We build the aggregator agent to combine intermediate results and generate a final structured response to the original task. We orchestrate the full hierarchical workflow, allowing our agent to autonomously plan, execute, and produce intelligent outputs in an end-to-end pipeline.<\/p>\n
In conclusion, we implemented a hierarchical multi-agent system that demonstrates structured planning, execution, and aggregation within a unified framework. We enabled the planner agent to break down complex problems, empowered the executor agent to perform reasoning and dynamic Python execution, and used the aggregator agent to generate refined, actionable outputs. By combining open-source language models, structured prompts, and tool-based execution, we created a powerful and extensible foundation for building advanced autonomous agents.<\/p>\n
\n
Check out the\u00a0Full Codes here<\/a>.\u00a0<\/strong>Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0120k+ ML SubReddit<\/a><\/strong>\u00a0and Subscribe to\u00a0our Newsletter<\/a><\/strong>. Wait! are you on telegram?\u00a0now you can join us on telegram as well.<\/a><\/strong><\/p>\n
The post A Coding Implementation to Build a Hierarchical Planner AI Agent Using Open-Source LLMs with Tool Execution and Structured Multi-Agent Reasoning<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"
In this tutorial, we build a h…<\/p>\n","protected":false},"author":1,"featured_media":29,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-485","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/485","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=485"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/485\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/media\/29"}],"wp:attachment":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=485"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=485"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=485"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}