[備忘録] Qwen3の”考える様子”を見てみよう – Google Colabで実装 #Python

# 思考プロセス表示型チャット（Google Colab用）

# 必要なパッケージのインストール
!pip install ollama -q
!pip install ipywidgets -q
!pip install nest_asyncio -q

# Ollamaのインストール (初回のみ実行)
!curl -fsSL https://ollama.com/install.sh | sh

# バックグラウンドでOllamaサーバーを起動し、モデルをダウンロード
!nohup ollama serve > /dev/null 2>&1 &
!ollama pull qwen3:4b

# 必要なライブラリをインポート
import re
import time
import nest_asyncio
from IPython.display import display, clear_output, HTML
import ipywidgets as widgets
from ollama import chat

# Jupyterでasyncを動作させるための設定
nest_asyncio.apply()

# 会話履歴の初期化（思考プロセスタグの使用法を指定）
conversation_history = [
    {"role": "system", "content": """
あなたは役立つアシスタントです。思考プロセスを示すために  タグを使ってください。
回答は必ず日本語で行ってください。思考プロセスはあなたの自然な形で構いません。

例：

ユーザー: 15 + 27は？


I need to calculate 15 + 27.
First, I'll add the units: 5 + 7 = 12
I'll write down 2 and carry the 1.
Then, I'll add the tens: 1 + 1 + 2 = 4
So the answer is 42.


15 + 27は42です。
"""}
]

# 思考プロセスの履歴保存のための変数
thinking_history = []

