RTX 4060 8GBで論文RAGを完全ローカル化した — BGE-M3 + Qwen2.5-32B + ChromaDB構築記

ArXivの論文を読むのにGPT-4oを使っていた。PDFを投げて「要約して」で30秒。便利だった。

ある日、社内の研究テーマに近い論文を50本まとめて処理しようとして手が止まった。セキュリティポリシー的に外部APIに投げていいのかこれ。上司に聞いたら案の定ダメで、じゃあローカルで全部やるしかないな、というのがこのプロジェクトの始まりだ。

llama.cppでQwen2.5-32Bを動かせることは前回の記事で確認済みだった。LLMはある。あとは「論文の中身を検索して、関連箇所をLLMに食わせる」仕組み——つまりRAGを組めばいい。言うのは簡単だが、VRAM 8GBという狭い箱の中でどう収めるかが問題だった。

ArXiv論文のテキスト抽出 — まずPDFからデータを抜く

最初にやったのは論文のテキスト抽出。ArXiv APIでPDFを落として、PyMuPDFでテキスト化する。

import arxiv
import fitz  # PyMuPDF
from pathlib import Path

def fetch_arxiv_papers(query: str, max_results: int = 20,
                       save_dir: str = "./papers") -> list[dict]:
    """ArXivから論文を検索・ダウンロード・テキスト抽出"""
    save_path = Path(save_dir)
    save_path.mkdir(exist_ok=True)

    client = arxiv.Client()
    search = arxiv.Search(
        query=query,
        max_results=max_results,
        sort_by=arxiv.SortCriterion.SubmittedDate
    )

    papers = []
    for result in client.results(search):
        pdf_path = save_path / f"{result.entry_id.split('/')[-1]}.pdf"
        if not pdf_path.exists():
            result.download_pdf(dirpath=str(save_path),
                              filename=pdf_path.name)

        text = ""
        with fitz.open(str(pdf_path)) as doc:
            for page in doc:
                text += page.get_text()

        papers.append({
            "title": result.title,
            "arxiv_id": result.entry_id,
            "abstract": result.summary,
            "authors": [a.name for a in result.authors],
            "published": result.published.isoformat(),
            "full_text": text,
            "categories": result.categories
        })
        print(f"  ✓ {result.title[:60]}...")

    return papers

papers = fetch_arxiv_papers(
    query="semiconductor AND (LLM OR deep learning OR anomaly detection)",
    max_results=25
)
print(f"\n取得完了: {len(papers)} 件")

25件で試したら、PyMuPDFのテキスト抽出が思ったより荒い。数式が多い論文はレイアウトが崩壊して、∑ の周辺がゴミ文字列になる。ここは割り切ることにした。数式の正確な抽出が欲しければ Nougat や GROBID を使うべきだが、今回は「検索して関連箇所を見つける」のが目的なので、本文テキストが取れれば十分だ。

ArXiv APIにはレート制限がある。25件なら問題ないが、100件以上引くなら time.sleep(3) を挟まないとBANされる。

BGE-M3を埋め込みモデルに選んだ理由 — 意外と悩んだ

テキストが取れたので次はベクトル化。ここで選択肢がいくつかある。

最初に試したのは multilingual-e5-large。ローカルで動くし多言語対応。だが日本語で質問を投げて英語論文を引っ張ってくる（cross-lingual retrieval）のテストをしたら、ヒット率が微妙だった。体感で3割くらい関係ない論文が上位に来る。

次に BGE-M3 を試した。BAAI（北京智源人工智能研究院）が出している埋め込みモデルで、100言語以上に対応、8192トークンまで入る。これに切り替えたら日本語クエリ→英語論文の検索精度が明らかに上がった。ベンチマーク上でもcross-lingual retrievalはトップクラスで、体感と数字が一致している。

OpenAIの text-embedding-3-small も当然候補だったが、そもそもAPIに投げたくないのがプロジェクトの動機なので却下。

from sentence_transformers import SentenceTransformer
import numpy as np

class BGEEmbedder:
    """BGE-M3 による多言語ベクトル化 (CUDA対応)"""

    def __init__(self, device: str = "cuda"):
        print("BGE-M3 をロード中...")
        self.model = SentenceTransformer(
            "BAAI/bge-m3",
            device=device
        )
        print(f"BGE-M3 ロード完了 (device={device})")

    def encode(self, texts: list[str], batch_size: int = 32) -> np.ndarray:
        embeddings = self.model.encode(
            texts,
            batch_size=batch_size,
            show_progress_bar=True,
            normalize_embeddings=True  # コサイン類似度用に正規化
        )
        return embeddings

RTX 4060でのエンコード速度:

