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)