小型SVLMがどんなもんか実験してみた

********

※LINE対応チャットボット版の
「LINEチャットボット屋」
いろんなチャットボットがあります。
ぜひ、ご覧ください!

***************

***************

スマホで動く小型SVLMは現実的なのか

ここ最近、AIの話題の中でよく見かけるようになったのが、「エッジデバイスで動く小型のマルチモーダルAI」です。

特に、防衛や産業用途の記事では、ドローンや携帯端末のような現場の機器で、小型の視覚言語モデル、いわゆるSVLMを動かし、その場で状況認識や報告を行うという構想が語られることが増えてきました。

たしかに考えてみると、現場で撮影した画像や動画をその場で簡単に理解し、短い報告文に変換できるAIがあれば便利です。

通信が不安定な場所でも最低限の認識ができ、そこから上位システムへ情報を渡せるからです。

では実際に、スマホのような比較的限られた計算資源の上で、そうした小型SVLMはどこまで現実的なのでしょうか。
今回は、その感触をつかむために試験的な実験を行ってみました。

小型SVLMとは何か、なぜ注目されているのか

SVLMは、画像や動画とテキストをまとめて扱えるマルチモーダルAIの中でも、比較的小さく、軽量化を意識したモデルのことです。

大規模なVLMやLLMがクラウド上で強力な推論を行うのに対し、小型SVLMはスマホやドローン、携帯端末などのエッジ側で動かすことを視野に入れています。

この小型SVLMが注目される理由ははっきりしています。
現場では、すべてのデータを毎回クラウドに送って解析するとは限らないからです。

通信の遅延、接続不良、セキュリティ上の制約などを考えると、まず端末側で「何が見えているか」を簡潔に把握し、必要な情報だけを上位へ渡す仕組みのほうが実用的な場面が多いのです。

ただし、ここで誤解してはいけないのは、小型SVLMは「何でも考えて判断する万能AI」ではないという点です。

むしろ現実的な役割は、「見て、短く説明する」「異常候補を挙げる」「報告文のたたき台を作る」といった一次処理に近いものだと考えられます。

Qwen系モデルで動画理解を試してみた

今回の実験では、Qwen系のマルチモーダルモデルをGoogle Colab上で動かし、短い動画を読み込ませて、その内容を説明させるテストを行いました。

使ったのは3Bクラスと7Bクラスのモデルです。

Qwen2.5-VL-7B-Instruct
Qwen2.5-VL-3B-Instruct

実験の狙いは、現時点の比較的小型なマルチモーダルモデルが、動画をどこまで理解し、どの程度自然な説明や推測ができるのかを見ることにありました。

入力したのは、白い球体がテーブルの端付近に置かれている短い動画です。

 

人間が見ると、球体の位置関係や動きの方向、落ちそうかどうか、といった点に自然と注意が向きます。

そこで、まずは「この動画で何が起きているか」といった基本的な説明をさせ、次に「この先どうなると思うか」といった続きを推測させる質問を投げました。

この種の実験で重要なのは、単に答えが流暢かどうかではありません。
本当に映像の内容を見て答えているのか、それとも一般論で埋めているのかを見分ける必要があります。

そのため、今回の実験では、シーン説明だけでなく、位置変化や未来予測のような少し難しい問いも加えてみました。

3Bと7Bを比べて見えた違い

実際に試してみると、3Bと7Bではかなり違いがありました。

まず、7Bのほうが全体として説明文が自然で、場面のまとまり方も良かったです。

白い球体、木のテーブル、端にある状態、といった特徴を比較的きれいに拾ってくれました。

少なくとも「何が映っているか」をざっくり説明させる用途では、7Bのほうが一段上だと感じました。

一方で、未来予測や物理的な推測になると、7Bでもまだ弱さが見えました。

たとえば「このあとどうなるか」と聞くと、ボールが落ちるかもしれない、止まるかもしれない、転がるかもしれない、といった一般論の候補を並べる傾向がありました。

これは動画を厳密に解析して未来を読んでいるというより、「テーブルの端にあるボールなら、こういう可能性がある」と常識的に補っている印象です。

つまり、7Bは3Bよりも動画の雰囲気理解や説明力は高いものの、物理予測まで高精度にできるわけではありませんでした。

