RAGベースFAQチャットボットのプロダクション構築ガイド：ベクトルDB選定から運用最適化まで

FAQドキュメント収集 -> 前処理/正規化 -> チャンキング -> エンベディング生成 -> ベクトルDB格納 -> メタデータインデキシング

ユーザー質問 -> クエリ前処理 -> エンベディング変換 -> ベクトル検索 + BM25 -> リランキング -> プロンプト構成 -> LLM回答生成 -> 後処理/ガードレール

"""
FAQ専用チャンキング戦略。
質問-回答ペアを1つの単位として維持しつつ、
Parent-Child構造で検索精度とコンテキストを同時に確保する。
"""
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from typing import List, Tuple
import re
import hashlib


def parse_faq_pairs(raw_text: str) -> List[Tuple[str, str, dict]]:
    """FAQ原文から質問-回答ペアを抽出する。"""
    faq_pattern = re.compile(
        r"(?:Q|질문)\s*[.:]\s*(.+?)\n+"
        r"(?:A|답변)\s*[.:]\s*(.+?)(?=\n(?:Q|질문)\s*[.:]|\Z)",
        re.DOTALL
    )
    pairs = []
    for i, match in enumerate(faq_pattern.finditer(raw_text)):
        question = match.group(1).strip()
        answer = match.group(2).strip()
        metadata = {
            "faq_id": hashlib.md5(question.encode()).hexdigest()[:8],
            "source_type": "faq",
            "question": question,
            "pair_index": i,
        }
        pairs.append((question, answer, metadata))
    return pairs


def create_faq_chunks(
    faq_pairs: List[Tuple[str, str, dict]],
    child_chunk_size: int = 200,
    child_chunk_overlap: int = 50,
) -> Tuple[List[Document], List[Document]]:
    """
    Parent-Childチャンキング戦略でFAQドキュメントを分割する。
    - Parent: 質問 + 全回答（LLMコンテキスト用）
    - Child: 回答を小さなチャンクに分割（検索精度用）
    """
    parent_docs = []
    child_docs = []

    child_splitter = RecursiveCharacterTextSplitter(
        chunk_size=child_chunk_size,
        chunk_overlap=child_chunk_overlap,
        separators=["\n\n", "\n", ". ", " "],
    )

    for question, answer, metadata in faq_pairs:
        # Parentドキュメント：質問 + 全回答
        parent_content = f"質問: {question}\n回答: {answer}"
        parent_id = metadata["faq_id"]
        parent_doc = Document(
            page_content=parent_content,
            metadata={**metadata, "doc_type": "parent", "parent_id": parent_id},
        )
        parent_docs.append(parent_doc)

        # Childドキュメント：回答を細分化して検索精度向上
        answer_chunks = child_splitter.split_text(answer)
        for j, chunk in enumerate(answer_chunks):
            child_content = f"質問: {question}\n回答の一部: {chunk}"
            child_doc = Document(
                page_content=child_content,
                metadata={
                    **metadata,
                    "doc_type": "child",
                    "parent_id": parent_id,
                    "chunk_index": j,
                },
            )
            child_docs.append(child_doc)

    return parent_docs, child_docs


# 使用例
raw_faq = """
Q: 返金処理は何日かかりますか？
A: 返金は申請日から営業日基準3-5日以内に処理されます。
クレジットカード決済の場合、カード会社の処理期間が追加で2-3日かかる場合があります。
銀行振込の場合、登録された口座に直接返金されます。

Q: 海外配送は可能ですか？
A: 現在、アメリカ、日本、中国、東南アジア地域への海外配送が可能です。
海外配送料は地域と重量によって異なり、関税は受取人負担です。
配送所要期間は地域によって7-14営業日です。
"""

pairs = parse_faq_pairs(raw_faq)
parents, children = create_faq_chunks(pairs)
print(f"Parentドキュメント: {len(parents)}件, Childドキュメント: {len(children)}件")

