この記事は全面的にLLMを用いて作成しています。
動作確認した上での記録です。
FAISS＋BM25＋ローカルLLMで自前のナレッジベースを作る
はじめに
社内の共有フォルダやNASには、膨大なWord/Excel/PowerPointファイルが眠っています。
「必要な情報がどこにあるかわからない」「検索しても目的の資料がヒットしない」…
そんな悩みを解消するために、RAG（Retrieval Augmented Generation） を自前で構築してみました。
この記事では、以下の構成で「ローカルに完結する文書検索システム」を実装する手順を紹介します。
- LibreOffice を利用したOffice文書のテキスト抽出
- チャンク化＋ノイズ除去で検索精度を高める前処理
- 埋め込み生成＋FAISS でベクトル検索
- BM25 を併用したハイブリッド検索
- LLMに渡すプロンプト生成まで
最終的には、ローカルLLM（例: gpt-oss-20b）に繋げれば、
自分だけのChatGPTライクな検索エンジンが完成します。
⚠ 注意
本記事のコードは、社内・個人のローカル環境での利用を想定しています。
実際の社内資料や個人情報は含まないダミーデータで試してください。
全体像
0. 環境など（ここだけ人間が記述）
この記事はchatGPT-5でコード作成、自分の環境で動作確認した後にchatGPT-5にまとめさせたものです。
データベースソフトは自分が慣れていないので、使わずRAGやLLMの使い方、出来そうなことを知るのが目的です。
ワードのdocファイルは、ライブラリの都合上docxに変換する必要があります。
オフィス文書の内、doc,docxだけを処理するプロトタイプです。
windowsPC
venv仮想環境でpython3.12
VScode利用（エディター内ターミナルはパワーシェルのvenv環境）
利用ライブラリ
requirements.txt
python-docx==1.1.2
tqdm==4.66.4
pandas==2.2.2
pywin32==306
# 埋め込みベクトル作成用
sentence-transformers
faiss-cpu
numpy
tqdm
# 訳語・略語に強化
rank-bm25
1. Office文書 → テキスト抽出
まずはWord/Excel/PowerPointをテキスト化します。
LibreOfficeをヘッドレスモードで呼び出すと、Officeが無くても安定して変換可能です。
# LibreOfficeの場所確認（Windows）
"C:\Program Files\LibreOffice\program\soffice.exe" --version
Pythonコード：chunk_office_safe.py
以下の処理を行います。
- LibreOfficeでTXT変換
- 不要な空白や全角スペースを削除（位置合わせスペースも除去）
- チャンク化（重複オーバーラップ付き）
-
chunks.jsonl形式で保存
# 主要ポイントのみ抜粋
def normalize_text_jp(s: str) -> str:
"""
日本語文書向けの軽量クリーニング:
- CRLF→LF
- 全角スペース→半角
- 行頭/行末の空白除去
- 連続スペース/タブの圧縮
- 連続空行を1つに
"""
s = s.replace("\r\n", "\n").replace("\r", "\n")
s = s.replace("\u3000", " ")
lines = []
for line in s.split("\n"):
line = line.strip()
line = re.sub(r"[ \t]+", " ", line)
lines.append(line)
return "\n".join(lines).strip()
▶ コード全文を見る
chunk_office_safe.py
# chunk_office_safe.py
import argparse, os, subprocess, json, time, re
from pathlib import Path
from datetime import datetime
from typing import Dict, Iterator, Optional
# ===== 固定パス（LibreOffice） =====
SOFFICE = r"C:\Program Files\LibreOffice\program\soffice.exe"
# ===== 動作パラメータ =====
MAX_SRC_MB = 100 # これ以上の .doc/.docx は変換スキップ
SLEEP_BETWEEN = 0.2 # 各ファイル処理間の小休止（I/Oスパイク抑制）
TXT_FILTER = "txt:Text" # LibreOfficeのTXT出力フィルタ
READ_CHARS_PER_STEP = 1_000_000 # テキスト読み込み単位（文字）
MAX_CHUNKS_PER_FILE = 100_000 # 異常系の暴走防止
OUTPUT_JSONL_ENCODING = "utf-8-sig" # JSONLはBOM付きUTF-8
# ===== 空白・改行 正規化 =====
_WS_MULTI_RE = re.compile(r"[ \t\u3000]{2,}") # 半角/全角スペース/タブの連続
_EOL_SPACE_RE = re.compile(r"[ \t]+(?=\r?\n)") # 行末の空白
_EMPTYLINES_RE = re.compile(r"(?:\r?\n){3,}") # 3行以上の連続空行
def normalize_text_jp(s: str) -> str:
"""
日本語文書向けの軽量クリーニング:
- CRLF→LF
- 全角スペース→半角
- 行頭/行末の空白除去
- 連続スペース/タブ/全角スペースの圧縮（1つに）
- 3連以上の空行を1つに圧縮（段落は保つ）
"""
if not s:
return s
s = s.replace("\r\n", "\n").replace("\r", "\n")
s = s.replace("\u3000", " ")
# 行単位で前後空白を削りつつ内部の長連続空白を圧縮
lines = []
for line in s.split("\n"):
line = line.strip() # 行頭/末の空白を排除（位置合わせスペースを除去）
line = _WS_MULTI_RE.sub(" ", line) # 連続空白を1つに
lines.append(line)
s = "\n".join(lines)
s = _EOL_SPACE_RE.sub("", s) # 行末の空白を削除
s = _EMPTYLINES_RE.sub("\n\n", s) # 空行は最大1つに
return s.strip()
def ensure_soffice() -> bool:
if not os.path.isfile(SOFFICE):
print(f"[ERROR] soffice not found: {SOFFICE}")
return False
return True
def soffice_convert(src: Path, out_dir: Path, filter_str: str) -> Optional[Path]:
out_dir.mkdir(parents=True, exist_ok=True)
dst = out_dir / (src.stem + ".txt")
cmd = [SOFFICE, "--headless", "--convert-to", filter_str, "--outdir", str(out_dir), str(src)]
res = subprocess.run(cmd, capture_output=True, text=True)
if res.returncode == 0 and dst.exists() and dst.stat().st_size > 0:
return dst
print(f"[WARN] convert fail: {src} code={res.returncode} stderr={res.stderr.strip()[:200]}")
return None
def human_mb(n: int) -> float:
try:
return n / (1024 * 1024)
except Exception:
return 0.0
def stream_chunks_from_file(txt_path: Path, encoding: str, max_chars: int, overlap: int) -> Iterator[str]:
"""
テキストをencodingで開き、max_chars上限でオーバーラップ付きに分割して逐次yield。
分割の直前に normalize_text_jp を適用し、不要空白と余計な改行を除去。
"""
buf = ""
produced = 0
errors_mode = "strict"
def yield_piece(piece: str):
nonlocal produced
piece = normalize_text_jp(piece)
if piece:
yield piece
produced += 1
try:
with txt_path.open("r", encoding=encoding, errors=errors_mode, newline="") as f:
while True:
chunk = f.read(READ_CHARS_PER_STEP)
if not chunk:
break
buf += chunk
while len(buf) >= max_chars:
cut = buf.rfind("\n", 0, max_chars)
if cut == -1 or cut < int(max_chars * 0.6):
cut = max_chars
piece = buf[:cut]
for y in yield_piece(piece):
yield y
if produced > MAX_CHUNKS_PER_FILE:
return
start_next = max(0, cut - overlap)
buf = buf[start_next:]
except UnicodeDecodeError:
errors_mode = "replace"
with txt_path.open("r", encoding=encoding, errors=errors_mode, newline="") as f:
while True:
chunk = f.read(READ_CHARS_PER_STEP)
if not chunk:
break
buf += chunk
while len(buf) >= max_chars:
cut = buf.rfind("\n", 0, max_chars)
if cut == -1 or cut < int(max_chars * 0.6):
cut = max_chars
piece = buf[:cut]
for y in yield_piece(piece):
yield y
if produced > MAX_CHUNKS_PER_FILE:
return
start_next = max(0, cut - overlap)
buf = buf[start_next:]
last = buf
piece = normalize_text_jp(last)
if piece:
yield piece
def detect_encoding(txt_path: Path) -> str:
with txt_path.open("rb") as fb:
head = fb.read(128 * 1024)
if head.startswith(b"\xef\xbb\xbf"):
return "utf-8-sig"
if head.startswith(b"\xff\xfe"):
return "utf-16-le"
if head.startswith(b"\xfe\xff"):
return "utf-16-be"
try:
head.decode("utf-8", errors="strict")
return "utf-8"
except UnicodeDecodeError:
pass
try:
head.decode("cp932", errors="strict")
return "cp932"
except UnicodeDecodeError:
pass
return "cp932"
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--input", required=True)
ap.add_argument("--out", required=True)
ap.add_argument("--max-chars", type=int, default=1400)
ap.add_argument("--overlap", type=int, default=200)
args = ap.parse_args()
in_dir = Path(args.input)
out_dir = Path(args.out); out_dir.mkdir(parents=True, exist_ok=True)
conv_dir = out_dir / "_txt"; conv_dir.mkdir(parents=True, exist_ok=True)
if not ensure_soffice():
return
targets = [p for p in in_dir.rglob("*")
if p.is_file() and p.suffix.lower() in (".doc", ".docx")]
filtered = []
for p in targets:
try:
size_mb = human_mb(p.stat().st_size)
except Exception:
size_mb = 0
if size_mb > MAX_SRC_MB:
print(f"[SKIP] large file > {MAX_SRC_MB}MB: {p} ({size_mb:.1f}MB)")
continue
filtered.append(p)
print(f"[INFO] files: {len(targets)} filtered: {len(filtered)} out={conv_dir}")
# 1) 変換
txt_files = []
for i, src in enumerate(filtered, 1):
print(f"[CONVERT] {i}/{len(filtered)} {src}")
dst = soffice_convert(src, conv_dir, TXT_FILTER)
if dst:
txt_files.append(dst)
time.sleep(SLEEP_BETWEEN)
# 2) JSONL 逐次書き出し
jsonl_path = out_dir / "chunks.jsonl"
summary_counts: Dict[str, int] = {}
with jsonl_path.open("w", encoding=OUTPUT_JSONL_ENCODING, newline="\n") as jf:
for i, txt in enumerate(txt_files, 1):
enc = detect_encoding(txt)
print(f"[CHUNK] {i}/{len(txt_files)} {txt.name} (encoding={enc})")
chunk_idx = 0
try:
for piece in stream_chunks_from_file(txt, enc, args.max_chars, args.overlap):
record = {
"file_path": str(txt),
"file_name": txt.name,
"ext": ".txt",
"heading": None,
"chunk_index": chunk_idx,
"text": piece,
"n_chars": len(piece),
"approx_tokens": max(1, len(piece)//2),
}
jf.write(json.dumps(record, ensure_ascii=False) + "\n")
chunk_idx += 1
except MemoryError:
print(f"[WARN] MemoryError while chunking: {txt}. Skipped remainder.")
continue
except Exception as e:
print(f"[WARN] chunk error ({txt}): {e}. Skipped remainder.")
continue
summary_counts[txt.name] = chunk_idx
# 3) summary.csv（件数のみ）
try:
import pandas as pd
rows = [{"file_name": k, "chunks": v} for k, v in summary_counts.items()]
from datetime import datetime as _dt
df = pd.DataFrame(rows)
df["generated_at"] = _dt.now().isoformat(timespec="seconds")
df.to_csv(out_dir / "summary.csv", index=False, encoding="utf-8-sig")
except Exception as e:
print(f"[WARN] failed to write summary.csv: {e}")
print(f"[DONE] JSONL: {jsonl_path}")
print(f"[DONE] CSV: {out_dir/'summary.csv'}")
if __name__ == "__main__":
main()
# 使い方 python .\chunk_office_safe.py --input .\tests --out .\output
実行例：
python .\chunk_office_safe.py --input .\tests --out .\output
結果フォルダ output/ には次が生成されます。
-
chunks.jsonl…… チャンク化された本文データ -
_txt/…… LibreOfficeが生成した中間TXT -
summary.csv…… ファイルごとのチャンク数サマリ
2. ベクトル検索（FAISS）
次に、テキストチャンクを埋め込みベクトル化してFAISSに登録します。
モデルは日本語にも強い BAAI/bge-m3 を使用します。
pip install sentence-transformers faiss-cpu numpy tqdm
python .\build_faiss_index.py --input .\output\chunks.jsonl --outdir .\index --model BAAI/bge-m3
▶ コード全文を見る
build_faiss_index.py
import argparse, json, os, faiss, numpy as np
from pathlib import Path
from typing import Dict, Iterable, List
from tqdm import tqdm
def batched(it: Iterable[dict], n: int):
buf = []
for r in it:
buf.append(r)
if len(buf) >= n:
yield buf
buf = []
if buf:
yield buf
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--input", required=True, help="chunks.jsonl（UTF-8/BOM可）")
ap.add_argument("--outdir", required=True, help="出力フォルダ")
ap.add_argument("--model", default="BAAI/bge-m3")
ap.add_argument("--batch", type=int, default=256)
ap.add_argument("--normalize", action="store_true", default=True)
args = ap.parse_args()
in_path = Path(args.input)
out_dir = Path(args.outdir); out_dir.mkdir(parents=True, exist_ok=True)
# 出力ファイル（すべて BOMなし UTF-8）
faiss_path = out_dir / "vectors.faiss"
meta_path = out_dir / "index_meta.jsonl" # 行ごとのメタ（本文プレビューのみ）
offs_path = out_dir / "index_meta.offsets.npy" # 各行の「開始位置」配列（int64, 0起点）
info_path = out_dir / "index_info.json"
# モデル
from sentence_transformers import SentenceTransformer
model = SentenceTransformer(args.model)
dim = model.get_sentence_embedding_dimension()
# FAISS（内積。正規化すればコサイン）
index = faiss.IndexFlatIP(dim)
# メタは「バイナリ」で開き、start = f.tell() をオフセットとして保存
meta_f = meta_path.open("wb")
offsets: List[int] = []
def gen_records():
# 入力は UTF-8(BOMあり/なし) どちらでもOK
with in_path.open("r", encoding="utf-8-sig", errors="strict") as f:
for line in f:
rec = json.loads(line)
txt = rec.get("text", "")
if not txt:
continue
yield rec
total = 0
for batch in batched(gen_records(), args.batch):
texts = [r["text"] for r in batch]
embs = model.encode(texts, batch_size=args.batch,
normalize_embeddings=True if args.normalize else False)
if isinstance(embs, list):
embs = np.array(embs, dtype="float32")
else:
embs = embs.astype("float32", copy=False)
index.add(embs)
for r in batch:
preview = r.get("text", "").replace("\r", " ").replace("\n", " ")
if len(preview) > 200:
preview = preview[:200] + " ..."
meta = {
"file_path": r.get("file_path"),
"file_name": r.get("file_name"),
"ext": r.get("ext"),
"heading": r.get("heading"),
"chunk_index": r.get("chunk_index"),
"n_chars": r.get("n_chars"),
"approx_tokens": r.get("approx_tokens"),
"preview": preview
}
# 書く前に開始位置を採取（BOMなしUTF-8）
start = meta_f.tell()
s = (json.dumps(meta, ensure_ascii=False) + "\n").encode("utf-8")
meta_f.write(s)
offsets.append(start)
total += len(batch)
meta_f.flush(); meta_f.close()
np.save(offs_path, np.array(offsets, dtype=np.int64))
faiss.write_index(index, str(faiss_path))
info: Dict = {
"model": args.model,
"dim": int(dim),
"normalize": bool(args.normalize),
"count": int(total),
"faiss_index": faiss_path.name,
"meta_file": meta_path.name,
"offsets_file": offs_path.name,
"encoding": "utf-8" # BOMなし
}
info_path.write_text(json.dumps(info, ensure_ascii=False, indent=2), encoding="utf-8")
print(f"[DONE] vectors={total} dim={dim}")
print(f"[OUT] {faiss_path}")
print(f"[OUT] {meta_path}")
print(f"[OUT] {offs_path}")
print(f"[OUT] {info_path}")
if __name__ == "__main__":
main()
3. 文章検索（BM25）
ベクトル検索は意味的な近さを見つけますが、
日本語の固有名詞などは単語一致が有効な場合もあります。
そこで、従来型のBM25も構築してハイブリッド検索を行います。
pip install rank-bm25
python .\build_bm25_index.py --input .\output\chunks.jsonl --out .\index\bm25.pkl
▶ コード全文を見る
build_bm25_index.py
import argparse, json, pickle
from pathlib import Path
from typing import List
from tqdm import tqdm
# 日本語でも効く簡易トークナイザ（文字bi-gram＋英数語）
import re
def tokenize(text: str) -> List[str]:
text = text.strip()
if not text:
return []
# 英数語（ASCIIの連続）はそのまま単語化
words = re.findall(r"[A-Za-z0-9_]+", text)
# それ以外は文字bi-gram（日本語向け、句読点は除外）
s = re.sub(r"[\s\r\n\t]", "", text)
s = re.sub(r"[、。,.，．・/／\-–—:：;；\(\)\[\]{}<>「」『』“”\"\']", "", s)
bigrams = [s[i:i+2] for i in range(len(s)-1)] if len(s) >= 2 else ([s] if s else [])
return words + bigrams
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--input", required=True, help="chunks.jsonl（UTF-8/BOM可）")
ap.add_argument("--out", required=True, help="出力 pickle（例 .\\index\\bm25.pkl）")
ap.add_argument("--max-docs", type=int, default=0, help="先頭N件だけ学習（0=全件）")
args = ap.parse_args()
inp = Path(args.input)
outp = Path(args.out)
outp.parent.mkdir(parents=True, exist_ok=True)
docs_tokens: List[List[str]] = []
meta_min: List[dict] = []
with inp.open("r", encoding="utf-8-sig", errors="strict") as f:
for i, line in enumerate(tqdm(f, desc="Reading chunks.jsonl")):
rec = json.loads(line)
text = rec.get("text", "")
if not text:
continue
docs_tokens.append(tokenize(text))
# 参照に必要な最小メタだけ保持（ファイル名・行番号）
meta_min.append({
"line_no": i, # chunks.jsonl 上の行番号（0起点）
"file_name": rec.get("file_name"),
"chunk_index": rec.get("chunk_index"),
})
if args.max_docs and len(docs_tokens) >= args.max_docs:
break
# Pickle 化（DB不要）
payload = {
"docs_tokens": docs_tokens, # List[List[str]]
"meta_min": meta_min, # List[dict]
"source_jsonl": str(inp.resolve())
}
with outp.open("wb") as pf:
pickle.dump(payload, pf)
print(f"[DONE] BM25 corpus saved: {outp} docs={len(docs_tokens)}")
if __name__ == "__main__":
main()
4. ハイブリッド検索＆RRF融合
- FAISS で「意味検索」上位を取得
- BM25 で「単語一致」上位を取得
- RRF（Reciprocal Rank Fusion）」で順位を融合
5. LLM用プロンプト生成
最後に、Top-kチャンクをまとめてRAGプロンプトを作成します。
このプロンプトをローカルLLMに入力すれば、参照元付きで回答が得られます。
python .\prepare_prompt.py `
--chunks .\output\chunks.jsonl `
--indexdir .\index `
--bm25 .\index\bm25.pkl `
--query "偏光フィルタの特性と測定手順" `
--topk 10 `
--context-chars 6000 `
--per-chunk-max 1200
▶ コード全文を見る
prepare_prompt.py
import argparse
import json
import pickle
import faiss
import numpy as np
import re
from pathlib import Path
from typing import Dict, List, Tuple
from rank_bm25 import BM25Okapi
# ===== プロンプト用 クリーニング =====
_WS_MULTI_RE = re.compile(r"[ \t\u3000]{2,}")
_EMPTYLINES_RE = re.compile(r"(?:\r?\n){3,}")
_EOL_SPACE_RE = re.compile(r"[ \t]+(?=\r?\n)")
def clean_for_prompt(s: str) -> str:
if not s:
return s
s = s.replace("\r\n", "\n").replace("\r", "\n")
s = s.replace("\u3000", " ")
# 行単位で前後空白を削り、連続空白を圧縮
lines = []
for line in s.split("\n"):
line = line.strip()
line = _WS_MULTI_RE.sub(" ", line)
lines.append(line)
s = "\n".join(lines)
s = _EOL_SPACE_RE.sub("", s)
# 3つ以上の空行は1つに（段落は保つ）
s = _EMPTYLINES_RE.sub("\n\n", s)
# 先頭・末尾の空行を除去
return s.strip()
# ===== 日本語向けの軽量トークナイザ（BM25用）=====
import re as _re
def tokenize(text: str) -> List[str]:
text = (text or "").strip()
if not text:
return []
words = _re.findall(r"[A-Za-z0-9_]+", text)
s = _re.sub(r"[\s\r\n\t]", "", text)
s = _re.sub(r"[、。,.，．・/／\\\-–—:：;；\(\)\[\]{}<>「」『』“”\"\']", "", s)
bigrams = [s[i:i+2] for i in range(len(s)-1)] if len(s) >= 2 else ([s] if s else [])
return words + bigrams
# ===== 付帯ユーティリティ =====
def load_info(indexdir: Path) -> Dict:
return json.loads((indexdir / "index_info.json").read_text(encoding="utf-8"))
def open_meta(indexdir: Path):
f = (indexdir / "index_meta.jsonl").open("rb")
offs = np.load(indexdir / "index_meta.offsets.npy")
return f, offs
def read_meta_line(f, offsets: np.ndarray, i: int) -> Dict:
if i < 0 or i >= len(offsets):
return {}
start = int(offsets[i])
f.seek(start)
line = f.readline()
if not line:
return {}
return json.loads(line.decode("utf-8"))
def read_chunk_text_by_lineno(chunks_jsonl: Path, line_no: int) -> Dict:
with chunks_jsonl.open("r", encoding="utf-8-sig") as f:
for i, line in enumerate(f):
if i == line_no:
return json.loads(line)
return {}
def rrf_fusion(faiss_list: List[int], bm25_list: List[int], k: int = 60, topn: int = 20) -> List[int]:
from collections import defaultdict
score = defaultdict(float)
for rank, idx in enumerate(faiss_list):
score[int(idx)] += 1.0 / (k + rank + 1)
for rank, idx in enumerate(bm25_list):
score[int(idx)] += 1.0 / (k + rank + 1)
fused = sorted(score.items(), key=lambda x: x[1], reverse=True)
return [idx for idx, _ in fused[:topn]]
def trim_text(s: str, limit: int) -> str:
if len(s) <= limit:
return s
return s[:limit] + " …"
def build_prompt(query: str, contexts: List[Tuple[Dict, Dict]], max_chars: int) -> str:
header = (
"あなたは技術文書を要約・参照するアシスタントです。以下の「コンテキスト」だけを根拠に、"
"質問に日本語で簡潔に回答し、最後に必ず参照元一覧（ファイル名とチャンク番号）を示してください。\n"
"推測は避け、根拠が足りない場合は『不明』と答えてください。\n"
)
qline = f"\n【質問】\n{clean_for_prompt(query)}\n"
# コンテキストは文書ごとに1行空ける。文書内の連続空行は圧縮済み。
ctx_lines = ["\n【コンテキスト】"]
for i, (meta, rec) in enumerate(contexts, start=1):
fn = rec.get("file_name")
ci = rec.get("chunk_index")
ctx = clean_for_prompt(rec.get("text", ""))
ctx_lines.append(f"--- 文書{i}：{fn}（chunk#{ci}） ---\n{ctx}")
ctx_block = "\n\n".join(ctx_lines) + "\n"
instruct = (
"\n【出力フォーマット】\n"
"回答本文。\n"
"\n"
"【参照元】\n"
"- <ファイル名> chunk#<番号>\n"
"- ...\n"
)
draft = clean_for_prompt(header + qline + ctx_block + instruct)
if len(draft) <= max_chars:
return draft
# 超過時はそのまま（選抜側で制御している想定）
return draft
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--chunks", required=True)
ap.add_argument("--indexdir", required=True)
ap.add_argument("--bm25", required=True)
ap.add_argument("--query", required=True)
ap.add_argument("--topk", type=int, default=10)
ap.add_argument("--context-chars", type=int, default=6000)
ap.add_argument("--per-chunk-max", type=int, default=1200)
args = ap.parse_args()
chunks_jsonl = Path(args.chunks)
indexdir = Path(args.indexdir)
bm25_pkl = Path(args.bm25)
# === FAISS ===
info = load_info(indexdir)
index = faiss.read_index(str(indexdir / info["faiss_index"]))
from sentence_transformers import SentenceTransformer
model = SentenceTransformer(info["model"])
meta_f, offsets = open_meta(indexdir)
# === BM25 ===
payload = pickle.loads(bm25_pkl.read_bytes())
docs_tokens: List[List[str]] = payload["docs_tokens"]
meta_min: List[dict] = payload["meta_min"]
bm25 = BM25Okapi(docs_tokens)
# === 検索 ===
q_emb = model.encode([args.query], normalize_embeddings=bool(info.get("normalize", True))).astype("float32")
D, I = index.search(q_emb, max(args.topk * 5, 20))
faiss_list = [int(x) for x in I[0] if x >= 0]
qtok = tokenize(args.query)
scores = bm25.get_scores(qtok)
bm25_rank = np.argsort(scores)[::-1]
bm25_list = list(bm25_rank[:max(len(faiss_list), args.topk * 5)])
fused_idx = rrf_fusion(faiss_list, bm25_list, k=60, topn=max(args.topk * 3, 30))
# === 選抜（文字数予算＆1チャンクあたりの上限）===
picked: List[Tuple[Dict, Dict]] = []
budget = args.context_chars
for idx in fused_idx:
if idx < 0 or idx >= len(offsets):
continue
meta = read_meta_line(meta_f, offsets, idx)
line_no = meta_min[idx]["line_no"] if idx < len(meta_min) else idx
rec = read_chunk_text_by_lineno(chunks_jsonl, line_no)
if not rec:
continue
text = clean_for_prompt(rec.get("text", ""))
if len(text) > args.per_chunk_max:
text = trim_text(text, args.per_chunk_max)
rec["text"] = text
need = len(text) + 200
if need > budget and picked:
break
picked.append((meta, rec))
budget -= need
if len(picked) >= args.topk:
break
prompt = build_prompt(args.query, picked, args.context_chars)
# === 保存 ===
outdir = Path("prompts"); outdir.mkdir(exist_ok=True)
(outdir / "latest_prompt.txt").write_text(prompt, encoding="utf-8")
bundle = {
"query": args.query,
"prompt": prompt,
"citations": [
{
"file_name": r.get("file_name"),
"chunk_index": r.get("chunk_index"),
"file_path": r.get("file_path"),
}
for (_, r) in picked
],
"model_hint": info["model"],
}
(outdir / "latest_prompt.json").write_text(json.dumps(bundle, ensure_ascii=False, indent=2), encoding="utf-8")
print(f"[DONE] prompts/latest_prompt.txt")
print(f"[DONE] prompts/latest_prompt.json")
if __name__ == "__main__":
main()
生成物：
-
prompts/latest_prompt.txt…… LLMにそのまま入力 -
prompts/latest_prompt.json…… 構造化データ（出典メタ付き）
6. ローカルLLMに接続する
ここまでの流れで準備した latest_prompt.txt を
ローカルLLM（例: gpt-oss-20b）に入力するだけです。
まとめ
| ステップ | ツール | 出力 |
|---|---|---|
| Office→TXT | LibreOffice | _txt/*.txt |
| 正規化＆チャンク化 | Python | chunks.jsonl |
| 埋め込み生成 | SentenceTransformers + FAISS | index/vectors.faiss |
| キーワード検索 | BM25 | index/bm25.pkl |
| プロンプト生成 | Python | prompts/latest_prompt.txt |
おわりに
RAGは「LLMの頭脳」と「検索エンジンの記憶」を組み合わせる強力なアプローチです。
完全ローカルでナレッジ活用ができるため、安心して社内で運用できます。
終わりに（人間が記述）
LLMで初めてのことにも挑戦しやすくなってきました。
今後は、
・ワード以外の文書も処理する。
・データベースを利用する。
・図や写真も検索対象とする。CLIP等を使う？
などの課題があると思います。