この違いはかなり重要で、マルチモーダルAIに何を期待するかを考えるうえで参考になります。

見たままを短く説明するのは得意でも、その先の世界の動きを正確に推論するのはまだ別の難しさがある、ということです。

スマホ実装でわかった課題

今回の実験を通して強く感じたのは、3Bでも7Bでも、そのままスマホに載せて快適に使うにはまだ重いということでした。

ColabのGPU上では動いても、スマホで同じように動画を処理し、数秒で応答を返し続けるとなると、発熱、メモリ、バッテリー、応答速度の面でかなり厳しくなります。

特に動画は重いです。
静止画1枚ならまだしも、複数フレームを処理しながらテキストを生成するとなると、見た目のパラメータ数以上に負荷を感じます。

そのため、3Bクラスをそのままスマホに持ち込んで万能に使うというよりは、もっと軽量化した1B前後のVLMに役割を限定して持たせるほうが現実的だと感じました。

ここでいう役割限定とは、「画像を見て1文で説明する」「特定対象の有無を答える」「位置の変化を短く報告する」といったものです。

逆に、「このあと何が起こるか」「なぜそうなるか」「物理的に自然か」といった深い推論は、端末側ではなく、本部側のより大きなLLMやルールベースの仕組みに任せたほうが現実的でしょう。

まとめ  スマホで全部考えるより「見て報告するAI」が現実的

今回の実験を通して見えてきたのは、小型SVLMがスマホでまったく使えないわけではない、ということでした。

むしろ、「何が見えているかを短く報告する」用途なら、かなり筋がいいと感じます。

7Bクラスではシーン説明の質が高く、3Bでも方向性を見るには十分な手応えがありました。

ただし、それは「スマホで全部考えるAI」が現実的だという意味ではありません。

現時点では、スマホ側の小型SVLMは前線の目とメモ係に近い役割が向いています。

画像や動画を見て、対象物や位置変化、異常候補を短くまとめ、その結果を本部の大きなLLMや分析システムへ渡す。
この分業構成のほうが、技術的にも運用面でも自然です。

今回の実験は、そうした未来の一端を確かめるものになりました。

今後は、3Bや7Bの能力をそのまま追うよりも、必要な機能だけをうまく削り、1B前後で実用的に動かす設計がより重要になってくるのかもしれません。

今回使ったPythonコード

* 表示のバグでインテントが崩れています。
下記のコードを使用する場合は、お気に入りのAIにコピペして「インテント直して」と言ってください。


# ============================================
# Qwen2.5-VL-7B-Instruct 動画テスト 1セル版
# ボタン送信対応版
# - mp4を1回アップロード
# - 同じ動画に何回でも質問可能
# - HFアクセストークンはテキストボックス入力
# - 空欄で送信すると終了
# ============================================

!pip -q install git+https://github.com/huggingface/transformers accelerate qwen-vl-utils decord ipywidgets

import os
import gc
import time
import torch
import platform
from google.colab import files
from IPython.display import display, clear_output
import ipywidgets as widgets

# ----------------------------
# GPU確認
# ----------------------------
print("Python:", platform.python_version())
print("PyTorch:", torch.__version__)
print("CUDA available:", torch.cuda.is_available())

if torch.cuda.is_available():
print("GPU:", torch.cuda.get_device_name(0))
!nvidia-smi
else:
raise RuntimeError("GPUが見つかりません。Colabの『ランタイムのタイプを変更』でGPUを選んでください。")

print("\n[注意]")
print("7B版は3B版よりかなり重いです。")
print("T4で厳しい場合は、L4/A100 か 3B版に戻すのがおすすめです。")

# ----------------------------
# HFトークン入力
# ----------------------------
print("\n=== Hugging Face アクセストークン入力 ===")
print("公開モデルなので空欄でも動くことがあります。必要なら入力してください。")

hf_token_widget = widgets.Password(
value='',
placeholder='hf_xxxxxxxxxxxxxxxxx',
description='HF Token:',
layout=widgets.Layout(width='650px')
)
display(hf_token_widget)

confirm_token_btn = widgets.Button(
description="トークン確定",
button_style='info'
)
token_status = widgets.Output()

