🐤

LangGraphのTool Callingを利用して、RAG Agentsを構築する（後編）

2025/03/04に公開

LangChain

はじめに

この記事の目的は、前編の記事をご覧ください。
では、早速前編記事のコードの解説を実施していきます。

LangChainのカスタムRetrieverクラスの実装だったり、LangChainの内部で類似度と距離がごっちゃになっている話など、中身を詳細に理解したい人にとっては面白い話もあると思うので、ぜひみてもらえると嬉しいです！

参考文献

（書籍のリンクはamazonアフィリエイトリンクです）

記事

LangChainからLangGraphによるAgent構築への移行方法

Chroma DBによる類似度検索のメソッド

カスタムRetrieverクラスの作成方法

LangChainでのドキュメントローダ

GoogleのGen AIモデル

書籍

LangChainとLangGraphによるRAG・AIエージェント［実践］入門
 ChatGPT/LangChainによるチャットシステム構築［実践］入門
LangChainを利用することで、あらゆるモデルを統一的なコードで実行できるようになります。
langchainに関しては、こちらの書籍を読めば大体のことはできるようになりますので、おすすめです。
また、現在推奨されているLangGraphでのRAG Agentを構築するcreate_react_agentに関しても説明されておりますし、さらに複雑なAgentsの構築方法やデザイン方法も網羅されており、とても勉強になります！

大規模言語モデル入門
 大規模言語モデル入門Ⅱ〜生成型LLMの実装と評価
よく紹介させていただいておりますが、こちらの書籍は、LLMのファインチューニングから、RLHF、RAG、分散学習にかけて、本当に幅広く解説されており、いつも参考にさせていただいております。
今回の記事で紹介したRAGの内容だけでなく、さらにその先であるRAGを前提としてInstruction Tuningについても触れており、とても面白いです。
LLMを取り扱っている方は、とりあえず買っておいても損はないと思います。
さまざまな章、ページで読めば読むほど新しい発見が得られる、スルメのような本だなといつも思っております。

LLMのファインチューニングとRAG ―チャットボット開発による実践
上記2冊の本よりもRAGやファインチューニングに絞って記載されている書籍です。だいぶ平易に書いてあるのでとてもわかりやすいと思いました。
また、本記事の内容ではないですが、RAGを実装する上でキーワード検索を加えたハイブリッド検索を検討することは一般的だと思います。本書はそこにも踏み込んで解説をしています。
また、キーワード検索でよく利用するBM25Retrieverが日本語のドキュメントに利用する際に一工夫がいるところなども紹介されており、使いやすい本だなと思いました。

成果物

下記のGithubをご覧ください。

コードの解説

ドキュメントをEmbeddingsモデルでベクトル化し、DBに格納する

実際のコードは下記になります。

定数定義

    # --- 定数定義 ---
    EM_MODEL_NAME = "models/text-embedding-004"
    RAG_FOLDER_PATH = "./inputs"
    CHROMA_DB = "./chroma/chroma_langchain_db"
    CHROMA_NAME = "example_collection"

利用するEmbeddingsモデルや、Chroma DBの名前、保存場所などを定義しています。
今回はEmbeddingsモデルとしてtext-embedding-004を利用しています。
理由は、無料だから以外ないです。

Embedingモデルを定義する

# テキスト エンベディング モデルを定義する (dense embedding)
embedding_model = GoogleGenerativeAIEmbeddings(model=EM_MODEL_NAME)

ここで、ドキュメントを埋め込みベクトル化する、Embeddingsモデルを定義しています。
下記の無料のモデルを利用します。

Chroma DBを定義する

vector_store = Chroma(
        collection_name=CHROMA_NAME,
        embedding_function=embedding_model,
        persist_directory=CHROMA_DB,  # Where to save data locally, remove if not necessary
    )

LangChainのラッパーを利用して、Chroma DBを定義します。
これだけで、ローカルに指定したDBを作成することができます。簡単ですね。

ここで、Embeddingsモデルを指定するので、ドキュメントを格納する際に、指定のメソッドを利用すれば、埋め込みベクトル化したうえで格納してくれます。

また、persist_directoryに、ローカルのパスを指定することで、そこにDBを永続化してくれるので、別のコードからも呼び出せるようになります。
今回の記事では、実際にRAGを利用してLLMに回答させるコードから呼び出すことになります。

ドキュメントを読み込み、チャンクやID、メタデータなどを作成する

