How to Build a T4-Friendly Autonomous Data Science Agent with DeepAnalyze-8B, Sandboxed Code Execution, and Iterative Analysis
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, “<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 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=[“</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 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”# 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 *

