Como construir um agente autônomo de ciência de dados compatível com T4 usando DeepAnalyze-8B, execução de código em sandbox e análise iterativa

In this tutorial, we build an autonomous data science agent around DeepAnalyze-8B and run it. We begin by preparing a stable runtime, installing the required machine-learning dependencies, and loading the DeepAnalyze tokenizer and model in 4-bit mode to keep the workflow practical on limited GPU memory. We then create a sandboxed execution environment that allows the model to generate Python code, execute it safely, observe the results, and continue its analysis in an agentic loop. By the end of the workflow, we give the agent a realistic multi-file e-commerce workspace and let it clean, join, analyze, visualize, and summarize the data as a structured analyst-grade report.

Installing DeepAnalyze-8B Runtime Dependencies

Copy CodeCopiedUse a different Browser

import os, sys, subprocess

os.environ["MPLBACKEND"] = "Agg"

def _pip(*args):

subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", *args])

_SETUP_FLAG = "/content/.da_ready"

if not os.path.exists(_SETUP_FLAG):

print("Installing dependencies (one-time). The runtime will RESTART; "

"just re-run this cell afterwards.\n")

_pip("-U", "transformers>=4.44", "accelerate>=0.30", "bitsandbytes>=0.43")

_pip("sentencepiece")

_pip("openpyxl")

_pip("--force-reinstall", "numpy==2.0.2")

open(_SETUP_FLAG, "w").close()

print("\nDependencies ready. Restarting runtime now...")

os.kill(os.getpid(), 9)

We start by preparing the Colab runtime with the required machine-learning dependencies for DeepAnalyze-8B. We install the transformer, acceleration, quantization, tokenizer, and spreadsheet libraries without disturbing the broader notebook workflow. We also pin NumPy and restart the runtime once to keep the environment clean and stable for the next execution.

Loading DeepAnalyze-8B in 4-Bit Mode

Copy CodeCopiedUse a different Browser

import re, io, glob, time, signal, contextlib, warnings, traceback

from threading import Thread

import numpy as np, pandas as pd

import torch

from transformers import (AutoModelForCausalLM, AutoTokenizer,

BitsAndBytesConfig, TextIteratorStreamer)

warnings.filterwarnings("ignore")

MODEL_ID = "RUC-DataLab/DeepAnalyze-8B"

USE_4BIT = True

COMPUTE_DT = torch.float16

assert torch.cuda.is_available(), (

"No GPU detected. In Colab: Runtime -> Change runtime type -> GPU.")

print("GPU:", torch.cuda.get_device_name(0), "| NumPy:", np.__version__)

print("\nLoading tokenizer & model (first run downloads ~16GB, be patient)...")

tok = AutoTokenizer.from_pretrained(MODEL_ID, trust_remote_code=True)

if tok.pad_token_id is None:

tok.pad_token = tok.eos_token

bnb = BitsAndBytesConfig(

load_in_4bit=True, bnb_4bit_quant_type="nf4",

bnb_4bit_compute_dtype=COMPUTE_DT, bnb_4bit_use_double_quant=True,

) if USE_4BIT else None