display(confirm_token_btn, token_status)

state = {
"hf_token": "",
"token_confirmed": False,
"video_path": None,
"history": [],
"ended": False,
}

def on_confirm_token_clicked(b):
with token_status:
clear_output()
state["hf_token"] = hf_token_widget.value.strip()
if state["hf_token"]:
os.environ["HUGGINGFACE_HUB_TOKEN"] = state["hf_token"]
print("HFトークンをセットしました。")
else:
print("HFトークン未入力のまま進めます。")
state["token_confirmed"] = True

confirm_token_btn.on_click(on_confirm_token_clicked)

print("\n上の『トークン確定』ボタンを押してから次へ進んでください。")

# ----------------------------
# モデル読み込み
# ----------------------------
from transformers import Qwen2_5_VLForConditionalGeneration, AutoProcessor
from qwen_vl_utils import process_vision_info

MODEL_NAME = "Qwen/Qwen2.5-VL-7B-Instruct"

dtype = torch.float16
if torch.cuda.is_available():
major, minor = torch.cuda.get_device_capability(0)
if major >= 8:
dtype = torch.bfloat16

print(f"\n=== モデル読み込み開始: {MODEL_NAME} ===")
print("dtype:", dtype)

hf_token = hf_token_widget.value.strip()

model_kwargs = {
"torch_dtype": dtype,
"device_map": "auto",
}
processor_kwargs = {}

if hf_token:
model_kwargs["token"] = hf_token
processor_kwargs["token"] = hf_token

model = Qwen2_5_VLForConditionalGeneration.from_pretrained(
MODEL_NAME,
**model_kwargs
)
processor = AutoProcessor.from_pretrained(
MODEL_NAME,
**processor_kwargs
)

print("モデル読み込み完了")

# ----------------------------
# mp4アップロード
# ----------------------------
print("\n=== mp4アップロード ===")
uploaded = files.upload()

video_path = None
for fn in uploaded.keys():
if fn.lower().endswith(".mp4"):
video_path = fn
break

if video_path is None:
raise ValueError("mp4ファイルが見つかりません。短い mp4 を1本アップロードしてください。")

state["video_path"] = video_path

print("動画ファイル:", video_path)
print("ファイルサイズ(MB):", round(os.path.getsize(video_path) / 1024 / 1024, 2))

# ----------------------------
# 推論関数
# ----------------------------
def ask_video(video_path, user_prompt, max_new_tokens=256, fps=1.0, max_pixels=360*420):
# 7Bは重いので fps と max_pixels を少し控えめにして負荷を下げる
messages = [
{
"role": "user",
"content": [
{
"type": "video",
"video": f"file://{os.path.abspath(video_path)}",
"fps": fps,
"max_pixels": max_pixels,
},
{
"type": "text",
"text": user_prompt
}
]
}
]

text = processor.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True
)

image_inputs, video_inputs = process_vision_info(messages)

inputs = processor(
text=[text],
images=image_inputs,
videos=video_inputs,
padding=True,
return_tensors="pt"
)

for k, v in inputs.items():
if hasattr(v, "to"):
inputs[k] = v.to(model.device)

start = time.time()
with torch.no_grad():
generated_ids = model.generate(
**inputs,
max_new_tokens=max_new_tokens,
do_sample=False
)
end = time.time()

generated_ids_trimmed = [
out_ids[len(in_ids):]
for in_ids, out_ids in zip(inputs["input_ids"], generated_ids)
]

output_text = processor.batch_decode(
generated_ids_trimmed,
skip_special_tokens=True,
clean_up_tokenization_spaces=False
)[0]

elapsed = end - start
return output_text, elapsed

# ----------------------------
# UI部品
# ----------------------------
print("\n=== 質問UI ===")
print("質問を入力して『送信』を押してください。")
print("空欄のまま『送信』を押すと終了します。")

prompt_widget = widgets.Textarea(
value="この動画で何が起きているか、2〜3文で日本語で説明してください。",
placeholder="ここに質問を書く。空欄で送信すると終了。",
description="Prompt:",
layout=widgets.Layout(width='900px', height='140px')
)

