大規模モデルRAGフルパイプライン実践：ゼロから本番級検索拡張生成システムを構築

AI与大数据

概要

  • RAG（検索拡張生成）は大規模モデルの本番展開のコアアーキテクチャ、ハルシネーション、知識の陳腐化、ドメイン欠落の3つの課題を解決
  • 本番級RAGの5つの重要ステップ：ドキュメント解析→チャンキング→Embedding→ベクトル検索→リランキング、各ステップに最適化の余地あり
  • スマートチャンキング戦略（セマンティックチャンキング+スライディングウィンドウ）は固定長チャンキングより検索再現率を20-30%向上
  • ハイブリッド検索（ベクトル+キーワード+ナレッジグラフ）は純粋ベクトル検索より再現率を15-25%向上
  • リランキングモデル（BGE-Reranker/Cohere Rerank）は最終回答精度を75%から90%+に向上
  • 本記事はドキュメント処理から本番デプロイまでの完全RAGソリューション、Python実装とパフォーマンスベンチマーク付き

目次

RAGが大規模モデル実装のコアアーキテクチャである理由

大規模モデルには3つの固有の欠陥がある：ハルシネーション（存在しない事実の生成）、知識の陳腐化（トレーニングデータカットオフ後の新情報へのアクセス不可）、ドメイン欠落（垂直領域の専門知識の不足）。RAGは生成前に外部ナレッジベースから関連コンテキストを検索し、Promptに注入することで、これら3つの問題を根本的に解決する。

┌──────────────────────────────────────────────────────────────────┐
│              RAGシステムフルパイプラインアーキテクチャ              │
│                                                                    │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐  │
│  │ 1.解析    │──→│ 2.チャンク│──→│ 3.Embed  │──→│ 4.ベクトル│  │
│  │ PDF/DOCX │    │ セマンティック│ │ パイプライン│  │ Milvus   │  │
│  │ HTML/MD  │    │ スライディング│ │ バッチ埋込│    │ HNSW     │  │
│  └──────────┘    └──────────┘    └──────────┘    └──────────┘  │
│                                                       │          │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐         │          │
│  │ 8.回答生成│←──│ 7.Prompt │←──│ 6.リランク│←────────┤          │
│  │ LLM生成  │    │ エンジニア│    │ BGE-Rank │  5.ハイブリッド│   │
│  │ 引用追跡  │    │ コンテキスト│ │ Top-K    │  Vec+BM25│         │
│  └──────────┘    └──────────┘    └──────────┘    └──────────┘  │
└──────────────────────────────────────────────────────────────────┘

RAG vs 純粋LLM主要指標比較

次元 純粋LLM RAG拡張LLM
事実精度 60-70% 90-95%
ハルシネーション率 15-30% 3-5%
ドメイン知識 一般 カスタマイズ可能
知識更新 再トレーニング必要 ナレッジベース増分更新
説明可能性 低 高（引用追跡）
コスト 高（大規模モデル推論） 中（検索+小規模モデル推論）

ドキュメント解析とスマートチャンキング

チャンキング戦略比較

戦略 再現率 コンテキスト完全性 実装複雑度
固定長(512 tokens) 65% 低（意味の切断） 簡単
段落チャンキング 72% 中 簡単
セマンティックチャンキング 85% 高 中程度
セマンティック+スライディングウィンドウ 92% 高 中程度

セマンティックチャンカー実装

class SemanticChunker:
    def __init__(self, embedding_client, max_chunk_tokens: int = 512, overlap_tokens: int = 64, similarity_threshold: float = 0.75):
        self.embedding_client = embedding_client
        self.max_chunk_tokens = max_chunk_tokens
        self.overlap_tokens = overlap_tokens
        self.similarity_threshold = similarity_threshold

    async def chunk_document(self, doc: ParsedDocument) -> list[Chunk]:
        sentences = self._split_sentences(doc.content)
        embeddings = await self._batch_embed(sentences)
        chunks = []
        current_chunk = [sentences[0]]
        current_tokens = self._count_tokens(sentences[0])

        for i in range(1, len(sentences)):
            similarity = self._cosine_similarity(embeddings[i - 1], embeddings[i])
            if similarity < self.similarity_threshold or current_tokens + self._count_tokens(sentences[i]) > self.max_chunk_tokens:
                chunk_content = ' '.join(current_chunk)
                chunks.append(Chunk(chunk_id=f"{doc.doc_id}_c{len(chunks)}", content=chunk_content, metadata={**doc.metadata, "chunk_index": len(chunks)}, token_count=current_tokens))
                overlap_start = max(0, len(current_chunk) - self._sentences_for_tokens(self.overlap_tokens))
                current_chunk = current_chunk[overlap_start:] + [sentences[i]]
                current_tokens = sum(self._count_tokens(s) for s in current_chunk)
            else:
                current_chunk.append(sentences[i])
                current_tokens += self._count_tokens(sentences[i])

        if current_chunk:
            chunks.append(Chunk(chunk_id=f"{doc.doc_id}_c{len(chunks)}", content=' '.join(current_chunk), metadata={**doc.metadata, "chunk_index": len(chunks)}, token_count=current_tokens))
        return chunks

Embeddingパイプラインとベクトルインデックス構築

バッチEmbeddingパイプライン

