class MetaAgent: def __init__(self, llm: Optional = None): self.llm = llm or LocalLLM() def _capability_heuristics(self, task: str) -> Dict: 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 = [] 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: 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, 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: 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.

class MetaAgent: def __init__(self, llm: Optional = None): self.llm = llm or LocalLLM() def _capability_heuristics(self, task: str) -> Dict: 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 = [] 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: 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, 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: 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.