max_tokens_widget = widgets.IntText(
value=180,
description='MaxTokens:',
layout=widgets.Layout(width='260px')
)

fps_widget = widgets.FloatText(
value=1.0,
description='FPS:',
layout=widgets.Layout(width='220px')
)

max_pixels_widget = widgets.IntText(
value=360 * 420,
description='MaxPixels:',
layout=widgets.Layout(width='260px')
)

send_button = widgets.Button(
description="送信",
button_style='success',
icon='paper-plane'
)

end_button = widgets.Button(
description="強制終了",
button_style='warning',
icon='stop'
)

status_output = widgets.Output()
answer_output = widgets.Output()
history_output = widgets.Output()

display(prompt_widget)
display(widgets.HBox([max_tokens_widget, fps_widget, max_pixels_widget]))
display(widgets.HBox([send_button, end_button]))
display(status_output)
display(answer_output)
display(history_output)

# ----------------------------
# ボタン動作
# ----------------------------
def redraw_history():
with history_output:
clear_output()
print("=== 質問履歴 ===")
if not state["history"]:
print("まだ質問はありません。")
return

for i, item in enumerate(state["history"], 1):
print(f"\n--- {i}回目 ---")
print("Prompt:", item["prompt"])
print("Output:", item["output"])
print("推論時間:", f'{item["elapsed"]:.2f} 秒')
print("fps:", item["fps"], "| max_pixels:", item["max_pixels"])

def on_send_clicked(b):
if state["ended"]:
with status_output:
clear_output()
print("すでに終了しています。")
return

user_prompt = prompt_widget.value.strip()
max_new_tokens = int(max_tokens_widget.value)
fps = float(fps_widget.value)
max_pixels = int(max_pixels_widget.value)

if user_prompt == "":
state["ended"] = True
send_button.disabled = True
prompt_widget.disabled = True
max_tokens_widget.disabled = True
fps_widget.disabled = True
max_pixels_widget.disabled = True

with status_output:
clear_output()
print("空欄送信を検出しました。終了します。")

with answer_output:
print("\n処理を終了しました。")
return

with status_output:
clear_output()
print("推論中です... 7B版なので少し時間がかかることがあります。")

try:
output_text, elapsed = ask_video(
video_path=state["video_path"],
user_prompt=user_prompt,
max_new_tokens=max_new_tokens,
fps=fps,
max_pixels=max_pixels
)

state["history"].append({
"prompt": user_prompt,
"output": output_text,
"elapsed": elapsed,
"max_new_tokens": max_new_tokens,
"fps": fps,
"max_pixels": max_pixels,
})

with answer_output:
print("\n==============================")
print("Prompt:")
print(user_prompt)
print("------------------------------")
print("Output:")
print(output_text)
print("------------------------------")
print(f"推論時間: {elapsed:.2f} 秒")
if torch.cuda.is_available():
print(f"GPU使用メモリ allocated: {torch.cuda.memory_allocated()/1024**3:.2f} GB")
print(f"GPU使用メモリ reserved : {torch.cuda.memory_reserved()/1024**3:.2f} GB")
print("==============================\n")

with status_output:
clear_output()
print(f"完了しました。質問回数: {len(state['history'])}")

redraw_history()

except Exception as e:
with status_output:
clear_output()
print("エラーが出ました。")
print(str(e))
print("\n対策例:")
print("- fps を 0.5 に下げる")
print("- MaxTokens を 128 に下げる")
print("- MaxPixels を 200*280 くらいに下げる")
print("- もっと短い mp4 にする")
print("- L4 / A100 に変える")

def on_end_clicked(b):
state["ended"] = True
send_button.disabled = True
prompt_widget.disabled = True
max_tokens_widget.disabled = True
fps_widget.disabled = True
max_pixels_widget.disabled = True

with status_output:
clear_output()
print("強制終了しました。")

gc.collect()
if torch.cuda.is_available():
torch.cuda.empty_cache()

send_button.on_click(on_send_clicked)
end_button.on_click(on_end_clicked)

print("\n準備完了。質問を書いて『送信』を押してください。")

****************

最近のデジタルアート作品を掲載!

X 旧ツイッターもやってます。
https://x.com/ison1232

