DeepAnalyze-8B を用いた T4 対応の自律型データサイエンスエージェント構築ガイド

このチュートリアルでは、DeepAnalyze-8B を中心とした自律型データサイエンスエージェントを構築し、実行します。まず、安定したランタイムの準備を行い、必要な機械学習依存ライブラリをインストールして、DeepAnalyze のトークナイザーとモデルを 4 ビットモードで読み込みます。これにより、限られた GPU メモリでも実用的なワークフローを実現できます。次に、モデルが Python コードを生成し、安全に実行し、結果を観察し、エージェントループ内で分析を継続できるサンドボックス化された実行環境を作成します。ワークフローの最後には、現実的なマルチファイル e コマースワークスペースをエージェントに与え、構造化されたアナリストグレードのレポートとしてデータのクリーニング、結合、分析、可視化、要約を行わせます。

DeepAnalyze-8B ランタイム依存ライブラリのインストール

コードをコピーしました。別のブラウザを使用してください

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)

DeepAnalyze-8B の必要な機械学習依存関係を備えた Colab ランタイムの準備から始めます。トランスフォーマー、アクセラレーション、量子化、トークナイザー、スプレッドシートライブラリをインストールしますが、ノートブック全体のワークフローには影響を与えません。また、NumPy を固定し、ランタイムを一度再起動して、次の実行のために環境をクリーンで安定した状態に保ちます。

4 ビットモードでの DeepAnalyze-8B の読み込み

コードをコピーしました

ブラウザを変更する

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

主要なライブラリをインポートし、DeepAnalyze-8B モデルを設定して、Colab 上で GPU が利用可能か確認します。モデルが T4 GPU でより快適に動作できるよう、トークナイザーを読み込み、4 ビット量子化（quantization）を準備します。その後、評価モードでモデルを読み込み、エージェントロジックに進む前に GPU メモリ使用量を確認します。

サンドボックス化されたコード実行器の構築

コードをコピーしました

ブラウザを変更してください

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)

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

私たちは、エージェントが生成したコードを実行するための永続的な Python ネームスペースを提供するサンドボックス化されたコード実行器を定義します。標準出力とエラーストリームをキャプチャすることで、すべての実行結果を推論ループにフィードバックできます。また、タイムアウトを設定し、長い出力は切り詰めることで、自律的なワークフローが制御可能で読みやすい状態を保ちます。

DeepAnalyze エージェントループの実装

コードをコピーしました（コピー済み）

別のブラウザを使用してください

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)

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

else:

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

finally:

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}

DeepAnalyze エージェントループを実装し、モデルの出力をストリーミングして生成されたコードを抽出し、ステップごとに実行します。特別なアクションタグを通じて、推論、コーディング、実行フィードバック、最終回答の間でモデルが交互に動作できるようにしています。過去の出力と実行結果に基づいて分析を洗練させるため、会話の完全な履歴を保持します。

E-Commerce Analysis Workspace の実行

コードをコピーしました（コピー済み）

別のブラウザを使用する

def _hsize(nbytes):

for u in ["B", "KB", "MB", "GB"]:

if nbytes < 1024:

return f"{nbytes:.2f} {u}"

nbytes /= 1024

return f"{nbytes:.2f} TB"

WORKSPACE = "/workspace/ecommerce_analysis"

os.makedirs(WORKSPACE, exist_ok=True)

existing_imgs = set(glob.glob(f"{WORKSPACE}/*.png"))

try:

for i in range(10): # max_rounds を上げてください

print(f"\n=== Round {i+1} ===")

result = run_agent(WORKSPACE, prompt_builder)

if "answer" in result and len(result["answer"]) > 50:

break

else:

raise RuntimeError("Max rounds reached without a valid 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"])

プロンプトビルダーを作成し、サンプルの E-Commerce ワークスペースを準備して、分析用の取引データと顧客ファイルを生成します。エージェントには、データのクリーニング、結合、探索、可視化、要約を行う完全な分析指示を与えます。最後にエージェントを実行し、最終レポートを表示するとともに、自律的な分析中に保存された PNG 画像をレンダリングします。

結論

結論として、DeepAnalyze-8B が単なるテキスト生成モデルを超えて活用できる様子を確認しました。私たちはこれを、ファイルに対して推論を行い、実行可能なコードを記述し、出力を検証し、最終的な洞察を生み出す反復型のデータ分析エージェントへと変換します。タスクの理解、コードの生成、その実行、そして実際の結果に基づいた分析の精緻化という、コアとなるエージェントのパターンを維持しつつ、ワークフローは軽量に保ちます。これにより、モデルが分析を記述するだけでなく、実際に実行を行い、視覚的な出力と簡潔な最終レポートの両方を返す、自律型のデータサイエンスノートブック構築のための基盤が提供されます。

完全なコードとノートブックをチェックしてください。また、Twitter でフォローしていただくこともお気軽にどうぞ。忘れずに 150k+ML SubReddit に参加し、ニュースレターも購読してください。待ってください！Telegram をご利用ですか？今なら Telegram でも私たちに参加できます。

GitHub リポジトリや Hugging Face ページ、製品リリース、ウェビナーなどのプロモーションでパートナーシップを結ぶ必要がある場合は、ご連絡ください。

本記事「DeepAnalyze-8B、サンドボックス化されたコード実行、反復分析を用いた T4 対応の自律型データサイエンスエージェントの構築方法」は、MarkTechPost で最初に公開されました。