def load_text_files_from_folder(folder_path):
    """
    指定したフォルダ内のすべてのテキストファイル(.txt)を読み込む関数
    
    :param folder_path: 読み込むフォルダのパス
    :return: 読み込んだドキュメントのリスト
    """
    # フォルダ内のすべての .txt ファイルを取得
    text_files = glob.glob(os.path.join(folder_path, "*.txt"))

    # すべてのテキストファイルを読み込む
    documents = []
    for file in text_files:
        loader = TextLoader(file)
        documents.extend(loader.load())  # 各ファイルの内容をリストに追加

    print(f"Loaded {len(documents)} documents from {folder_path}")
    return documents  # 読み込んだドキュメントのリストを返す

def main():

    ・・・

    # テキストファイルを読み込む
    documents = load_text_files_from_folder(RAG_FOLDER_PATH)
    
    # チャンクに分割
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=500, 
        chunk_overlap=100,
        separators=[
        "\n\n",
        "\n",
        " ",
        ".",
        ",",
        "\u200b",  # Zero-width space
        "\uff0c",  # Fullwidth comma
        "\u3001",  # Ideographic comma
        "\uff0e",  # Fullwidth full stop
        "\u3002",  # Ideographic full stop
        "",
    ],)
    
    doc_splits = text_splitter.split_documents(documents)
    
    # チャンクのテキスト部分を抽出
    texts = [doc.page_content for doc in doc_splits]
    
    # optional IDs とメタデータ
    ids = ["i_" + str(i + 1) for i in range(len(texts))]
    metadatas = [{"my_metadata": i} for i in range(len(texts))]

　　・・・

ドキュメントの読み込み

ここでは、まずload_text_files_from_folder関数を利用して、ローカルにおいてあるテキストファイルをDocument形式で読み込みます。

関数の中ではTextLoaderを利用して、テキストファイルを読み込んでいます。これもLangChainの便利機能です。
そのほかにもPDFやCSVなどを読み込むためのクラスも実装されています。詳しくは下記をご覧ください。

上記を利用することで、ファイルの拡張子やメタデータに合わせて、フォルダ内のさまざまなファイルを読み取ることができます。

得られたデータは、documents.extend(loader.load())により、documentsに結合されていきます。

チャンク分割

続いて、得られた文章データはチャンクに分割されます。
今回はRecursiveCharacterTextSplitterを利用しています。

こちらを利用することで、文章中のセパレータ（separators）に反応して分割してくれます。
したがって、（あまりにも上記のセパレータが出現しない文章では難しいですが）ある程度、段落などの文脈にしたがって文章を分割してくれます。
また、下記の記事を参考に、separatorsを増やしています。日本語だと増やした方が良さそうです。（今回の実験の範囲では変わらなかったですが）

また、今回は文章量が少ないかつ1ファイルのため、500文字単位で分割、100文字のオーバーラップを行います。
500文字以内の最大部分にて、上記のセパレータが反応した箇所で文章が分割されます。

テキスト、ID、メタデータの設定

最終的に下記の部分で、テキスト部分を抽出したり、IDやメタデータを振っています。

    doc_splits = text_splitter.split_documents(documents)

    # チャンクのテキスト部分を抽出
    texts = [doc.page_content for doc in doc_splits]

    # optional IDs とメタデータ
    ids = ["i_" + str(i + 1) for i in range(len(texts))]
    metadatas = [{"my_metadata": i} for i in range(len(texts))]

今回は、かなり適当にIDやメタデータを振っていますが、LangGraphのエージェントにRAGを実施させる場合、このIDやメタデータの情報も与えることができます。
したがって、ドキュメントの情報（例えば文章のファイル名やページ数など）をきちんと指定してあげることで、顧客のQ&A対応の際に、一次情報を提示してあげることもできると思います。

ベクトル化してDBに格納

    # ---- dense embedding ----
    dense_embeddings = embedding_model.embed_documents(texts)
    
    # embeddingsの中身を確認
    print("dense embeddings（一部）:", dense_embeddings[0][:5])  # 最初の埋め込みを確認
    print("dense embeddings length:", len(dense_embeddings))

    #https://github.com/langchain-ai/langchain/blob/5d581ba22c68ab46818197da907278c1c45aad41/libs/partners/chroma/langchain_chroma/vectorstores.py#L502
    result = vector_store.add_texts(
        texts=texts,
        metadatas=metadatas,
        ids=ids,
    )

上記にてドキュメントをdense_embeddingsにベクトル化しています。
その上で、Chroma DBにadd_textsメソッドを利用して、ベクトルを格納しています。

