How to Build a Self-Designing Meta-Agent That Automatically Constructs, Instantiates, and Refines Task-Specific AI Agents

class MetaAgent:
   def __init__(self, llm: Optional[LocalLLM] = None):
       self.llm = llm or LocalLLM()


   def _capability_heuristics(self, task: str) -> Dict[str, Any]:
       t = task.lower()


       needs_data = any(k in t for k in ["csv", "dataframe", "pandas", "dataset", "table", "excel"])
       needs_math = any(k in t for k in ["calculate", "compute", "probability", "equation", "optimize", "derivative", "integral"])
       needs_writing = any(k in t for k in ["write", "draft", "email", "cover letter", "proposal", "summarize", "rewrite"])
       needs_analysis = any(k in t for k in ["analyze", "insights", "trend", "compare", "benchmark"])
       needs_memory = any(k in t for k in ["long", "multi-step", "remember", "plan", "workflow", "pipeline"])


       return {
           "needs_data": needs_data,
           "needs_math": needs_math,
           "needs_writing": needs_writing,
           "needs_analysis": needs_analysis,
           "needs_memory": needs_memory,
       }


   def design(self, task_description: str) -> AgentConfig:
       caps = self._capability_heuristics(task_description)
       tools = default_tool_registry()


       selected: List[ToolSpec] = []
       selected.append(ToolSpec(
           name="calc",
           description="Evaluate a safe mathematical expression (no arbitrary code).",
           inputs_schema={"type":"object","properties":{"expression":{"type":"string"}}, "required":["expression"]}
       ))
       selected.append(ToolSpec(
           name="text_stats",
           description="Compute basic statistics about a text blob (words, lines, unique words).",
           inputs_schema={"type":"object","properties":{"text":{"type":"string"}}, "required":["text"]}
       ))
       if caps["needs_data"]:
           selected.append(ToolSpec(
               name="csv_profile",
               description="Load a CSV from a local path and print a quick profile (head, describe).",
               inputs_schema={"type":"object","properties":{"path":{"type":"string"},"n_rows":{"type":"integer"}}, "required":["path"]}
           ))


       if caps["needs_memory"] or caps["needs_analysis"] or caps["needs_data"]:
           mem = MemorySpec(kind="retrieval_tfidf", max_items=250, retrieval_k=6)
       else:
           mem = MemorySpec(kind="scratchpad", max_items=120, retrieval_k=5)


       if caps["needs_analysis"] or caps["needs_data"] or caps["needs_memory"]:
           planner = PlannerSpec(kind="react", max_steps=12, temperature=0.2)
       else:
           planner = PlannerSpec(kind="react", max_steps=8, temperature=0.2)


       objective = "Solve the user task with tool use when helpful; produce a clean final response."
       cfg = AgentConfig(
           agent_name="AutoDesignedAgent",
           objective=objective,
           planner=planner,
           memory=mem,
           tools=selected,
           output_style="concise",
       )


       for ts in selected:
           if not tools.has(ts.name):
               raise RuntimeError(f"Tool selected but not registered: {ts.name}")


       return cfg


   def instantiate(self, cfg: AgentConfig) -> AgentRuntime:
       tools = default_tool_registry()
       if cfg.memory.kind == "retrieval_tfidf":
           mem = TfidfRetrievalMemory(max_items=cfg.memory.max_items, retrieval_k=cfg.memory.retrieval_k)
       else:
           mem = ScratchpadMemory(max_items=cfg.memory.max_items)
       return AgentRuntime(config=cfg, llm=self.llm, tools=tools, memory=mem)


   def evaluate(self, task: str, answer: str) -> Dict[str, Any]:
       a = (answer or "").strip().lower()
       flags = {
           "empty": len(a) == 0,
           "generic": any(p in a for p in ["i can't", "cannot", "missing", "provide more details", "parser fallback"]),
           "mentions_max_steps": "max steps" in a,
       }
       score = 1.0
       if flags["empty"]: score -= 0.6
       if flags["generic"]: score -= 0.25
       if flags["mentions_max_steps"]: score -= 0.2
       score = max(0.0, min(1.0, score))
       return {"score": score, "flags": flags}


   def refine(self, cfg: AgentConfig, eval_report: Dict[str, Any], task: str) -> AgentConfig:
       new_cfg = cfg.model_copy(deep=True)


       if eval_report["flags"]["generic"] or eval_report["flags"]["mentions_max_steps"]:
           new_cfg.planner.max_steps = min(18, new_cfg.planner.max_steps + 6)
           new_cfg.planner.temperature = min(0.35, new_cfg.planner.temperature + 0.05)
           if new_cfg.memory.kind != "retrieval_tfidf":
               new_cfg.memory.kind = "retrieval_tfidf"
               new_cfg.memory.max_items = max(new_cfg.memory.max_items, 200)
               new_cfg.memory.retrieval_k = max(new_cfg.memory.retrieval_k, 6)


       t = task.lower()
       if any(k in t for k in ["csv", "dataframe", "pandas", "dataset", "table"]):
           if not any(ts.name == "csv_profile" for ts in new_cfg.tools):
               new_cfg.tools.append(ToolSpec(
                   name="csv_profile",
                   description="Load a CSV from a local path and print a quick profile (head, describe).",
                   inputs_schema={"type":"object","properties":{"path":{"type":"string"},"n_rows":{"type":"integer"}}, "required":["path"]}
               ))


       return new_cfg


   def build_and_run(self, task: str, improve_rounds: int = 1, verbose: bool = True) -> Tuple[str, AgentConfig]:
       cfg = self.design(task)
       agent = self.instantiate(cfg)


       if verbose:
           print("\n==============================")
           print("META-AGENT: DESIGNED CONFIG")
           print("==============================")
           print(cfg.model_dump_json(indent=2))


       ans = agent.run(task, verbose=verbose)
       report = self.evaluate(task, ans)


       if verbose:
           print("\n==============================")
           print("EVALUATION REPORT")
           print("==============================")
           print(json.dumps(report, indent=2))
           print("\n==============================")
           print("FINAL ANSWER")
           print("==============================")
           print(ans)


       for r in range(improve_rounds):
           if report["score"] >= 0.85:
               break
           cfg = self.refine(cfg, report, task)
           agent = self.instantiate(cfg)
           if verbose:
               print(f"\n\n==============================")
               print(f"SELF-IMPROVEMENT ROUND {r+1}: UPDATED CONFIG")
               print("==============================")
               print(cfg.model_dump_json(indent=2))
           ans = agent.run(task, verbose=verbose)
           report = self.evaluate(task, ans)
           if verbose:
               print("\nEVAL:", json.dumps(report, indent=2))
               print("\nANSWER:\n", ans)


       return ans, cfg


meta = MetaAgent()


examples = [
   "Design an agent workflow to summarize a long meeting transcript and extract action items. Keep it concise.",
   "I have a local CSV at /content/sample.csv. Profile it and tell me the top 3 insights.",
   "Compute the monthly payment for a $12,000 loan at 8% APR over 36 months. Show the formula briefly.",
]


print("\n==============================")
print("RUNNING A QUICK DEMO TASK")
print("==============================")
demo_task = examples[2]
_ = meta.build_and_run(demo_task, improve_rounds=1, verbose=True)