# LangChain + Pinecone で構築するプロダクションRAGシステム：ハルシネーション対策と精度改善の実践

> 検証環境ではなく本番運用レベルのRAGシステム構築ガイド。LangChain + Pinecone + FastAPIで実装したハルシネーション対策5選、精度評価手法、コスト最適化戦略を実コード付きで解説します。

- 公開日: 2025-01-10
- 著者: 友田 陽大
- タグ: AI, RAG, LangChain, Pinecone, OpenAI, Python, FastAPI, 機械学習, NLP
- URL: https://tomodahinata.com/blog/langchain-pinecone-production-rag-system
- カテゴリ: 生成AI・LLM・RAG
- 総合ガイド: https://tomodahinata.com/blog/vercel-ai-sdk-production-llm-apps-streaming-tools-rag

## 要点

- ハルシネーション対策は多層防御──ソース引用強制・信頼度スコア・メタデータフィルタ・人間レビューを重ねる
- Retrieval精度はチャンクサイズ最適化が鍵で、実験ではchunk_size=512が精度と速度のバランス最適だった
- RAGの精度はRecall@5・Precision@5・MRRで定量評価し、継続モニタリングする
- コストはモデル選択（GPT-3.5への部分ダウングレード）とキャッシングで最適化する
- Pineconeのネームスペースで論理分離し、検索時も同じネームスペースを指定してデータ漏洩を防ぐ

---

「RAGシステムを本番環境に投入したいが、ハルシネーション（幻覚）問題が解決できない」
「LangChainのチュートリアルは動いたが、精度が低すぎて実用に耐えない」
「コスト試算が不明で、スケール時の予算が読めない」

AI機能を既存SaaSに統合しようとする開発責任者や、RAGシステムをPoCから本番化したいエンジニアが直面する、これらの課題。私も同じ壁にぶつかりました。

この記事では、**LangChain + Pinecone + FastAPI**で構築したプロダクションレベルのRAGシステムの実装を公開します。単なるチュートリアルではなく、**ハルシネーション対策5選**、**精度評価の定量手法**、**月1万クエリでのコスト試算**まで、本番運用の現実を共有します。

---

## 前提：RAGシステムとは何か

### RAG (Retrieval-Augmented Generation) の基本

RAGは、**大規模言語モデル（LLM）に外部知識を検索・注入して回答精度を高める技術**です。

**従来のLLM（ChatGPT等）の限界**:
- 学習データの日付以降の情報を知らない
- 企業の内部ドキュメントにアクセスできない
- ハルシネーション（事実と異なる回答）が発生

**RAGによる解決**:
1. ユーザーの質問に関連する文書を**ベクトルDBから検索** (Retrieval)
2. 検索結果をプロンプトに**注入**してLLMに送信 (Augmentation)
3. LLMが検索結果を基に**正確な回答を生成** (Generation)

---

## システムアーキテクチャ全体像

```
┌─────────────┐
│ ユーザー    │
└──────┬──────┘
       │ HTTP Request
       ▼
┌──────────────────────────┐
│ FastAPI (APIゲートウェイ) │
│ - 入力バリデーション      │
│ - 認証・認可              │
│ - レート制限              │
└──────┬───────────────────┘
       │
       ▼
┌──────────────────────────┐
│ LangChain                 │
│ - クエリの埋め込み生成    │
│ - Pinecone検索            │
│ - プロンプト構築          │
│ - OpenAI API呼び出し      │
└──────┬───────────────────┘
       │
       ├─────────────┐
       ▼             ▼
┌────────────┐  ┌──────────────┐
│ Pinecone   │  │ OpenAI API   │
│ ベクトルDB │  │ (GPT-4)      │
│ - 文書検索 │  │ - 回答生成   │
└────────────┘  └──────────────┘
```

### 技術スタック