add_textsメソッドは、LangChainにおいて大体のDBで実装されているので、そのまま使えることが多いですが、この辺りの実装はコントリビュータによって異なることもあるため、大元のコードを一旦確認することをお勧めします。
もっと使いやすいメソッドもあったりするので。

格納できているか確認

ここまでで、完了ではありますが、本当にDBに格納できているのかを確認しました。

    #　以下は、chroma DBに保存されたデータの中身を確認するためのコード
    # 全データの取得 (ドキュメントとメタデータだけ取得する例)
    data = vector_store._collection.get(include=["documents", "metadatas","embeddings"])

    print("Documents（3件）:", data["documents"][:3])
    print("Metadatas（全件）:", data["metadatas"])
    print("Embeddings（一部）:", data["embeddings"][0][:5])

こちらを実行すると、Chroma DBに格納されたベクトルデータのドキュメント情報とメタデータ情報を取得できます。

DBに保存された内容から、LLMがRAGで質問に回答する

実際のコードは下記になります。

こちらも同様に解説していきます。

定数定義

 --- 定数定義 ---
LLM_MODEL_NAME = "gemini-2.0-flash-001" 
EM_MODEL_NAME = "models/text-embedding-004"
CHROMA_DB = "./chroma/chroma_langchain_db"
CHROMA_NAME = "example_collection"

# 質問文
query = "16歳未満のユーザーが海外から当社サービスを利用した場合、親権者が同意していないときはどう扱われますか？ そのときデータは国外にも保存される可能性がありますか？"
    

ここでは、前回のコードで設定したChroma DBの情報と、利用するEmbeddingsモデル、LLMモデルを指定します。
また、ユーザからの質問queryもここで定義しておきます。

ちなみに、この質問は、プライバシーポリシーの第8条と第10条に回答がのっている質問になります。

モデル、DBの定義

    # テキスト エンベディング モデルを定義する (dense embedding)
    embedding_model = GoogleGenerativeAIEmbeddings(model=EM_MODEL_NAME)

    # Chroma
    vector_store = Chroma(
        collection_name=CHROMA_NAME,
        embedding_function=embedding_model,
        persist_directory=CHROMA_DB,  # Where to save data locally, remove if not necessary
    )

    # Chatモデル (LLM)
    llm = ChatGoogleGenerativeAI(
        model=LLM_MODEL_NAME,
        temperature=0.2,
        max_tokens=512,
    )

基本的に、上のコードと同じです。
LLMは、gemini-2.0-flash-001を選定しています。無料のモデルです。

DBから検索するロジックを実装する

class VectorSearchRetriever(BaseRetriever):
    """
    ベクトル検索を行うためのRetrieverクラス。
    """
    vector_store: SkipValidation[Any]
    embedding_model: SkipValidation[Any]
    k: int = 5 # 返すDocumentのチャンク数

    class Config:
        arbitrary_types_allowed = True

    def _get_relevant_documents(self, query: str) -> List[Document]:
        # Dense embedding
        embedding = self.embedding_model.embed_query(query)
        search_results = self.vector_store.similarity_search_by_vector_with_relevance_scores(
            embedding=embedding,
            k=self.k,
        )

        # Document のリストだけ取り出す
        return [doc for doc, _ in search_results]

    async def _aget_relevant_documents(self, query: str) -> List[Document]:
        return self._get_relevant_documents(query)

・・・

def main():
    ・・・
    # DenceRetrieverを用意
    dence_retriever = VectorSearchRetriever(
        vector_store=vector_store,
        embedding_model=embedding_model,
        k=5,
    )

    ・・・

LangChainでRAGを実装する場合には、Retrieverという便利なクラスを利用することになります。
今回利用したChroma DBであれば、as_retrieverメソッドという超便利なメソッドがあるので、下記のように簡単にRetrieverクラスとして利用できます。

dence_retriever = vector_store.as_retriever()

しかしながら、Google Cloudのいろんなサービスを見ていると、必ずしも.as_retriever()メソッドに対応しているベクトルストアだけでは無いように見えています。
（今後、どんどん対応はされていくとは思うのですが）

しかしながら、.as_retriever()メソッドに対応していなくても、自分でカスタムRetrieverクラスを作成して、同じことをすることがLangChainでは可能です。
そこで、自分の勉強のためにも、それを実施することにしました。

カスタムRetrieverクラスの作り方は下記に詳しく記載されています。