インスタグラムはこちら
https://www.instagram.com/nanahati555/

***************

GoogleのGemma-2Bを使ってRAGを試してみた

********

※LINE対応チャットボット版の
「LINEチャットボット屋」
いろんなチャットボットがあります。
ぜひ、ご覧ください!

***************

***************

小型モデルを回答&分析に使うメリット

小型の生成AIモデルで、どこまで実用的なことができるのかを試します。

そこで今回は、GoogleのGemma-2Bを使って、RAGの簡単なテストをしてみました。

小型モデルを使う利点は、大型モデルと違って、社内にシステムを置けるので、情報漏洩がしづらいと言うところです

chataGPTなどの大型モデルを使うと、社内の資料をアップロードしないといけないので、情報が外に漏れてしまいます。

こういったことを気にしないで活用できるのが、小型モデルの良いところでしょう。

さて、RAGは外部の資料を検索して、その内容をもとに回答&分析を作る仕組みです。

モデルが元から知っている知識だけで答えるのではなく、自分で資料を参照しながら返答するので、うまく作れれば特定の文書に強いAIを作ることができます。

有名なシステムでわかりやすいのはGoogleの出しているnotebook LMです。
ただしあちらはサーバ上で動いているので、やはり社内などの文章をアップロードすると、情報が漏れるって言う、いいんだか悪いんだかが起きますね。

そこで今回は、大がかりな本番システムを作らず、まずはGoogle Colab上で動くシンプルな構成で、Gemma-2Bのような比較的小さなモデルでもRAGらしい動きができるのかを確かめてみました。

そもそもRAGとは何か

RAGは、日本語では「検索拡張生成」などと呼ばれます。

ざっくり言えば、まず質問に関連する文章を資料の中から探し、その検索結果を生成AIに渡して回答や分析をさせる仕組みです。

普通のチャットAIは、学習済みの知識をもとに答えます。
しかしRAGを使うと、手元のテキストファイルや社内文書、マニュアル、議事録、取引先データなどを参照しながら答えさせることができます。

つまり、モデル自体を追加学習しなくても、ある程度その場で知識を補えるわけです。

この仕組みの面白いところは、モデルのサイズがそれほど大きくなくても、参照資料がしっかりしていれば役立つ回答&分析ができることです。

逆に言えば、モデルそのものの性能だけでなく、資料の分け方や検索精度がかなり重要になります。
今回のテストでも、まさにそこがポイントになりました。

今回の環境と使用した諸々

今回の実験環境はGoogle Colabです。
ローカル環境を整えなくても始めやすく、ちょっとしたRAGの試作にはかなり便利です。

役割としては、Gemma-2Bが回答生成を担当し、SentenceTransformerがテキストの埋め込みを作り、FAISSがベクトル検索を担当します。

Gradioは簡単な操作画面を作るために使いました。
これにより、Hugging Faceのトークンを入力してモデルを初期化し、テキストファイルをアップロードし、質問を入力して回答を得る、という一連の流れをブラウザ上で試せるようにしました。

最初はとにかく最小構成で動かすことを優先し、1セルで完結するコードにしていました。

こういう形は試行錯誤しやすい反面、後から見ると雑な実装になりやすく、精度面でも改善の余地が多いですね。

実装してみてつまずいたポイント

実際に組んでみると、最初から順調に進んだわけではありません。まず単純な構文ミスがありました。

関数定義の引数のところに余計な文字が混ざっていて、それだけでエラーになります。
こういうミスは地味ですが、Colabで一気に書いていると意外と起きやすいです。

ファイル読み込みや空入力に対する例外処理も必要でした。
こうした部分をちゃんと整えるだけでも、実験コードとしてかなり扱いやすくなります。

RAGは検索や生成の精度に目が行きがちですが、その前にまずシステムが落ちずに動かすことだと、あらためて感じました。

なぜ最初は精度があまり出なかったのか

動くようになってから試してみると、精度はあまり高くありませんでした。

質問に対して少しずれた回答をしたり、資料に書かれていないことをうまく拾えなかったりします。

最初はGemma-2Bという小型モデルの限界かなと思ったのですが、見直してみると、原因はモデルだけではありませんでした。

