はじめに

Google Colab上でGradioを使って、Qwen3-4Bモデルの思考プロセスが見えるチャットボットを構築する方法を紹介します。Gradio 5.xに対応し、ブラウザ上でAIの「考え方」をリアルタイムで可視化できる実装になっています。

主な特徴・完成イメージ

Google ColabのGPU環境で動作します。ブラウザ上でGradioの直感的なインターフェースを使って操作でき、タグで推論過程も確認できます。最新のGradio 5.xに対応し、生成失敗時の自動リトライ機能も搭載しています。以下の動画がアプリの動作デモです。

環境セットアップ

必要あればGoogle Colabの新しいセルで以下を実行してください：
筆者がためしたときはgradioをインストールしました。

!pip install transformers gradio torch

注意: GPU環境を有効にすることを強く推奨します（ランタイム > ランタイムのタイプを変更 > T4 GPU）。

実装

Hugging Faceのサンプルコードをもとに実装

import gradio as gr
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
import time

class QwenChatbot:
    def __init__(self, model_name="Qwen/Qwen3-4B"):
        print("モデルを読み込み中...")
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        
        # パディングトークンの設定
        if self.tokenizer.pad_token is None:
            self.tokenizer.pad_token = self.tokenizer.eos_token
        
        self.model = AutoModelForCausalLM.from_pretrained(
            model_name,
            torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
            device_map="auto" if torch.cuda.is_available() else None
        )
        self.history = []
        
        # モデル情報の表示
        total_params = sum(p.numel() for p in self.model.parameters())
        print(f"モデルの読み込みが完了しました!")
        print(f"パラメータ数: {total_params:,}")
        print(f"使用デバイス: {next(self.model.parameters()).device}")

    def generate_response(self, user_input, max_attempts=3):
        # 履歴が長くなりすぎた場合の管理
        if len(self.history) > 20:
            self.history = self.history[-20:]
        
        messages = self.history + [{"role": "user", "content": user_input}]
        
        for attempt in range(max_attempts):
            print(f"🔄 生成試行 {attempt + 1}/{max_attempts}")
            
            text = self.tokenizer.apply_chat_template(
                messages,
                tokenize=False,
                add_generation_prompt=True
            )

            inputs = self.tokenizer(text, return_tensors="pt", truncation=True, max_length=2048)
            if torch.cuda.is_available():
                inputs = inputs.to(self.model.device)
            
            input_length = inputs.input_ids.shape[1]
            print(f"入力トークン数: {input_length}")
            
            with torch.no_grad():
                response_ids = self.model.generate(
                    **inputs, 
                    max_new_tokens=min(512, 2048 - input_length),
                    do_sample=True,
                    temperature=0.7,
                    top_p=0.9,
                    pad_token_id=self.tokenizer.eos_token_id,
                    early_stopping=False,
                    num_beams=1
                )[0][len(inputs.input_ids[0]):].tolist()

            response = self.tokenizer.decode(response_ids, skip_special_tokens=True)
            
            # 思考部分と最終回答を分離
            thinking, final_answer = self.parse_thinking_response(response)
            
            # 最終回答が十分にある場合は成功
            if final_answer and len(final_answer.strip()) > 10:
                print(f"✅ 適切な回答を取得: {len(final_answer)}文字")
                break
            elif thinking and len(thinking.strip()) > 0:
                print(f"⚠️ 思考のみで最終回答が不足: {len(final_answer)}文字")
                continue
            else:
                print(f"⚠️ 思考も回答も不十分、再試行")
                continue

        # Update history
        self.history.append({"role": "user", "content": user_input})
        self.history.append({"role": "assistant", "content": response})

        return response
    
    def parse_thinking_response(self, response):
        """思考部分と最終回答を分離し、不完全な応答を検出"""
        if "<think>" in response and "</think>" in response:
            think_start = response.find("<think>") + 7
            think_end = response.find("</think>")
            thinking = response[think_start:think_end].strip()
            final_answer = response[think_end + 8:].strip()
            
            if len(final_answer) < 5:
                thinking += f"\n\n⚠️ [システム] 最終回答が短すぎます: '{final_answer}'"
                final_answer = "申し訳ありません。完全な回答を生成できませんでした。"
            
            return thinking, final_answer
        elif "<think>" in response and "</think>" not in response:
            think_start = response.find("<think>") + 7
            incomplete_thinking = response[think_start:].strip()
            thinking = incomplete_thinking + "\n\n⚠️ [システム] 思考が途中で止まりました"
            final_answer = "思考処理が完了しませんでした。もう一度お試しください。"
            return thinking, final_answer
        else:
            if len(response.strip()) < 5:
                return "", "申し訳ありません。適切な回答を生成できませんでした。"
            return "", response

    def clear_history(self):
        """履歴をクリア"""
        self.history = []

# グローバルなチャットボットインスタンス
chatbot = QwenChatbot()