こちらを確認すると、どうもBaseRetrieverクラスを継承した上で、同期メソッドの_get_relevant_documentsと非同期メソッドの_aget_relevant_documentsを実装すれば良さそうです。

_get_relevant_documentsメソッドでは、ユーザの質問内容のベクトル化と、そのベクトルとDBに格納されているベクトルの類似度を計算して、該当するドキュメントを取得する処理を実装すれば良さそうです。

Chroma DBのLangChain実装元を確認すると、similarity_search_by_vector_with_relevance_scoresメソッドなど、さまざまなメソッドが利用できます。

例えばchroma DBで実装されているのは下記です。

similarity_search
- クエリ自体を入力とし、類似度の高いk件のドキュメントを取得するメソッドです。
- 内部でクエリをEmbeddingsモデルに入力し、ベクトル化しています。
- 内部的には、similarity_search_with_scoreメソッドが呼ばれており、その中のドキュメント部分だけが返却されます
similarity_search_with_score
- クエリ自体を入力とし、類似度の高いk件のドキュメントを取得します。さらに追加で計算した類似度も出力するメソッドです。
  - ただし、ここの類似度は距離になっており、値が小さいほど類似しています。
  - コサイン距離なら（1-コサイン類似度）です。
similarity_search_by_vector
- 自分らでEmbeddingしたベクトルを入力とし、類似度の高いk件のドキュメントを取得するメソッドです。
similarity_search_by_vector_with_relevance_scores
- 自分らでEmbeddingしたベクトルを入力とし、類似度の高いk件のドキュメントを取得します。さらに追加で計算した類似度（距離）も出力するメソッドです。
  - こちらも距離が小さいほど、類似しています。
similarity_search_with_vectors
- クエリ自体を入力とし、類似度の高いk件のドキュメントを取得します。さらに返却時に、ドキュメントの埋め込みベクトル自体の出力します。
similarity_search_by_image（画像系）
- 画像を入力すると、その画像に類似した画像をk件出力するメソッドです
- これを利用するためには、vector_storeに設定するEnbeddingsモデルが、画像の埋め込みに対応している必要があります
similarity_search_by_image_with_relevance_score（画像系）
- 上記メソッドとほぼ同様ですが、類似度（距離）も出力します。
- こちらももちろん、距離の数値が小さい方が類似度が高いです。
max_marginal_relevance_search（MMR系）
- MMR（Maximal marginal relevance optimizes）という手法を利用して、類似する文章を取得する際に、他の文章との多様性を考慮しながら、文章を取得するメソッドです。
- クエリ分を入力し、一旦fetch_k件のドキュメントを取得したのちに、MMRを用いて、k件に絞ります。この時内容が類似しているドキュメントから削除される形です
max_marginal_relevance_search_by_vector（MMR系）
- 上のメソッドと同様ですが、あらかじめEmbeddingしたベクトルを利用して検索するメソッドです。

今回は、なんでもいいんですが、メソッド名が長い方がかっこいいので、similarity_search_by_vector_with_relevance_scoresメソッドを利用しています。

実際の実装では、下記のように利用しています。

    def _get_relevant_documents(self, query: str) -> List[Document]:
        # Dense embedding
        embedding = self.embedding_model.embed_query(query)
        search_results = self.vector_store.similarity_search_by_vector_with_relevance_scores(
            embedding=embedding,
            k=self.k,
        )

        # Document のリストだけ取り出す
        return [doc for doc, _ in search_results]

メソッド通り、先にあらかじめembedding_model.embed_queryにより、クエリをベクトル化しています。
その後、similarity_search_by_vector_with_relevance_scoresメソッドを利用して、k件のドキュメントを取得します。
その後、距離は不要なので、Documentだけをreturnしています。
（なら、「最初からsimilarity_searchメソッドだけでいいじゃん」というのはその通りです）

ただ、今後の拡張性が期待できるかな？という希望だけで実装しています。

最後にmain関数内にて、下記のように利用しています。

# DenceRetrieverを用意
    dence_retriever = VectorSearchRetriever(
        vector_store=vector_store,
        embedding_model=embedding_model,
        k=5,
    )

重ねてお伝えしますが、下記の実装で十分です。（というか等価です）

dence_retriever = vector_store.as_retriever(search_kwargs={'k': 5})

（補足：細かい話）　as_retrieverの実装

ちなみにas_retrieverには、3つの設定が可能です。

"similarity" (default)
"mmr"
"similarity_score_threshold".

ここで、similarityでは、内部的にsimilarity_searchメソッドが最終的に呼ばれます。