一番大きかったのは、テキストの分割方法です。

最初は単純に文字数で区切っていたので、文の途中で分かれたり、見出しと本文が離れたりして、検索しにくい断片が大量にできていました。
これでは、検索で拾う文章の質が下がってしまいます。

もう一つは埋め込みモデルです。最初は英語寄りの埋め込みモデルを使っていたため、日本語のテキストを扱うには相性がいまひとつでした。

さらに、検索時に取る件数が少なすぎると、たまたま外したときにそのまま精度低下につながります。

つまり、生成モデルの前段階である検索部分に改善余地がかなりあったわけです。

チャンク分割や検索方法を見直してみた

そこで、精度改善版ではまずチャンク分割を見直しました。
文字数で機械的に切るのではなく、段落ベースでまとめ、長すぎる段落だけ追加で分割するようにします。

これだけでも、意味のまとまりが保たれやすくなり、検索の質が上がります。
さらに overlap を入れて、前後の文脈が多少つながるようにしました。

埋め込みモデルも、多言語対応の intfloat/multilingual-e5-base に変更。
日本語の資料を扱うなら、こうした多言語系のモデルのほうが安定感があります。

加えて、E5系の推奨に合わせて、文書側には passage:、質問側には query: という接頭辞を付けて埋め込みを作るようにしました。

検索部分では、FAISSのIndexFlatL2からIndexFlatIPに変更。
埋め込みを正規化して使うなら、内積ベースのほうが相性が良いケースがあります。

また、検索件数も top_k=3 から top_k=5 に増やしました。これにより、関連文脈を取りこぼしにくくなりました。

さらに、デバッグのために「実際に検索で拾った文脈をそのまま表示する機能」も付けました。これがかなり便利で、精度が悪い原因が検索側なのか、生成側なのかを切り分けやすくなりました。

小型モデルでRAGを試す面白さ

今回試してみて感じたのは、Gemma-2Bのような小型モデルでも、RAGの基本構成を理解したり、簡単な実験をしたりするには十分面白いということです。

もちろん、大型モデルに比べると読解力や回答の安定性には差がありますし、複雑な質問では限界も見えます。

ただし、RAGではモデル本体だけでなく、資料の整え方、チャンク分割、埋め込みモデル、検索方法などが全体の精度に大きく効きます。

つまり、小型モデルだから駄目と決めつけるのではなく、周辺設計を丁寧に詰めることでかなり改善できる余地があります。

今回のテストは、まさにそのことを実感する機会になりました。
最初は「何となく動いた」段階でしたが、検索部分をきちんと見直すことで、回答の納得感がかなり変わってきます。

RAGは単なる生成AIの応用ではなく、検索設計まで含めた総合戦だとよく分かります。

もしこれから試すなら、まずは小さく作って動かし、検索結果の確認機能を入れながら、少しずつ精度を詰めていくのがおすすめです。

Gemma-2Bを使った今回の実験は、その第一歩としてかなり面白い実験でした。

必要以上に大きな構成にせず、まずは手軽に試してみる。それだけでも、RAGの勘所はかなりつかめると思います。

今回使用したコード

*表示のバグでインデントが崩れています。
コピーしてお気に入りのAIに直してもらってください。


# -*- coding: utf-8 -*-
"""
Gemma-2B RAG System (All-in-One Cell) - Accuracy Improved Version
Google Colab 用
"""

# ==========================================
# 1. ライブラリのインストール
# ==========================================
!pip install -q transformers sentence-transformers faiss-cpu gradio accelerate

# ==========================================
# 2. インポート
# ==========================================
import os
import re
import faiss
import gradio as gr
import numpy as np
import torch

from transformers import AutoTokenizer, AutoModelForCausalLM
from sentence_transformers import SentenceTransformer

# ==========================================
# 3. グローバル変数
# ==========================================
model = None
tokenizer = None
embedder = None
index = None
chunked_texts = []

# ==========================================
# 4. ユーティリティ
# ==========================================
def get_torch_dtype():
if torch.cuda.is_available():
return torch.float16
return torch.float32

def get_model_input_device():
try:
return next(model.parameters()).device
except Exception:
return torch.device("cuda" if torch.cuda.is_available() else "cpu")

