PostgreSQLを使った本番運用向けRAGシステムの構築: 完全実装ガイド
ほとんどのRAG（Retrieval-Augmented Generation）システムは、本番環境で予測可能な技術的理由により失敗します：リトリーブ品質の低下、出典情報の欠如、現実のクエリの変動に対処できないこと。 このガイドは、実際に機能する本番運用向けRAGシステムを構築する方法を示します。
なぜほとんどのRAGシステムは失敗するのか
本番環境でのRAGの失敗は、3つの核となる技術的問題に起因します：
1. 純粋なベクター検索の制約
ベクトルの類似性は、必ずしも人間の関連性と一致しません。 「APIのレート制限」というクエリは、ユーザーが「1000リクエスト/時間」に関する正確な情報を求めている場合に「リクエストのスロットリングガイドライン」を返してしまうことがあります。
2. 出典情報の欠如
確認できない回答は信頼されません。 明確な出典がないと、正しくても信頼性が低く感じられます。
3. 単一の検索戦略
ベクター検索だけ、またはキーワード検索だけに頼ると、重要な結果を見逃します。 実際の質問には、意味理解と正確な用語の一致の両方が必要です。
アーキテクチャ概要: PostgreSQL + pgvector
私たちのアーキテクチャは、ベクトル検索と従来の検索の両方を、単一で信頼性の高いデータベース内で処理するために、pgvector拡張を備えたPostgreSQLを使用します:
…コードブロックは省略…ステップ1: データベーススキーマ
…SQLコードブロックは省略…必須インデックスの作成:
…SQLコードブロックは省略…ステップ2: ハイブリッド検索の実装
私たちのシステムの核は、意味理解とキーワード検索を組み合わせた関数です:
…SQLコードブロックは省略…ステップ3: ドキュメント処理パイプライン
文脈を保持するスマートチャンク化:
import asyncpg
import openai
from typing import List, Dict
import re
from dataclasses import dataclass
@dataclass
class DocumentChunk:
content: str
section_title: str
chunk_index: int
token_count: int
class DocumentProcessor:
def __init__(self, db_url: str, openai_api_key: str):
self.db_url = db_url
self.client = openai.AsyncOpenAI(api_key=openai_api_key)
def _create_smart_chunks(self, content: str, max_tokens: int = 512) -> List[DocumentChunk]:
"""Create chunks that respect document structure."""
chunks = []
# Split by major sections (headers)
sections = re.split(r'
\s*#{1,3}\s+', content)
current_section = "Introduction"
for section in sections:
if not section.strip():
continue
for chunk in chunks:
chunk_embedding = await self._generate_embedding(chunk.content)
await conn.execute("""
INSERT INTO document_chunks
(document_id, chunk_index, content, embedding, token_count, section_title)
VALUES ($1, $2, $3, $4, $5, $6)
""", doc_id, chunk.chunk_index, chunk.content,
chunk_embedding, chunk.token_count, chunk.section_title)
return True
Step 4: 出典を付与するクエリパイプライン
import json
from typing import List, Dict
class ProductionRAGSystem:
def __init__(self, db_url: str, openai_api_key: str):
self.db_url = db_url
self.client = openai.AsyncOpenAI(api_key=openai_api_key)
async def answer_question(self, question: str) -> Dict:
"""ユーザーの質問に回答するための主要エントリポイント。"""
# クエリの埋め込みを生成
embedding_response = await self.client.embeddings.create(
model="text-embedding-3-large",
input=question
)
query_embedding = embedding_response.data[0].embedding
# 関連するチャンクを取得
chunks = await self._retrieve_chunks(question, query_embedding)
# 品質ゲート - 適切な出典を確保
high_quality_chunks = [c for c in chunks if c['combined_score'] > 0.5]
if len(high_quality_chunks) < 2:
return {
"answer": "この質問に自信を持って答えるのに十分な関連情報が見つかりませんでした。",
"confidence": 0.0,
"sources": [],
"suggestion": "より具体的な用語で質問を言い換えてください。"
}
# 出典付きの回答を生成
return await self._generate_attributed_answer(question, high_quality_chunks[:5])
実装の要点
本番環境のRAGシステムを構築するには:
- ハイブリッド検索: 意味的検索とキーワード検索を組み合わせて関連性を高める
- スマートチャンク化: 文書の構造と文脈を保持する
- 出典表示: 検証可能な引用を常に提供する
- 品質ゲート: 信頼度の閾値を設定して不良な回答を防ぐ
- パフォーマンス最適化: キャッシュ、接続プーリング、データベースチューニング
- 体系的なテスト: 実際の指標を用いた包括的評価
この実装は、本番環境で実際に機能するRAGシステムの堅固な基盤を提供し、実際のユーザーの問い合わせを高い精度と透明性で処理します。
完全なコードは、特定のユースケースに合わせてデプロイおよびカスタマイズできる実用例として利用可能です。高度な機能の最適化を行う前に、取得品質、出典表示、ユーザーの信頼といった基本を正しく抑えることに集中してください。