"""
PineconeベクトルDB設定とFAQドキュメントインデキシング。
ネームスペースでドキュメントタイプを分離し、メタデータフィルタリングを活用する。
"""
from pinecone import Pinecone, ServerlessSpec
from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain_core.documents import Document
from typing import List
import os
import time


def setup_pinecone_index(
    index_name: str = "faq-chatbot",
    dimension: int = 1536,
    metric: str = "cosine",
) -> None:
    """Pineconeインデックスを作成する。既に存在すればスキップする。"""
    pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])

    existing_indexes = [idx.name for idx in pc.list_indexes()]
    if index_name not in existing_indexes:
        pc.create_index(
            name=index_name,
            dimension=dimension,
            metric=metric,
            spec=ServerlessSpec(cloud="aws", region="us-east-1"),
        )
        # インデックスが準備されるまで待機
        while not pc.describe_index(index_name).status["ready"]:
            time.sleep(1)
        print(f"インデックス'{index_name}'作成完了")
    else:
        print(f"インデックス'{index_name}'は既に存在")


def index_faq_documents(
    parent_docs: List[Document],
    child_docs: List[Document],
    index_name: str = "faq-chatbot",
) -> PineconeVectorStore:
    """
    Parent-Child構造のFAQドキュメントをPineconeにインデキシングする。
    - Childドキュメント：'search'ネームスペース（検索用）
    - Parentドキュメント：'context'ネームスペース（LLMコンテキスト用）
    """
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

    # Childドキュメントインデキシング（検索対象）
    child_store = PineconeVectorStore.from_documents(
        documents=child_docs,
        embedding=embeddings,
        index_name=index_name,
        namespace="search",
    )
    print(f"Childドキュメント{len(child_docs)}件インデキシング完了（namespace: search）")

    # Parentドキュメントインデキシング（コンテキスト提供用）
    parent_store = PineconeVectorStore.from_documents(
        documents=parent_docs,
        embedding=embeddings,
        index_name=index_name,
        namespace="context",
    )
    print(f"Parentドキュメント{len(parent_docs)}件インデキシング完了（namespace: context）")

    return child_store


# 実行
setup_pinecone_index()
child_vectorstore = index_faq_documents(parents, children)

"""
LangChainベースFAQチャットボット実装。
Parent-Child検索 + カスタムプロンプト + 会話履歴管理を統合する。
"""
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_core.documents import Document
from typing import List, Dict
import os


# 1. コンポーネント初期化
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

child_store = PineconeVectorStore(
    index_name="faq-chatbot",
    embedding=embeddings,
    namespace="search",
)
parent_store = PineconeVectorStore(
    index_name="faq-chatbot",
    embedding=embeddings,
    namespace="context",
)


# 2. Parent-Child検索器の実装
def retrieve_with_parent_lookup(query: str, k: int = 3) -> List[Document]:
    """
    Childチャンクで検索した後、マッチしたParentドキュメントを返す。
    これにより検索精度はChildレベル、コンテキストはParentレベルで確保される。
    """
    # Step 1: Childチャンクで類似度検索
    child_results = child_store.similarity_search(query, k=k * 2)

    # Step 2: 重複排除してユニークなparent_idを抽出
    seen_parent_ids = set()
    unique_parent_ids = []
    for doc in child_results:
        pid = doc.metadata.get("parent_id")
        if pid and pid not in seen_parent_ids:
            seen_parent_ids.add(pid)
            unique_parent_ids.append(pid)
        if len(unique_parent_ids) >= k:
            break

    # Step 3: Parentドキュメント検索
    parent_results = parent_store.similarity_search(
        query,
        k=k,
        filter={"parent_id": {"$in": unique_parent_ids}},
    )

    return parent_results


