Gemma 4：E2Bからエッジへ、31Bからクラウドへ — 4モデルの選び方と実装パターン

from transformers import AutoProcessor, AutoModelForCausalLM
import torch
 
model = AutoModelForCausalLM.from_pretrained(
    "google/gemma-4-31b-it",
    torch_dtype=torch.bfloat16,
    device_map="auto"
)
processor = AutoProcessor.from_pretrained("google/gemma-4-31b-it")
 
text = "Explain quantum computing in simple terms."
inputs = processor(text=text, return_tensors="pt")
outputs = model.generate(**inputs, max_new_tokens=256)
print(processor.decode(outputs[0], skip_special_tokens=True))

from PIL import Image
 
# ローカルファイルまたはURL から画像を読み込み
image = Image.open("path/to/image.jpg")
 
# または URL から直接：
# from urllib.request import urlopen
# image = Image.open(urlopen("https://example.com/image.jpg"))
 
text = "What is shown in this image?"
inputs = processor(
    text=text,
    images=image,
    return_tensors="pt"
)
outputs = model.generate(**inputs, max_new_tokens=512)
print(processor.decode(outputs[0], skip_special_tokens=True))

images = [Image.open(f"image_{i}.jpg") for i in range(3)]
 
text = "Compare these three images and describe their differences."
inputs = processor(
    text=text,
    images=images,
    return_tensors="pt"
)
outputs = model.generate(**inputs, max_new_tokens=1024)

from PIL import Image
 
def resize_image_smart(image, target_resolution=512):
    """アスペクト比を保持してリサイズ"""
    width, height = image.size
    aspect = width / height
    
    if aspect > 1:  # 横長
        new_width = target_resolution
        new_height = int(target_resolution / aspect)
    else:  # 縦長
        new_height = target_resolution
        new_width = int(target_resolution * aspect)
    
    return image.resize((new_width, new_height), Image.Resampling.LANCZOS)
 
image = Image.open("high_res_image.jpg")
image_resized = resize_image_smart(image, target_resolution=512)

# 必要なライブラリのインストール
pip install transformers torch pillow datasets
 
# GPU サポート（CUDA 12.1）
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121
 
# メモリ効率化のための追加パッケージ
pip install bitsandbytes  # 量子化
pip install accelerate     # マルチ GPU サポート

from transformers import AutoModelForCausalLM, AutoProcessor, BitsAndBytesConfig
import torch
 
# 8-bit 量子化の設定
bnb_config = BitsAndBytesConfig(
    load_in_8bit=True,
    llm_int8_threshold=200.0,
    llm_int8_has_fp16_weight=False
)
 
model = AutoModelForCausalLM.from_pretrained(
    "google/gemma-4-31b-it",
    quantization_config=bnb_config,
    device_map="auto"
)
processor = AutoProcessor.from_pretrained("google/gemma-4-31b-it")

model = AutoModelForCausalLM.from_pretrained(
    "google/gemma-4-31b-it",
    dtype=torch.bfloat16,
    device_map="auto"
)

def generate_text(prompt, max_length=256, temperature=0.7, top_p=0.95):
    """テキスト生成の基本関数"""
    inputs = processor(text=prompt, return_tensors="pt").to(model.device)
    
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=max_length,
            temperature=temperature,
            top_p=top_p,
            do_sample=True
        )
    
    return processor.decode(outputs[0], skip_special_tokens=True)
 
# 使用例
response = generate_text("技術的負債を減らすためのベストプラクティスは？")
print(response)

from transformers import TextIteratorStreamer
from threading import Thread
 