| レイヤー | 技術 | 役割 |
|---------|------|------|
| **API層** | FastAPI 0.109+ | REST API提供、バリデーション |
| **オーケストレーション** | LangChain 0.1+ | RAGフロー制御、プロンプト管理 |
| **ベクトルDB** | Pinecone | 埋め込みベクトル検索 |
| **LLM** | OpenAI GPT-4 Turbo | 回答生成 |
| **埋め込みモデル** | text-embedding-3-small | ベクトル化（1536次元） |
| **デプロイ** | AWS ECS Fargate | コンテナ実行環境 |

---

## ハルシネーション対策5選

### 対策①：ソース引用の強制

**問題**: LLMが検索結果を無視して、学習データから回答を生成

**解決策**: プロンプトで「必ず検索結果からのみ回答せよ」と明示

```python
from langchain.prompts import PromptTemplate

# ハルシネーション防止プロンプト
PROMPT_TEMPLATE = """あなたは正確な情報提供を重視するAIアシスタントです。

以下の検索結果のみを使って、ユーザーの質問に回答してください。
検索結果に含まれない情報は「情報が見つかりませんでした」と回答してください。
推測や憶測は絶対に含めないでください。

検索結果:
{context}

ユーザーの質問: {question}

回答の形式:
1. 回答内容（検索結果から引用）
2. 引用元（ソースのファイル名とページ番号）

回答:"""

prompt = PromptTemplate(
    template=PROMPT_TEMPLATE,
    input_variables=["context", "question"]
)
```

**効果**: ハルシネーション率 40% → 10%（自社テストデータで測定）

---

### 対策②：信頼度スコアの付与

**問題**: LLMが自信なく回答しても、ユーザーには判別できない

**解決策**: 回答に信頼度スコア（0.0〜1.0）を付与

```python
from langchain.llms import OpenAI
from langchain.chains import LLMChain
from pydantic import BaseModel, Field

class AnswerWithConfidence(BaseModel):
    """回答と信頼度スコア"""
    answer: str = Field(description="ユーザーへの回答")
    confidence: float = Field(
        description="信頼度スコア（0.0〜1.0）。検索結果に明確な回答がある場合1.0、推測の場合0.5以下",
        ge=0.0,
        le=1.0
    )
    sources: list[str] = Field(description="引用元のドキュメントID")

# OpenAI Function Callingで構造化出力
from langchain.output_parsers import PydanticOutputParser

parser = PydanticOutputParser(pydantic_object=AnswerWithConfidence)

# プロンプトに指示を追加
format_instructions = parser.get_format_instructions()
```

**活用例**:
- `confidence < 0.7`: 「この回答は推測を含む可能性があります」と警告表示
- `confidence < 0.5`: 人間オペレーターにエスカレーション

---

### 対策③：Retrieval精度向上（チャンクサイズ最適化）

**問題**: 文書の分割方法が不適切で、文脈が失われる

**解決策**: チャンクサイズとオーバーラップの最適化

```python
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 最適なチャンク設定（実験で決定）
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,        # トークン数（経験的に512が最適）
    chunk_overlap=50,      # オーバーラップ（文脈保持）
    length_function=len,
    separators=["\n\n", "\n", "。", ".", " ", ""]  # 日本語対応
)

# 文書の分割
chunks = text_splitter.split_text(document_text)

# Pineconeへのアップロード
from langchain.vectorstores import Pinecone
from langchain.embeddings import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vectorstore = Pinecone.from_texts(
    texts=chunks,
    embedding=embeddings,
    index_name="my-knowledge-base",
    namespace="product-docs"
)
```

**パラメータ選定の実験結果**:

| チャンクサイズ | Recall@5 | 応答時間 |
|--------------|---------|---------|
| 256 | 0.65 | 1.2秒 |
| **512** | **0.82** | **1.5秒** |
| 1024 | 0.78 | 2.1秒 |
| 2048 | 0.71 | 3.5秒 |

**結論**: chunk_size=512が精度と速度のバランス最適

---

### 対策④：メタデータフィルタリング

**問題**: 無関係な文書が検索結果に混入（例：古いバージョンのドキュメント）

