clase MetaAgent: def __init__(self, llm: Opcional[LocalLLM] = Ninguno): self.llm = llm o LocalLLM() def _capability_heuristics(self, tarea: str) -> Dict[str, Any]: t = tarea.inferior() necesita_datos = cualquiera(k en t para k en [“csv”, “dataframe”, “pandas”, “dataset”, “table”, “excel”]) need_math = cualquiera(k en t para k en [“calculate”, “compute”, “probability”, “equation”, “optimize”, “derivative”, “integral”]) necesita_escritura = cualquiera(k en t para k en [“write”, “draft”, “email”, “cover letter”, “proposal”, “summarize”, “rewrite”]) análisis_necesidades = cualquiera(k en t para k en [“analyze”, “insights”, “trend”, “compare”, “benchmark”]) necesita_memoria = cualquiera(k en t para k en [“long”, “multi-step”, “remember”, “plan”, “workflow”, “pipeline”]) return { “needs_data”: need_data, “needs_math”: need_math, “needs_writing”: need_writing, “needs_analysis”: need_analysis, “needs_memory”: need_memory, } def design(self, task_description: str) -> AgentConfig: caps = self._capability_heuristics(task_description) herramientas = default_tool_registry() seleccionado: Lista[ToolSpec] = []
selected.append(ToolSpec( name=”calc”, descripción=”Evaluar una expresión matemática segura (sin código arbitrario).”, inputs_schema={“type”:”object”,”properties”:{“expression”:{“type”:”string”}}, “required”:[“expression”]} )) selected.append(ToolSpec( name=”text_stats”, descripción=”Calcular estadísticas básicas sobre un blob de texto (palabras, líneas, palabras únicas).”, inputs_schema={“type”:”object”,”properties”:{“text”:{“type”:”string”}}, “required”:[“text”]} )) si mayúsculas[“needs_data”]: selected.append(ToolSpec( name=”csv_profile”, descripción=”Cargue un CSV desde una ruta local e imprima un perfil rápido (head, describe).”, inputs_schema={“type”:”object”,”properties”:{“path”:{“type”:”string”},”n_rows”:{“type”:”integer”}}, “required”:[“path”]} )) si mayúsculas[“needs_memory”] o gorras[“needs_analysis”] o gorras[“needs_data”]: mem = MemorySpec(kind=”retrieval_tfidf”, max_items=250, retrieval_k=6) else: mem = MemorySpec(kind=”scratchpad”, max_items=120, retrieval_k=5) si está en mayúsculas[“needs_analysis”] o gorras[“needs_data”] o gorras[“needs_memory”]: planificador = PlannerSpec(kind=”react”, max_steps=12, temperatura=0.2) else: planificador = PlannerSpec(kind=”react”, max_steps=8, temperatura=0.2) objetivo = “Resolver la tarea del usuario con el uso de herramientas cuando sea útil; producir una respuesta final limpia.” cfg = AgentConfig( nombre_agente=”AutoDesignedAgent”, objetivo=objetivo, planificador=planificador, memoria=mem, herramientas=seleccionado, estilo_salida=”concise”, ) para ts en seleccionado: si no, herramientas.has(ts.nombre): elevar RuntimeError(f”Herramienta seleccionada pero no registrada: {ts.nombre}”) return cfg def instanciate(self, cfg: AgentConfig) -> AgentRuntime: herramientas = 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, herramientas=tools, memoria=mem) def evaluar(self, tarea: str, respuesta: str) -> Dict[str, Any]: a = (respuesta o “”).strip().lower() flags = { “vacío”: len(a) == 0, “genérico”: cualquiera(p en a para p en [“i can’t”, “cannot”, “missing”, “provide more details”, “parser fallback”]), “mentions_max_steps”: “pasos máximos” en a, } puntuación = 1,0 si hay banderas[“empty”]: puntuación -= 0,6 si hay banderas[“generic”]: puntuación -= 0,25 si hay banderas[“mentions_max_steps”]: puntuación -= 0,2 puntuación = max(0,0, min(1,0, puntuación)) return {“score”: puntuación, “flags”: flags} def refine(self, cfg: AgentConfig, eval_report: Dict[str, Any]tarea: str) -> AgentConfig: new_cfg = cfg.model_copy(deep=True) si eval_report[“flags”][“generic”] o informe_eval[“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() si existe(k en t para k en [“csv”, “dataframe”, “pandas”, “dataset”, “table”]): si no hay ninguno(ts.name == “csv_profile” for ts in new_cfg.tools): new_cfg.tools.append(ToolSpec( name=”csv_profile”, descripción=”Cargue un CSV desde una ruta local e imprima un perfil rápido (head, describe).”, inputs_schema={“tipo”:”objeto”,”properties”:{“ruta”:{“tipo”:”cadena”},”n_rows”:{“tipo”:”entero”}}, “requerido”:[“path”]} )) devolver new_cfg def build_and_run(self, tarea: str, mejorar_rounds: int = 1, detallado: bool = True) -> Tupla[str, AgentConfig]: cfg = self.design(tarea) agente = self.instantiate(cfg) si es detallado: print(“\n===============================”) print(“META-AGENTE: CONFIGURACIÓN DISEÑADA”) print(“===============================”) print(cfg.model_dump_json(indent=2)) ans = agent.run(tarea, detallado=detallado) report = self.evaluate(tarea, ans) si detallado: print(“\n===============================”) print(“INFORME DE EVALUACIÓN”) print(“===============================”) print(json.dumps(report, indent=2)) print(“\n===============================”) print(“RESPUESTA FINAL”) print(“===============================”) print(ans) para r en el rango(improve_rounds): si informe[“score”] >= 0.85: break cfg = self.refine(cfg, report, task) agent = self.instantiate(cfg) if detallado: print(f”\n\n================================”) print(f”RONDA DE MEJORA PERSONAL {r+1}: CONFIGURACIÓN ACTUALIZADA”) print(“==============================”) print(cfg.model_dump_json(indent=2)) ans = agent.run(tarea, detallado=detallado) report = self.evaluate(tarea, respuesta) si detallado: print(“\nEVAL:”, json.dumps(report, indent=2)) print(“\nRESPUESTA:\n”, ans) return ans, cfg meta = MetaAgent() ejemplos = [
“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.”,
]
selected.append(ToolSpec( name=”calc”, descripción=”Evaluar una expresión matemática segura (sin código arbitrario).”, inputs_schema={“type”:”object”,”properties”:{“expression”:{“type”:”string”}}, “required”:[“expression”]} )) selected.append(ToolSpec( name=”text_stats”, descripción=”Calcular estadísticas básicas sobre un blob de texto (palabras, líneas, palabras únicas).”, inputs_schema={“type”:”object”,”properties”:{“text”:{“type”:”string”}}, “required”:[“text”]} )) si mayúsculas[“needs_data”]: selected.append(ToolSpec( name=”csv_profile”, descripción=”Cargue un CSV desde una ruta local e imprima un perfil rápido (head, describe).”, inputs_schema={“type”:”object”,”properties”:{“path”:{“type”:”string”},”n_rows”:{“type”:”integer”}}, “required”:[“path”]} )) si mayúsculas[“needs_memory”] o gorras[“needs_analysis”] o gorras[“needs_data”]: mem = MemorySpec(kind=”retrieval_tfidf”, max_items=250, retrieval_k=6) else: mem = MemorySpec(kind=”scratchpad”, max_items=120, retrieval_k=5) si está en mayúsculas[“needs_analysis”] o gorras[“needs_data”] o gorras[“needs_memory”]: planificador = PlannerSpec(kind=”react”, max_steps=12, temperatura=0.2) else: planificador = PlannerSpec(kind=”react”, max_steps=8, temperatura=0.2) objetivo = “Resolver la tarea del usuario con el uso de herramientas cuando sea útil; producir una respuesta final limpia.” cfg = AgentConfig( nombre_agente=”AutoDesignedAgent”, objetivo=objetivo, planificador=planificador, memoria=mem, herramientas=seleccionado, estilo_salida=”concise”, ) para ts en seleccionado: si no, herramientas.has(ts.nombre): elevar RuntimeError(f”Herramienta seleccionada pero no registrada: {ts.nombre}”) return cfg def instanciate(self, cfg: AgentConfig) -> AgentRuntime: herramientas = 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, herramientas=tools, memoria=mem) def evaluar(self, tarea: str, respuesta: str) -> Dict[str, Any]: a = (respuesta o “”).strip().lower() flags = { “vacío”: len(a) == 0, “genérico”: cualquiera(p en a para p en [“i can’t”, “cannot”, “missing”, “provide more details”, “parser fallback”]), “mentions_max_steps”: “pasos máximos” en a, } puntuación = 1,0 si hay banderas[“empty”]: puntuación -= 0,6 si hay banderas[“generic”]: puntuación -= 0,25 si hay banderas[“mentions_max_steps”]: puntuación -= 0,2 puntuación = max(0,0, min(1,0, puntuación)) return {“score”: puntuación, “flags”: flags} def refine(self, cfg: AgentConfig, eval_report: Dict[str, Any]tarea: str) -> AgentConfig: new_cfg = cfg.model_copy(deep=True) si eval_report[“flags”][“generic”] o informe_eval[“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() si existe(k en t para k en [“csv”, “dataframe”, “pandas”, “dataset”, “table”]): si no hay ninguno(ts.name == “csv_profile” for ts in new_cfg.tools): new_cfg.tools.append(ToolSpec( name=”csv_profile”, descripción=”Cargue un CSV desde una ruta local e imprima un perfil rápido (head, describe).”, inputs_schema={“tipo”:”objeto”,”properties”:{“ruta”:{“tipo”:”cadena”},”n_rows”:{“tipo”:”entero”}}, “requerido”:[“path”]} )) devolver new_cfg def build_and_run(self, tarea: str, mejorar_rounds: int = 1, detallado: bool = True) -> Tupla[str, AgentConfig]: cfg = self.design(tarea) agente = self.instantiate(cfg) si es detallado: print(“\n===============================”) print(“META-AGENTE: CONFIGURACIÓN DISEÑADA”) print(“===============================”) print(cfg.model_dump_json(indent=2)) ans = agent.run(tarea, detallado=detallado) report = self.evaluate(tarea, ans) si detallado: print(“\n===============================”) print(“INFORME DE EVALUACIÓN”) print(“===============================”) print(json.dumps(report, indent=2)) print(“\n===============================”) print(“RESPUESTA FINAL”) print(“===============================”) print(ans) para r en el rango(improve_rounds): si informe[“score”] >= 0.85: break cfg = self.refine(cfg, report, task) agent = self.instantiate(cfg) if detallado: print(f”\n\n================================”) print(f”RONDA DE MEJORA PERSONAL {r+1}: CONFIGURACIÓN ACTUALIZADA”) print(“==============================”) print(cfg.model_dump_json(indent=2)) ans = agent.run(tarea, detallado=detallado) report = self.evaluate(tarea, respuesta) si detallado: print(“\nEVAL:”, json.dumps(report, indent=2)) print(“\nRESPUESTA:\n”, ans) return ans, cfg meta = MetaAgent() ejemplos = [
“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(“EJECUTANDO UNA TAREA DE DEMOSTRACIÓN RÁPIDA”) print(“================================”) demo_task = ejemplos[2]
_ = meta.build_and_run(demo_task, mejorar_rondas=1, detallado=Verdadero)