# 3. プロンプト設計
FAQ_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """あなたは顧客FAQを専門に回答するAIアシスタントです。
以下のルールを必ず遵守してください：

1. 提供されたFAQドキュメントに基づいてのみ回答してください。
2. FAQドキュメントに回答がない場合は「該当する質問への回答が見つかりませんでした。カスタマーセンター（1234-5678）にお問い合わせください。」と案内してください。
3. 推測したりFAQにない情報を生成したりしないでください。
4. 回答時に関連FAQドキュメントの出典を併せて案内してください。
5. 親切かつ簡潔に回答してください。

参考FAQドキュメント：
{context}"""),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{question}"),
])


# 4. チェーン構成
def format_docs(docs: List[Document]) -> str:
    """検索されたドキュメントをプロンプトに含める形式に変換する。"""
    formatted = []
    for i, doc in enumerate(docs, 1):
        source_info = doc.metadata.get("faq_id", "unknown")
        formatted.append(
            f"[FAQ-{source_info}]\n{doc.page_content}"
        )
    return "\n\n---\n\n".join(formatted)


faq_chain = (
    {
        "context": RunnableLambda(
            lambda x: format_docs(retrieve_with_parent_lookup(x["question"]))
        ),
        "question": RunnablePassthrough() | RunnableLambda(lambda x: x["question"]),
        "chat_history": RunnableLambda(lambda x: x.get("chat_history", [])),
    }
    | FAQ_PROMPT
    | llm
    | StrOutputParser()
)


# 5. 実行
response = faq_chain.invoke({
    "question": "返金処理にどのくらいかかりますか？",
    "chat_history": [],
})
print(response)

"""
BM25 + Denseハイブリッド検索実装。
Reciprocal Rank Fusion（RRF）で2つの検索結果を結合する。
"""
from langchain_community.retrievers import BM25Retriever
from langchain_core.documents import Document
from typing import List, Dict, Tuple
import numpy as np


class HybridRetriever:
    """BM25とベクトル検索を結合するハイブリッド検索器。"""

    def __init__(
        self,
        vector_store,
        documents: List[Document],
        bm25_k: int = 10,
        vector_k: int = 10,
        rrf_k: int = 60,
        alpha: float = 0.5,
    ):
        self.vector_store = vector_store
        self.bm25_retriever = BM25Retriever.from_documents(
            documents, k=bm25_k
        )
        self.vector_k = vector_k
        self.rrf_k = rrf_k
        self.alpha = alpha  # 0=BM25 only, 1=Dense only

    def _reciprocal_rank_fusion(
        self,
        bm25_results: List[Document],
        vector_results: List[Document],
    ) -> List[Tuple[Document, float]]:
        """RRFアルゴリズムで2つの検索結果を結合する。"""
        doc_scores: Dict[str, Tuple[Document, float]] = {}

        # BM25結果にRRFスコアを付与
        for rank, doc in enumerate(bm25_results):
            doc_key = doc.page_content[:100]
            score = (1 - self.alpha) / (self.rrf_k + rank + 1)
            if doc_key in doc_scores:
                doc_scores[doc_key] = (
                    doc,
                    doc_scores[doc_key][1] + score,
                )
            else:
                doc_scores[doc_key] = (doc, score)

        # Dense結果にRRFスコアを付与
        for rank, doc in enumerate(vector_results):
            doc_key = doc.page_content[:100]
            score = self.alpha / (self.rrf_k + rank + 1)
            if doc_key in doc_scores:
                doc_scores[doc_key] = (
                    doc,
                    doc_scores[doc_key][1] + score,
                )
            else:
                doc_scores[doc_key] = (doc, score)

        # RRFスコア基準で降順ソート
        sorted_results = sorted(
            doc_scores.values(), key=lambda x: x[1], reverse=True
        )
        return sorted_results

    def retrieve(self, query: str, top_k: int = 5) -> List[Document]:
        """ハイブリッド検索を実行する。"""
        # 並列検索（プロダクションではasyncio使用推奨）
        bm25_results = self.bm25_retriever.invoke(query)
        vector_results = self.vector_store.similarity_search(
            query, k=self.vector_k
        )

        # RRFで結合
        fused = self._reciprocal_rank_fusion(bm25_results, vector_results)

        return [doc for doc, score in fused[:top_k]]