def clean_text(text):
if not text:
return ""
text = text.replace("\r\n", "\n").replace("\r", "\n")
text = re.sub(r"\n{3,}", "\n\n", text)
text = re.sub(r"[ \t]+", " ", text)
return text.strip()

# ==========================================
# 5. システム初期化・リセット
# ==========================================
def initialize_system(hf_token):
global model, tokenizer, embedder, index, chunked_texts

index = None
chunked_texts = []

try:
model_id = "google/gemma-2b-it"

if tokenizer is None:
tokenizer = AutoTokenizer.from_pretrained(
model_id,
token=hf_token if hf_token else None
)

if model is None:
dtype = get_torch_dtype()
model = AutoModelForCausalLM.from_pretrained(
model_id,
token=hf_token if hf_token else None,
device_map="auto",
torch_dtype=dtype
)
model.eval()

# 精度重視で small → base に変更
if embedder is None:
embedder = SentenceTransformer("intfloat/multilingual-e5-base")

dimension = embedder.get_sentence_embedding_dimension()

# normalize_embeddings=True と相性のよい Inner Product 検索
index = faiss.IndexFlatIP(dimension)

return "システム初期化完了(データはリセットされました。新しいファイルをアップロードしてください)"

except Exception as e:
return f"初期化エラー: {str(e)}"

# ==========================================
# 6. 精度改善版テキスト分割
# - 段落ベース
# - 長すぎる段落のみ追加分割
# ==========================================
def split_text(text, chunk_size=500, overlap=100):
text = clean_text(text)
if not text:
return []

paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
chunks = []
current = ""

for para in paragraphs:
# 今の current に追加できるなら追加
if len(current) + len(para) + 2 <= chunk_size: current += ("\n\n" + para) if current else para else: if current: chunks.append(current) # 段落自体が長すぎるならスライド分割 if len(para) > chunk_size:
start = 0
while start < len(para): end = min(start + chunk_size, len(para)) chunk = para[start:end].strip() if chunk: chunks.append(chunk) if end >= len(para):
break
start += chunk_size - overlap
current = ""
else:
current = para

if current:
chunks.append(current)

# 空要素除去
chunks = [c.strip() for c in chunks if c and c.strip()]
return chunks

# ==========================================
# 7. ファイル読み込み・保存
# ==========================================
def process_and_store_file(file_obj):
global index, chunked_texts, embedder

if index is None:
return "先に [Initialize / Reset System] ボタンを押してください。"

if file_obj is None:
return "ファイルが選択されていません。"

try:
file_path = getattr(file_obj, "name", None)
if file_path is None:
file_path = file_obj

with open(file_path, "r", encoding="utf-8") as f:
text = f.read()

except UnicodeDecodeError:
return "エラー: ファイルの文字コードがUTF-8ではありません。"
except Exception as e:
return f"ファイル読み込みエラー: {str(e)}"

text = clean_text(text)
if not text:
return "エラー: ファイルの中身が空です。"

chunks = split_text(text, chunk_size=500, overlap=100)

if not chunks:
return "テキストを分割できませんでした。ファイル内容を確認してください。"

try:
passages = [f"passage: {c}" for c in chunks]
embeddings = embedder.encode(
passages,
convert_to_numpy=True,
normalize_embeddings=True,
show_progress_bar=False
).astype("float32")

index.add(embeddings)
chunked_texts.extend(chunks)

except Exception as e:
return f"埋め込み保存エラー: {str(e)}"

return (
f"完了: {os.path.basename(file_path)} から "
f"{len(chunks)} チャンクの情報を記憶しました。"
)

# ==========================================
# 8. 検索
# - top_k を 5 に増量
# - デバッグ用にスコア付き検索も返せるように
# ==========================================
def retrieve_context(query, top_k=5, return_debug=False):
global index, chunked_texts, embedder

if index is None or index.ntotal == 0:
return "" if not return_debug else []

if not query or not query.strip():
return "" if not return_debug else []

try:
query_embedding = embedder.encode(
[f"query: 次の質問に関連する箇所を資料から探してください。質問: {query}"],
convert_to_numpy=True,
normalize_embeddings=True,
show_progress_bar=False
).astype("float32")