**解決策**: Pineconeのメタデータフィルタで絞り込み

```python
from datetime import datetime

# 文書のメタデータ付きアップロード
metadata = {
    "source": "product_manual_v2.pdf",
    "page": 42,
    "version": "2.0",
    "category": "API仕様",
    "last_updated": "2024-12-01"
}

vectorstore.add_texts(
    texts=[chunk_text],
    metadatas=[metadata]
)

# 検索時のフィルタリング
from langchain.vectorstores import Pinecone

vectorstore = Pinecone.from_existing_index(
    index_name="my-knowledge-base",
    embedding=embeddings,
    namespace="product-docs"
)

# 最新バージョンのみ検索
results = vectorstore.similarity_search(
    query="APIの認証方法は？",
    k=5,
    filter={
        "version": {"$eq": "2.0"},
        "category": {"$eq": "API仕様"}
    }
)
```

**効果**: 無関係な結果の混入率 30% → 5%

---

### 対策⑤：人間レビューのフィードバックループ

**問題**: ハルシネーションを検出しても、自動改善されない

**解決策**: 人間のフィードバックをシステムに反映

```python
from pydantic import BaseModel
from datetime import datetime

class UserFeedback(BaseModel):
    """ユーザーフィードバック"""
    query_id: str
    is_helpful: bool
    is_accurate: bool
    comment: str | None = None
    timestamp: datetime

# フィードバック収集API
from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.post("/api/feedback")
async def submit_feedback(feedback: UserFeedback):
    # フィードバックをDBに保存
    await save_feedback_to_db(feedback)

    # 精度が低い回答を検出
    if not feedback.is_accurate:
        # アラート送信（Slack等）
        await send_alert_to_team(feedback)

        # 該当する文書の再確認フラグ
        await flag_document_for_review(feedback.query_id)

    return {"status": "success"}

# 週次での精度分析
async def analyze_feedback_trends():
    feedbacks = await get_feedbacks_last_week()

    accuracy_rate = sum(f.is_accurate for f in feedbacks) / len(feedbacks)

    if accuracy_rate < 0.85:
        # 精度低下アラート
        await send_weekly_report(accuracy_rate)
```

**運用フロー**:
1. 回答に👍/👎ボタン設置
2. 不正確な回答を人間が確認
3. 該当文書の改善 or チャンク分割調整
4. 週次レポートで精度トレンド監視

---

## Pineconeインデックス設計

### ネームスペース分離戦略

Pineconeでは単一インデックス内で**ネームスペース**を使い、論理的にデータを分離できます。

```python
import pinecone

# Pinecone初期化
pinecone.init(
    api_key="your-api-key",
    environment="us-west1-gcp"
)

# インデックス作成（初回のみ）
pinecone.create_index(
    name="my-knowledge-base",
    dimension=1536,  # text-embedding-3-smallの次元数
    metric="cosine"   # コサイン類似度
)

# ネームスペース別のデータ投入
namespaces = {
    "product-docs": "製品ドキュメント",
    "api-reference": "API仕様書",
    "faq": "よくある質問",
    "internal-kb": "社内ナレッジベース（社員のみ）"
}

# 例：製品ドキュメントの投入
vectorstore = Pinecone.from_texts(
    texts=product_doc_chunks,
    embedding=embeddings,
    index_name="my-knowledge-base",
    namespace="product-docs"
)
```

### マルチテナント対応

B2B SaaSで顧客ごとにデータを分離する場合:

```python
# 顧客ごとのネームスペース
customer_namespace = f"customer-{customer_id}"

vectorstore = Pinecone.from_existing_index(
    index_name="my-knowledge-base",
    embedding=embeddings,
    namespace=customer_namespace
)

# 検索時も同じネームスペースを指定
results = vectorstore.similarity_search(
    query=user_query,
    k=5,
    namespace=customer_namespace  # データ漏洩防止
)
```

---

## 精度評価の定量手法

### 評価指標