下記のように追っていきます。

上記がas_retrieverメソッドです。ここでは最終的にVectorStoreRetrieverクラスを作成します。

VectorStoreRetrieverクラスは、私が上記で実施していたように、Retrieverクラスとして実装されています。（つまり本質的にやっていることは同じ）
したがって、_get_relevant_documentsメソッドを見れば、どんな処理をしているかがわかります。

中身は下記のようになっています。

    def _get_relevant_documents(
        self, query: str, *, run_manager: CallbackManagerForRetrieverRun, **kwargs: Any
    ) -> list[Document]:
        _kwargs = self.search_kwargs | kwargs
        if self.search_type == "similarity":
            docs = self.vectorstore.similarity_search(query, **_kwargs)
        elif self.search_type == "similarity_score_threshold":
            docs_and_similarities = (
                self.vectorstore.similarity_search_with_relevance_scores(
                    query, **_kwargs
                )
            )
            docs = [doc for doc, _ in docs_and_similarities]
        elif self.search_type == "mmr":
            docs = self.vectorstore.max_marginal_relevance_search(query, **_kwargs)
        else:
            msg = f"search_type of {self.search_type} not allowed."
            raise ValueError(msg)
        return docs

ここで、similarityとsimilarity_score_thresholdとmmrの選択肢が現れます。

ここで、self.search_type == "similarity"を指定しているとき（つまりDefault）では、similarity_searchメソッドが呼ばれていることがわかります。
したがって、私の実装と等価なわけです。

（補足：細かい話2）　類似度と距離の混在

内部実装を読んでいると、距離だが類似度だが、入り乱れていてメチャクチャ混乱します。

例えば、「細かい話1」の例で言うとsimilarity_score_thresholdのパターンを追っていくとわかります。

上記の、as_retrieverメソッドの実装のコメントを見ると、下記のように利用することを想定しているようです。

# Only retrieve documents that have a relevance score
# Above a certain threshold
docsearch.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={'score_threshold': 0.8}
)

結論を言うと、上記の書き方においては、閾値は「類似度」です。
つまり、1が最も類似していて、0が最も離れています。

はあ？？？？
さっきは、距離だから0が最も類似度が高いって言ってたじゃねえか！！

はい、おっしゃる通りです。
なぜこれが起こるのかは、おそらくですがDBごとに実装が異なっており、うまく間に変換が挟まっているからです。

実際に見ていきましょう。

search_type="similarity_score_threshold"を指定した場合は、similarity_search_with_relevance_scoresメソッドが呼ばれます。

これは、下記のような実装になっています。

    def similarity_search_with_relevance_scores(
        self,
        query: str,
        k: int = 4,
        **kwargs: Any,
    ) -> list[tuple[Document, float]]:
        """Return docs and relevance scores in the range [0, 1].

        0 is dissimilar, 1 is most similar.

        Args:
            query: Input text.
            k: Number of Documents to return. Defaults to 4.
            **kwargs: kwargs to be passed to similarity search. Should include:
                score_threshold: Optional, a floating point value between 0 to 1 to
                    filter the resulting set of retrieved docs.

        Returns:
            List of Tuples of (doc, similarity_score).
        """
        score_threshold = kwargs.pop("score_threshold", None)

        docs_and_similarities = self._similarity_search_with_relevance_scores(
            query, k=k, **kwargs
        )
        if any(
            similarity < 0.0 or similarity > 1.0
            for _, similarity in docs_and_similarities
        ):
            warnings.warn(
                "Relevance scores must be between"
                f" 0 and 1, got {docs_and_similarities}",
                stacklevel=2,
            )

        if score_threshold is not None:
            docs_and_similarities = [
                (doc, similarity)
                for doc, similarity in docs_and_similarities
                if similarity >= score_threshold
            ]
            if len(docs_and_similarities) == 0:
                logger.warning(
                    "No relevant docs were retrieved using the relevance score"
                    f" threshold {score_threshold}"
                )
        return docs_and_similarities

後半部分を読めばわかるように、similarity >= score_thresholdである文章だけを最終的に取得します。
そして、上部のコメントに記載されている通り、「0 is dissimilar, 1 is most similar.」となります。

さて、実際にドキュメントを取得しているメソッドは_similarity_search_with_relevance_scoresです。

