15 KiB
15 KiB
In [ ]:
# Install dependencies — skip torch, Colab sudah punya
!pip install -qU \
'transformers>=5.7.0' \
accelerate \
sentencepiece \
psutil \
'pillow<11' \
requests \
matplotlib \
tabulate \
av \
einops \
2>&1 | tail -3
print("Install selesai")In [ ]:
import os, sys, json, time, gc, warnings
from pathlib import Path
from datetime import datetime
from IPython.display import display, Video
import torch
import psutil
import numpy as np
import matplotlib.pyplot as plt
from tabulate import tabulate
from PIL import Image
import requests
from io import BytesIO
warnings.filterwarnings("ignore")
print(f"Python : {sys.version}")
print(f"PyTorch : {torch.__version__}")
print(f"CUDA avail : {torch.cuda.is_available()}")
if torch.cuda.is_available():
print(f"CUDA device : {torch.cuda.get_device_name(0)}")
print(f"CUDA VRAM : {torch.cuda.get_device_properties(0).total_memory / 1e9:.1f} GB")
print(f"CUDA cap : {torch.cuda.get_device_capability()}")
try:
import transformers
print(f"transformers : {transformers.__version__}")
except:
print("transformers : NOT INSTALLED")In [ ]:
MODEL_ID = "openbmb/MiniCPM-V-4.6"
CACHE_DIR = None # Set ke "drive/MyDrive/hf_cache" untuk persistent
DOWNSAMPLE_MODE = "16x" # "16x" = efisien, "4x" = detail lebih tinggi
print(f"Model ID: {MODEL_ID}")
print("Loading...")
t0 = time.perf_counter()
from transformers import AutoProcessor, AutoModelForImageTextToText
processor = AutoProcessor.from_pretrained(MODEL_ID, cache_dir=CACHE_DIR)
print(f"Processor loaded in {time.perf_counter()-t0:.1f}s")
load_start = time.perf_counter()
model = AutoModelForImageTextToText.from_pretrained(
MODEL_ID,
torch_dtype=torch.bfloat16,
device_map="auto",
cache_dir=CACHE_DIR,
)
load_time = time.perf_counter() - load_start
print(f"\nModel loaded in {load_time:.1f}s")
total_params = sum(p.numel() for p in model.parameters())
print(f"Parameters: {total_params/1e9:.2f}B")
print(f"Device: {model.device}, Dtype: {model.dtype}")In [ ]:
if torch.cuda.is_available():
vram_used = torch.cuda.memory_allocated() / 1e9
print(f"VRAM allocated: {vram_used:.2f} GB")
ram_used = psutil.Process(os.getpid()).memory_info().rss / 1e9
print(f"RAM used: {ram_used:.2f} GB")In [ ]:
def throughput(prompt, max_tokens=256, runs=3):
msgs = [{"role": "user", "content": prompt}]
inputs = processor.apply_chat_template(msgs, tokenize=True, return_dict=True,
return_tensors="pt", add_generation_prompt=True).to(model.device)
inp_len = inputs["input_ids"].shape[-1]
lats, toks = [], []
for _ in range(runs):
start = time.perf_counter()
with torch.no_grad():
out = model.generate(**inputs, max_new_tokens=max_tokens, do_sample=True, temperature=0.7)
elapsed = time.perf_counter() - start
gen = out[0][inp_len:]
lats.append(elapsed)
toks.append(len(gen))
tps = [t/l for t,l in zip(toks, lats)]
return {"prompt": prompt[:60]+"...", "tokens": int(np.mean(toks)),
"latency": float(np.mean(lats)), "tps": float(np.mean(tps))}
prompts = {
"Simple QA": "What is the capital of Indonesia?",
"Math": "If a train travels at 120 km/h and another at 80 km/h toward each other from 500 km apart, how long until they meet?",
"Code": "Write a Python function to find the longest palindromic substring.",
}
results = []
for name, p in prompts.items():
r = throughput(p, runs=2)
results.append(r)
print(f"{name}: {r['tokens']} tok, {r['latency']:.1f}s, {r['tps']:.1f} tok/s")In [ ]:
mmlu = [
{"q": "What is the time complexity of binary search?", "o": ["A. O(n)", "B. O(log n)", "C. O(n log n)", "D. O(1)"], "a": "B"},
{"q": "Which planet has the strongest surface gravity?", "o": ["A. Earth", "B. Mars", "C. Jupiter", "D. Saturn"], "a": "C"},
{"q": "In C++, which keyword prevents overriding?", "o": ["A. static", "B. const", "C. final", "D. override"], "a": "C"},
{"q": "Probability of drawing a red ball from 3 red + 5 blue?", "o": ["A. 3/5", "B. 3/8", "C. 5/8", "D. 1/2"], "a": "B"},
{"q": "What does mitochondria do?", "o": ["A. Protein", "B. Energy (ATP)", "C. Lipid", "D. DNA"], "a": "B"},
]
ok = 0
for q in mmlu:
prompt = f"{q['q']}\n\n" + "\n".join(q["o"]) + "\n\nAnswer with a single letter:"
inputs = processor.apply_chat_template([{"role":"user","content":prompt}],
tokenize=True, return_dict=True, return_tensors="pt", add_generation_prompt=True).to(model.device)
with torch.no_grad():
out = model.generate(**inputs, max_new_tokens=8, do_sample=False)
ans = processor.decode(out[0][inputs["input_ids"].shape[-1]:], skip_special_tokens=True).strip()
cor = q["a"] in ans.upper()[:1]
if cor: ok += 1
print(f" {'OK' if cor else 'NO'} Expected={q['a']} Got={ans[:20]} | {q['q'][:50]}")
print(f"\nAccuracy: {ok}/{len(mmlu)} = {ok/len(mmlu)*100:.0f}%")In [ ]:
for name, prompt in [
("Binary Search", "Write Python `binary_search(arr, target)` returning index or -1."),
("Fibonacci", "Write Python `fib(n)` for nth Fibonacci using DP."),
]:
print(f"\n{'='*40}\n{name}\n{'='*40}")
inputs = processor.apply_chat_template([{"role":"user","content":prompt}],
tokenize=True, return_dict=True, return_tensors="pt", add_generation_prompt=True).to(model.device)
with torch.no_grad():
out = model.generate(**inputs, max_new_tokens=512, temperature=0.2, do_sample=False)
print(processor.decode(out[0][inputs["input_ids"].shape[-1]:], skip_special_tokens=True)[:400])In [ ]:
print("Test image understanding (16x downsample / efisien)...")
try:
url = "https://huggingface.co/datasets/openbmb/DemoCase/resolve/main/refract.png"
img = Image.open(BytesIO(requests.get(url, timeout=30).content))
display(img.resize((250, 180)))
msgs = [{"role":"user","content":[
{"type":"image","image":img},
{"type":"text","text":"What causes this phenomenon?"}
]}]
inputs = processor.apply_chat_template(msgs, tokenize=True, add_generation_prompt=True,
return_dict=True, return_tensors="pt", downsample_mode=DOWNSAMPLE_MODE,
).to(model.device)
t0 = time.perf_counter()
with torch.no_grad():
out = model.generate(**inputs, downsample_mode=DOWNSAMPLE_MODE, max_new_tokens=128, do_sample=True)
resp = processor.decode(out[0][inputs["input_ids"].shape[-1]:], skip_special_tokens=True)
print(f"Time: {time.perf_counter()-t0:.1f}s")
print(f"Answer: {resp[:300]}")
except Exception as e:
print(f"ERROR: {e}")
print("\n---\nTest image understanding (4x downsample / detail tinggi)...")
try:
url2 = "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/p-blog/candy.JPG"
img2 = Image.open(BytesIO(requests.get(url2, timeout=30).content))
display(img2.resize((250, 180)))
msgs2 = [{"role":"user","content":[
{"type":"image","image":img2},
{"type":"text","text":"What animal is on the candy?"}
]}]
inputs2 = processor.apply_chat_template(msgs2, tokenize=True, add_generation_prompt=True,
return_dict=True, return_tensors="pt", downsample_mode="4x",
).to(model.device)
t0 = time.perf_counter()
with torch.no_grad():
out2 = model.generate(**inputs2, downsample_mode="4x", max_new_tokens=128, do_sample=True)
resp2 = processor.decode(out2[0][inputs2["input_ids"].shape[-1]:], skip_special_tokens=True)
print(f"Time: {time.perf_counter()-t0:.1f}s")
print(f"Answer: {resp2[:300]}")
except Exception as e:
print(f"ERROR: {e}")In [ ]:
def needle(pos):
needle_str = "The secret code is BLUE-42-GREEN."
filler = "The quick brown fox jumps over the lazy dog. Python is versatile. "
sents = [filler] * 50
if pos == "early": sents.insert(0, needle_str)
elif pos == "middle": sents.insert(len(sents)//2, needle_str)
else: sents.append(needle_str)
prompt = f"Read the text and answer.\n\nText: {' '.join(sents)}\n\nQ: What is the secret code? Answer with code only."
inputs = processor.apply_chat_template([{"role":"user","content":prompt}],
tokenize=True, return_dict=True, return_tensors="pt", add_generation_prompt=True).to(model.device)
t0 = time.perf_counter()
with torch.no_grad():
out = model.generate(**inputs, max_new_tokens=16, do_sample=False)
resp = processor.decode(out[0][inputs["input_ids"].shape[-1]:], skip_special_tokens=True).strip()
return {"pos": pos, "tokens": inputs["input_ids"].shape[-1], "resp": resp[:80],
"correct": "BLUE-42-GREEN" in resp, "time": time.perf_counter()-t0}
print("Needle-in-Haystack:")
for p in ["early", "middle", "late"]:
r = needle(p)
print(f"{p.upper()}: {r['tokens']} tok, correct={'OK' if r['correct'] else 'NO'}, {r['time']:.1f}s")In [ ]:
rows = [
["Model", MODEL_ID],
["Parameters", f"{total_params/1e9:.2f}B"],
["Device", str(model.device)],
["Dtype", str(model.dtype)],
]
if torch.cuda.is_available():
rows.append(["VRAM", f"{torch.cuda.memory_allocated()/1e9:.2f} GB"])
rows.append(["RAM", f"{ram_used:.2f} GB"])
rows.append(["Load Time", f"{load_time:.1f}s"])
if results:
rows.append(["Throughput", f"{np.mean([r['tps'] for r in results]):.1f} tok/s"])
rows.append(["MMLU", f"{ok}/{len(mmlu)} ({ok/len(mmlu)*100:.0f}%)"])
if 'nh' in dir():
rows.append(["Needle/Haystack", f"{sum(1 for r in nh if r['correct'])}/{len(nh)} correct"])
print("="*60)
print(" MINICPM-V 4.6 — BENCHMARK SUMMARY")
print("="*60)
print(tabulate(rows, headers=["Metric","Value"], tablefmt="grid"))
print("="*60)In [ ]:
del model, processor
gc.collect()
if torch.cuda.is_available():
torch.cuda.empty_cache()
print(f"VRAM: {torch.cuda.memory_allocated()/1e9:.2f} GB")
print("Done.")