def _hsize(nbytes): para u en [“B”, “KB”, “MB”, “GB”]: si nbytes < 1024: return f"{nbytes:.1f}{u}" nbytes /= 1024 return f"{nbytes:.1f}TB" def build_prompt(instrucción, espacio de trabajo): exts = (".csv", ".xlsx", ".xls", ".json", ".xml", ".yaml", ".yml", ".txt", Archivos ".md", ".tsv", ".db", ".sqlite") = ordenados (f para f en os.listdir(espacio de trabajo) si f.lower().endswith(exts)) líneas = [f'File {i+1}: {{"name": "{f}", "size": "'
f'{_hsize(os.path.getsize(os.path.join(workspace, f)))}"}}'
for i, f in enumerate(files)]
return f"# Instrucción\n{instrucción}\n\n# Datos\n" + "\n".join(lines) WORKSPACE = "/content/da_workspace" os.makedirs(WORKSPACE, exist_ok=True) rng = np.random.default_rng(42); N = 2500 categorías = ["Electronics", "Home", "Fashion", "Books", "Toys"]
fechas = pd.to_datetime("2024-01-01") + pd.to_timedelta(rng.integers(0, 365, N), unit="D") tx = pd.DataFrame({ "order_id": np.arange(100000, 100000 + N), "fecha": fechas, "customer_id": rng.integers(1, 601, N), "categoría": rng.choice(categorías, N, p=[.3, .2, .25, .15, .1]), "región": rng.choice(["North", "South", "East", "West"]N), "cantidad": rng.integers(1, 6, N), "precio unitario": np.round(rng.gamma(3, 12, N) + 5, 2), "descuento": np.round(rng.choice([0, .05, .1, .15, .2]N), 2), }) tx["revenue"] = np.round(tx.cantidad * tx.precio_unidad * (1 - tx.descuento), 2) tx.loc[rng.choice(N, 60, replace=False), "unit_price"] = np.nan tx.to_csv(f"{WORKSPACE}/transactions.csv", index=False) pd.DataFrame({ "customer_id": np.arange(1, 601), "signup_year": rng.choice([2021, 2022, 2023, 2024]600), "segmento": rng.choice(["Consumer", "Corporate", "Home Office"]600), "age": rng.integers(18, 70, 600), }).to_excel(f"{WORKSPACE}/customers.xlsx", index=False) print("Archivos de espacio de trabajo:", os.listdir(WORKSPACE)) INSTRUCTION = ( "Realice un análisis de extremo a extremo de este conjunto de datos de comercio electrónico. Explore y " "úna los archivos, limpie cualquier calidad de los datos problemas, analice las tendencias de ingresos a lo largo del " "tiempo, por región, categoría y segmento de clientes, e identifique los " "impulsores clave de ingresos. Cree al menos una visualización clara y GUARDE " "como un archivo PNG en el directorio actual. Termine con un " "informe de calidad de analista conciso y bien estructurado con 2 o 3 recomendaciones prácticas". DeepAnalyzeAgent(modelo, tok, temperatura=0.5) t0 = time.time() resultado = agente.run(INSTRUCTION, WORKSPACE, max_rounds=12, max_new_tokens=3072, exec_timeout=120) print(f"\n\nHecho en {time.time()-t0:.0f}s | " f"{suma(1 para k,_ en resultado['trace'] if k=='execute')} ejecuciones de código") intente: desde IPython.display visualización de importación, Markdown, Imagen si es resultado["answer"]: display(Markdown("## 📊 Informe final\n\n" + resultado["answer"])) else: print("No se produjo ningún bloque (intente generar max_rounds).") for img in sorted(set(glob.glob(f"{WORKSPACE}/*.png")) - existing_imgs): print("Figura:", img); mostrar (Imagen (nombre de archivo = img)) excepto Excepción: imprimir ("\n===== INFORME FINAL =====\n", resultado["answer"])