これは下記のような実装になっています。

    def _similarity_search_with_relevance_scores(
        self,
        query: str,
        k: int = 4,
        **kwargs: Any,
    ) -> list[tuple[Document, float]]:
        """Default similarity search with relevance scores. Modify if necessary
        in subclass.
        Return docs and relevance scores in the range [0, 1].

        0 is dissimilar, 1 is most similar.

        Args:
            query: Input text.
            k: Number of Documents to return. Defaults to 4.
            **kwargs: kwargs to be passed to similarity search. Should include:
                score_threshold: Optional, a floating point value between 0 to 1 to
                    filter the resulting set of retrieved docs

        Returns:
            List of Tuples of (doc, similarity_score)
        """
        relevance_score_fn = self._select_relevance_score_fn()
        docs_and_scores = self.similarity_search_with_score(query, k, **kwargs)
        return [(doc, relevance_score_fn(score)) for doc, score in docs_and_scores]

ここにおいて、実際に文章を取得しているのはsimilarity_search_with_scoreは、今見ている継承元であるVectorStoreクラスでは実装されていません。（証拠）

つまり、このVectorStoreクラスを継承している、Chroma DBのクラスであるclass Chroma(VectorStore)の実装を利用します。

つまり下記の実装であり、前述した通り、スコアは「距離」として計算されています。

similarity_search_with_score
- クエリ自体を入力とし、類似度の高いk件のドキュメントを取得します。さらに追加で計算した類似度も出力するメソッドです。
- ただし、ここの類似度は距離になっており、値が小さいほど類似しています。
- コサイン距離なら（1-コサイン類似度）です。

さて、この乖離を埋めるのが、上記コードの残りの部分である、下記の実装です。

return [(doc, relevance_score_fn(score)) for doc, score in docs_and_scores]

よく見ると、スコアの部分に、relevance_score_fn関数が適用されています。
そしてこの関数は、下記のように定義されています。

relevance_score_fn = self._select_relevance_score_fn()

この_select_relevance_score_fnメソッドは、やはりVectorStoreクラスでは実装されていません。（証拠）

これは、DBごとに、「類似度なのか」「距離なのか」の実装が違うからです。

では、今回のclass Chroma(VectorStore)の実装では、下記のようになっています。

    def _select_relevance_score_fn(self) -> Callable[[float], float]:
        """Select the relevance score function based on collections distance metric.

        The most similar documents will have the lowest relevance score. Default
        relevance score function is euclidean distance. Distance metric must be
        provided in `collection_metadata` during initialization of Chroma object.
        Example: collection_metadata={"hnsw:space": "cosine"}. Available distance
        metrics are: 'cosine', 'l2' and 'ip'.

        Returns:
            The relevance score function.

        Raises:
            ValueError: If the distance metric is not supported.
        """
        if self.override_relevance_score_fn:
            return self.override_relevance_score_fn

        distance = "l2"
        distance_key = "hnsw:space"
        metadata = self._collection.metadata

        if metadata and distance_key in metadata:
            distance = metadata[distance_key]

        if distance == "cosine":
            return self._cosine_relevance_score_fn
        elif distance == "l2":
            return self._euclidean_relevance_score_fn
        elif distance == "ip":
            return self._max_inner_product_relevance_score_fn
        else:
            raise ValueError(
                "No supported normalization function"
                f" for distance metric of type: {distance}."
                "Consider providing relevance_score_fn to Chroma constructor."
            )

コードを見るとdistance = "l2"がデフォルトのようなので、self._euclidean_relevance_score_fnが利用されているようです。

（この時点で、察しの良い方は、「何かしらの関数を渡すと言うことは、スコアが別の値に変更されるんだな」と言うのがわかると思います。もし、そのままスコアを利用して良い（つまりスコア=類似度）なら、恒等関数で良いわけですから）

こちらのメソッドは、VectorStoreクラスで実装されています。

    @staticmethod
    def _euclidean_relevance_score_fn(distance: float) -> float:
        """Return a similarity score on a scale [0, 1]."""
        # The 'correct' relevance function
        # may differ depending on a few things, including:
        # - the distance / similarity metric used by the VectorStore
        # - the scale of your embeddings (OpenAI's are unit normed. Many
        #  others are not!)
        # - embedding dimensionality
        # - etc.
        # This function converts the Euclidean norm of normalized embeddings
        # (0 is most similar, sqrt(2) most dissimilar)
        # to a similarity function (0 to 1)
        return 1.0 - distance / math.sqrt(2)

最後の行でわかりますが、「距離」が「類似度」に変換されていることがわかります。

こう言う処理が、入っているので、「類似度」なのか「距離」なのかは、元実装を見ながら考慮して利用する必要があります。