model = AutoModelForCausalLM.from_pretrained(

MODEL_ID, quantization_config=bnb, device_map="auto",

torch_dtype=COMPUTE_DT, low_cpu_mem_usage=True, trust_remote_code=True,

model.eval()

print("Model loaded. VRAM used: %.1f GB" % (torch.cuda.memory_allocated()/1e9))

We import the main libraries, configure the DeepAnalyze-8B model, and verify that a GPU is available in Colab. We load the tokenizer and prepare 4-bit quantization so the model can fit more comfortably on a T4 GPU. We then load the model in evaluation mode and confirm GPU memory usage before moving on to the agent logic.

Building the Sandboxed Code Executor

Copy CodeCopiedUse a different Browser

class CodeSandbox:

def __init__(self, timeout=120, max_chars=6000):

self.ns = {"__name__": "__main__"}

self.timeout, self.max_chars = timeout, max_chars

def _run(self, code):

with contextlib.redirect_stdout(io.StringIO()) as out, \

contextlib.redirect_stderr(io.StringIO()) as err:

exec(compile(code, "", "exec"), self.ns)

return out.getvalue() + err.getvalue()

def execute(self, code):

def _handler(signum, frame):

raise TimeoutError(f"Execution exceeded {self.timeout}s")

prev = signal.signal(signal.SIGALRM, _handler)

signal.alarm(self.timeout)

out = self._run(code)

result = out if out.strip() else "[Executed successfully, no stdout]"

except Exception as e:

tb = traceback.format_exc().splitlines()

loc = next((l.strip() for l in tb if '""' in l), "")

result = f"[Error]\n{loc}\n{type(e).__name__}: {e}".strip()

signal.alarm(0)

signal.signal(signal.SIGALRM, prev)

if len(result) > self.max_chars:

result = result[:self.max_chars] + "\n...[output truncated]..."

return result

We define a sandboxed code executor that gives the agent a persistent Python namespace for running generated code. We capture standard output and error streams so that every execution result can be passed back into the reasoning loop. We also enforce a timeout and truncate long outputs to keep the autonomous workflow controlled and readable.

Implementing the DeepAnalyze Agentic Loop

Copy CodeCopiedUse a different Browser

class DeepAnalyzeAgent:

def __init__(self, model, tok, temperature=0.5, top_p=0.95):

self.model, self.tok = model, tok

self.temperature, self.top_p = temperature, top_p

def _stream_generate(self, context, max_new_tokens):

inputs = self.tok(context, return_tensors="pt",

add_special_tokens=False).to(self.model.device)

streamer = TextIteratorStreamer(self.tok, skip_prompt=True,

skip_special_tokens=False)

kwargs = dict(

**inputs, max_new_tokens=max_new_tokens, do_sample=True,

temperature=self.temperature, top_p=self.top_p,

stop_strings=[""], tokenizer=self.tok, streamer=streamer,

pad_token_id=self.tok.pad_token_id, eos_token_id=self.tok.eos_token_id,

Thread(target=self.model.generate, kwargs=kwargs).start()

pieces = []

for chunk in streamer:

pieces.append(chunk); print(chunk, end="", flush=True)

return "".join(pieces)

@staticmethod

def _extract_code(delta):

if "" in delta and "" not in delta:

delta += ""

m = re.search(r"(.*?)", delta, re.DOTALL)

return None

code = m.group(1).strip()

fenced = re.search(r"```(?:python)?(.*?)```", code, re.DOTALL)

return (fenced.group(1) if fenced else code).strip()

def run(self, instruction, workspace, max_rounds=12,

max_new_tokens=3072, exec_timeout=120):

prompt = build_prompt(instruction, workspace)

prefix = self.tok.apply_chat_template(

[{"role": "user", "content": prompt}],

tokenize=False, add_generation_prompt=True)

sandbox = CodeSandbox(timeout=exec_timeout)

full, trace = prefix, []

cwd0 = os.getcwd(); os.chdir(workspace)

for r in range(max_rounds):

print(f"\n\n{'='*70}\n ROUND {r+1}\n{'='*70}")

delta = self._stream_generate(full, max_new_tokens)

full += delta

trace.append(("model", delta))

if "" in delta:

print("\n\n[Agent finished: produced]"); break

code = self._extract_code(delta)

if code is None:

print("\n\n[Agent stopped: no further action]"); break

output = sandbox.execute(code)

print(f"\n\n--- ---\n{output}\n--- ---")

full += f"\n\n{output}\n\n"

trace.append(("execute", output))

print(f"\n\n[Reached max_rounds={max_rounds}]")

os.chdir(cwd0)

answer = ""

if "" in full:

answer = full.split("")[-1]

answer = re.sub(r"", "", answer)

answer = re.sub(r"<[｜|][^>]*?[｜|]>", "", answer).strip()

return {"full": full, "trace": trace, "answer": answer}

We implement the DeepAnalyze agent loop, which streams model outputs, extracts the generated code, and executes it step by step. We allow the model to alternate between reasoning, coding, execution feedback, and final answering through special action tags. We maintain the full conversation trace so the agent can refine its analysis based on previous outputs and execution results.

Running the E-Commerce Analysis Workspace

Copy CodeCopiedUse a different Browser

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")

from IPython.display import display, Markdown, Image

if result["answer"]:

display(Markdown("## Final Report\n\n" + result["answer"]))

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"])

We create the prompt builder, prepare a sample e-commerce workspace, and generate transaction and customer files for analysis. We give the agent a complete analytical instruction that asks it to clean, join, explore, visualize, and summarize the dataset. We finally run the agent, display its final report, and render any saved PNG figures produced during the autonomous analysis.

In conclusion, we saw how DeepAnalyze-8B can be used as more than a simple text-generation model: we turn it into an iterative data-analysis agent that reasons over files, writes executable code, inspects outputs, and produces final insights. We keep the workflow lightweight while still preserving the core agentic pattern of understanding the task, generating code, executing it, and refining the analysis based on real results. It provides us with a foundation for building autonomous data-science notebooks in which the model not only describes an analysis but also actively performs it and returns both visual outputs and a concise final report.

Check out the FULL CODES with NOTEBOOK. Also, feel free to follow us on Twitter and don’t forget to join our 150k+ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.

Need to partner with us for promoting your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar etc.? Connect with us

The post How to Build a T4-Friendly Autonomous Data Science Agent with DeepAnalyze-8B, Sandboxed Code Execution, and Iterative Analysis appeared first on MarkTechPost.