Una implementación de codificación para crear un agente de IA de planificación jerárquica utilizando LLM de código abierto con ejecución de herramientas y razonamiento estructurado de múltiples agentes

def agente_ejecutor(paso: Dict[str, Any]contexto: dictado[str, Any]) -> Resultado del paso: step_id = int(step.get(“id”, 0)) title = step.get(“title”, f”Step {step_id}”) herramienta = step.get(“tool”, “llm”) ctx_compact = { “meta”: context.get(“meta”), “supuestos”: context.get(“supuestos”, []), “resultados_anteriores”: [
{“step_id”: r.step_id, “title”: r.title, “tool”: r.tool, “output”: r.output[:1500]} para r en contexto.get(“resultados”, []) ], } if tool == “python”: code = llm_chat( EXECUTOR_SYSTEM, user=( f”Step:\n{json.dumps(step, indent=2)}\n\n” f”Context:\n{json.dumps(ctx_compact, indent=2)}\n\n” f”Escribe código Python que complete el paso. Genera SÓLO código.” ), max_new_tokens=700, temperatura=0.2, ) py = run_python(código) out = []
out.append(“PYTHON_CODE:\n” + código) out.append(“\nEXECUTION_OK: ” + str(py[“ok”])) si py[“stdout”]: out.append(“\nSTDOUT:\n” + py[“stdout”]) si py[“error”]: out.append(“\nERROR:\n” + py[“error”]) return StepResult(step_id=step_id, título=título, herramienta=herramienta, salida=”\n”.join(out)) result_text = llm_chat( EXECUTOR_SYSTEM, user=( f”Step:\n{json.dumps(step, indent=2)}\n\n” f”Context:\n{json.dumps(ctx_compact, indent=2)}\n\n” f”Devuelve el resultado del paso.” ), max_new_tokens=700, temperatura=0.3, ) return StepResult(step_id=step_id, título=título, herramienta=herramienta, salida=result_text) def aggregator_agent(tarea: str, plan: Dict[str, Any]resultados: Lista[StepResult]) -> str: carga útil = { “tarea”: ​​tarea, “plan”: plan, “resultados”: [{“step_id”: r.step_id, “title”: r.title, “tool”: r.tool, “output”: r.output[:2500]} para r en resultados]} return llm_chat( AGGREGATOR_SYSTEM, user=f”Combina todo en la respuesta final.\n\nINPUT:\n{json.dumps(payload, indent=2)}”, max_new_tokens=900, Temperature=0.2, ) def run_hierarchical_agent(task: str, detallado: bool = True) -> dictar[str, Any]: plan = planificador_agente(tarea) si es detallado: print(“\n=====================”) print(“PLAN (del Planificador)”) print(“=====================”) print(json.dumps(plan, sangría=2)) contexto = { “meta”: plan.get(“meta”, tarea), “supuestos”: plan.get(“supuestos”, []), “resultados”: []} resultados: Lista[StepResult] = []
para el paso en plan.get(“pasos”, []): res = executor_agent(paso, contexto) resultados.append(res) contexto[“results”].append(res) si es detallado: print(“\n——————–“) print(f”PASO {res.step_id}: {res.title} [tool={res.tool}]”) print(“——————–“) print(res.output) final = agregador_agent(tarea, plan, resultados) si es detallado: print(“\n====================”) print(“FINAL (del agregador)”) print(“=====================”) print(final) return {“tarea”: tarea, “plan”: plan, “results”: results, “final”: final} demo_task = “”” Cree una lista de verificación práctica para lanzar un pequeño sistema multiagente en Python para coordinar la logística: – Un agente planificador que descompone las tareas – Dos agentes ejecutores (enrutamiento + inventario) – Un almacén de memoria simple para decisiones pasadas Manténgalo liviano y ejecutable en Colab. “”” _ = run_hierarchical_agent(demo_task, verbose=True) print(“\n\nEscriba su propia tarea (o presione Enter para omitir):”) user_task = input().strip() if user_task: _ = run_hierarchical_agent(user_task, verbose=True)