（補足：細かい話3）　embed_documentsとembed_queryの違い

LangChainではテキストを埋め込みベクトルに変換させるときに、主に二つの方法を利用すると思います。

ベクトル化する対象が、検索させる側のドキュメントの場合は下記

embeddings = embedding_model.embed_documents(texts)

ベクトル化する対象が、ユーザのクエリの場合は下記

embedding = embedding_model.embed_query(query)

なぜ、使い分けているのかというと、Embeddingモデルの中には、用途によって違うベクトルを出力するようになっているからです。
なぜなら、RAGにおいて、私たちが取得したい文章は、「質問文と近い文章」ではなく、「想定される回答が含まれている文章」であるはずだからです。

したがって、質問文をそのままベクトル化するよりも、質問文の回答文を生成し、その文章をベクトル化する方が、結果として検索精度が上がるように思います。
このような特殊な処理が可能な埋め込みモデルの場合、上記のように使い分けることに価値があります。

そして、Googleの埋め込みモデルは、上記のような処理が入っているようです。

詳細は、上記を見ていいただければわかると思いますが、下記の2つの結果が異なるようです。

query_embeddings = GoogleGenerativeAIEmbeddings(
    model="models/embedding-001", task_type="retrieval_query"
)
query_vecs = [query_embeddings.embed_query(q) for q in [query, query_2, answer_1]]

doc_embeddings = GoogleGenerativeAIEmbeddings(
    model="models/embedding-001", task_type="retrieval_document"
)
doc_vecs = [doc_embeddings.embed_query(q) for q in [query, query_2, answer_1]]

違いは、task_typeです。
これがretrieval_queryの場合は、ユーザのクエリを想定しているため、質問文から想定される回答文章のベクトルに近いベクトルが出力されます。
一方で、retrieval_documentの場合は、検索されるドキュメントの方を想定しているため、そのままベクトル化されます。

賢いですね。使う側は何も考えずに、ユーザのクエリならembed_query、ドキュメントならembed_documentsを使っていきましょう。

より、詳細な説明は、下記も併せてご覧ください。

実際にchainでLLMに回答させる

細かい話を読んでくださった方、ありがとうございます。本題に戻ります。

ここまでで、Retrieverクラスは作成できたので、あとは普通のLCEL記法でChainを組んで利用するだけです。
具体的には下記の実装になります。

    # Prompt 定義
    prompt_template = """
あなたは、株式会社asapに所属するAIアシスタントです。
ユーザからサービスに関連する質問や雑談を振られた場合に、適切な情報を提供することもが求められています。
ユーザからサービスに関する情報を質問された場合は、下記の情報源から情報を取得して回答してください。
下記の情報源から情報を取得して回答する場合は、ユーザが一時情報を確認できるように、取得した情報の文章も追加で出力してください。

情報源
{context}
"""
    prompt = ChatPromptTemplate.from_messages(
        [
            ("system", prompt_template),
            ("human", "{query}"),
        ]
    )


    # チェーンを定義（retriever で文脈を取り、Prompt に当てはめて、LLM へ）
    chain = (
        {"context": dence_retriever, "query": RunnablePassthrough()}
        | prompt
        | llm
        | StrOutputParser()
    )

    # 実行
    print("===== DenseRetriever の実行結果 =====")
    dense_docs = dence_retriever.invoke(query)
    print("\nDenseRetrieved Documents:", dense_docs)

    print("\n================= LLMの実行結果 =================")
    result = chain.invoke(query)
    print(result)

プロンプトは、企業のQ&A botとして利用することを想定したプロンプトを記載しています。そして、作成したdence_retrieverをchainに組み込む形で実装しています。

LCEL記法の詳細な説明に関しては、記事を書いておりますので、そちらもご参照ください。

Agentsを利用して、DBに保存された内容から、LLMがRAGで質問に回答する

では、実際にAgentを実装します。
Agentといっても、主に利用するのはTool Callingです。
RAGの機能をToolとして設定して、あとはLLMがToolを使うか否かを判断すると言う、よくあるTool Callingの形です。

現在では、Tool Callingを利用する実装は、LangGraphを利用したAgentとして実装することが推奨されているので、Agentと呼んでいます。

該当するコードは下記になります。

基本的には、search_rag_documents_local.pyと中身は近いので、差分だけ解説します。

RetrieverをToolとして設定

    # DenceRetrieverを用意
    dence_retriever = VectorSearchRetriever(
        vector_store=vector_store,
        embedding_model=embedding_model,
        k=5,
    )

    tool = dence_retriever.as_tool(
    name="Document_Search_Tool",
    description="サービスの規約に関する情報を取得するためのtoolです。"
    )