| 指標 | 定義 | 目標値 |
|-----|------|-------|
| **Recall@5** | 上位5件に正解が含まれる割合 | > 0.85 |
| **Precision@5** | 上位5件のうち関連性ある文書の割合 | > 0.70 |
| **MRR** (Mean Reciprocal Rank) | 正解の順位の逆数の平均 | > 0.75 |
| **Accuracy** | 回答の正確性（人間評価） | > 0.90 |

### 評価スクリプト

```python
from typing import List, Tuple
import numpy as np

class RAGEvaluator:
    """RAGシステムの精度評価"""

    def __init__(self, test_data: List[Tuple[str, List[str]]]):
        """
        Args:
            test_data: [(質問, 正解ドキュメントIDのリスト), ...]
        """
        self.test_data = test_data

    def calculate_recall_at_k(
        self,
        retrieved_docs: List[str],
        ground_truth: List[str],
        k: int = 5
    ) -> float:
        """Recall@K を計算"""
        top_k = set(retrieved_docs[:k])
        relevant = set(ground_truth)

        if not relevant:
            return 0.0

        return len(top_k & relevant) / len(relevant)

    def calculate_precision_at_k(
        self,
        retrieved_docs: List[str],
        ground_truth: List[str],
        k: int = 5
    ) -> float:
        """Precision@K を計算"""
        top_k = set(retrieved_docs[:k])
        relevant = set(ground_truth)

        if not top_k:
            return 0.0

        return len(top_k & relevant) / k

    def calculate_mrr(
        self,
        retrieved_docs: List[str],
        ground_truth: List[str]
    ) -> float:
        """Mean Reciprocal Rank を計算"""
        for rank, doc_id in enumerate(retrieved_docs, 1):
            if doc_id in ground_truth:
                return 1.0 / rank
        return 0.0

    def evaluate(self, rag_system) -> dict:
        """RAGシステム全体を評価"""
        recalls = []
        precisions = []
        mrrs = []

        for question, ground_truth in self.test_data:
            # RAGシステムで検索
            retrieved = rag_system.retrieve(question, k=10)
            retrieved_ids = [doc.metadata["id"] for doc in retrieved]

            # 指標計算
            recalls.append(self.calculate_recall_at_k(retrieved_ids, ground_truth, k=5))
            precisions.append(self.calculate_precision_at_k(retrieved_ids, ground_truth, k=5))
            mrrs.append(self.calculate_mrr(retrieved_ids, ground_truth))

        return {
            "recall@5": np.mean(recalls),
            "precision@5": np.mean(precisions),
            "mrr": np.mean(mrrs)
        }

# 使用例
test_data = [
    ("APIキーの取得方法は？", ["doc_123", "doc_456"]),
    ("料金プランの違いは？", ["doc_789"]),
    # ... 100件以上のテストケース
]

evaluator = RAGEvaluator(test_data)
metrics = evaluator.evaluate(my_rag_system)

print(f"Recall@5: {metrics['recall@5']:.2f}")
print(f"Precision@5: {metrics['precision@5']:.2f}")
print(f"MRR: {metrics['mrr']:.2f}")
```

### 実プロジェクトでの評価結果

**改善前**:
- Recall@5: 0.68
- Precision@5: 0.52
- MRR: 0.61

**改善後（対策①〜④適用）**:
- Recall@5: **0.87** (+28%)
- Precision@5: **0.78** (+50%)
- MRR: **0.81** (+33%)

---

## コスト試算：月1万クエリでの実例

### OpenAI API コスト

```
【前提】
- クエリ数: 10,000回/月
- 平均プロンプト長: 2,000トークン（検索結果5件 + 質問）
- 平均レスポンス長: 500トークン
- モデル: GPT-4 Turbo

【計算】
入力: 10,000クエリ × 2,000トークン × $0.01/1K = $200
出力: 10,000クエリ × 500トークン × $0.03/1K = $150
合計: $350/月
```

### Pinecone コスト

