はじめに
以前の記事「GPUなしで作るローカルRAG入門（LM Studio + LangChain + Chroma）」では、TXT / Markdown / PDFを対象にしたオフライン動作のRAGシステムを構築しました。
本記事では、前回の構成に対して以下2点の改善を加えます。
- 対応ファイル形式の拡張: docx / xlsx / pptx / csv に対応
- 検索精度の向上: BM25とベクトル検索を組み合わせたHybrid Searchの導入
前回記事の構成が前提となります。変更対象は ingest.py → ingest_v2.py、rag.py → rag_hybrid.py の2ファイルです。
追加パッケージ
pip install docx2txt openpyxl python-pptx rank-bm25
前回からの追加は以上の4パッケージのみです。
ファイル形式の拡張（ingest_v2.py）
対応形式の整理
| 形式 | ローダー | ロード単位 |
|---|---|---|
| docx |
Docx2txtLoader（LangChain標準） |
ファイル全体で1ドキュメント |
| csv |
CSVLoader（LangChain標準） |
1行で1ドキュメント |
| xlsx | カスタム実装（openpyxl） | 1行で1ドキュメント（複数シート対応） |
| pptx | カスタム実装（python-pptx） | 1スライドで1ドキュメント |
xlsx と pptx については、LangChainに UnstructuredExcelLoader 等の標準ローダーが存在しますが、unstructured は依存パッケージが非常に重く、オフライン環境との相性も悪いため、openpyxl と python-pptx を直接使ったカスタム実装を採用しています。
ハマりどころ
CSVのBOM問題
ExcelでCSVを「名前を付けて保存」すると、BOM付きUTF-8（utf-8-sig）になります。CSVLoader のデフォルトは encoding='utf-8' のため、そのままだと先頭フィールド名に \ufeff が混入し、インデックスに登録されても検索でヒットしなくなります。
# encoding='utf-8' のままだと先頭フィールドが壊れる
\ufeffインシデントID: INC-2024-0001
# encoding='utf-8-sig' にすれば正常
インシデントID: INC-2024-0001
encoding='utf-8-sig' を指定することで解消できます。
XLSXのタイトル行問題
Excelファイルの先頭行にシートタイトル（「社内ナレッジ — ユーザー台帳」等）が入っているケースがあります。rows[0] を無条件にヘッダーとして扱うと、フィールド名が壊れた状態でインデックスに登録されます。
# タイトル行をヘッダーとして誤認識した場合
社内ナレッジ — ユーザー台帳: USR-005
: 山田 花子
# 正しくタイトル行をスキップした場合
ユーザーID: USR-005
氏名: 山田 花子
「1セルのみ値がある先頭行」をタイトル行と判定してスキップするロジックを追加しています。
Officeの一時ロックファイル問題（Windows）
WordやPowerPointでファイルを開いている間、Windowsは ~$rag_sample.pptx のような一時ロックファイルを自動生成します。path.rglob("*") でこのファイルを拾ってしまうと PackageNotFoundError でクラッシュします。ループの先頭で ~$ 始まりのファイルをスキップすることで回避できます。
コード（ingest_v2.py）
import os
os.environ["HF_HUB_OFFLINE"] = "1"
from pathlib import Path
import openpyxl
from pptx import Presentation as PptxPresentation
from langchain_core.documents import Document
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import (
TextLoader,
PyPDFLoader,
Docx2txtLoader,
CSVLoader,
)
DOCS_DIR = "./docs"
VECTOR_DIR = "./vector_db"
EMBED_MODEL = "intfloat/multilingual-e5-large"
def _is_title_row(row: tuple) -> bool:
"""
タイトル行の判定:
最初のセルのみ値があり、残りがすべてNoneの場合はタイトル行とみなす
（例: '社内ナレッジ — ユーザー台帳', None, None, ...）
"""
non_none = [v for v in row if v is not None]
return len(non_none) == 1
def load_xlsx(filepath: str) -> list[Document]:
"""
Excelの各行を1ドキュメントとして読み込む。
シートが複数ある場合はすべて対象とする。
先頭行がタイトル行（1セルのみ記載）の場合は2行目をヘッダーとして扱う。
"""
wb = openpyxl.load_workbook(filepath)
docs = []
for sheet_name in wb.sheetnames:
ws = wb[sheet_name]
rows = list(ws.iter_rows(values_only=True))
if len(rows) < 2:
continue
if _is_title_row(rows[0]):
header_row = rows[1]
data_rows = rows[2:]
else:
header_row = rows[0]
data_rows = rows[1:]
headers = [str(h) if h is not None else "" for h in header_row]
for row in data_rows:
if all(v is None for v in row):
continue
row_text = "\n".join(
f"{h}: {v}" for h, v in zip(headers, row)
if v is not None and h
)
if row_text.strip():
docs.append(Document(
page_content=row_text,
metadata={"source": filepath, "sheet": sheet_name},
))
return docs
def load_pptx(filepath: str) -> list[Document]:
"""PowerPointの各スライドを1ドキュメントとして読み込む。"""
prs = PptxPresentation(filepath)
docs = []
for i, slide in enumerate(prs.slides, 1):
texts = []
for shape in slide.shapes:
if shape.has_text_frame:
for para in shape.text_frame.paragraphs:
t = para.text.strip()
if t:
texts.append(t)
if texts:
docs.append(Document(
page_content="\n".join(texts),
metadata={"source": filepath, "slide": i},
))
return docs
def load_documents(docs_dir: str) -> list[Document]:
docs = []
path = Path(docs_dir)
for file in path.rglob("*"):
# Officeの一時ロックファイルをスキップ（~$で始まるファイル）
if file.name.startswith("~$"):
continue
if file.suffix == ".txt":
loader = TextLoader(str(file), encoding="utf-8")
docs.extend(loader.load())
print(f" [TXT] {file}")
elif file.suffix == ".md":
loader = TextLoader(str(file), encoding="utf-8")
docs.extend(loader.load())
print(f" [MD] {file}")
elif file.suffix == ".pdf":
loader = PyPDFLoader(str(file))
docs.extend(loader.load())
print(f" [PDF] {file}")
elif file.suffix == ".docx":
loader = Docx2txtLoader(str(file))
docs.extend(loader.load())
print(f" [DOCX] {file}")
elif file.suffix in (".xlsx", ".xlsm"):
docs.extend(load_xlsx(str(file)))
print(f" [XLSX] {file}")
elif file.suffix == ".pptx":
docs.extend(load_pptx(str(file)))
print(f" [PPTX] {file}")
elif file.suffix == ".csv":
# encoding='utf-8-sig' でBOM付きUTF-8（Excelで保存したCSV）に対応
loader = CSVLoader(str(file), encoding="utf-8-sig")
docs.extend(loader.load())
print(f" [CSV] {file}")
return docs
def split_documents(docs: list[Document]) -> list[Document]:
splitter = RecursiveCharacterTextSplitter(
chunk_size=400,
chunk_overlap=80,
separators=["\n\n", "\n", "。", "、", " ", ""],
)
return splitter.split_documents(docs)
def build_vectordb(chunks: list[Document]) -> Chroma:
embeddings = HuggingFaceEmbeddings(
model_name=EMBED_MODEL,
model_kwargs={"device": "cpu"},
encode_kwargs={"normalize_embeddings": True},
)
vectordb = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory=VECTOR_DIR,
)
return vectordb
if __name__ == "__main__":
print("=== ドキュメントロード ===")
docs = load_documents(DOCS_DIR)
print(f"ロード完了: {len(docs)} ドキュメント\n")
print("=== チャンク分割 ===")
chunks = split_documents(docs)
print(f"チャンク数: {len(chunks)}\n")
print("=== VectorDB構築 ===")
vectordb = build_vectordb(chunks)
print(f"登録完了: {vectordb._collection.count()} チャンク\n")
print("=== 登録内容サンプル ===")
for i, chunk in enumerate(chunks[:3]):
print(f"\n[{i + 1}] {chunk.page_content[:80]}...")
検索精度の向上（rag_hybrid.py）
なぜベクトル検索だけでは不十分か
前回の構成（ベクトル検索のみ）で以下のクエリを試した結果です。
質問: INC-2024-0004はどんな障害ですか
回答: ドキュメントに記載がありません
質問: 課題・リスクは
回答: ドキュメントに記載がありません
ドキュメントには INC-2024-0004 の情報が登録されているにもかかわらず、ヒットしませんでした。
原因はベクトル検索の特性にあります。ベクトル検索は意味的な類似度で検索するため、INC-2024-0004 のようなIDコードや 課題・リスク のような短いキーワードとの類似度計算が弱くなります。こうしたキーワード一致が重要なケースにはBM25が有効です。
BM25とHybrid Search
BM25（Best Match 25）は語の出現頻度と文書頻度に基づくキーワード検索アルゴリズムです。IDコードや固有名詞との一致に強く、ベクトル検索と相補的な関係にあります。
今回はBM25とベクトル検索の結果をRRF（Reciprocal Rank Fusion）で統合します。RRFは各リストの順位から score = 1 / (k + rank) を算出して合算する手法です（k=60 は論文推奨値）。スコアの絶対値に依存しないため、BM25とベクトル検索のスコールスケール差を調整する必要がなく、ハイパーパラメータのチューニングが不要です。
ハマりどころ
BM25のデフォルトトークナイザが日本語非対応
BM25Retriever のデフォルト実装は text.split()（スペース区切り）です。日本語はスペースなしで連続するため、INC-2024-0004はどんな障害ですか が1トークンの塊になり、コーパス側の INC-2024-0004 と一致しません。結果としてBM25が何もヒットさせられず、ベクトル検索のみで動いていた前回と実質的に同じ動作になります。
# デフォルトのトークナイズ結果
"INC-2024-0004はどんな障害ですか" → ['INC-2024-0004はどんな障害ですか'] # 1トークンになってしまう
"課題・リスクは" → ['課題・リスクは']
preprocess_func にカスタムトークナイザを渡すことで解決します。
def japanese_tokenizer(text: str) -> list[str]:
"""
BM25用日本語トークナイザ。追加パッケージ不要。
- ASCII英数字（IDコード等）はそのまま抽出
- 日本語テキストは文字バイグラム（2文字）で分割
"""
ascii_tokens = re.findall(r'[A-Za-z0-9][A-Za-z0-9\-_]*', text)
jp_text = re.sub(r'[A-Za-z0-9\s]', '', text)
jp_text = re.sub(r'[\x00-\x7F]', '', jp_text)
bigrams = [jp_text[i:i+2] for i in range(len(jp_text) - 1)]
return ascii_tokens + bigrams
トークナイズの動作例です。
"INC-2024-0004はどんな障害ですか" → ['INC-2024-0004', 'はど', 'どん', 'んな', 'な障', '障害', ...]
"課題・リスクは" → ['課題', '題・', '・リ', 'リス', 'スク', 'クは']
IDコードはそのまま1トークンとして抽出され、コーパス側と一致するようになります。MeCab等の形態素解析器を使えばより高精度になりますが、追加パッケージが必要になるため、今回はオフライン構成のシンプルさを優先してバイグラム方式を採用しています。
動作確認
同一クエリで変更前後の結果を比較しました。
変更前（ベクトル検索のみ）
質問: INC-2024-0004はどんな障害ですか
回答: ドキュメントに記載がありません
質問: 課題・リスクは
回答: ドキュメントに記載がありません
変更後（Hybrid Search）
質問: INC-2024-0004はどんな障害ですか
回答: インシデントID: INC-2024-0004のタイトルは「VPN接続タイムアウト」です。
質問: 課題・リスクは
回答: 課題・リスクは以下の通りです。
【指摘①】APIトークンのハードコーディングリスク
標準テンプレートの一部にAPIキーを直接埋め込む実装例が含まれていました。（以下略）
IDコードや短いキーワードによるクエリでもヒットするようになりました。
コード（rag_hybrid.py）
import os
os.environ["HF_HUB_OFFLINE"] = "1"
import re
import httpx
from typing import List
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_chroma import Chroma
from langchain_community.retrievers import BM25Retriever
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
_embeddings = None
_vectordb = None
_chain = None
def japanese_tokenizer(text: str) -> list[str]:
"""
BM25用日本語トークナイザ。追加パッケージ不要。
- ASCII英数字（IDコード等）はそのまま抽出
- 日本語テキストは文字バイグラム（2文字）で分割
"""
ascii_tokens = re.findall(r'[A-Za-z0-9][A-Za-z0-9\-_]*', text)
jp_text = re.sub(r'[A-Za-z0-9\s]', '', text)
jp_text = re.sub(r'[\x00-\x7F]', '', jp_text)
bigrams = [jp_text[i:i+2] for i in range(len(jp_text) - 1)]
return ascii_tokens + bigrams
class HybridRetriever(BaseRetriever):
"""
BM25とベクトル検索をRRF（Reciprocal Rank Fusion）で統合するRetriever。
langchainのバージョンによってインポートパスが変わるEnsembleRetrieverへの
依存を避けるため、同等の処理を自前実装している。
"""
bm25: BM25Retriever
vector_retriever: object
k: int = 5
def _get_relevant_documents(
self, query: str, *, run_manager: CallbackManagerForRetrieverRun
) -> List[Document]:
bm25_docs = self.bm25.invoke(query)
vector_docs = self.vector_retriever.invoke(query)
return self._rrf_merge(bm25_docs, vector_docs)
def _rrf_merge(
self,
bm25_docs: List[Document],
vector_docs: List[Document],
rrf_k: int = 60,
) -> List[Document]:
scores: dict[str, float] = {}
doc_map: dict[str, Document] = {}
for rank, doc in enumerate(bm25_docs):
key = doc.page_content
scores[key] = scores.get(key, 0) + 1 / (rrf_k + rank + 1)
doc_map[key] = doc
for rank, doc in enumerate(vector_docs):
key = doc.page_content
scores[key] = scores.get(key, 0) + 1 / (rrf_k + rank + 1)
doc_map[key] = doc
ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)
return [doc_map[key] for key, _ in ranked[: self.k]]
def _chroma_to_documents(vectordb: Chroma) -> List[Document]:
"""ChromaDBの全ドキュメントをDocumentリストに変換する。"""
result = vectordb.get()
return [
Document(page_content=text, metadata=meta)
for text, meta in zip(result["documents"], result["metadatas"])
]
def get_rag_chain():
global _embeddings, _vectordb, _chain
if _chain is not None:
return _chain
# 1. Embedding（初回のみweightsロード）
_embeddings = HuggingFaceEmbeddings(
model_name="intfloat/multilingual-e5-large",
model_kwargs={"device": "cpu"},
encode_kwargs={"normalize_embeddings": True},
)
# 2. VectorDB
_vectordb = Chroma(
persist_directory="./vector_db",
embedding_function=_embeddings,
)
# 3. Hybrid Retriever（BM25 + ベクトル検索）
vector_retriever = _vectordb.as_retriever(
search_type="mmr",
search_kwargs={"k": 3, "fetch_k": 10},
)
all_docs = _chroma_to_documents(_vectordb)
bm25_retriever = BM25Retriever.from_documents(
all_docs,
k=3,
preprocess_func=japanese_tokenizer,
)
retriever = HybridRetriever(
bm25=bm25_retriever,
vector_retriever=vector_retriever,
k=5,
)
# 4. Prompt
prompt = ChatPromptTemplate.from_messages([
("system",
"あなたは社内ナレッジベースのアシスタントです。\n"
"【厳守ルール】\n"
"1. 以下のコンテキストに書かれている情報だけを使って回答すること\n"
"2. コンテキストに存在しない情報は絶対に追加しないこと\n"
"3. 推測・補完・一般知識による回答は禁止\n"
"4. コンテキストに情報がない場合は「ドキュメントに記載がありません」とだけ答えること\n"
"5. 回答はコンテキストの表現をそのまま使うこと\n\n"
"コンテキスト:\n{context}"),
("human", "{question}"),
])
# 5. LLM（LM Studio互換API）
llm = ChatOpenAI(
base_url="http://localhost:1234/v1",
api_key="lm-studio",
model="qwen2.5-7b-instruct@q5_k_m",
temperature=0.1,
max_tokens=512,
max_retries=1,
http_client=httpx.Client(timeout=180.0),
)
# 6. Chain
_chain = (
{"context": retriever | _format_docs, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
print("[RAG] 初期化完了（Hybrid Search: BM25 + ベクトル検索）")
return _chain
def _format_docs(docs: List[Document]) -> str:
seen = set()
unique_docs = []
for doc in docs:
if doc.page_content not in seen:
seen.add(doc.page_content)
unique_docs.append(doc)
result = []
total = 0
for doc in unique_docs:
if total + len(doc.page_content) > 1500:
break
result.append(doc.page_content)
total += len(doc.page_content)
return "\n\n---\n\n".join(result)
if __name__ == "__main__":
chain = get_rag_chain()
print("終了するには 'exit' または 'quit' と入力してください。\n")
while True:
try:
q = input("\n質問: ").strip()
if not q:
continue
if q.lower() in ("exit", "quit"):
print("終了します。")
break
answer = chain.invoke(q)
print(f"\n回答: {answer}")
except KeyboardInterrupt:
print("\n終了します。")
break
実行手順
# VectorDBを削除して再インデックス
rm -r vector_db
python ingest_v2.py
# Hybrid Searchで起動
python rag_hybrid.py
まとめ
今回加えた変更を整理します。
ファイル形式の拡張
| 追加形式 | 主な注意点 |
|---|---|
| docx | 特になし |
| csv |
encoding='utf-8-sig' でBOM付きUTF-8（Excelで保存したCSV）に対応する |
| xlsx | 先頭行がタイトル行の場合は2行目をヘッダーとして扱う |
| pptx |
~$ 始まりの一時ロックファイルをスキップする（Windows） |
Hybrid Search
| 変更点 | 内容 |
|---|---|
| BM25トークナイザ | デフォルト（スペース区切り）から日本語対応のバイグラム方式に変更 |
| Retriever統合 | BM25とベクトル検索の結果をRRFで統合 |
| 効果 | IDコードや短いキーワードのクエリでもヒットするようになった |