【実践】ElasticSearch × RAG 入門 (5) – LangChainを用いてRAGを実装してみよう

前回の記事では、ElasticSearchのインデックスを実際にpythonで構築し、ベクトル検索を実践してみました。

【実践】ElasticSearch × RAG 入門 (4) – 埋め込みモデルを用いたHybrid Searchへ

前回の記事では、Elasticsearchを用いたBM25ベースの全文検索手法について解説しました。今回はより高度な検索体験を提供するために、ベクトル（Embedding）検索について詳しく説明します。これにより、ユーザーの意図や文脈を考慮...

www.next-ai-agent.com

2025.05.21

今回がこのシリーズの最終記事です。いよいよelastic searchでragを実践してみます。

1. LangChainを使わずにRAGを構築するメリット

LangChainは強力なフレームワークですが、すべてのケースで必要とは限りません。シンプルな構成であれば、LangChainを使わずにElasticsearchとOpenAIのAPIを直接組み合わせて、RAG（Retrieval-Augmented Generation）を実装する方が柔軟かつ制御しやすいこともあります。

• 自由度の高い設計が可能
• 実装の中身が見えやすく、デバッグが容易
• 不要な抽象化がない分、パフォーマンスやコストを最適化しやすい

2. Elasticsearch × OpenAIで作るハイブリッドRAGシステム（LangChainなし）

以下は、LangChainを使わずにハイブリッド検索とRAGの応答生成を実現するPythonコードです。

import os
from typing import List, Dict, Any

import openai
from elasticsearch import Elasticsearch
from langchain_openai import OpenAIEmbeddings

openai.api_key = os.getenv("OPENAI_API_KEY")  # 環境変数はよしなに指定してください

INDEX_NAME = "botchan"
es = Elasticsearch("http://localhost:9200")
embedding_model = OpenAIEmbeddings(model="text-embedding-3-small")

def create_hybrid_query(query: str, embedding_model, vector_weight: float = 0.7, keyword_weight: float = 0.3, size: int = 5):
    query_vec = embedding_model.embed_query(query)
    return {
        "bool": {
            "should": [
                {
                    "knn": {
                        "field": "line_embedding",
                        "query_vector": query_vec,
                        "k": size,
                        "num_candidates": 100,
                        "boost": vector_weight
                    }
                },
                {
                    "match": {
                        "line": {
                            "query": query,
                            "boost": keyword_weight
                        }
                    }
                }
            ]
        }
    }

def hybrid_search(query: str, k: int = 5, vector_weight: float = 0.7, keyword_weight: float = 0.3) -> List[Dict[str, Any]]:
    hybrid_query = create_hybrid_query(query, embedding_model, vector_weight, keyword_weight, size=k)
    body = {"query": hybrid_query, "size": k}
    results = es.search(index=INDEX_NAME, body=body)
    return [
        {
            "text": hit["_source"]["line"],
            "line_number": hit["_source"]["line_number"],
            "score": hit["_score"],
            "search_type": "hybrid"
        }
        for hit in results["hits"]["hits"]
    ]

def answer_with_rag(question: str, k: int = 5) -> str:
    search_results = hybrid_search(question, k=k)
    search_results.sort(key=lambda x: x["line_number"])
    context = "\n".join([r["text"] for r in search_results])

    system_prompt = """
    あなたは日本文学「坊っちゃん」の専門家です。提供されたテキスト断片のみに基づいて質問に答えてください。
    """
    user_prompt = f"""
    以下のテキスト断片を参考にして質問に答えてください。

    テキスト断片:
    ---
    {context}
    ---

    質問: {question}
    """

    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        temperature=0.3,
    )

    return response.choices[0].message.content

# 実行例
if __name__ == "__main__":
    query = "庭には何の木が生えている？"
    print("=== ハイブリッド検索のテスト ===")
    results = hybrid_search(query, k=5)
    for i, result in enumerate(results, 1):
        print(f"{i}. {result['text']} (score: {result['score']:.4f})")

    print("\n=== RAGによる回答 ===")
    answer = answer_with_rag(query)
    print(f"回答: {answer}")

# 出力例:
=== ハイブリッド検索のテスト ===
「庭には何の木が生えている？」に関する検索結果:
1. 庭を東へ二十歩に行き尽つくすと、南上がりにいささかばかりの菜園があって、真中まんなかに栗くりの木が一本立っている。これは命より大事な栗だ。実の熟する時分は起き抜けに背戸せどを出て落ちた奴を拾ってきて、... (スコア: 2.3623)
=== RAGを使った質問応答のテスト ===

質問: 庭には何の木が生えている？
回答: 庭には栗の木が一本生えています。

期待する結果ですね🎉

3. 同じ処理をLangChainで実装するには？

def answer_with_rag_chain(question: str, k: int = 5) -> str:
    """
    LangChainのチェーンを使用してRAGを実装した質問応答関数
    
    Args:
        question: 質問文
        k: 使用するコンテキストの数
    
    Returns:
        質問への回答
    """
    # 質問に関連するコンテキストを検索
    search_results = hybrid_search(
        query=question,
        k=k,
        vector_weight=0.7,
        keyword_weight=0.3
    )
    
    # 検索結果を行番号順にソート（文脈を維持するため）
    search_results.sort(key=lambda x: x["line_number"])
    
    # コンテキストを構築
    context = "\n".join([result["text"] for result in search_results])
    
    # プロンプトテンプレートを作成
    prompt = ChatPromptTemplate.from_messages([
        ("system", "あなたは日本文学「坊っちゃん」の専門家です。提供されたテキスト断片のみに基づいて質問に答えてください。"),
        ("user", """以下のテキスト断片を参考にして質問に答えてください。

テキスト断片:
---
{context}
---

質問: {question}
""")
    ])
    
    # LLMChainを作成
    chain = prompt | llm  # この記法をLCEL(LangChain Expression Language)といいます
    
    # チェーンを実行して回答を生成
    response = chain.invoke({"context": context, "question":question})
    
    return response

LangChainを使うと、LLMとの連携や履歴管理が簡潔になりますが、検索部分の制御性はやや失われます。
実際、langchain公式のRetrieverを使うともう少し実装がスッキリしますが、langchainオリジナルのDocumentクラスで検索結果を扱うことになるので、構造化された情報を扱うのが大変になります。

4. ハイブリッド検索の設計ポイント

• vector_weight と keyword_weight のバランス調整で検索品質が変わる
• Embeddingモデルの精度も影響がある
• weightは0.5, 0.5でそこまでいじらなくても良い。Mappingの構造のほうが圧倒的に重要

4. LangChainとの比較（まとめ）

項目	LangChainあり	LangChainなし
実装の自由度	抽象化が強く制限あり	自由自在で柔軟な設計が可能
デバッグのしやすさ	抽象化により複雑	実装が明示的でトレースしやすい
セットアップ	多機能だが学習コスト高	シンプルで最小構成から始められる
拡張性	多くのコンポーネントが用意されている	拡張には自作が必要

おわりに

入門（1）　〜　入門（５）まで、シリーズを通してelasticsearchを用いたRAGの方法を紹介してきました。

RAGでは検索の性能が非常に重要です。
ElasticSearchを用いれば、複雑な検索設定も簡単に構築することができます。

実際のRAGアプリケーションで、あまりタスクが解けない場合は

• 「クエリに対してうまく検索ができているのか？」
• 「検索の後のプロンプトでは正しく指示が記述されているか？」

をうまく切り分けてエラー分析をすることが重要です。

ともあれ、お疲れ様でした🎉