テキスト言語	速度
英語	~1,200 chunks/min
日本語	~950 chunks/min
多言語混在	~820 chunks/min

VRAMはピークで約2.5GB。ここが後で問題になる。

ちなみに上の数字はCUDAで正しく動かした場合の話。最初にBGE-M3を回したとき、エンコードが異様に遅かった。1,000チャンクで10分以上かかる。「BGE-M3ってこんなもんか？」と一瞬思ったが、さすがにおかしい。タスクマネージャーのリソースモニターを開いたらGPU使用率が0%で、CPUが55%程度で回っていた。100%に張り付いていたらもっと早く気づけたと思うが、中途半端に余裕があるせいで「なんか遅いな」程度の違和感しかなかった。device="cuda" を指定しているのに、実際にはCPUで実行されていた。

原因はPyTorchのCUDAビルドが入っていなかったこと。pip install torch だけだとプラットフォームによってはCPU版が入る。PyTorchは torch.cuda.is_available() が False でもエラーを吐かずにCPUフォールバックするので、コード側からは何も異常が見えない。SentenceTransformer もログに Using device: cpu と出してはいるが、大量のログに埋もれて見逃していた。リソースモニターを開く習慣がなかったら、CPUのまま「ローカルRAGは遅い」と結論づけていたと思う。pip install torch --index-url https://download.pytorch.org/whl/cu121 でCUDAビルドに入れ替えたら、同じ処理が65秒で終わった。文字通り10倍の差。

RAG用チャンキング設計で思ったより苦労した

ベクトル化の前にテキストをチャンクに切る必要がある。最初は「512トークンで等分すればいいだろ」と雑にやったら、検索結果がいまいちだった。

原因はチャンクの境界がセクションの途中でぶった切られること。「Methods」セクションの後半と「Results」セクションの前半が1チャンクに混在すると、どっちのクエリにも中途半端にしかヒットしない。

結局、オーバーラップを50ワード入れて、短すぎるチャンク（50ワード未満）は捨てる方式に落ち着いた。

import re

class PaperChunker:
    def __init__(self, chunk_size: int = 512, overlap: int = 50):
        self.chunk_size = chunk_size
        self.overlap = overlap

    def chunk(self, text: str, metadata: dict) -> list[dict]:
        text = re.sub(r'\n{3,}', '\n\n', text)
        text = re.sub(r' {2,}', ' ', text)

        words = text.split()
        chunks = []

        for i in range(0, len(words), self.chunk_size - self.overlap):
            chunk_words = words[i:i + self.chunk_size]
            if len(chunk_words) < 50:
                continue

            chunk_text = " ".join(chunk_words)
            chunks.append({
                "text": chunk_text,
                "metadata": {
                    **metadata,
                    "chunk_index": len(chunks),
                    "word_count": len(chunk_words)
                }
            })

        return chunks

chunker = PaperChunker(chunk_size=512, overlap=50)
all_chunks = []

for paper in papers:
    meta = {
        "title": paper["title"],
        "arxiv_id": paper["arxiv_id"],
        "authors": ", ".join(paper["authors"][:3]),
        "published": paper["published"]
    }
    chunks = chunker.chunk(paper["full_text"], meta)
    all_chunks.extend(chunks)

print(f"総チャンク数: {len(all_chunks)}")

25論文で約1,000チャンク。ここでもう一つ気づいたことがある。Abstractを独立チャンクとして別途追加すると検索精度がかなり変わる。 Abstractは論文全体の要約として情報密度が高く、クエリとの意味的な距離が近くなりやすい。これは後から入れたハックだが、効果は大きかった。

for paper in papers:
    abstract_chunk = {
        "text": f"Title: {paper['title']}\nAbstract: {paper['abstract']}",
        "metadata": {**meta, "chunk_type": "abstract"}
    }
    all_chunks.append(abstract_chunk)

ChromaDBをベクトルDBに採用した経緯 — FAISSかQdrantか

ベクトルの保存先。候補は3つあった。

FAISS: Metaの定番。速い。でも永続化を自分で書かないといけない（pickle or ファイルI/O）
Qdrant: 高機能で好印象だったが、Dockerを立てないといけない。このプロジェクトにはオーバーキル
ChromaDB: PersistentClient で1行で永続化。軽い

「週末で組み終わりたい」というモチベーションだったので、セットアップが最も軽いChromaDBにした。

import chromadb
from chromadb.config import Settings

