A Coding Implementation to Construct a Hierarchical Planner AI Agent Utilizing Open-Supply LLMs with Software Execution and Structured Multi-Agent Reasoning

def executor_agent(step: Dict[str, Any], context: Dict[str, Any]) -> StepResult:
   step_id = int(step.get("id", 0))
   title = step.get("title", f"Step {step_id}")
   device = step.get("device", "llm")


   ctx_compact = {
       "purpose": context.get("purpose"),
       "assumptions": context.get("assumptions", []),
       "prior_results": [
           {"step_id": r.step_id, "title": r.title, "tool": r.tool, "output": r.output[:1500]}
           for r in context.get("outcomes", [])
       ],
   }


   if device == "python":
       code = llm_chat(
           EXECUTOR_SYSTEM,
           consumer=(
               f"Step:n{json.dumps(step, indent=2)}nn"
               f"Context:n{json.dumps(ctx_compact, indent=2)}nn"
               f"Write Python code that completes the step. Output ONLY code."
           ),
           max_new_tokens=700,
           temperature=0.2,
       )
       py = run_python(code)
       out = []
       out.append("PYTHON_CODE:n" + code)
       out.append("nEXECUTION_OK: " + str(py["ok"]))
       if py["stdout"]:
           out.append("nSTDOUT:n" + py["stdout"])
       if py["error"]:
           out.append("nERROR:n" + py["error"])
       return StepResult(step_id=step_id, title=title, device=device, output="n".be part of(out))


   result_text = llm_chat(
       EXECUTOR_SYSTEM,
       consumer=(
           f"Step:n{json.dumps(step, indent=2)}nn"
           f"Context:n{json.dumps(ctx_compact, indent=2)}nn"
           f"Return the step consequence."
       ),
       max_new_tokens=700,
       temperature=0.3,
   )
   return StepResult(step_id=step_id, title=title, device=device, output=result_text)




def aggregator_agent(job: str, plan: Dict[str, Any], outcomes: Record[StepResult]) -> str:
   payload = {
       "job": job,
       "plan": plan,
       "outcomes": [{"step_id": r.step_id, "title": r.title, "tool": r.tool, "output": r.output[:2500]} for r in outcomes],
   }
   return llm_chat(
       AGGREGATOR_SYSTEM,
       consumer=f"Mix every little thing into the ultimate reply.nnINPUT:n{json.dumps(payload, indent=2)}",
       max_new_tokens=900,
       temperature=0.2,
   )




def run_hierarchical_agent(job: str, verbose: bool = True) -> Dict[str, Any]:
   plan = planner_agent(job)


   if verbose:
       print("n====================")
       print("PLAN (from Planner)")
       print("====================")
       print(json.dumps(plan, indent=2))


   context = {
       "purpose": plan.get("purpose", job),
       "assumptions": plan.get("assumptions", []),
       "outcomes": [],
   }


   outcomes: Record[StepResult] = []
   for step in plan.get("steps", []):
       res = executor_agent(step, context)
       outcomes.append(res)
       context["results"].append(res)


       if verbose:
           print("n--------------------")
           print(f"STEP {res.step_id}: {res.title}  [tool={res.tool}]")
           print("--------------------")
           print(res.output)


   ultimate = aggregator_agent(job, plan, outcomes)
   if verbose:
       print("n====================")
       print("FINAL (from Aggregator)")
       print("====================")
       print(ultimate)


   return {"job": job, "plan": plan, "outcomes": outcomes, "ultimate": ultimate}




demo_task = """
Create a sensible guidelines to launch a small multi-agent system in Python for coordinating logistics:
- One planner agent that decomposes duties
- Two executor brokers (routing + stock)
- A easy reminiscence retailer for previous choices
Maintain it light-weight and runnable in Colab.
"""


_ = run_hierarchical_agent(demo_task, verbose=True)


print("nnType your personal job (or press Enter to skip):")
user_task = enter().strip()
if user_task:
   _ = run_hierarchical_agent(user_task, verbose=True)