def _hsize(nbytes):
for u in ["B", "KB", "MB", "GB"]:
if nbytes < 1024: return f"{nbytes:.1f}{u}"
nbytes /= 1024
return f"{nbytes:.1f}TB"
def build_prompt(instruction, workspace):
exts = (".csv", ".xlsx", ".xls", ".json", ".xml", ".yaml", ".yml",
".txt", ".md", ".tsv", ".db", ".sqlite")
files = sorted(f for f in os.listdir(workspace) if f.lower().endswith(exts))
lines = [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"# Instruction\n{instruction}\n\n# Data\n" + "\n".join(lines)
WORKSPACE = "/content/da_workspace"
os.makedirs(WORKSPACE, exist_ok=True)
rng = np.random.default_rng(42); N = 2500
categories = ["Electronics", "Home", "Fashion", "Books", "Toys"]
dates = 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), "date": dates,
"customer_id": rng.integers(1, 601, N),
"category": rng.choice(categories, N, p=[.3, .2, .25, .15, .1]),
"region": rng.choice(["North", "South", "East", "West"], N),
"quantity": rng.integers(1, 6, N),
"unit_price": np.round(rng.gamma(3, 12, N) + 5, 2),
"discount": np.round(rng.choice([0, .05, .1, .15, .2], N), 2),
})
tx["revenue"] = np.round(tx.quantity * tx.unit_price * (1 - tx.discount), 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),
"segment": rng.choice(["Consumer", "Corporate", "Home Office"], 600),
"age": rng.integers(18, 70, 600),
}).to_excel(f"{WORKSPACE}/customers.xlsx", index=False)
print("Workspace files:", os.listdir(WORKSPACE))
INSTRUCTION = (
"Perform an end-to-end analysis of this e-commerce dataset. Explore and "
"join the files, clean any data-quality issues, analyze revenue trends over "
"time, by region, category, and customer segment, and identify the key "
"drivers of revenue. Create at least one clear visualization and SAVE it as "
"a PNG file in the current directory. Finish with a concise, well-structured "
"analyst-grade report of your findings and 2-3 actionable recommendations."
)
existing_imgs = set(glob.glob(f"{WORKSPACE}/*.png"))
agent = DeepAnalyzeAgent(model, tok, temperature=0.5)
t0 = time.time()
result = agent.run(INSTRUCTION, WORKSPACE, max_rounds=12,
max_new_tokens=3072, exec_timeout=120)
print(f"\n\nDone in {time.time()-t0:.0f}s | "
f"{sum(1 for k,_ in result['trace'] if k=='execute')} code executions")
try:
from IPython.display import display, Markdown, Image
if result["answer"]:
display(Markdown("## 📊 Final Report\n\n" + result["answer"]))
else:
print("No block produced (try raising max_rounds).")
for img in sorted(set(glob.glob(f"{WORKSPACE}/*.png")) - existing_imgs):
print("Figure:", img); display(Image(filename=img))
except Exception:
print("\n===== FINAL REPORT =====\n", result["answer"])
Credit: Source link