# UIコンポーネントの作成 - 色合いを見やすく調整
title = widgets.HTML(value="
'color: #2c3e50;'>思考プロセス表示型チャット")
instructions = widgets.HTML(value="""
"background-color: #e1f5fe; padding: 10px; margin: 10px 0; border-radius: 5px; border-left: 5px solid #0288d1; color: #01579b;">
使い方:

下のテキストエリアにメッセージを入力して「送信」をクリックしてください
アシスタントのリアルタイムの思考プロセスが表示されます
回答は日本語で返信されます
思考プロセス履歴の表示/非表示を切り替えることができます


""")

text_input = widgets.Textarea(
    placeholder='ここにメッセージを入力...',
    layout=widgets.Layout(width='100%', height='80px')
)
send_button = widgets.Button(
    description="送信",
    button_style='primary',
    icon='paper-plane'
)
clear_button = widgets.Button(
    description="履歴クリア",
    button_style='danger',
    icon='trash'
)
show_thinking_history = widgets.Checkbox(
    value=True,
    description='思考プロセス履歴を表示',
    indent=False
)

output_area = widgets.Output(layout=widgets.Layout(height='400px', max_height='600px', overflow='auto'))
thinking_area = widgets.Output(layout=widgets.Layout(height='200px', max_height='300px', overflow='auto', border='1px solid #bdbdbd'))
thinking_history_area = widgets.Output(layout=widgets.Layout(height='200px', max_height='600px', overflow='auto', border='1px solid #bdbdbd'))

# ラベルの作成
thinking_label = widgets.HTML(value="
'color: #2c3e50; margin-bottom: 5px;'>思考プロセス (リアルタイム):")
chat_label = widgets.HTML(value="
'color: #2c3e50; margin-bottom: 5px;'>チャット履歴:")
thinking_history_label = widgets.HTML(value="
'color: #2c3e50; margin-bottom: 5px;'>思考プロセス履歴:")

# 思考内容を抽出してフォーマットする関数
def format_thinking_content(text):
    pattern = r'(.*?)'
    matches = re.findall(pattern, text, re.DOTALL)
    if matches:
        return matches[0].strip()
    return ""

# 思考タグを最終回答から削除する関数
def remove_thinking_tags(text):
    return re.sub(r'.*?', '', text, flags=re.DOTALL).strip()

# 会話履歴をクリアする関数
def clear_history(b):
    global conversation_history, thinking_history
    conversation_history = [
        {"role": "system", "content": """
あなたは役立つアシスタントです。思考プロセスを示すために  タグを使ってください。
回答は必ず日本語で行ってください。思考プロセスはあなたの自然な形で構いません。

例：

ユーザー: 15 + 27は？


I need to calculate 15 + 27.
First, I'll add the units: 5 + 7 = 12
I'll write down 2 and carry the 1.
Then, I'll add the tens: 1 + 1 + 2 = 4
So the answer is 42.


15 + 27は42です。
"""}
    ]
    thinking_history = []
    
    with output_area:
        clear_output(wait=True)
        display(HTML("
'padding: 10px; color: #004d40; background-color: #e0f2f1; border-radius: 5px;'>チャット履歴をクリアしました"))
    
    with thinking_history_area:
        clear_output(wait=True)
        display(HTML("
'padding: 10px; color: #004d40; background-color: #e0f2f1; border-radius: 5px;'>思考履歴をクリアしました"))

# 思考履歴の表示/非表示を切り替える関数
def toggle_thinking_history(change):
    if change['new']:
        thinking_history_box.layout.display = 'flex'
    else:
        thinking_history_box.layout.display = 'none'

# メッセージ送信を処理する関数
def on_send_button_clicked(b):
    user_input = text_input.value.strip()
    if not user_input:
        return
    
    # 処理中は入力を無効化
    text_input.disabled = True
    send_button.disabled = True
    
    # 入力フィールドをクリア
    text_input.value = ""
    
    # ユーザーメッセージを会話に追加して表示
    conversation_history.append({"role": "user", "content": user_input})
    
    with output_area:
        display(HTML(f"
'margin: 10px; padding: 10px; background-color: #e3f2fd; border-radius: 5px; border-left: 4px solid #2196F3; color: #0d47a1;'>あなた: {user_input}"))
    
    # モデルからの応答を取得
    thinking_content = ""
    response_content = ""
    in_thinking = False
    
    # 思考中インジケータを開始
    with thinking_area:
        clear_output(wait=True)
        thinking_html = HTML("
'margin: 10px; padding: 10px; background-color: #f1f8e9; border-radius: 5px; border-left: 4px solid #8bc34a; color: #33691e;'>アシスタント考え中...")
        display(thinking_html)
    
    # レスポンスをストリーミング
    try:
        for chunk in chat(model="qwen3:4b", messages=conversation_history, stream=True):
            content = chunk["message"]["content"] if "message" in chunk and "content" in chunk["message"] else ""
            
            if "" in content and not in_thinking:
                in_thinking = True
                thinking_content = ""
            
            if in_thinking:
                thinking_content += content
                
                # 思考表示を更新
                with thinking_area:
                    clear_output(wait=True)
                    formatted_thinking = thinking_content.replace("", "").replace("", "")
                    display(HTML(f"
'margin: 10px; padding: 10px; background-color: #f1f8e9; border-radius: 5px; border-left: 4px solid #8bc34a; color: #33691e;'>思考プロセス:
'white-space: pre-wrap; background-color: #fafafa; padding: 8px; border-radius: 4px; color: #37474f;'>{formatted_thinking}

"))
                if "" in content:

                    in_thinking = False
                    # 思考履歴に追加

                    if formatted_thinking:

                        thinking_history.append({

                            "question": user_input,

                            "thinking": formatted_thinking

                        })
                        # 思考履歴表示を更新

                        with thinking_history_area:

                            clear_output(wait=True)

                            for i, item in enumerate(thinking_history):

                                display(HTML(f"""
'margin: 10px; padding: 10px; background-color: #fff8e1; border-radius: 5px; border-left: 4px solid #ffc107; margin-bottom: 15px;'>

'font-weight: bold; color: #bf360c; margin-bottom: 5px;'>質問 {i+1}: {item['question']}
'background-color: #fafafa; padding: 8px; border-radius: 4px; color: #37474f;'>

'white-space: pre-wrap; margin: 0;'>{item['thinking']}

                                
                                """))

            else:

                response_content += content
                # レスポンス表示を更新

                with output_area:

                    # 最新のメッセージを含む全履歴を表示

                    clear_output(wait=True)

                    # 過去のメッセージを再表示

                    for i, msg in enumerate(conversation_history):

                        if msg["role"] == "user":

                            display(HTML(f"
'margin: 10px; padding: 10px; background-color: #e3f2fd; border-radius: 5px; border-left: 4px solid #2196F3; color: #0d47a1;'>あなた: {msg['content']}
"))

                        elif msg["role"] == "assistant" and i > 0:  # システムメッセージをスキップ

                            display(HTML(f"
'margin: 10px; padding: 10px; background-color: #e8f5e9; border-radius: 5px; border-left: 4px solid #4caf50; color: #1b5e20;'>アシスタント: {remove_thinking_tags(msg['content'])}
"))
                    # 現在のレスポンスを表示

                    display(HTML(f"
'margin: 10px; padding: 10px; background-color: #e8f5e9; border-radius: 5px; border-left: 4px solid #4caf50; color: #1b5e20;'>アシスタント: {response_content}
"))
        # アシスタントのレスポンスを会話履歴に追加

        full_response = thinking_content + response_content

        conversation_history.append({"role": "assistant", "content": full_response})
    except Exception as e:

        with output_area:

            display(HTML(f"
'margin: 10px; padding: 10px; background-color: #ffebee; border-radius: 5px; border-left: 4px solid #f44336; color: #b71c1c;'>エラー: {str(e)}
"))
    finally:

        # 入力を再有効化

        text_input.disabled = False

        send_button.disabled = False

        # 思考エリアをクリア

        with thinking_area:

            clear_output()
# イベントリスナーを設定

send_button.on_click(on_send_button_clicked)

clear_button.on_click(clear_history)

show_thinking_history.observe(toggle_thinking_history, names='value')
# Enterキーでも送信できるように（Shift+Enterは改行用に予約）

def on_key_press(widget, event):

    if event.get('type') == 'keydown' and event.get('key') == 'Enter' and not event.get('shiftKey'):

        on_send_button_clicked(None)

        return True

    return False
text_input.observe(on_key_press, names=['_keydown'])
# ボタンをグループ化

button_box = widgets.HBox([send_button, clear_button], layout=widgets.Layout(justify_content='space-between'))
# 思考履歴ボックスを作成

thinking_history_box = widgets.VBox([

    thinking_history_label,

    thinking_history_area

], layout=widgets.Layout(display='flex', margin='10px 0'))
# 最終的なチャットUIを構築

chat_ui = widgets.VBox([

    title,

    instructions,

    widgets.HBox([chat_label]),

    output_area,

    widgets.HBox([thinking_label]),

    thinking_area,

    widgets.HBox([show_thinking_history]),

    thinking_history_box,

    widgets.HBox([text_input]),

    button_box

], layout=widgets.Layout(width='100%', padding='10px'))
# チャットインターフェースを表示

display(chat_ui)