壊れないRAGナレッジベースを作る：遅延初期化・冪等インジェスト・グレースフルデグラデーション
RAGのプロダクション運用で遍歴した問題とその対処パターンを紹介。ベクトルストアの過延初期化、SHA1による冪等インジェスト、見出し階層保持チャンキング、グレースフルデグレーデーションの5パターンを、Python/ChromaDB/LangChainの実装例とともに解説します。
2026.07.28
はじめに
社内AI自動化ツールにRAG（Retrieval-Augmented Generation）機能を組み込みました。Markdownドキュメントをベクトルストアに取り込み、チャットで質問するとソース付きで回答する仕組みです。
しかしRAGは壊れやすい。ベクトルストアの初期化失敗、エンベディングAPIのタイムアウト、再起動時のチャンク重複など、プロダクション運用で遭遇した問題と、それらに対処するために確立したパターンを紹介します。
前提・環境
- Python 3.12, FastAPI
- ベクトルストア: ChromaDB (ローカル永続化)
- エンベディング: OpenAI
text-embedding-3-small - チャンキング: LangChain
MarkdownHeaderTextSplitter+RecursiveCharacterTextSplitter
パターン1: 遅延初期化（Lazy Initialization）
ベクトルストアの初期化をアプリ起動時ではなく、初回リクエスト時に行うパターンです。
rag/vectorstore.py
def get_chroma(collection: str = KB_COLLECTION) -> Chroma:
pdir = _persist_dir(collection)
os.makedirs(pdir, exist_ok=True)
try:
vec = Chroma(
collection_name=collection,
persist_directory=pdir,
embedding_function=get_embeddings(),
)
# ドキュメント数をログに記録（ベストエフォート）
count = None
try:
underlying = getattr(vec, "_collection", None)
if underlying and hasattr(underlying, "count"):
count = int(underlying.count())
except Exception:
count = None
logger.info(json.dumps({
"message": "chroma_initialized",
"persist_directory": pdir,
"doc_count": count,
}))
return vec
except Exception as e:
logger.error(json.dumps({
"message": "chroma_init_failed",
"error": str(e),
}))
raise
チャットエージェント側で遅延インジェストを行います。
chat_agents/kb.py
# コレクションが空の場合のみインジェスト
try:
if count_docs(KB_COLLECTION) <= 0:
yield ChatEvent("log", {"step": "kb", "message": "KB empty; ingesting"})
ingest_kb(KB_COLLECTION)
except Exception as exc:
logger.warning("kb_ingest_failed: %s", exc)
# インジェスト失敗時はRAGなしで続行
なぜ遅延初期化するのか:
- アプリ起動時にエンベディングAPIが利用不可でも、他の機能（チャット、Jira連携等）は正常に使える
- ベクトルストアのセットアップには数秒〜数十秒かかるため、起動時間を短縮できる
- ナレッジベース機能を使わないユーザーには初期化コストがかからない
パターン2: 冪等インジェスト（Deterministic Chunk IDs）
再起動するたびにドキュメントを再インジェストすると、同じチャンクが重複して格納される問題が発生します。検索結果に同じ内容が複数回表示され、コンテキストウィンドウを無駄に消費します。
解決策は決定論的なチャンクIDです。ファイル名・見出しパス・バイトオフセットからSHA1ハッシュを生成し、同じ内容のチャンクは同じIDになるようにします。
rag/vectorstore.py
import hashlib
def _deterministic_id(md: dict[str, Any]) -> str:
"""チャンクのメタデータから決定論的なIDを生成する。"""
basis = (
f"{md.get('filename', '')}|"
f"{md.get('heading_path', '')}|"
f"{md.get('start_index', '')}|"
f"{md.get('end_index', '')}"
)
# セキュリティ用途ではなくデータ識別用のためSHA1で十分
return hashlib.sha1(basis.encode("utf-8")).hexdigest()
インジェスト時にこのIDを使ってChromaDBに追加します。
docs = [Document(page_content=c["text"], metadata=c["metadata"]) for c in chunks]
ids = [_deterministic_id(d.metadata) for d in docs]
vec.add_documents(docs, ids=ids)
ChromaDBは同じIDのドキュメントを追加すると上書き（upsert）するため、再起動してもチャンクが重複しません。
パターン3: 見出し階層を保持するチャンキング
Markdownのチャンキングは「テキストを一定長で分割」するだけでは不十分です。見出しの階層構造を保持することで、検索結果に文脈が残ります。
rag/splitter.py
from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter
def split_markdown_file(path: str, chunk_size: int = 800, chunk_overlap: int = 120) -> list[dict[str, Any]]:
with open(path, encoding="utf-8") as f:
raw = f.read()
fm, body = parse_frontmatter(raw)
# Step 1: 見出しで論理的に分割
md_splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=[
("#", "Header 1"),
("##", "Header 2"),
("###", "Header 3"),
]
)
header_docs = md_splitter.split_text(body)
# Step 2: 各セクションをさらに文字数で分割
char_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
add_start_index=True,
)
out = []
filename = os.path.basename(path)
for doc in header_docs:
heading_path = _build_heading_path(doc.metadata)
chunks = char_splitter.create_documents([doc.page_content])
for c in chunks:
meta = {"filename": filename}
if heading_path:
meta["heading_path"] = heading_path
# バイトオフセットを保存（決定論的ID生成に使用）
start_index = c.metadata.get("start_index")
if isinstance(start_index, int):
meta["start_index"] = start_index
meta["end_index"] = start_index + len(c.page_content)
# フロントマターのメタデータを伝播
if fm.get("source_url"):
meta["source_url"] = fm["source_url"]
out.append({"text": c.page_content, "metadata": meta})
return out
見出しパスの構築:
def _build_heading_path(md_meta: dict[str, str]) -> str:
"""Header 1 > Header 2 > Header 3 のようなパスを構築する。"""
keys = [k for k in md_meta if k.lower().startswith("header ")]
keys.sort(key=lambda k: int(k.split()[1]))
parts = [str(md_meta[k]).strip() for k in keys if md_meta[k].strip()]
return " > ".join(parts)
これにより、チャンクのメタデータにheading_path: "セットアップ > 環境構築 > Dockerの設定"のような情報が付与されます。検索結果の表示やLLMへのコンテキスト提供に活用できます。
パターン4: グレースフルデグラデーション
RAGの各段階（初期化→インジェスト→検索→コンテキスト注入）のどこで失敗しても、チャット機能自体は動き続けるようにします。
ファイルレベル: 1ファイルの失敗で全体を止めない
for path in files:
try:
chunks = split_markdown_file(path)
# ... インジェスト処理
except Exception as e:
logger.error(json.dumps({
"message": "split_markdown_failed",
"path": path,
"error": str(e),
}))
continue # このファイルをスキップして次へ
エージェントレベル: RAGなしでも応答する
# KB agentの処理フロー
GROUNDS = "" # RAGコンテキスト（空の場合もある）
if rag_is_ready():
results = rag_retrieve(content, k=5)
# ... 検索結果をGROUNDSに格納
# ... sources/chunksイベントを送出
# GROUNDSが空でもLLMは応答できる（ナレッジベースなしの一般回答になる）
gen = gen_fn(
prompt=f"{GROUNDS}\n\n{prompt}" if GROUNDS else prompt,
model=meta.model,
)
デグラデーションの段階:
| 障害箇所 | 動作 |
|---|---|
| エンベディングAPI | RAGなしで一般回答 |
| 一部のファイルの解析 | 解析できたファイルのみでRAG |
| ChromaDB初期化 | RAGなしで一般回答 |
| 検索で0件ヒット | RAGなしで一般回答 |
パターン5: ソース引用
検索結果をLLMに渡す際、引用フォーマットを指定してソースの透明性を確保します。
# 検索結果からプレビューを構築
previews = []
for c in chunks_payload[:5]:
loc = ""
if c.get("startLine") is not None and c.get("endLine") is not None:
loc = f":{c['startLine']}-{c['endLine']}"
previews.append(f"[{c['filename']}{loc}] {c['preview']}")
# LLMへのコンテキスト注入
GROUNDS = (
"Use the following knowledge base excerpts when relevant.\n"
"Cite inline using [filename:startLine-endLine]. If unsure, say so.\n"
+ "\n".join(previews)
)
フロントエンドには検索結果のメタデータもSSEイベントで送信し、UIでソースカードを表示します。
# ソース情報をSSEで送信
yield ChatEvent("sources", sources_payload)
yield ChatEvent("chunks", {
"chunks": chunks_payload,
"document_set": "knowledge-base",
"strategy": "vector",
})
各チャンクのペイロード:
{
"id": chunk_id, # SHA1ハッシュ
"filename": "setup.md",
"url": "https://...", # フロントマターのsource_url
"heading_path": "セットアップ > Docker",
"score": 0.87, # 類似度スコア
"preview": "Dockerの設定は...", # 先頭240文字
}
まとめ
| パターン | 目的 | 実装 |
|---|---|---|
| 遅延初期化 | 起動時間短縮、障害分離 | 初回リクエスト時にインジェスト |
| 冪等インジェスト | 再起動時のチャンク重複防止 | SHA1による決定論的チャンクID |
| 見出し階層保持 | 検索結果の文脈保持 | 2段階チャンキング（見出し→文字数） |
| グレースフルデグラデーション | RAG障害時もチャット継続 | 各段階でtry-except + フォールバック |
| ソース引用 | 回答の透明性 | [filename:startLine-endLine]形式 |
最も重要な学びは、RAGは「検索して注入する」だけでなく、「検索できなかったとき何をするか」の設計が本体だということです。ベクトルストア、エンベディングAPI、ファイル解析のどれが壊れてもユーザーに応答を返せる設計にすることで、RAGは「あれば便利、なくても動く」機能として安定運用できます。