class PaperVectorStore:
    def __init__(self, persist_dir: str = "./chroma_db"):
        self.client = chromadb.PersistentClient(
            path=persist_dir,
            settings=Settings(anonymized_telemetry=False)
        )
        self.collection = self.client.get_or_create_collection(
            name="arxiv_papers",
            metadata={"hnsw:space": "cosine"}
        )

    def add_chunks(self, chunks: list[dict], vectors: np.ndarray):
        ids = [f"chunk_{i}" for i in range(len(chunks))]
        documents = [c["text"] for c in chunks]
        metadatas = [c["metadata"] for c in chunks]

        batch_size = 500
        for i in range(0, len(ids), batch_size):
            end = min(i + batch_size, len(ids))
            self.collection.add(
                ids=ids[i:end],
                documents=documents[i:end],
                embeddings=vectors[i:end].tolist(),
                metadatas=metadatas[i:end]
            )

        print(f"ChromaDB に {len(ids)} チャンクを保存")

    def search(self, query_vector: np.ndarray, top_k: int = 5) -> list[dict]:
        results = self.collection.query(
            query_embeddings=query_vector.tolist(),
            n_results=top_k,
            include=["documents", "metadatas", "distances"]
        )

        hits = []
        for i in range(len(results["ids"][0])):
            hits.append({
                "text": results["documents"][0][i],
                "metadata": results["metadatas"][0][i],
                "similarity": 1 - results["distances"][0][i]
            })

        return hits

embedder = BGEEmbedder(device="cuda")
texts = [c["text"] for c in all_chunks]
vectors = embedder.encode(texts, batch_size=32)

store = PaperVectorStore(persist_dir="./paper_rag_db")
store.add_chunks(all_chunks, vectors)

ここまでは順調だった。25論文、約1,000チャンクのインデックス構築が1分半で完了。

で、ChromaDBのAPIに文句がある。 query の返り値が results["documents"][0][i] というネスト構造になっていて、なぜ1段余分にリストが挟まっているのか。複数クエリのバッチ対応用の次元らしいが、単一クエリしか投げない場面が大半なのに毎回 [0] を書かされるのは純粋にイライラする。

あと anonymized_telemetry=False を明示しないとテレメトリが飛ぶのもどうかと思う。ローカルで完結させたくてChromaDB使ってるのに、黙ってデータを送信するのは思想が矛盾している。0.5系でも変わっていなかった。

ただ、永続化の手軽さとHNSWの検索速度は文句なしに良い。1,000チャンクの検索が1ms未満。ディスク使用量も50MB程度。不満はあるが乗り換えるほどではない。

ローカルRAGの回答生成 — Qwen2.5-32Bに検索結果を渡す

検索ができたので、あとは検索結果をQwen2.5-32Bに渡して回答を生成する。llama-serverがOpenAI互換APIで立ち上がっているので、OpenAI SDKをそのまま使える。

from openai import OpenAI

class LocalPaperRAG:
    def __init__(self, store: PaperVectorStore,
                 embedder: BGEEmbedder,
                 llm_base_url: str = "http://localhost:8080/v1"):
        self.store = store
        self.embedder = embedder
        self.llm = OpenAI(base_url=llm_base_url, api_key="dummy")

    def ask(self, question: str, top_k: int = 5,
            max_tokens: int = 1024) -> dict:
        q_vec = self.embedder.encode([question])
        hits = self.store.search(q_vec, top_k=top_k)

        context_parts = []
        sources = set()
        for hit in hits:
            title = hit["metadata"].get("title", "Unknown")
            sources.add(title)
            context_parts.append(
                f"[論文: {title}]\n{hit['text']}"
            )
        context = "\n\n---\n\n".join(context_parts)

        response = self.llm.chat.completions.create(
            model="qwen2.5-32b",
            messages=[
                {
                    "role": "system",
                    "content": (
                        "あなたは論文読解の専門家です。"
                        "提供されたコンテキストに基づいて正確に回答してください。"
                        "コンテキストに情報がない場合は「該当する情報が見つかりません」と答えてください。"
                        "回答の根拠となる論文タイトルを必ず明記してください。"
                    )
                },
                {
                    "role": "user",
                    "content": f"## コンテキスト\n{context}\n\n## 質問\n{question}"
                }
            ],
            max_tokens=max_tokens,
            temperature=0.3
        )

        return {
            "answer": response.choices[0].message.content,
            "sources": list(sources),
            "chunks_used": len(hits),
            "tokens": response.usage.total_tokens
        }

rag = LocalPaperRAG(store=store, embedder=embedder)

temperature=0.3 にしているのは、論文ベースの回答にハルシネーションが混ざると致命的だから。温度を下げるのがいちばん手っ取り早い抑止策だ。