上記の部分で、作成したカスタムRetrieverクラスをToolとして設定しています。
Toolとして設定するには、as_toolメソッドを利用するだけです。簡単ですね。

nameはToolの名前を入力する部分です。基本的に英語のみしか受け付けていません。また空白はNGです。
descriptionはToolの説明を入力する部分です。日本語、空白ともにOKですが、LLMはこの部分の説明をみて、クエリに対して、どのToolを利用するかを判断することになります。
したがって、詳細かつ明確に記載する必要があることに注意してください。

プロンプトの定義

    # Prompt 定義
    prompt_template = """
あなたは、株式会社asapに所属するAIアシスタントです。
ユーザからサービスに関連する質問や雑談を振られた場合に、適切な情報を提供することもが求められています。
ユーザからサービスに関する情報を質問された場合は、toolから情報を取得して回答してください。
toolから情報を取得して回答する場合は、ユーザが一時情報を確認できるように、取得した情報の文章も追加で出力してください。
"""
    prompt = ChatPromptTemplate.from_messages(
        [
            ("system", prompt_template),
            ('human', '{messages}'),
        ]
    )

以前のコードとは少し異なっていますが、基本的には同じです。
違うのはシステムプロンプトから、情報源（{context}）の記載がなくなりました。
Toolを実行して、外部情報を取得したら、queryの後に取得するようになりますので、ここでは明示的に記載する必要はありません。

またLangGraphを利用しているため、messagesが必要になります。それに併せて、ここでmessagesを利用しています。

Agentの定義

    # エージェントの作成
    agent = create_react_agent(
        model=llm,
        tools=[tool],
        state_modifier=prompt
    )

上記の部分で、Agentを定義しています。
LangGraphでは、基本的にcreate_react_agentを利用することになります。
このreactは、フロントエンドのreactではなく、ReActの意味です。

ReActは、まだ論文を読んでいないので、一旦解説記事を置いておきます。そのうちちゃんと勉強します。

重要なのは、上記のように設定することで、LLMモデルとTool（RAG）とプロンプトをAgentとして設定できたことになります。

Agentsの実行

    print("\n================= エージェントの実行結果 =================")
    result = agent.invoke({'messages':query})
    print_agent_result_details(result)

    print("\n================= 最終的な出力結果 =================")
    print(result['messages'][-1].content)

上記で、Agentを実行して、途中経過も含めて表示しています。print_agent_result_details関数は、このコード内で私が書いている関数で、Agentの見にくい出力を見やすく表示するだけの関数です。

Agentの実行もChainと同様にinvokeで良いです。
稀に、runメソッドを利用している記事がありますが、これは古い非推奨の書き方なので注意してください（一敗）

まとめ

前後編でだいぶ長かったかと思いますが、みていただけた方ありがとうございました。

今回は、ローカルのDBとローカルのドキュメントを利用しました。
実際にRAGを導入する際は、Google Cloudなどのクラウドサービス上のドキュメントやDBを利用することになると思います。

次回以降は、そちらでの実装方法を解説したいと思います。
（私が苦労したので、備忘録として残したいのはそっち・・・）

では、次回もぜひご覧ください！

学習書籍

（書籍のリンクはamazonアフィリエイトリンクです）

Discussion

ログインするとコメントできます

はじめに

参考文献

記事

書籍

成果物

コードの解説

ドキュメントをEmbeddingsモデルでベクトル化し、DBに格納する

定数定義

Embedingモデルを定義する

Chroma DBを定義する

ドキュメントを読み込み、チャンクやID、メタデータなどを作成する

ベクトル化してDBに格納

格納できているか確認

DBに保存された内容から、LLMがRAGで質問に回答する

定数定義

モデル、DBの定義

DBから検索するロジックを実装する

（補足：細かい話） as_retrieverの実装

（補足：細かい話2） 類似度と距離の混在

（補足：細かい話3） embed_documentsとembed_queryの違い

実際にchainでLLMに回答させる

Agentsを利用して、DBに保存された内容から、LLMがRAGで質問に回答する

RetrieverをToolとして設定

プロンプトの定義

Agentの定義

Agentsの実行

まとめ

学習書籍

Discussion

（補足：細かい話）　as_retrieverの実装

（補足：細かい話2）　類似度と距離の混在

（補足：細かい話3）　embed_documentsとembed_queryの違い