```
【前提】
- インデックス数: 1個
- ベクトル数: 100,000件
- プラン: Starter ($70/月 for 100K vectors)

【計算】
基本料金: $70/月
```

### 埋め込み生成コスト (OpenAI)

```
【前提】
- 新規文書追加: 1,000件/月
- 平均文書長: 1,000トークン
- モデル: text-embedding-3-small

【計算】
1,000文書 × 1,000トークン × $0.00002/1K = $0.02/月
（ほぼ無視できる）
```

### インフラコスト (AWS ECS Fargate)

```
【前提】
- Fargate Task: 0.5 vCPU, 1GB RAM
- 稼働時間: 24時間 × 30日

【計算】
Fargate: $35/月
ALB: $23/月
CloudWatch Logs: $3/月
合計: $61/月
```

### **総コスト**

```
OpenAI API (GPT-4 Turbo): $350/月
Pinecone: $70/月
OpenAI Embeddings: $0.02/月（無視）
AWS インフラ: $61/月
────────────────────────
合計: 約 $481/月（約68,000円）
```

### コスト削減戦略

#### 戦略①：GPT-3.5 Turboへの部分的ダウングレード

```python
# 簡単な質問はGPT-3.5、複雑な質問はGPT-4
def select_model(query: str, context: str) -> str:
    # トークン数で判定
    total_tokens = len(query.split()) + len(context.split())

    if total_tokens < 500:
        return "gpt-3.5-turbo"  # $0.001/1K (入力)
    else:
        return "gpt-4-turbo"
```

**効果**: 50%のクエリをGPT-3.5に → コスト30%削減（$350 → $245）

#### 戦略②：キャッシング

```python
import hashlib
from functools import lru_cache

class RAGCache:
    """よくある質問をキャッシュ"""

    def __init__(self, redis_client):
        self.redis = redis_client
        self.ttl = 3600  # 1時間

    def get_cache_key(self, query: str) -> str:
        return f"rag_cache:{hashlib.md5(query.encode()).hexdigest()}"

    async def get(self, query: str) -> str | None:
        key = self.get_cache_key(query)
        return await self.redis.get(key)

    async def set(self, query: str, answer: str):
        key = self.get_cache_key(query)
        await self.redis.setex(key, self.ttl, answer)

# 使用例
cache = RAGCache(redis_client)

async def rag_with_cache(query: str) -> str:
    # キャッシュ確認
    cached = await cache.get(query)
    if cached:
        return cached

    # RAG実行
    answer = await rag_system.query(query)

    # キャッシュ保存
    await cache.set(query, answer)

    return answer
```

**効果**: キャッシュヒット率30% → API呼び出し30%削減（$350 → $245）

---

## 本番運用の落とし穴とエラーハンドリング

### 落とし穴①：OpenAI APIのレート制限

**問題**: 急激なトラフィック増でAPI呼び出し制限に引っかかる

**解決策**: Exponential Backoff + Retry

```python
import asyncio
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type
)
from openai import RateLimitError

@retry(
    retry=retry_if_exception_type(RateLimitError),
    wait=wait_exponential(multiplier=1, min=4, max=60),
    stop=stop_after_attempt(5)
)
async def call_openai_with_retry(prompt: str) -> str:
    """OpenAI API呼び出し（リトライ付き）"""
    response = await openai_client.chat.completions.create(
        model="gpt-4-turbo",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=500
    )
    return response.choices[0].message.content
```

### 落とし穴②：Pinecone検索タイムアウト

**問題**: 大量ベクトル検索時にタイムアウト（>10秒）

**解決策**: タイムアウト設定 + フォールバック

```python
import asyncio

async def search_with_timeout(
    vectorstore,
    query: str,
    k: int = 5,
    timeout: float = 5.0
) -> List[Document]:
    """タイムアウト付き検索"""
    try:
        result = await asyncio.wait_for(
            vectorstore.asimilarity_search(query, k=k),
            timeout=timeout
        )
        return result
    except asyncio.TimeoutError:
        # フォールバック：キーワード検索
        return await fallback_keyword_search(query, k=k)
```