Qwen2.5-32Bを使っている理由はシンプルで、前回の記事でRTX 4060での動作実績があるから。Qwen3.5-27Bがリリースされているのでそっちも試したいが、RAG用途での回答品質比較がまだできていない。27Bならパラメータ数が減ってVRAMに余裕が出るし、アーキテクチャ改良で品質も上がっているはず。次の記事のネタだ。

VRAM 8GBでRAGを動かす壁にぶつかる

ここで最大の問題が発生した。

BGE-M3をCUDAでロードすると約2.5GBのVRAMを食う。Qwen2.5-32Bがngl=60で7.6GB使っている。合計10.1GB。8GBのカードに入るわけがない。

最初「両方同時にGPUで動かせばいいだろ」と思っていた自分が甘かった。

結局こうした：

# インデックス構築時: llama-serverを止めてBGE-M3にGPUを使わせる
# → エンコード完了後、llama-serverを再起動
# 検索時: 1クエリだけのベクトル化はCPUで十分速い（数十ms）
embedder_for_query = BGEEmbedder(device="cpu")

時分割だ。インデックス構築（バッチエンコード）は頻繁にやるものじゃないから、そのときだけLLMサーバーを止めてBGE-M3にGPU全振り。検索フェーズでは新しいクエリのベクトル化が1件だけなのでCPUで十分。

もう一つの方法として、BGE-M3を最初からCPUで動かす手もある。速度は約1/4に落ちて、1,000チャンクのエンコードが4分以上かかるようになるが、llama-serverを止めなくて済む。我慢できるならこっちのほうが運用は楽。

ArXiv論文検索の精度を実際に確かめる

result = rag.ask("半導体故障解析(FA)でLLMを活用する際の主な課題は何か？")
print(f"回答:\n{result['answer']}\n")
print(f"参照論文: {result['sources']}")

result = rag.ask("時系列異常検知においてN-BEATSとTransformerベースの手法の違いは？")
print(f"回答:\n{result['answer']}\n")

# BGE-M3の真価: 日本語クエリで英語論文を引く
result = rag.ask("半導体製造プロセスにおけるAI活用の最新動向は？")
print(f"回答:\n{result['answer']}\n")

3つ目の質問が面白い結果を出した。日本語で聞いているのに、英語論文の該当箇所がちゃんとTop-5に入ってくる。BGE-M3を選んだ甲斐があった。

パフォーマンスの数字:

処理	時間
論文25本のテキスト抽出	~30秒
チャンキング (1,050チャンク)	<1秒
BGE-M3 エンコード	~65秒
ChromaDB 保存	~2秒
インデックス構築合計	~100秒
クエリ (検索+回答生成)	~55秒

55秒のうち54秒はQwen2.5-32Bの生成待ち。検索自体は1ms未満。ここを速くしたければ max_tokens を絞るか7Bクラスに落とすかだが、論文の技術的な内容を扱うなら32Bの回答品質は7Bと比べて段違いなので、待つ価値はある。

ローカルRAG構築の全体像 — 200行で動くシステム

結果的にできあがったシステムは以下の構成になった:

構成要素	役割	VRAM
BGE-M3	多言語ベクトル化	~2.5GB (バッチ時)
ChromaDB	ベクトル検索・永続化	ディスクのみ
Qwen2.5-32B (llama.cpp)	回答生成	~7.6GB (ngl=60)

コード全体で約200行。依存ライブラリ6個。丸一日かからず動くところまで持っていけた。

最初は「ローカルRAGなんて大変そうだ」と思っていたが、実際に組んでみると個々のパーツは枯れた技術の組み合わせで、いちばん苦労したのはVRAMのやりくりというハードウェア制約の部分だった。ソフトウェアのレイヤーでは、BGE-M3もChromaDBもllama.cppも、それぞれ十分に安定している。

VRAM 8GB環境でのRAG改善 — 次に潰したい課題

HyDE（Hypothetical Document Embeddings）が気になっている。クエリをそのままベクトル化するんじゃなくて、一度LLMに「この質問に答える架空の文書」を生成させてからベクトル化する。クエリと文書の意味空間のギャップを埋められるらしい。ただ、LLMを1回余分に叩くので応答時間が倍になる。8GBのVRAMで55秒が110秒——さすがにきつい。

BGE-M3にはDense以外にSparseとColBERTのスコアリングもあるのに、今はDenseしか使っていない。Multi-vector retrievalを入れれば精度は上がるはずだが、ChromaDBがSparseベクトルをネイティブサポートしていないので別のストレージが必要になる。ここを回す方法を考え中。

あとQwen3.5-27Bへの乗り換え検証。27Bなら8GBでもnglを攻められるし、9Bとの品質差も気になる。モデルサイズ別のRAG回答品質比較——これは単体で1本書けるネタなのでまた別の記事で。