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
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
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
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, "<cell>", "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)
try:
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 '"<cell>"' in l), "")
result = f"[Error]n{loc}n{type(e).__name__}: {e}".strip()
finally:
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
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=["</Code>"], 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 "<Code>" in delta and "</Code>" not in delta:
delta += "</Code>"
m = re.search(r"<Code>(.*?)</Code>", delta, re.DOTALL)
if not m:
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)
try:
for r in range(max_rounds):
print(f"nn{'='*70}n ROUND {r+1}n{'='*70}")
delta = self._stream_generate(full, max_new_tokens)
full += delta
trace.append(("model", delta))
if "<Answer>" in delta:
print("nn[Agent finished: <Answer> produced]"); break
code = self._extract_code(delta)
if code is None:
print("nn[Agent stopped: no further action]"); break
output = sandbox.execute(code)
print(f"nn--- <Execute> ---n{output}n--- </Execute> ---")
full += f"n<Execute>n{output}n</Execute>n"
trace.append(("execute", output))
else:
print(f"nn[Reached max_rounds={max_rounds}]")
finally:
os.chdir(cwd0)
answer = ""
if "<Answer>" in full:
answer = full.split("<Answer>")[-1]
answer = re.sub(r"</?Answer>", "", 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
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"# Instructionn{instruction}nn# Datan" + "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"nnDone 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 Reportnn" + result["answer"]))
else:
print("No <Answer> 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.
Conclusion
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.