def stream_generate(prompt, max_length=512):
    """ストリーミング出力の実装"""
    inputs = processor(text=prompt, return_tensors="pt").to(model.device)
    
    streamer = TextIteratorStreamer(
        processor,
        skip_special_tokens=True,
        skip_prompt=True
    )
    
    generation_kwargs = dict(
        inputs,
        max_new_tokens=max_length,
        streamer=streamer,
        temperature=0.7,
        top_p=0.95
    )
    
    # バックグラウンドスレッドで生成開始
    thread = Thread(target=model.generate, kwargs=generation_kwargs)
    thread.start()
    
    # リアルタイムで出力
    for text in streamer:
        print(text, end="", flush=True)
    
    thread.join()
 
# 使用例
stream_generate("Pythonで非同期プログラミングを始めるには？")

# macOS / Linux / Windows
curl -fsSL https://ollama.ai/install.sh | sh

# E2B の実行
ollama run gemma-4-e2b
 
# E4B の実行
ollama run gemma-4-e4b
 
# 26B A4B の実行（GPU 推奨）
ollama run gemma-4-26b
 
# 31B の実行（高性能 GPU 推奨）
ollama run gemma-4-31b

import requests
import json
 
def ollama_generate(prompt, model="gemma-4-31b"):
    """Ollama 経由でテキスト生成"""
    url = "http://localhost:11434/api/generate"
    
    payload = {
        "model": model,
        "prompt": prompt,
        "stream": False
    }
    
    response = requests.post(url, json=payload)
    result = json.loads(response.text)
    
    return result["response"]
 
# 使用例
response = ollama_generate("マイクロサービスアーキテクチャの利点は？")
print(response)

import json
 
# 利用可能な関数を定義
tools = [
    {
        "name": "get_weather",
        "description": "指定した都市の天気情報を取得します",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "都市名（例：東京）"
                },
                "units": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "温度単位"
                }
            },
            "required": ["city"]
        }
    },
    {
        "name": "search_web",
        "description": "Web 検索を実行します",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "検索クエリ"
                }
            },
            "required": ["query"]
        }
    }
]

def agent_loop(initial_prompt, max_iterations=10):
    """関数呼び出しを含むエージェントループ"""
    messages = [{"role": "user", "content": initial_prompt}]
    
    for iteration in range(max_iterations):
        # モデルに推論させる
        response = model.generate(
            messages=messages,
            tools=tools,
            temperature=0.7
        )
        
        # 出力がテキストか関数呼び出しかを判定
        if response.get("type") == "function_call":
            func_name = response["function_name"]
            func_args = response["arguments"]
            
            # 関数を実行（実装はここで省略）
            result = execute_function(func_name, func_args)
            
            # 結果をメッセージに追加
            messages.append({
                "role": "assistant",
                "content": f"関数 {func_name} を実行しました"
            })
            messages.append({
                "role": "user",
                "content": f"結果：{result}"
            })
        else:
            # テキスト応答が返された（終了）
            return response["content"]
    
    return "最大反復回数に達しました"
 
# 使用例
response = agent_loop("東京の天気を調べて、明日の服装をアドバイスしてください")
print(response)

from peft import get_peft_model, LoraConfig, TaskType
from transformers import Trainer, TrainingArguments
 
# QLoRA 設定（メモリ効率最適化）
lora_config = LoraConfig(
    r=16,  # LoRA ランク
    lora_alpha=32,
    target_modules=["q_proj", "v_proj", "k_proj", "o_proj"],
    lora_dropout=0.05,
    bias="none",
    task_type=TaskType.CAUSAL_LM
)
 
# モデルに LoRA を適用
model = get_peft_model(model, lora_config)
 
# ファインチューニング設定
training_args = TrainingArguments(
    output_dir="./results",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    per_device_eval_batch_size=4,
    gradient_accumulation_steps=4,
    learning_rate=2e-4,
    warmup_steps=100,
    weight_decay=0.01,
    logging_steps=100,
    save_steps=500,
    eval_steps=500,
    fp16=True  # 混合精度学習
)
 
# トレーナーのセットアップ
trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
    eval_dataset=eval_dataset,
    data_collator=data_collator
)
 
# ファインチューニング実行
trainer.train()