# 使用例
hybrid_retriever = HybridRetriever(
    vector_store=child_store,
    documents=children,  # BM25用の原本ドキュメント
    alpha=0.6,  # Dense重み60%
)
results = hybrid_retriever.retrieve("エラーコード P4021 の解決方法")

                    +------------------+
                    |   Load Balancer  |
                    +--------+---------+
                             |
              +--------------+--------------+
              |                             |
    +---------v---------+    +---------v---------+
    |  API Server (1)   |    |  API Server (2)   |
    |  FastAPI + Uvicorn|    |  FastAPI + Uvicorn|
    +---------+---------+    +---------+---------+
              |                        |
    +---------v------------------------v---------+
    |              Redis Cache                    |
    |   (クエリエンベディングキャッシュ、回答キャッシュ) |
    +-----+-------------+-------------+----------+
          |             |             |
+---------v---+ +-------v-----+ +----v----------+
| Pinecone    | | BM25 Index  | | LLM API       |
| (Dense)     | | (Sparse)    | | (OpenAI/Azure)|
+-------------+ +-------------+ +---------------+

"""
RAGASベースFAQチャットボット品質評価。
Golden Datasetに対して自動的に評価を実行し、指標を算出する。
"""
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from ragas.llms import LangchainLLMWrapper
from ragas.embeddings import LangchainEmbeddingsWrapper
from datasets import Dataset
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from typing import List, Dict
import json
from datetime import datetime


def prepare_evaluation_dataset(
    test_cases: List[Dict],
    retriever,
    chain,
) -> Dataset:
    """
    テストケースをRAGAS評価形式に変換する。
    各質問に対して実際の検索と回答生成を実行する。
    """
    eval_data = {
        "question": [],
        "answer": [],
        "contexts": [],
        "ground_truth": [],
    }

    for case in test_cases:
        question = case["question"]

        # 実際の検索を実行
        retrieved_docs = retriever.retrieve(question, top_k=5)
        contexts = [doc.page_content for doc in retrieved_docs]

        # 実際の回答を生成
        answer = chain.invoke({
            "question": question,
            "chat_history": [],
        })

        eval_data["question"].append(question)
        eval_data["answer"].append(answer)
        eval_data["contexts"].append(contexts)
        eval_data["ground_truth"].append(case["expected_answer"])

    return Dataset.from_dict(eval_data)


def run_ragas_evaluation(dataset: Dataset) -> Dict:
    """RAGAS評価を実行し結果を返す。"""
    eval_llm = LangchainLLMWrapper(ChatOpenAI(model="gpt-4o", temperature=0))
    eval_embeddings = LangchainEmbeddingsWrapper(
        OpenAIEmbeddings(model="text-embedding-3-small")
    )

    result = evaluate(
        dataset=dataset,
        metrics=[
            faithfulness,
            answer_relevancy,
            context_precision,
            context_recall,
        ],
        llm=eval_llm,
        embeddings=eval_embeddings,
    )

    # 結果を保存
    report = {
        "timestamp": datetime.now().isoformat(),
        "dataset_size": len(dataset),
        "metrics": {
            "faithfulness": float(result["faithfulness"]),
            "answer_relevancy": float(result["answer_relevancy"]),
            "context_precision": float(result["context_precision"]),
            "context_recall": float(result["context_recall"]),
        },
    }

    # デプロイゲート：すべての指標が閾値以上であること
    thresholds = {
        "faithfulness": 0.9,
        "answer_relevancy": 0.85,
        "context_precision": 0.8,
        "context_recall": 0.9,
    }
    report["deployment_gate"] = all(
        report["metrics"][k] >= v for k, v in thresholds.items()
    )

    with open(f"eval_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json", "w") as f:
        json.dump(report, f, indent=2, ensure_ascii=False)

    return report


