The tutorial demonstrates how to construct an autonomous data science agent using the DeepAnalyze-8B model, running it within a sandboxed environment that permits safe code execution and iterative analysis.
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)
The process begins by configuring the Colab runtime with the necessary machine-learning libraries for DeepAnalyze-8B. We install the transformer, acceleration, quantization, tokenizer, and spreadsheet packages without disrupting the wider notebook structure. NumPy is pinned to version 2.0.2, and the runtime restarts once to ensure a clean environment for subsequent 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 core libraries and configure the DeepAnalyze-8B model, checking that a GPU is available within Colab. The tokenizer is loaded, and 4-bit quantization is applied to allow the model to run on a T4 GPU with more available memory. The model loads in evaluation mode, and we verify the GPU memory usage before proceeding 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
This section defines a sandboxed code executor that provides the agent with a persistent Python namespace for running generated scripts. Standard output and error streams are captured so that every execution result feeds back into the reasoning loop. We enforce a timeout and truncate long outputs to maintain control over the autonomous workflow and keep the logs 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(maxSource Read original →Related articles
