# 従来のPyTorch実装（概念的なコード）classStandardAttention(nn.Module):defforward(self,x):# Query, Key, Valueの計算q=self.q_proj(x)k=self.k_proj(x)v=self.v_proj(x)# スケーリングscores=torch.matmul(q,k.transpose(-2,-1))/math.sqrt(self.head_dim)# Softmaxattn_weights=F.softmax(scores,dim=-1)# 重み付け和output=torch.matmul(attn_weights,v)returnoutput

# UnslothAI風の最適化実装（概念的なコード）importtritonimporttriton.languageastl@triton.jitdeffused_attention_kernel(q_ptr,k_ptr,v_ptr,out_ptr,seq_len,head_dim,BLOCK_SIZE:tl.constexpr):# ブロック単位での効率的な計算pid=tl.program_id(0)# 全ての計算を単一カーネル内で実行# メモリアクセスを最小化# 中間結果をレジスタに保持# 実際の実装はより複雑ですが、# 基本的な考え方は全ての操作を融合すること

classDynamicQuantization:def__init__(self,model):self.model=modelself.quantization_cache={}defforward(self,x,layer_name):# レイヤーの重要度を動的に評価importance_score=self.evaluate_layer_importance(x,layer_name)ifimportance_score>0.8:# 重要なレイヤーは高精度で計算weight=self.dequantize_to_fp16(self.model.get_weight(layer_name))elifimportance_score>0.5:# 中程度の重要度は8bit精度weight=self.dequantize_to_int8(self.model.get_weight(layer_name))else:# 低重要度は4bit精度のままweight=self.model.get_weight(layer_name)returnself.compute_with_weight(x,weight)defevaluate_layer_importance(self,x,layer_name):# 入力の分散やレイヤーの位置などから重要度を計算# 実際の実装はより洗練されていますinput_variance=torch.var(x)layer_position=self.get_layer_position(layer_name)# ヒューリスティックな重要度計算importance=input_variance*(1-layer_position/self.total_layers)returnimportance.item()

# UnslothAI風のLoRA最適化実装classOptimizedLoRALayer(nn.Module):def__init__(self,in_features,out_features,rank=16):super().__init__()self.rank=rank# LoRAの低ランク行列self.lora_A=nn.Parameter(torch.zeros(rank,in_features))self.lora_B=nn.Parameter(torch.zeros(out_features,rank))# 初期化戦略の最適化nn.init.kaiming_uniform_(self.lora_A,a=math.sqrt(5))nn.init.zeros_(self.lora_B)# スケーリング係数self.scaling=1.0/rankdefforward(self,x,base_weight):# 基本の線形変換とLoRA計算を融合# メモリ効率的な実装ifself.training:# トレーニング時は融合カーネルを使用returnself.fused_forward(x,base_weight)else:# 推論時は最適化された別の経路returnself.inference_forward(x,base_weight)@torch.jit.scriptdeffused_forward(self,x,base_weight):# JITコンパイルによる最適化base_output=F.linear(x,base_weight)lora_output=F.linear(F.linear(x,self.lora_A),self.lora_B)returnbase_output+self.scaling*lora_output

