背景
Topic Sentence: RAGの性能を最大限に引き出すには、ベクトル検索の精度が鍵を握ります。
Evidence / Data: RAG実装においては、検索精度が低いとLLMが誤った情報を生成するケースがあります。
Reasoning: 特に日本語文書では、チャンクの分割方法や埋め込みモデルの選定が精度に大きく影響します。
Example: 日本語に適した文節単位のチャンク分割を採用すると、回答品質が向上しました。
Wrap-up: 本記事では、日本語文書を対象にしたベクトル検索の実装方法と、精度を高めるためのチャンク戦略を解説します。
対象読者
- Python 中級者
- LLM や RAG の基本概念を理解している方
- 社内ナレッジ検索やFAQボットを構築したいエンジニア
環境
| 項目 | バージョン |
|---|---|
| OS | Ubuntu 22.04 |
| ランタイム | Python 3.10.12 |
| 主要ライブラリ | langchain_community==0.3.24, openai==1.82.0, sudachipy==0.6.10 |
| 埋め込みモデル | text-embedding-3-small |
事前準備
- requirements.txtの準備
langchain_community langchain openai azure-core numpy sudachipy sudachidict-core - 仮想環境の作成とライブラリインストール
python -m venv venv source venv/bin/activate pip install -r requirements.txt
手順
1. 文書の読み込みと前処理
話題文: 日本語文書を読み込み、前処理を行います。
根拠: 文書の構造を理解し、意味のまとまりを保つための準備です。
解説: markdown文書を読み込んでいますが、PDFやTXTなどの形式に応じて読み込み方法を変えることもできます。
具体例:
# langchainのTextLoaderでMarkdownファイルを読み込み
loader = TextLoader(args.input, encoding="utf-8")
documents = loader.load()
markdown_content = "\n".join([doc.page_content for doc in documents])
まとめ: 文書が読み込めたら、次はチャンクに分割します。
2. 日本語に適したチャンク分割
話題文: 文の意味を保ったままチャンクに分割します。
根拠: 英語と違い、日本語はスペースで単語が区切られないため、適切な分割が重要です。
解説: pythonで利用できる形態素解析器としてはMecab,Sudachiが挙げられますが、辞書がpipで一体で提供され扱いやすいSudachiを採用しました。SudachiPyの辞書からトークナイザー（形態素解析器）を作成し、テキストを1行ずつ処理し、「。」「！」「？」などの句点で文ごとに分割、各文をSudachiPyで形態素解析し、指定したSplitModeで分割します。
具体例:
def split_into_chunks_with_sudachipy(text, splitmode="C"):
"""SudachiPyで文節単位に分割し、チャンクリストを返す"""
token_obj = dictionary.Dictionary().create()
mode_map = {
"A": tokenizer.Tokenizer.SplitMode.A,
"B": tokenizer.Tokenizer.SplitMode.B,
"C": tokenizer.Tokenizer.SplitMode.C,
}
mode_key = splitmode.upper()
mode = mode_map.get(mode_key, tokenizer.Tokenizer.SplitMode.C)
print(f"[DEBUG] SudachiPy SplitMode: {mode_key} ({mode})") # デバッグ出力
# 文末の句点で分割
sentences = []
buf = ""
for line in text.splitlines():
for c in line:
buf += c
if c in "。！？\n":
sentences.append(buf.strip())
buf = ""
if buf:
sentences.append(buf.strip())
buf = ""
# 文節ごとに分割
chunks = []
for sentence in sentences:
morphemes = token_obj.tokenize(sentence, mode)
chunk = "".join([m.surface() for m in morphemes])
if chunk:
chunks.append(chunk)
return chunks
まとめ: チャンクができたら、次はベクトル化して検索できるようにします。
3. 埋め込み生成とベクトルストアの構築
話題文: チャンクをベクトルに変換し、検索可能な構造に保存します。
根拠: LLMが参照するためには、意味的に近いチャンクを高速に検索する必要があります。
解説: テキストのリストを受け取り、それぞれのテキストをAzure OpenAIの埋め込みモデルでベクトル化したうえで、チャンクとエンベディングをペアにしてファイルへ保存します。
具体例:
def get_embeddings(texts):
"""テキストリストをエンベディングする"""
client = openai.AzureOpenAI(
api_key=AZURE_OPENAI_API_KEY,
azure_endpoint=AZURE_OPENAI_ENDPOINT,
api_version="2024-10-21"
)
embeddings = []
for text in texts:
response = client.embeddings.create(
input=text,
model=EMBEDDING_MODEL,
)
embeddings.append(response.data[0].embedding)
return embeddings
まとめ: これで検索準備が整いました。次は実際にクエリを投げてみます。
4. クエリによるベクトル検索
話題文: ユーザーの質問に対して、関連するチャンクを検索します。
根拠: RAGの「Retrieval」部分に該当し、生成精度に直結します。
解説: クエリをベクトル化し、類似度の高いチャンクを取得します。
具体例:
def search_similar_chunks(query, vector_file, top_k=3):
"""クエリを埋め込みに変換し、類似チャンクを返す"""
# ベクトルデータの読み込み
with open(vector_file, "r", encoding="utf-8") as f:
vector_data = json.load(f)
# クエリを埋め込みに変換
query_embedding = get_embeddings([query])[0]
# 類似度計算
results = []
for item in vector_data:
score = cosine_similarity(query_embedding, item["embedding"])
results.append({"chunk": item["chunk"], "score": score})
# スコア順にソートして上位を返す
results.sort(key=lambda x: x["score"], reverse=True)
return results[:top_k]
まとめ: 検索結果をLLMに渡すことで、RAGの生成精度が向上します。
結果
- 最終的な実装です。
import os
import json
import argparse
from langchain_community.document_loaders import TextLoader
import openai
import numpy as np
from sudachipy import tokenizer
from sudachipy import dictionary
from langchain.text_splitter import RecursiveCharacterTextSplitter # 追加
# Azure OpenAIの設定
AZURE_OPENAI_ENDPOINT = "https://エンドポイント.openai.azure.com/"
AZURE_OPENAI_API_KEY = "APIキー"
EMBEDDING_MODEL = "text-embedding-3-small"
# 出力ベクトルファイル
OUTPUT_VECTOR_FILE = "vectors.json"
def save_vectors(vectors, file_path):
"""ベクトルをJSONファイルに保存する"""
with open(file_path, "w", encoding="utf-8") as f:
json.dump(vectors, f, ensure_ascii=False, indent=4)
def get_embeddings(texts):
"""テキストリストをエンベディングする"""
client = openai.AzureOpenAI(
api_key=AZURE_OPENAI_API_KEY,
azure_endpoint=AZURE_OPENAI_ENDPOINT,
api_version="2024-10-21"
)
embeddings = []
for text in texts:
response = client.embeddings.create(
input=text,
model=EMBEDDING_MODEL,
)
embeddings.append(response.data[0].embedding)
return embeddings
def cosine_similarity(vec1, vec2):
"""コサイン類似度を計算"""
vec1 = np.array(vec1)
vec2 = np.array(vec2)
return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))
def search_similar_chunks(query, vector_file, top_k=3):
"""クエリを埋め込みに変換し、類似チャンクを返す"""
# ベクトルデータの読み込み
with open(vector_file, "r", encoding="utf-8") as f:
vector_data = json.load(f)
# クエリを埋め込みに変換
query_embedding = get_embeddings([query])[0]
# 類似度計算
results = []
for item in vector_data:
score = cosine_similarity(query_embedding, item["embedding"])
results.append({"chunk": item["chunk"], "score": score})
# スコア順にソートして上位を返す
results.sort(key=lambda x: x["score"], reverse=True)
return results[:top_k]
def split_into_chunks_with_sudachipy(text, splitmode="C"):
"""SudachiPyで文節単位に分割し、チャンクリストを返す"""
token_obj = dictionary.Dictionary().create()
mode_map = {
"A": tokenizer.Tokenizer.SplitMode.A,
"B": tokenizer.Tokenizer.SplitMode.B,
"C": tokenizer.Tokenizer.SplitMode.C,
}
mode_key = splitmode.upper()
mode = mode_map.get(mode_key, tokenizer.Tokenizer.SplitMode.C)
print(f"[DEBUG] SudachiPy SplitMode: {mode_key} ({mode})") # デバッグ出力
# 文末の句点で分割
sentences = []
buf = ""
for line in text.splitlines():
for c in line:
buf += c
if c in "。！？\n":
sentences.append(buf.strip())
buf = ""
if buf:
sentences.append(buf.strip())
buf = ""
# 文節ごとに分割
chunks = []
for sentence in sentences:
morphemes = token_obj.tokenize(sentence, mode)
chunk = "".join([m.surface() for m in morphemes])
if chunk:
chunks.append(chunk)
return chunks
def split_into_chunks_with_langchain(text, chunk_size=500, chunk_overlap=50):
"""langchainのRecursiveCharacterTextSplitterでチャンク分割"""
splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", "。", "？", "！", " "]
)
return splitter.split_text(text)
def main():
# コマンドライン引数の解析
parser = argparse.ArgumentParser(description="Markdownファイルをベクトル化・検索します。")
parser.add_argument(
"--input",
type=str,
default="input.md",
help="入力Markdownファイルのパス (デフォルト: input.md)"
)
parser.add_argument(
"--search",
type=str,
default=None,
help="キーワード検索クエリ"
)
parser.add_argument(
"--topk",
type=int,
default=3,
help="検索時に返す上位件数 (デフォルト: 3)"
)
parser.add_argument(
"--sudachi_splitmode",
type=str,
default="C",
choices=["A", "B", "C"],
help="SudachiPyのSplitMode（A:細かい, B:中間, C:大きい文節, デフォルト:C）"
)
parser.add_argument(
"--chunk_method",
type=str,
default="sudachi",
choices=["sudachi", "langchain"],
help="チャンク化方法: sudachi または langchain (デフォルト: sudachi)"
)
parser.add_argument(
"--chunk_size",
type=int,
default=500,
help="langchainチャンク化時のchunk_size (デフォルト: 500)"
)
parser.add_argument(
"--chunk_overlap",
type=int,
default=50,
help="langchainチャンク化時のchunk_overlap (デフォルト: 50)"
)
args = parser.parse_args()
if args.search:
# ベクトル検索モード
results = search_similar_chunks(args.search, OUTPUT_VECTOR_FILE, top_k=args.topk)
print(f"クエリ: {args.search}")
for i, r in enumerate(results, 1):
print(f"\n--- Top {i} (score={r['score']:.4f}) ---\n{r['chunk']}\n")
return
# langchainのTextLoaderでMarkdownファイルを読み込み
loader = TextLoader(args.input, encoding="utf-8")
documents = loader.load()
markdown_content = "\n".join([doc.page_content for doc in documents])
# チャンク化方法の切り替え
if args.chunk_method == "sudachi":
chunks = split_into_chunks_with_sudachipy(markdown_content, splitmode=args.sudachi_splitmode)
else:
chunks = split_into_chunks_with_langchain(
markdown_content,
chunk_size=args.chunk_size,
chunk_overlap=args.chunk_overlap
)
# チャンクをエンベディング
embeddings = get_embeddings(chunks)
# チャンクとエンベディングをペアにして保存
vector_data = [{"chunk": chunk, "embedding": embedding} for chunk, embedding in zip(chunks, embeddings)]
save_vectors(vector_data, OUTPUT_VECTOR_FILE)
print(f"ベクトルを {OUTPUT_VECTOR_FILE} に保存しました。")
if __name__ == "__main__":
main()
- 実装により、日本語文書から意味のまとまりを保ったチャンクを作成し、ベクトル検索が可能になりました。
例えば、以下のような架空の剣術に関するテキストをベクトル化して、意味の近い部分を抽出できるようになりました。
# 江戸異伝剣法録 〜影に生き、影に斬る〜
**著：民明書房刊**
---
## 序章：剣の時代、影の剣法
江戸時代、泰平の世にあっても、剣は人の心を映す鏡であり続けた。表の剣術が道場で磨かれる一方、裏の剣法は密やかに伝承され、時に幕府の密命を帯び、時に復讐の刃として闇を駆けた。
本書では、歴史の表舞台に現れることのなかった、幻の剣法たちを記録する。
---
## 第一章：霞流影走剣（かすみりゅう・かげばしりけん）
- **創始者**：霞之介（かすみのすけ）
- **発祥地**：肥後国・阿蘇山麓
- **特徴**：霧の中での戦闘を想定した「視覚撹乱剣法」
> 「敵の目を欺くのではない。己の存在を霞ませるのだ」
> 〜霞之介『影走剣口伝』より
霞流は、霧や煙の中での戦闘を想定し、足音を消す歩法「霞歩（かほ）」と、残像を生む斬撃「影走（かげばしり）」を極意とする。幕末には新撰組の一部隊がこの流派を学んでいたという記録もある（※民明書房調べ）。
---
## 第二章：無音一閃流（むおんいっせんりゅう）
- **創始者**：音無斎（おとなしさい）
- **発祥地**：江戸・深川
- **特徴**：完全無音の抜刀術と、音を封じる呼吸法
この流派の剣士は、足音、衣擦れ、息遣いすらも消し、まるで幽霊のように敵に近づく。抜刀と同時に勝負が決するため、「一閃」の名が冠されている。
> 「音がある限り、心は読まれる」
> 〜音無斎『無音剣法秘伝』より
---
## 第三章：逆風斬月流（ぎゃくふうざんげつりゅう）
- **創始者**：風間月影（かざま・つきかげ）
- **発祥地**：信濃国・諏訪湖畔
- **特徴**：風を読む剣法。逆風を利用して斬撃を加速させる。
この流派は、風の流れを読む「風眼（ふうがん）」と、風に逆らって斬る「逆風斬（ぎゃくふうざん）」を極意とする。風間月影は「風の剣士」として諸国を放浪し、数多の剣豪と立ち会ったという。
---
## 終章：剣は心なり
これらの剣法は、いずれも現代には伝わっていない。だが、剣に込められた思想や哲学は、今もなお、武道の根底に息づいている。
> 「剣は斬るためにあらず。己を映す鏡なり」
> 〜民明書房『江戸異伝剣法録』より
---
## 出典
- 民明書房『江戸異伝剣法録』
- 民明書房『影走剣口伝』『無音剣法秘伝』ほか
- クエリに対して関連のある文が返されるようになりました。
例えば、上記文章でクエリとして「静寂」を与えると、次のような結果が返ってきます。
--- Top 1 (score=0.4017) ---
## 第二章：無音一閃流（むおんいっせんりゅう）
--- Top 2 (score=0.3436) ---
- **特徴**：完全無音の抜刀術と、音を封じる呼吸法
--- Top 3 (score=0.2910) ---
この流派の剣士は、足音、衣擦れ、息遣いすらも消し、まるで幽霊のように敵に近づく。
- sudachiによる形態素解析をもちいた手法のほうが、langchainのRecursiveCharacterTextSplitterを用いた方法よりも意味的な類似項を返却してくれることが多いです。
考察
Topic Sentence: チャンク分割の方法を日本語に適したものに変えることで、回答品質が向上しました。
Reasoning: 文脈を保ったチャンクは、LLMがより正確に回答を生成するための鍵となります。
Wrap-up: 今後は、形態素解析ではなくtransformerを用いた手法を試してみたいと考えています。
まとめ & 次のステップ
- 日本語文書におけるRAGの精度は、チャンク分割とベクトル化の工夫で大きく向上します。
- 読者には、自分の持つデータソースに合わせたチャンク戦略を試すことをおすすめします。
Discussion