class EmbeddingPipeline:
    def __init__(self, model: str = "BAAI/bge-large-zh-v1.5", batch_size: int = 64):
        self.client = AsyncOpenAI(base_url="http://localhost:8000/v1")
        self.model = model
        self.batch_size = batch_size

    async def embed_chunks(self, chunks: list[Chunk]) -> list[dict]:
        all_embeddings = []
        for i in range(0, len(chunks), self.batch_size):
            batch = chunks[i:i + self.batch_size]
            texts = [f"検索クエリ：{c.content}" for c in batch]
            response = await self.client.embeddings.create(model=self.model, input=texts)
            for j, item in enumerate(response.data):
                all_embeddings.append({"chunk_id": batch[j].chunk_id, "content": batch[j].content, "metadata": batch[j].metadata, "embedding": item.embedding, "token_count": batch[j].token_count})
        return all_embeddings

ハイブリッド検索とリランキング

ハイブリッド検索：ベクトル+BM25

class HybridRetriever:
    def __init__(self, vector_store: RAGVectorStore, bm25: BM25Retriever, vector_weight: float = 0.7, bm25_weight: float = 0.3):
        self.vector_store = vector_store
        self.bm25 = bm25
        self.vector_weight = vector_weight
        self.bm25_weight = bm25_weight

    async def search(self, query: str, query_embedding: list[float], top_k: int = 10) -> list[dict]:
        vector_results = self.vector_store.search(query_embedding, top_k=top_k * 2)
        bm25_results = self.bm25.search(query, top_k=top_k * 2)
        merged: dict[str, dict] = {}
        for r in vector_results: merged[r["chunk_id"]] = {**r, "vector_score": r["score"], "bm25_score": 0.0}
        for r in bm25_results:
            if r["chunk_id"] in merged: merged[r["chunk_id"]]["bm25_score"] = r["score"]
            else: merged[r["chunk_id"]] = {**r, "vector_score": 0.0, "bm25_score": r["score"]}
        max_vector = max((m["vector_score"] for m in merged.values()), default=1.0) or 1.0
        max_bm25 = max((m["bm25_score"] for m in merged.values()), default=1.0) or 1.0
        for m in merged.values(): m["combined_score"] = self.vector_weight * m["vector_score"] / max_vector + self.bm25_weight * m["bm25_score"] / max_bm25
        return sorted(merged.values(), key=lambda x: x["combined_score"], reverse=True)[:top_k]

リランキング

class Reranker:
    def __init__(self, llm_client, model: str = "BAAI/bge-reranker-v2-m3"):
        self.llm = llm_client
        self.model = model

    async def rerank(self, query: str, candidates: list[dict], top_k: int = 5) -> list[dict]:
        pairs = [[query, c["content"][:512]] for c in candidates]
        scores = await self._compute_scores(pairs)
        for i, candidate in enumerate(candidates): candidate["rerank_score"] = scores[i]
        candidates.sort(key=lambda x: x["rerank_score"], reverse=True)
        return candidates[:top_k]

RAGフルパイプライン最適化戦略

RAGフルパイプラインパフォーマンスベンチマーク

ステップ レイテンシ(P50) レイテンシ(P99) 備考
ドキュメント解析(PDF 10ページ) 500ms 1.5s PyMuPDF
セマンティックチャンキング(10ページ) 2s 5s Embedding呼び出し含む
バッチEmbedding(64チャンク) 800ms 2s BGE-large
ベクトル検索(top-10) 5ms 15ms Milvus HNSW
BM25検索(top-10) 2ms 5ms メモリインデックス
ハイブリッド融合 1ms 3ms スコア正規化
リランキング(top-5) 200ms 500ms BGE-Reranker
LLM生成(7B) 1.5s 3s Qwen2.5-7B
エンドツーエンドRAG 3s 6s フルパイプライン

検索再現率比較

検索方式 Top-5再現率 Top-10再現率 Top-20再現率
純粋ベクトル検索 72% 82% 88%
純粋BM25 65% 75% 80%
ハイブリッド検索 82% 90% 95%
ハイブリッド+リランキング 88% 94% 97%

本番デプロイとオブザーバビリティ

RAGサービスK8sデプロイ

apiVersion: apps/v1
kind: Deployment
metadata:
  name: rag-service
  namespace: ai-rag
spec:
  replicas: 2
  selector:
    matchLabels:
      app: rag-service
  template:
    metadata:
      labels:
        app: rag-service
    spec:
      containers:
        - name: rag-api
          image: myregistry/rag-service:v1.0
          ports:
            - containerPort: 8000
          resources:
            requests:
              cpu: "2"
              memory: 4Gi
            limits:
              cpu: "4"
              memory: 8Gi
          env:
            - name: MILVUS_URI
              value: "http://milvus:19530"
            - name: LLM_API_BASE
              value: "http://vllm-qwen2-72b:8000/v1"

まとめと関連記事

RAGは大規模モデル本番展開のコアアーキテクチャである。5つの重要ステップ（解析→チャンキング→Embedding→検索→リランキング）の各ステップに最適化の余地がある。セマンティックチャンキング+スライディングウィンドウは固定チャンキングより再現率を20-30%向上し、ハイブリッド検索は純粋ベクトル検索より再現率を15-25%向上し、リランキングは最終精度を75%から90%+に向上させる。

開発の要点振り返り：

  1. ドキュメント解析：PDFはPyMuPDF、Markdownは見出し別セクション、HTMLはBeautifulSoupでノイズ除去
  2. スマートチャンキング：セマンティック+スライディングウィンドウ、max_chunk_tokens=512, overlap_tokens=64
  3. Embedding：BGE-large-zh-v1.5、バッチ64、クエリプレフィックス「検索クエリ：」
  4. ハイブリッド検索：ベクトル重み0.7 + BM25重み0.3、スコア正規化後加重融合
  5. リランキング：BGE-Reranker-v2-m3、top-20候補からtop-5を精選

関連記事：

権威参考：

ブラウザローカルツールを無料で試す →

#大模型RAG系统#RAG生产部署#向量检索RAG#知识库构建#RAG全链路优化#2026