classHierarchicalMemoryPool:def__init__(self,total_memory):self.total_memory=total_memoryself.pools={'critical':MemoryPool(total_memory*0.3),# 重要な計算用'standard':MemoryPool(total_memory*0.5),# 通常の計算用'temporary':MemoryPool(total_memory*0.2)# 一時的な計算用}defallocate(self,size,priority='standard'):# 優先度に基づいてメモリを割り当てpool=self.pools[priority]ifpool.available()>=size:returnpool.allocate(size)else:# メモリが不足している場合は他のプールから借用returnself.borrow_from_other_pools(size,priority)defoptimize_allocation(self,computation_graph):# 計算グラフを分析して最適なメモリ配置を決定critical_ops=self.identify_critical_operations(computation_graph)foropincomputation_graph:ifopincritical_ops:op.memory_priority='critical'elifop.is_temporary():op.memory_priority='temporary'else:op.memory_priority='standard'

classMultiQueryAttention(nn.Module):def__init__(self,embed_dim,num_heads):super().__init__()self.embed_dim=embed_dimself.num_heads=num_headsself.head_dim=embed_dim//num_heads# Queryは各ヘッドごとself.q_proj=nn.Linear(embed_dim,embed_dim)# KeyとValueは共有（ヘッド数分作らない）self.k_proj=nn.Linear(embed_dim,self.head_dim)self.v_proj=nn.Linear(embed_dim,self.head_dim)self.out_proj=nn.Linear(embed_dim,embed_dim)defforward(self,x):batch_size,seq_len,_=x.shape# Query計算q=self.q_proj(x).reshape(batch_size,seq_len,self.num_heads,self.head_dim)q=q.transpose(1,2)# [batch, heads, seq, dim]# Key, Value計算（共有）k=self.k_proj(x).unsqueeze(1)# [batch, 1, seq, dim]v=self.v_proj(x).unsqueeze(1)# [batch, 1, seq, dim]# Attentionスコア計算scores=torch.matmul(q,k.transpose(-2,-1))/math.sqrt(self.head_dim)attn_weights=F.softmax(scores,dim=-1)# 出力計算attn_output=torch.matmul(attn_weights,v)attn_output=attn_output.transpose(1,2).reshape(batch_size,seq_len,self.embed_dim)returnself.out_proj(attn_output)

classSlidingWindowAttention(nn.Module):def__init__(self,embed_dim,num_heads,window_size=4096):super().__init__()self.window_size=window_sizeself.embed_dim=embed_dimself.num_heads=num_headsdefforward(self,x,positions):batch_size,seq_len,_=x.shape# 各位置について、window_size内のトークンのみに注意を向けるattention_mask=self.create_sliding_window_mask(seq_len,self.window_size)# 通常のAttention計算attn_output=self.compute_attention(x,attention_mask)# グローバルトークンの追加（重要な情報を保持）global_tokens=self.extract_global_tokens(x,positions)returnself.merge_local_and_global(attn_output,global_tokens)defcreate_sliding_window_mask(self,seq_len,window_size):# 各位置から見て、前後window_size/2の範囲のみ1にするmask=torch.zeros(seq_len,seq_len)foriinrange(seq_len):start=max(0,i-window_size//2)end=min(seq_len,i+window_size//2+1)mask[i,start:end]=1returnmask

classMultiModalEncoder(nn.Module):def__init__(self,config):super().__init__()self.config=config# テキストエンコーダーself.text_encoder=TextEncoder(config.text_config)# 画像エンコーダーself.image_encoder=VisionTransformer(config.vision_config)# モダリティ統合層self.modality_fusion=ModalityFusion(config.fusion_config)defforward(self,inputs):encoded_features=[]# テキスト入力の処理if'text'ininputs:text_features=self.text_encoder(inputs['text'])encoded_features.append(('text',text_features))# 画像入力の処理if'image'ininputs:image_features=self.image_encoder(inputs['image'])# 画像特徴を言語モデルの次元に投影image_features=self.project_image_features(image_features)encoded_features.append(('image',image_features))# モダリティの統合fused_features=self.modality_fusion(encoded_features)returnfused_featuresdefproject_image_features(self,image_features):# 画像特徴量をテキストと同じ次元空間に投影returnself.image_projection(image_features)

# GPUの確認nvidia-smi# CUDAバージョンの確認nvcc--version# Pythonバージョンの確認（3.9以上推奨）python--version

# 基本的な開発ツールsudoapt updatesudoaptinstall-y build-essential cmake git wget# Python開発環境sudoaptinstall-y python3-dev python3-pip python3-venv# CUDA関連（既にインストールされていない場合）# CUDA 12.1の例wget https://developer.download.nvidia.com/compute/cuda/repos/ubuntu2204/x86_64/cuda-keyring_1.0-1_all.debsudodpkg-i cuda-keyring_1.0-1_all.debsudoapt updatesudoaptinstall-y cuda-12-1

# プロジェクトディレクトリの作成mkdir ~/gemma3_unsloth_projectcd ~/gemma3_unsloth_project# 仮想環境の作成python3-m venv venv# 仮想環境の有効化sourcevenv/bin/activate# pipのアップグレードpipinstall--upgrade pip setuptools wheel

# PyTorchのインストール（CUDA 12.1対応版）pipinstalltorch torchvision torchaudio--index-url https://download.pytorch.org/whl/cu121# 基本的な依存関係pipinstallnumpy pandas matplotlib jupyter notebook ipywidgets# Hugging Face関連pipinstalltransformers datasets accelerate tokenizers sentencepiece# 量子化関連pipinstallbitsandbytes# UnslothAI本体pipinstallunsloth# 最新のGemma 3対応のためのtransformersアップデートpipinstall--no-deps git+https://github.com/huggingface/transformers@v4.49.0-Gemma-3

# こちらを使ってください!pipinstall"unsloth[colab-new] @ git+https://github.com/unslothai/unsloth.git"# test_environment.pyfromunslothimportFastModel# unslothを最初にインポートimporttorchimporttransformersimportsysdefcheck_environment():"""環境の動作確認を行う関数"""print("=== 環境診断開始 ===\n")# Python バージョンprint(f"Python version:{sys.version}")# PyTorchprint(f"\nPyTorch version:{torch.__version__}")print(f"CUDA available:{torch.cuda.is_available()}")iftorch.cuda.is_available():print(f"CUDA version:{torch.version.cuda}")print(f"GPU device:{torch.cuda.get_device_name(0)}")print(f"GPU memory:{torch.cuda.get_device_properties(0).total_memory/1024**3:.2f} GB")# Transformersprint(f"\nTransformers version:{transformers.__version__}")# UnslothAIの基本的な機能テストtry:print("\n=== UnslothAI機能テスト ===")# 小さなモデルでテスト（メモリ節約のため）model,tokenizer=FastModel.from_pretrained(model_name="unsloth/gemma-3-4b-it",# 1Bモデルでテストmax_seq_length=512,load_in_4bit=True,)print("✓ モデルの読み込み成功")# 簡単な推論テストtest_text="こんにちは, Gemma! 調子はどうです?"inputs=tokenizer(test_text,return_tensors="pt").to("cuda")withtorch.no_grad():outputs=model.generate(**inputs,max_length=50)response=tokenizer.decode(outputs[0],skip_special_tokens=True)print(f"✓ 推論テスト成功")print(f"入力:{test_text}")print(f"出力:{response[:100]}...")# 最初の100文字のみ表示exceptExceptionase:print(f"✗ エラーが発生しました:{str(e)}")returnFalseprint("\n=== 環境診断完了 ===")returnTrueif__name__=="__main__":success=check_environment()ifsuccess:print("\n✅ すべてのテストに合格しました！")else:print("\n❌ 一部のテストに失敗しました。エラーメッセージを確認してください。")

# エラー例: "CUDA out of memory"# 対処法: バッチサイズを小さくするか、より小さいモデルを使用# メモリ使用量を監視するユーティリティdefprint_gpu_memory():iftorch.cuda.is_available():print(f"GPU Memory:{torch.cuda.memory_allocated()/1024**3:.2f}GB /"f"{torch.cuda.get_device_properties(0).total_memory/1024**3:.2f}GB")# メモリをクリアするdefclear_gpu_memory():iftorch.cuda.is_available():torch.cuda.empty_cache()importgcgc.collect()

# transformersとunslothのバージョン競合が発生した場合pip uninstall-y transformers unslothpipinstall--no-deps unslothpipinstall--no-deps git+https://github.com/huggingface/transformers@v4.49.0-Gemma-3

importpandasaspdimportjsonfromdatasetsimportDataset,DatasetDictfromtypingimportList,Dict,AnyclassCustomDatasetPreparer:"""カスタムデータセットの準備クラス"""def__init__(self,data_path:str):self.data_path=data_pathself.conversations=[]defload_customer_support_data(self)->List[Dict[str,Any]]:"""カスタマーサポートデータの読み込み"""# CSVファイルからデータを読み込む例df=pd.read_csv(self.data_path)conversations=[]for_,rowindf.iterrows():# データを対話形式に変換conversation={"conversations":[{"role":"system","content":"あなたは親切で知識豊富なカスタマーサポートアシスタントです。"},{"role":"user","content":row['customer_question']},{"role":"assistant","content":row['support_answer']}]}# データの品質チェックifself.validate_conversation(conversation):conversations.append(conversation)returnconversationsdefvalidate_conversation(self,conversation:Dict[str,Any])->bool:"""対話データの妥当性チェック"""# 必須フィールドの確認if"conversations"notinconversation:returnFalse# 各ターンの検証forturninconversation["conversations"]:if"role"notinturnor"content"notinturn:returnFalse# 内容の長さチェックiflen(turn["content"].strip())<5:returnFalse# 不適切な内容のフィルタリングifself.contains_inappropriate_content(turn["content"]):returnFalsereturnTruedefcontains_inappropriate_content(self,text:str)->bool:"""不適切な内容のチェック（簡易版）"""# 実際の実装では、より洗練されたフィルタリングを行うinappropriate_keywords=["spam","inappropriate","xxx"]returnany(keywordintext.lower()forkeywordininappropriate_keywords)defaugment_data(self,conversations:List[Dict[str,Any]])->List[Dict[str,Any]]:"""データ拡張"""augmented_conversations=[]forconvinconversations:# オリジナルデータを追加augmented_conversations.append(conv)# バリエーションを生成# 敬語レベルの変更formal_conv=self.convert_to_formal(conv)ifformal_conv:augmented_conversations.append(formal_conv)# 言い換えバージョンparaphrased_conv=self.paraphrase_conversation(conv)ifparaphrased_conv:augmented_conversations.append(paraphrased_conv)returnaugmented_conversationsdefconvert_to_formal(self,conversation:Dict[str,Any])->Dict[str,Any]:"""カジュアルな表現をフォーマルに変換"""# 実装例（簡易版）formal_conv=json.loads(json.dumps(conversation))# Deep copyforturninformal_conv["conversations"]:ifturn["role"]=="assistant":# 簡易的な敬語変換turn["content"]=turn["content"].replace("です。","でございます。")turn["content"]=turn["content"].replace("ます。","ます。")returnformal_convdefparaphrase_conversation(self,conversation:Dict[str,Any])->Dict[str,Any]:"""対話の言い換えバージョンを生成"""# 実際の実装では、別のモデルを使った言い換えを行う# ここでは簡易的な実装returnNone# 省略defcreate_dataset(self)->DatasetDict:"""最終的なデータセットの作成"""# データの読み込みconversations=self.load_customer_support_data()# データ拡張augmented_conversations=self.augment_data(conversations)# 訓練/検証データの分割split_idx=int(len(augmented_conversations)*0.9)train_data=augmented_conversations[:split_idx]val_data=augmented_conversations[split_idx:]# Hugging Face Dataset形式に変換train_dataset=Dataset.from_list(train_data)val_dataset=Dataset.from_list(val_data)returnDatasetDict({'train':train_dataset,'validation':val_dataset})# 使用例preparer=CustomDatasetPreparer("customer_support_data.csv")dataset=preparer.create_dataset()print(f"訓練データ数:{len(dataset['train'])}")print(f"検証データ数:{len(dataset['validation'])}")

fromunslothimportFastModelfromunsloth.chat_templatesimportget_chat_template,train_on_responses_onlyfromtransformersimportTrainingArgumentsfromtrlimportSFTTrainer,SFTConfigimporttorchfromdatetimeimportdatetimeimportosclassGemmaFineTuner:"""Gemma 3ファインチューニング用クラス"""def__init__(self,model_size="4b",experiment_name=None):self.model_size=model_sizeself.experiment_name=experiment_nameorf"gemma3_{model_size}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"self.output_dir=f"./outputs/{self.experiment_name}"# 出力ディレクトリの作成os.makedirs(self.output_dir,exist_ok=True)defload_model_and_tokenizer(self):"""モデルとトークナイザーの読み込み"""print(f"Loading Gemma 3{self.model_size} model...")# モデルサイズに応じた設定model_configs={"1b":{"model_name":"unsloth/gemma-3-1b-it","max_seq_length":8192,"r":8,"lora_alpha":16},"4b":{"model_name":"unsloth/gemma-3-4b-it","max_seq_length":8192,"r":16,"lora_alpha":32},"12b":{"model_name":"unsloth/gemma-3-12b-it","max_seq_length":4096,# メモリ制約のため短縮"r":32,"lora_alpha":64}}config=model_configs[self.model_size]# モデルの読み込みself.model,self.tokenizer=FastModel.from_pretrained(model_name=config["model_name"],max_seq_length=config["max_seq_length"],load_in_4bit=True,dtype=torch.float16,)# LoRAの設定self.model=FastModel.get_peft_model(self.model,finetune_vision_layers=False,finetune_language_layers=True,finetune_attention_modules=True,finetune_mlp_modules=True,r=config["r"],lora_alpha=config["lora_alpha"],lora_dropout=0.1,# 過学習防止bias="none",use_gradient_checkpointing=True,# メモリ効率化random_state=42,max_seq_length=config["max_seq_length"],)# チャットテンプレートの設定self.tokenizer=get_chat_template(self.tokenizer,chat_template="gemma-3",)print(f"Model loaded successfully!")self.print_model_info()defprint_model_info(self):"""モデル情報の表示"""total_params=sum(p.numel()forpinself.model.parameters())trainable_params=sum(p.numel()forpinself.model.parameters()ifp.requires_grad)print(f"\n=== Model Information ===")print(f"Total parameters:{total_params:,}")print(f"Trainable parameters:{trainable_params:,}")print(f"Trainable ratio:{trainable_params/total_params*100:.2f}%")print(f"========================\n")defprepare_training_arguments(self,num_epochs=3,batch_size=2):"""訓練引数の準備"""# 動的な学習率とバッチサイズの調整base_learning_rate=2e-4effective_batch_size=batch_size*4# gradient_accumulation_stepsを考慮# モデルサイズに応じた調整ifself.model_size=="12b":base_learning_rate*=0.5# 大きいモデルは学習率を下げるtraining_args=SFTConfig(output_dir=self.output_dir,# 基本的な訓練パラメータnum_train_epochs=num_epochs,per_device_train_batch_size=batch_size,per_device_eval_batch_size=batch_size,gradient_accumulation_steps=4,gradient_checkpointing=True,# 学習率スケジューリングlearning_rate=base_learning_rate,lr_scheduler_type="cosine",warmup_ratio=0.1,# 最適化設定optim="adamw_8bit",# 8bit AdamWで省メモリ化weight_decay=0.01,max_grad_norm=0.3,# ロギングと保存logging_steps=10,save_steps=100,eval_steps=100,save_total_limit=3,load_best_model_at_end=True,# その他の設定report_to="tensorboard",logging_dir=f"{self.output_dir}/logs",# Mixed Precision Trainingfp16=torch.cuda.is_available(),bf16=False,# A100以外では無効化# データセット関連dataset_text_field="text",max_seq_length=self.model.config.max_position_embeddings,dataset_num_proc=4,# データ処理の並列化# Early Stoppingmetric_for_best_model="eval_loss",greater_is_better=False,)returntraining_argsdefcreate_trainer(self,train_dataset,eval_dataset,training_args):"""トレーナーの作成"""trainer=SFTTrainer(model=self.model,tokenizer=self.tokenizer,train_dataset=train_dataset,eval_dataset=eval_dataset,args=training_args,# コールバック関数callbacks=[# カスタムコールバックを追加可能],)# レスポンスのみの学習設定trainer=train_on_responses_only(trainer,instruction_part="<start_of_turn>user\n",response_part="<start_of_turn>model\n",)returntrainerdeftrain(self,train_dataset,eval_dataset,num_epochs=3,batch_size=2):"""ファインチューニングの実行"""# モデルとトークナイザーの読み込みself.load_model_and_tokenizer()# データセットの前処理print("Preprocessing datasets...")train_dataset=self.preprocess_dataset(train_dataset)eval_dataset=self.preprocess_dataset(eval_dataset)# 訓練引数の準備training_args=self.prepare_training_arguments(num_epochs,batch_size)# トレーナーの作成trainer=self.create_trainer(train_dataset,eval_dataset,training_args)# GPUメモリの状態を表示iftorch.cuda.is_available():print(f"\nGPU Memory before training:{torch.cuda.memory_allocated()/1024**3:.2f}GB")# 訓練の実行print("\nStarting fine-tuning...")train_result=trainer.train()# 結果の保存self.save_results(trainer,train_result)returntrainer,train_resultdefpreprocess_dataset(self,dataset):"""データセットの前処理"""defformatting_func(examples):texts=[]forconversationsinexamples["conversations"]:text=self.tokenizer.apply_chat_template(conversations,tokenize=False,add_generation_prompt=False)texts.append(text)return{"text":texts}# バッチ処理で高速化dataset=dataset.map(formatting_func,batched=True,num_proc=4,remove_columns=dataset.column_names)returndatasetdefsave_results(self,trainer,train_result):"""訓練結果の保存"""# モデルの保存print("\nSaving model...")trainer.save_model(f"{self.output_dir}/final_model")# LoRAアダプターのみの保存self.model.save_pretrained(f"{self.output_dir}/lora_adapters")self.tokenizer.save_pretrained(f"{self.output_dir}/lora_adapters")# マージされたモデルの保存（オプション）ifself.model_sizein["1b","4b"]:# メモリに余裕がある場合のみprint("Saving merged model...")self.model.save_pretrained_merged(f"{self.output_dir}/merged_model",self.tokenizer,save_method="merged_16bit")# 訓練統計の保存importjsonwithopen(f"{self.output_dir}/training_stats.json","w")asf:json.dump({"training_loss":train_result.training_loss,"metrics":train_result.metrics,"global_step":train_result.global_step,"experiment_name":self.experiment_name},f,indent=2)print(f"\nAll results saved to{self.output_dir}")# 使用例if__name__=="__main__":# データセットの準備（前述のCustomDatasetPreparerを使用）preparer=CustomDatasetPreparer("data/customer_support.csv")dataset=preparer.create_dataset()# ファインチューニングの実行finetuner=GemmaFineTuner(model_size="4b",experiment_name="customer_support_v1")trainer,results=finetuner.train(train_dataset=dataset["train"],eval_dataset=dataset["validation"],num_epochs=3,batch_size=2)

importtorchfromtransformersimportAutoModelForCausalLM,AutoTokenizer,TextStreamerfromtypingimportOptional,List,Dict,AnyimporttimeclassGemmaInferenceEngine:"""Gemma 3推論エンジン"""def__init__(self,model_path:str,device:str="cuda"):self.model_path=model_pathself.device=deviceself.model=Noneself.tokenizer=Nonedefload_model(self,load_in_4bit:bool=True):"""モデルの読み込み"""print(f"Loading model from{self.model_path}...")# 量子化設定ifload_in_4bit:fromtransformersimportBitsAndBytesConfigquantization_config=BitsAndBytesConfig(load_in_4bit=True,bnb_4bit_compute_dtype=torch.float16,bnb_4bit_use_double_quant=True,bnb_4bit_quant_type="nf4")else:quantization_config=None# モデルの読み込みself.model=AutoModelForCausalLM.from_pretrained(self.model_path,quantization_config=quantization_config,device_map="auto",torch_dtype=torch.float16,low_cpu_mem_usage=True)# トークナイザーの読み込みself.tokenizer=AutoTokenizer.from_pretrained(self.model_path)# パディングトークンの設定ifself.tokenizer.pad_tokenisNone:self.tokenizer.pad_token=self.tokenizer.eos_tokenprint("Model loaded successfully!")defgenerate_response(self,prompt:str,max_new_tokens:int=512,temperature:float=0.7,top_p:float=0.95,top_k:int=50,do_sample:bool=True,repetition_penalty:float=1.1,stream:bool=False)->str:"""レスポンスの生成"""# 入力の準備messages=[{"role":"user","content":prompt}]# チャットテンプレートの適用formatted_prompt=self.tokenizer.apply_chat_template(messages,tokenize=False,add_generation_prompt=True)# トークナイズinputs=self.tokenizer(formatted_prompt,return_tensors="pt",truncation=True,max_length=self.model.config.max_position_embeddings).to(self.device)# ストリーミング設定ifstream:streamer=TextStreamer(self.tokenizer,skip_prompt=True,skip_special_tokens=True)else:streamer=None# 生成パラメータgeneration_config={"max_new_tokens":max_new_tokens,"temperature":temperature,"top_p":top_p,"top_k":top_k,"do_sample":do_sample,"repetition_penalty":repetition_penalty,"pad_token_id":self.tokenizer.pad_token_id,"eos_token_id":self.tokenizer.eos_token_id,"streamer":streamer}# 推論の実行start_time=time.time()withtorch.no_grad():outputs=self.model.generate(**inputs,**generation_config)generation_time=time.time()-start_time# デコードresponse=self.tokenizer.decode(outputs[0][inputs["input_ids"].shape[1]:],skip_special_tokens=True)# 統計情報num_tokens=outputs.shape[1]-inputs["input_ids"].shape[1]tokens_per_second=num_tokens/generation_timeifnotstream:print(f"\nGeneration stats:")print(f"- Tokens generated:{num_tokens}")print(f"- Time:{generation_time:.2f}s")print(f"- Speed:{tokens_per_second:.2f} tokens/s")returnresponsedefbatch_generate(self,prompts:List[str],batch_size:int=4,**generation_kwargs)->List[str]:"""バッチ推論"""responses=[]foriinrange(0,len(prompts),batch_size):batch_prompts=prompts[i:i+batch_size]# バッチ用のメッセージ準備batch_messages=[[{"role":"user","content":prompt}]forpromptinbatch_prompts]# チャットテンプレートの適用formatted_prompts=[self.tokenizer.apply_chat_template(messages,tokenize=False,add_generation_prompt=True)formessagesinbatch_messages]# パディングを考慮したトークナイズinputs=self.tokenizer(formatted_prompts,return_tensors="pt",padding=True,truncation=True,max_length=self.model.config.max_position_embeddings).to(self.device)# バッチ生成withtorch.no_grad():outputs=self.model.generate(**inputs,**generation_kwargs)# デコードforj,outputinenumerate(outputs):response=self.tokenizer.decode(output[inputs["input_ids"][j].shape[0]:],skip_special_tokens=True)responses.append(response)returnresponsesdefcreate_interactive_session(self):"""対話的セッション"""print("\n=== Gemma 3 Interactive Session ===")print("Type'exit' to quit,'clear' to reset conversation")print("===================================\n")conversation_history=[]whileTrue:user_input=input("\nYou:").strip()ifuser_input.lower()=='exit':breakelifuser_input.lower()=='clear':conversation_history=[]print("Conversation cleared.")continue# 会話履歴に追加conversation_history.append({"role":"user","content":user_input})# フルコンテキストでプロンプト作成full_prompt=self.tokenizer.apply_chat_template(conversation_history,tokenize=False,add_generation_prompt=True)# レスポンス生成print("\nGemma:",end="",flush=True)response=self.generate_response(user_input,# ここは簡略化のため最新の入力のみstream=True,temperature=0.7,max_new_tokens=256)# 会話履歴に追加conversation_history.append({"role":"assistant","content":response})# メモリ管理（履歴が長くなりすぎた場合）iflen(conversation_history)>20:conversation_history=conversation_history[-10:]# 使用例if__name__=="__main__":# 推論エンジンの初期化engine=GemmaInferenceEngine("./outputs/customer_support_v1/merged_model")engine.load_model(load_in_4bit=True)# 単一の推論response=engine.generate_response("製品の返品方法を教えてください。",temperature=0.7,max_new_tokens=256)print(f"Response:{response}")# バッチ推論test_prompts=["注文のキャンセル方法は？","配送状況を確認したい","支払い方法を変更できますか？"]batch_responses=engine.batch_generate(test_prompts,batch_size=2,temperature=0.7,max_new_tokens=200)forprompt,responseinzip(test_prompts,batch_responses):print(f"\nQ:{prompt}")print(f"A:{response}")# 対話的セッション# engine.create_interactive_session()

importtorchimporttimeimportpsutilimportGPUtilfromtypingimportDict,List,Anyimportpandasaspdimportmatplotlib.pyplotaspltimportseabornassnsfromdataclassesimportdataclassimportjson@dataclassclassBenchmarkResult:"""ベンチマーク結果を格納するデータクラス"""model_name:stroptimization:strbatch_size:intsequence_length:intthroughput:float# tokens/secondlatency:float# secondsmemory_usage:float# GBaccuracy_score:floatclassUnslothBenchmark:"""UnslothAIベンチマーククラス"""def__init__(self):self.results=[]self.device=torch.device("cuda"iftorch.cuda.is_available()else"cpu")defmeasure_gpu_memory(self)->float:"""GPU メモリ使用量を測定"""iftorch.cuda.is_available():returntorch.cuda.memory_allocated()/1024**3# GBreturn0.0defmeasure_system_resources(self)->Dict[str,float]:"""システムリソースの測定"""resources={"cpu_percent":psutil.cpu_percent(interval=1),"ram_usage_gb":psutil.virtual_memory().used/1024**3,"ram_percent":psutil.virtual_memory().percent}iftorch.cuda.is_available():gpus=GPUtil.getGPUs()ifgpus:gpu=gpus[0]resources.update({"gpu_memory_used_gb":gpu.memoryUsed/1024,"gpu_memory_percent":gpu.memoryUtil*100,"gpu_utilization":gpu.load*100,"gpu_temperature":gpu.temperature})returnresourcesdefbenchmark_training(self,model,train_dataloader,num_steps:int=100,optimization_type:str="unsloth")->BenchmarkResult:"""訓練性能のベンチマーク"""print(f"\nBenchmarking{optimization_type} training...")# 初期リソース測定torch.cuda.empty_cache()initial_memory=self.measure_gpu_memory()# ウォームアップfori,batchinenumerate(train_dataloader):ifi>=5:breakoutputs=model(**batch)loss=outputs.lossloss.backward()torch.cuda.synchronize()# 実際のベンチマークstart_time=time.time()total_tokens=0forstep,batchinenumerate(train_dataloader):ifstep>=num_steps:breakstep_start=time.time()# Forward passoutputs=model(**batch)loss=outputs.loss# Backward passloss.backward()# トークン数のカウントtotal_tokens+=batch["input_ids"].numel()ifstep%10==0:step_time=time.time()-step_startcurrent_memory=self.measure_gpu_memory()print(f"Step{step}:{step_time:.3f}s, Memory:{current_memory:.2f}GB")torch.cuda.synchronize()total_time=time.time()-start_time# メトリクスの計算throughput=total_tokens/total_timeavg_latency=total_time/num_stepspeak_memory=self.measure_gpu_memory()memory_usage=peak_memory-initial_memoryresult=BenchmarkResult(model_name=model.config.model_type,optimization=optimization_type,batch_size=train_dataloader.batch_size,sequence_length=batch["input_ids"].shape[1],throughput=throughput,latency=avg_latency,memory_usage=memory_usage,accuracy_score=0.0# 別途評価)self.results.append(result)returnresultdefbenchmark_inference(self,model,tokenizer,test_prompts:List[str],optimization_type:str="unsloth")->BenchmarkResult:"""推論性能のベンチマーク"""print(f"\nBenchmarking{optimization_type} inference...")# ウォームアップfor_inrange(3):inputs=tokenizer(test_prompts[0],return_tensors="pt").to(self.device)withtorch.no_grad():_=model.generate(**inputs,max_new_tokens=50)torch.cuda.synchronize()# 実際のベンチマークtotal_time=0total_tokens_generated=0initial_memory=self.measure_gpu_memory()forpromptintest_prompts:inputs=tokenizer(prompt,return_tensors="pt").to(self.device)start_time=time.time()withtorch.no_grad():outputs=model.generate(**inputs,max_new_tokens=128,do_sample=True,temperature=0.7)torch.cuda.synchronize()generation_time=time.time()-start_timetotal_time+=generation_time# 生成されたトークン数num_generated=outputs.shape[1]-inputs["input_ids"].shape[1]total_tokens_generated+=num_generated# メトリクスの計算throughput=total_tokens_generated/total_timeavg_latency=total_time/len(test_prompts)peak_memory=self.measure_gpu_memory()memory_usage=peak_memory-initial_memoryresult=BenchmarkResult(model_name=model.config.model_type,optimization=optimization_type,batch_size=1,sequence_length=128,throughput=throughput,latency=avg_latency,memory_usage=memory_usage,accuracy_score=0.0)self.results.append(result)returnresultdefcompare_optimizations(self):"""異なる最適化手法の比較"""# 結果をDataFrameに変換df=pd.DataFrame([vars(r)forrinself.results])# 可視化fig,axes=plt.subplots(2,2,figsize=(15,12))# スループット比較sns.barplot(data=df,x="optimization",y="throughput",ax=axes[0,0])axes[0,0].set_title("Throughput Comparison (tokens/second)")axes[0,0].set_ylabel("Tokens per Second")# レイテンシ比較sns.barplot(data=df,x="optimization",y="latency",ax=axes[0,1])axes[0,1].set_title("Latency Comparison")axes[0,1].set_ylabel("Latency (seconds)")# メモリ使用量比較sns.barplot(data=df,x="optimization",y="memory_usage",ax=axes[1,0])axes[1,0].set_title("Memory Usage Comparison")axes[1,0].set_ylabel("Memory Usage (GB)")# 総合スコア（正規化して計算）df['normalized_throughput']=df['throughput']/df['throughput'].max()df['normalized_latency']=1-(df['latency']/df['latency'].max())df['normalized_memory']=1-(df['memory_usage']/df['memory_usage'].max())df['overall_score']=(df['normalized_throughput']+df['normalized_latency']+df['normalized_memory'])/3sns.barplot(data=df,x="optimization",y="overall_score",ax=axes[1,1])axes[1,1].set_title("Overall Performance Score")axes[1,1].set_ylabel("Score (0-1)")plt.tight_layout()plt.savefig("benchmark_results.png",dpi=300)plt.show()# 詳細な統計情報を出力print("\n=== Benchmark Summary ===")foroptindf['optimization'].unique():opt_data=df[df['optimization']==opt]print(f"\n{opt}:")print(f"  Average Throughput:{opt_data['throughput'].mean():.2f} tokens/s")print(f"  Average Latency:{opt_data['latency'].mean():.3f}s")print(f"  Average Memory:{opt_data['memory_usage'].mean():.2f}GB")print(f"  Overall Score:{opt_data['overall_score'].mean():.3f}")defsave_results(self,filename:str="benchmark_results.json"):"""結果の保存"""results_dict={"timestamp":time.strftime("%Y-%m-%d %H:%M:%S"),"system_info":{"gpu":torch.cuda.get_device_name(0)iftorch.cuda.is_available()else"CPU","cuda_version":torch.version.cuda,"pytorch_version":torch.__version__,},"results":[vars(r)forrinself.results]}withopen(filename,"w")asf:json.dump(results_dict,f,indent=2)print(f"\nResults saved to{filename}")# 実際のベンチマーク実行例defrun_comprehensive_benchmark():"""包括的なベンチマークの実行"""benchmark=UnslothBenchmark()# テストプロンプトtest_prompts=["機械学習の基本的な概念を説明してください。","Pythonでクイックソートを実装する方法を教えてください。","地球温暖化の原因と対策について述べてください。","健康的な生活習慣について助言をください。","最新のAI技術のトレンドを教えてください。"]# 異なる最適化設定でのベンチマークoptimizations=[("standard",False,False),# 標準実装("flash_attention",True,False),# Flash Attention のみ("unsloth",True,True),# UnslothAI フル最適化]foropt_name,use_flash,use_unslothinoptimizations:print(f"\n{'='*50}")print(f"Testing{opt_name} optimization")print(f"{'='*50}")# モデルの読み込み（各最適化設定で）ifuse_unsloth:fromunslothimportFastModelmodel,tokenizer=FastModel.from_pretrained("unsloth/gemma-3-4b-it",max_seq_length=2048,load_in_4bit=True,)else:# 標準的な読み込みfromtransformersimportAutoModelForCausalLM,AutoTokenizermodel=AutoModelForCausalLM.from_pretrained("google/gemma-3-4b-it",torch_dtype=torch.float16,device_map="auto",use_flash_attention_2=use_flash)tokenizer=AutoTokenizer.from_pretrained("google/gemma-3-4b-it")# 推論ベンチマークinference_result=benchmark.benchmark_inference(model,tokenizer,test_prompts,optimization_type=opt_name)print(f"\n{opt_name} Results:")print(f"  Throughput:{inference_result.throughput:.2f} tokens/s")print(f"  Latency:{inference_result.latency:.3f}s")print(f"  Memory:{inference_result.memory_usage:.2f}GB")# メモリクリーンアップdelmodelif'tokenizer'inlocals():deltokenizertorch.cuda.empty_cache()# 結果の比較と保存benchmark.compare_optimizations()benchmark.save_results()if__name__=="__main__":run_comprehensive_benchmark()