# 実行例
test_cases = [
    {
        "question": "返金処理は何日かかりますか？",
        "expected_answer": "返金は申請日から営業日基準3-5日以内に処理されます。",
    },
    {
        "question": "海外配送料はいくらですか？",
        "expected_answer": "海外配送料は地域と重量によって異なり、関税は受取人負担です。",
    },
]

# eval_dataset = prepare_evaluation_dataset(test_cases, hybrid_retriever, faq_chain)
# report = run_ragas_evaluation(eval_dataset)
# print(f"デプロイゲート通過: {report['deployment_gate']}")

"""
FAQチャットボット運用モニタリング。
リクエスト別指標収集、異常検知、アラート送信を担当する。
"""
import time
import logging
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime
from prometheus_client import (
    Counter,
    Histogram,
    Gauge,
    start_http_server,
)

logger = logging.getLogger(__name__)

# Prometheusメトリクス定義
REQUEST_COUNT = Counter(
    "faq_chatbot_requests_total",
    "Total FAQ chatbot requests",
    ["status", "category"],
)
RESPONSE_LATENCY = Histogram(
    "faq_chatbot_response_seconds",
    "Response latency in seconds",
    buckets=[0.5, 1.0, 2.0, 3.0, 5.0, 10.0],
)
RETRIEVAL_LATENCY = Histogram(
    "faq_chatbot_retrieval_seconds",
    "Retrieval latency in seconds",
    buckets=[0.05, 0.1, 0.2, 0.5, 1.0],
)
LLM_TOKENS_USED = Counter(
    "faq_chatbot_llm_tokens_total",
    "Total LLM tokens consumed",
    ["type"],  # prompt, completion
)
FALLBACK_RATE = Gauge(
    "faq_chatbot_fallback_rate",
    "Current fallback (no answer) rate",
)
ACTIVE_REQUESTS = Gauge(
    "faq_chatbot_active_requests",
    "Currently processing requests",
)


@dataclass
class RequestMetrics:
    """単一リクエストのメトリクスを収集するコンテキストマネージャー。"""
    question: str
    start_time: float = field(default_factory=time.time)
    retrieval_time: Optional[float] = None
    llm_time: Optional[float] = None
    total_time: Optional[float] = None
    status: str = "success"
    is_fallback: bool = False
    prompt_tokens: int = 0
    completion_tokens: int = 0

    def record_retrieval(self):
        self.retrieval_time = time.time() - self.start_time

    def record_llm_start(self):
        self._llm_start = time.time()

    def record_llm_end(self, prompt_tokens: int, completion_tokens: int):
        self.llm_time = time.time() - self._llm_start
        self.prompt_tokens = prompt_tokens
        self.completion_tokens = completion_tokens

    def finalize(self):
        self.total_time = time.time() - self.start_time

        # Prometheusメトリクスを記録
        REQUEST_COUNT.labels(
            status=self.status, category="faq"
        ).inc()
        RESPONSE_LATENCY.observe(self.total_time)

        if self.retrieval_time:
            RETRIEVAL_LATENCY.observe(self.retrieval_time)

        LLM_TOKENS_USED.labels(type="prompt").inc(self.prompt_tokens)
        LLM_TOKENS_USED.labels(type="completion").inc(
            self.completion_tokens
        )

        # 構造化ロギング
        logger.info(
            "faq_request_completed",
            extra={
                "question_preview": self.question[:50],
                "total_time_ms": round(self.total_time * 1000),
                "retrieval_time_ms": round(
                    (self.retrieval_time or 0) * 1000
                ),
                "status": self.status,
                "is_fallback": self.is_fallback,
                "tokens": self.prompt_tokens + self.completion_tokens,
            },
        )


# Prometheusメトリクスサーバー起動
# start_http_server(8001)  # /metricsエンドポイント公開