### 落とし穴③：メモリリーク

**問題**: LangChainのオブジェクトがメモリに残り続ける

**解決策**: 明示的なクリーンアップ

```python
from contextlib import asynccontextmanager

@asynccontextmanager
async def rag_session():
    """RAGセッション管理（リソース自動解放）"""
    # 初期化
    vectorstore = Pinecone.from_existing_index(...)
    llm = ChatOpenAI(...)

    try:
        yield vectorstore, llm
    finally:
        # クリーンアップ
        del vectorstore
        del llm
        import gc
        gc.collect()

# 使用例
async def query_rag(user_query: str) -> str:
    async with rag_session() as (vectorstore, llm):
        # RAG処理
        results = await vectorstore.asimilarity_search(user_query)
        # ...
    # 自動的にリソース解放
```

---

## FastAPI統合：本番APIの実装

```python
from fastapi import FastAPI, HTTPException, Depends
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from typing import AsyncGenerator
import asyncio

app = FastAPI(title="Production RAG API")

class QueryRequest(BaseModel):
    query: str = Field(..., min_length=1, max_length=500)
    namespace: str = Field(default="product-docs")
    max_results: int = Field(default=5, ge=1, le=10)

class QueryResponse(BaseModel):
    answer: str
    confidence: float
    sources: list[dict]
    processing_time_ms: float

# RAGシステムの初期化（起動時）
@app.on_event("startup")
async def startup_event():
    global rag_system
    rag_system = initialize_rag_system()

@app.post("/api/query", response_model=QueryResponse)
async def query_endpoint(request: QueryRequest):
    """RAGクエリエンドポイント"""
    import time
    start_time = time.time()

    try:
        # RAG実行
        result = await rag_system.query(
            query=request.query,
            namespace=request.namespace,
            k=request.max_results
        )

        processing_time = (time.time() - start_time) * 1000

        return QueryResponse(
            answer=result.answer,
            confidence=result.confidence,
            sources=[
                {
                    "id": src.id,
                    "title": src.metadata.get("title"),
                    "page": src.metadata.get("page"),
                    "score": src.score
                }
                for src in result.sources
            ],
            processing_time_ms=processing_time
        )

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# ストリーミングレスポンス（リアルタイム回答）
@app.post("/api/query/stream")
async def query_stream_endpoint(request: QueryRequest):
    """ストリーミングRAGクエリ"""

    async def generate() -> AsyncGenerator[str, None]:
        async for chunk in rag_system.query_stream(request.query):
            yield f"data: {chunk}\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream"
    )

# ヘルスチェック
@app.get("/health")
async def health_check():
    return {"status": "healthy", "service": "rag-api"}
```

---

## まとめ：プロダクションRAGの成功要因

### 技術的ポイント

1. **ハルシネーション対策は多層防御**: プロンプト + スコア + 人間レビュー
2. **精度評価は定量化必須**: Recall/Precision/MRRで継続モニタリング
3. **コストは運用設計で最適化**: キャッシング + モデル選択 + スケーリング

### 運用的ポイント

1. **フィードバックループが命**: 人間の評価を週次で反映
2. **エラーハンドリングは過剰なくらいに**: リトライ + タイムアウト + フォールバック
3. **段階的リリース**: 社内限定 → β版 → 全体公開

### 実績データ（私のプロジェクト）

- **精度**: Recall@5 0.87, Accuracy 0.92（人間評価）
- **レスポンス時間**: 平均1.8秒（95パーセンタイル 3.2秒）
- **コスト**: 月1万クエリで約7万円（予算内）
- **稼働率**: 99.8%（月間ダウンタイム < 1時間）

---

## 次のステップ

RAGシステムの実装でお悩みなら、お気軽にご相談ください。LangChain/Pinecone/OpenAIの実践知を、あなたのプロジェクトに活かします。

[お問い合わせはこちら](/contact)

---