k = min(top_k, index.ntotal)
scores, indices = index.search(query_embedding, k)

results = []
for score, idx in zip(scores[0], indices[0]):
if idx != -1 and 0 <= idx < len(chunked_texts):
results.append({
"score": float(score),
"text": chunked_texts[idx]
})

if return_debug:
return results

return "\n---\n".join([r["text"] for r in results])

except Exception:
return "" if not return_debug else []

# ==========================================
# 9. 検索結果確認用
# ==========================================
def preview_retrieved_context(query):
if model is None or tokenizer is None:
return "エラー: システムが初期化されていません。"

if not query or not query.strip():
return "質問を入力してください。"

results = retrieve_context(query, top_k=5, return_debug=True)

if not results:
return "関連する情報が見つかりませんでした。"

lines = []
for i, item in enumerate(results, start=1):
lines.append(f"[候補 {i}] score={item['score']:.4f}\n{item['text']}")

return "\n\n==============================\n\n".join(lines)

# ==========================================
# 10. 回答生成
# ==========================================
def generate_answer(query):
global model, tokenizer

if model is None or tokenizer is None:
return "エラー: システムが初期化されていません。"

if not query or not query.strip():
return "質問を入力してください。"

context = retrieve_context(query, top_k=5)

if not context:
return "アップロードされた資料の中に、関連する情報が見つかりませんでした。"

prompt = f"""あなたは資料の内容に忠実なアシスタントです。
以下の[参照資料]だけを根拠に回答してください。
資料にないことは「資料には書かれていません」と答えてください。
推測や想像で補完してはいけません。
回答は簡潔かつ具体的に述べ、必要なら箇条書きで示してください。

[参照資料]
{context}

[質問]
{query}

[回答]
"""

try:
input_device = get_model_input_device()
inputs = tokenizer(
prompt,
return_tensors="pt",
truncation=True,
max_length=2048
)
inputs = {k: v.to(input_device) for k, v in inputs.items()}

with torch.no_grad():
outputs = model.generate(
**inputs,
max_new_tokens=256,
do_sample=False
)

generated_text = tokenizer.decode(outputs[0], skip_special_tokens=True)
answer = generated_text[len(prompt):].strip()

if not answer:
return "回答を生成できませんでした。"

return answer

except Exception as e:
return f"回答生成エラー: {str(e)}"

# ==========================================
# 11. UI構築
# ==========================================
with gr.Blocks() as demo:
gr.Markdown("## Gemma-2B RAG System(精度改善版)")

with gr.Group():
gr.Markdown("### 1. 初期化(リセット)")
with gr.Row():
token_input = gr.Textbox(
label="Hugging Face Token",
type="password",
placeholder="hf_..."
)
init_btn = gr.Button("Initialize / Reset System")
init_status = gr.Textbox(label="Status", interactive=False)
init_btn.click(fn=initialize_system, inputs=token_input, outputs=init_status)

gr.Markdown("---")

with gr.Group():
gr.Markdown("### 2. テキストファイルをアップロード")
file_input = gr.File(label="Upload Text File (.txt)", file_types=[".txt"])
upload_status = gr.Textbox(label="Upload Status", interactive=False)
file_input.upload(fn=process_and_store_file, inputs=file_input, outputs=upload_status)

gr.Markdown("---")

with gr.Group():
gr.Markdown("### 3. 質問")
query_input = gr.Textbox(
label="Ask Question",
placeholder="例: この資料の要点は?"
)

with gr.Row():
preview_btn = gr.Button("Preview Retrieved Context")
submit_btn = gr.Button("Submit")

preview_output = gr.Textbox(label="Retrieved Context Preview", lines=14)
answer_output = gr.Textbox(label="Answer", lines=10)

preview_btn.click(fn=preview_retrieved_context, inputs=query_input, outputs=preview_output)
submit_btn.click(fn=generate_answer, inputs=query_input, outputs=answer_output)

demo.launch(debug=True, share=True)

****************

最近のデジタルアート作品を掲載!

X 旧ツイッターもやってます。
https://x.com/ison1232

インスタグラムはこちら
https://www.instagram.com/nanahati555/

**************

PAGE TOP