def respond_to_message(message, chat_history, thinking_display):
    """Gradio 5.x用のチャット応答関数（思考表示付き）"""
    if not message.strip():
        return chat_history, "", thinking_display, ""
    
    try:
        print(f"🔄 処理開始: '{message[:50]}...'")
        
        if len(message) > 200:
            error_msg = "⚠️ メッセージが長すぎます。200文字以内でお試しください。"
            chat_history.append({"role": "user", "content": message})
            chat_history.append({"role": "assistant", "content": error_msg})
            return chat_history, "", thinking_display, ""
        
        # ボットからの応答を取得
        start_time = time.time()
        full_response = chatbot.generate_response(message)
        end_time = time.time()
        
        print(f"✅ 処理完了: {end_time - start_time:.2f}秒")
        
        # 思考部分と最終回答を分離
        thinking, final_answer = chatbot.parse_thinking_response(full_response)
        
        # 思考エリアの更新
        if thinking:
            current_time = time.strftime("%H:%M:%S")
            thinking_entry = f"[{current_time}] 💭 質問: {message}\n{thinking}\n{'-'*50}\n"
            thinking_display = thinking_entry + thinking_display
        
        # 最終回答の確認と処理
        if not final_answer or len(final_answer.strip()) < 5:
            if thinking:
                final_answer = "思考は完了しましたが、最終回答の生成が不完全でした。"
            else:
                final_answer = "申し訳ありません。適切な回答を生成できませんでした。"
        
        # チャット履歴に追加
        chat_history.append({"role": "user", "content": message})
        chat_history.append({"role": "assistant", "content": final_answer})
        
        return chat_history, "", thinking_display, ""
    
    except Exception as e:
        print(f"❌ エラー詳細: {str(e)}")
        error_msg = f"エラーが発生しました: {str(e)}"
        chat_history.append({"role": "user", "content": message})
        chat_history.append({"role": "assistant", "content": error_msg})
        return chat_history, "", thinking_display, ""

def clear_chat():
    """チャット履歴をクリア"""
    chatbot.clear_history()
    return [], ""

def create_gradio_interface():
    """Gradio 5.x用のインターフェースを作成"""
    with gr.Blocks(title="Qwen3-4B Chatbot") as demo:
        gr.Markdown("""
        # 🤖 Qwen3-4B Chatbot
        
        Qwen3-4Bモデルを使用したチャットボットです。思考プロセスを可視化できます。
        """)
        
        # チャットボットコンポーネント
        chatbot_display = gr.Chatbot(
            value=[],
            height=300,
            show_label=False,
            type="messages"
        )
        
        # 思考プロセス表示エリア
        thinking_display = gr.Textbox(
            label="🤔 思考プロセス",
            value="",
            lines=8,
            max_lines=15,
            interactive=False,
            visible=True
        )
        
        # 入力欄と送信ボタン
        with gr.Row():
            msg_textbox = gr.Textbox(
                placeholder="ここにメッセージを入力してください...",
                container=False,
                scale=4,
                show_label=False
            )
            send_button = gr.Button("送信", scale=1, variant="primary")
        
        # クリアボタン
        clear_button = gr.Button("履歴をクリア", variant="secondary")
        
        # 例文
        gr.Examples(
            examples=[
                "How many r's in strawberries?",
                "Then, how many r's in blueberries?", 
                "Really?"
            ],
            inputs=msg_textbox
        )
        
        # イベントハンドラー
        msg_textbox.submit(
            respond_to_message,
            inputs=[msg_textbox, chatbot_display, thinking_display],
            outputs=[chatbot_display, msg_textbox, thinking_display, msg_textbox]
        )
        
        send_button.click(
            respond_to_message,
            inputs=[msg_textbox, chatbot_display, thinking_display],
            outputs=[chatbot_display, msg_textbox, thinking_display, msg_textbox]
        )
        
        clear_button.click(
            clear_chat,
            outputs=[chatbot_display, thinking_display]
        )
    
    return demo

def run_chatbot():
    """チャットボットを起動"""
    print("Gradioインターフェースを起動中...")
    
    try:
        demo = create_gradio_interface()
        demo.launch(share=False, debug=False, quiet=True)
        
    except Exception as e:
        print(f"❌ エラーが発生しました: {e}")

# メイン実行
if __name__ == "__main__":
    run_chatbot()

注意事項： share=Trueでは誰でもアクセス可能になります。セキュリティ上、動作確認後はFalseに変更することを推奨します。この動作確認ではshare=Trueで確認しました。

実行方法

上記のコードを新しいGoogle Colabセルにコピーして実行するだけです。Gradioが自動的にWebインターフェースを起動し、ブラウザ上でチャットボットが利用できます。

実行の流れ:

モデルのダウンロードと読み込み（初回のみ数分かかります）
Gradioインターフェースの起動
ブラウザ上でチャットボットが利用可能に

使用例

実際に以下のような質問で思考プロセスを確認できます：

"How many r's in strawberries?"
"なぜ1+1=2なのですか？"
"複雑な数学の問題を解いてください"

技術的工夫点

このシステムでは、タグで推論過程を構造化し、思考と最終回答を分けて表示できます。

不完全な応答を検出して最大3回まで自動リトライし、エラー時のフォールバック機能も備えています。

メモリ効率のため履歴を20エントリまでに制限し、torch.float16による省メモリ実行を採用。

最新のGradio 5.xに対応し、直感的なユーザーインターフェースを提供します。

実行方法

# Google Colabで実行
if __name__ == "__main__":
    run_chatbot()

まとめ

Google Colab、Gradio、Qwen3-4Bを組み合わせることで、AIの推論過程を可視化できる仕組みを簡単に構築できました。

Google Colabで環境構築不要でGPUを利用し、Gradioでコード中心のWebインターフェースを構築しました

タグを解析して推論過程を構造化抽出し、多段階チェックとリトライで応答品質を担保しています。

あとがき

上記の記事ではOllama経由で試したが、ColabでWebUIとして実装することができなかった。今回はTransformers経由でWebUIとして実装してみた。