LoginSignup
1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@MinadukiSekina(水無月 せきな)

LangChainチュートリアル② - セマンティック検索の実装とAPI化

Posted at

はじめに

どうも、水無月せきなです。

LangChainのチュートリアルに取り組み始めました。
前回の記事では、Chat models and promptsをベースに実装してみたお話をしました。

今回は、同じチュートリアルのSemantic searchを元にやってみたことを書こうと思います。

この記事でわかること

  • チュートリアルの要点と実践コード
  • FastAPIを用いたセマンティック検索APIの作成方法

リポジトリ

開発環境の準備

開発環境

Windows 11 24H2
WSL2
Docker Desktop 4.43.1
Cursor 1.2.2
Python 3.12

使用したテンプレート

最近お世話になりっぱなしの、下記テンプレートリポジトリを使用しました。

ただ、Claude Code 利用時にclaude code error: eacces: permission denied, mkdir '/home/vscode/.claude/todos'というエラーが発生したため、postCreateCommandを修正しました。

.devcontainer/devcontainer.json
-	"postCreateCommand": "uv sync",
+   "postCreateCommand": "uv sync && sudo chown -R vscode:vscode /home/vscode/.claude",

ライブラリのインストール

LangChain・pypdf

PDFを読み込むため、pypdfもインストールします。

インストールするコマンド
uv add langchain-community pypdf

Langchain-core

今回、ベクトルDBにインメモリを使用するためインストールします。

uv add langchain-core

LangSmith

LLMの応答をトレースするために入れます。

uv add langsmith

GoogleGenerativeAIEmbeddings

文書やユーザーからのクエリをembedding（ベクトル表現へ変換）するために必要です。
Googleである必要性はないので、お好みのモデルを使用してください。

使えるモデルの一覧はこちらにあります。

インストールするコマンド
uv add langchain-google-genai

python-dotenv

.envから環境変数を読み込むために入れます。

インストールするコマンド
uv add python-dotenv

FastAPI

今回はLangServeではなく、こちらを使います。

uv add "fastapi[standard]"

チュートリアルをやってみる

事前準備

Gemini API キー・LangSmith API キーの取得と設定

前回の記事に方法は記載していますので、そちらをご参照ください。

ドキュメントを読み込む

LangChainは文書に関するデータを表現するDocumentクラスを提供しています。

また、LangChainはドキュメント用のローダーも提供しているので、それを使用して読み込むことができます。

src/semantic_search/main.py
from langchain_community.document_loaders import PyPDFLoader

file_path = "src/semantic_search/example_data/nke-10k-2023.pdf"
loader = PyPDFLoader(file_path)

# ドキュメントの読み込み
docs = loader.load()

print(len(docs))
print(f"{docs[0].page_content[:200]}\n")
print(docs[0].metadata)

※ファイルのパスは、実行時のディレクトリからの参照パスのようです。

PyPDFLoader はPDFのページごとに一つのDocumentオブジェクトに読み込むため、ファイル名やページ数に簡単にアクセスできます。

ただ、LLMの回答に用いるにはページ単位だと大きすぎるため、text splittersを用いてチャンクに分割します。

テキストのチャンク化（分割）

テキストを分割していくつかのまとまりに変えます。この時、前後を少し重複する形で分割することで、文脈からの遊離を軽減します。

基本的なテキスト分割のユースケースでは、RecursiveCharacterTextSplitterが推奨されるようです。

src/semantic_search/main.py
# 読み込み部分は割愛

# text splitters の定義
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000, chunk_overlap=200, add_start_index=True
)
# テキストの分割を実行
all_splits = text_splitter.split_documents(docs)

print(len(all_splits))

ベクトルに変換

関連性による検索を行うために、ベクトルへの変換を行います。
この時に使うのが Embedding models です。

src/semantic_search/main.py
# テキストの分割までは割愛

embeddings = GoogleGenerativeAIEmbeddings(model="models/embedding-001")

vector_1 = embeddings.embed_query(all_splits[0].page_content)
vector_2 = embeddings.embed_query(all_splits[1].page_content)

assert len(vector_1) == len(vector_2)
print(f"Generated vectors of length {len(vector_1)}\n")
print(vector_1[:10])

ベクトルデータベースの構築
VectorStoreオブジェクトが、テキストやDocumentオブジェクトのベクトル保存、検索を担う。
LangChainには多種のベクトルデータベースが統合されている。
今回はインメモリでやってみる。

ベクトルDBの構築とクエリの実行

ベクトル表現を保持するベクトルDBを作ります。
VectorStoreがテキストやDocumentオブジェクトのベクトル保存、検索を担っています。

LangChainで使用できるベクトルDBは色々あるのですが、今回はインメモリDBを使います。

# テキストの分割までは割愛

# embedding modelの定義
embeddings = GoogleGenerativeAIEmbeddings(model="models/embedding-001")

# ベクトルDBのインスタンス化
vector_store = InMemoryVectorStore(embeddings)

# ベクトルDBへドキュメントを追加（同時にベクトル変換）
ids = vector_store.add_documents(documents=all_splits)

# 類似度検索
results = vector_store.similarity_search(
    "How many distribution centers does Nike have in the US?"
)

print(results[0])
実行結果 ```bash page_content='operations. We also lease an office complex in Shanghai, China, our headquarters for our Greater China geography, occupied by employees focused on implementing our wholesale, NIKE Direct and merchandising strategies in the region, among other functions. In the United States, NIKE has eight significant distribution centers. Five are located in or near Memphis, Tennessee, two of which are owned and three of which are leased. Two other distribution centers, one located in Indianapolis, Indiana and one located in Dayton, Tennessee, are leased and operated by third-party logistics providers. One distribution center for Converse is located in Ontario, California, which is leased. NIKE has a number of distribution facilities outside the United States, some of which are leased and operated by third-party logistics providers. The most significant distribution facilities outside the United States are located in Laakdal,' metadata={'producer': 'EDGRpdf Service w/ EO.Pdf 22.0.40.0', 'creator': 'EDGAR Filing HTML Converter', 'creationdate': '2023-07-20T16:22:00-04:00', 'title': '0000320187-23-000039', 'author': 'EDGAR Online, a division of Donnelley Financial Solutions', 'subject': 'Form 10-K filed on 2023-07-20 for the period ending 2023-05-31', 'keywords': '0000320187-23-000039; ; 10-K', 'moddate': '2023-07-20T16:22:08-04:00', 'source': 'src/semantic_search/example_data/nke-10k-2023.pdf', 'total_pages': 107, 'page': 26, 'page_label': '27', 'start_index': 804} ```

単純な類似度検索の他にも、一緒にスコアを返したりもできるようです。

VectorStoreのRunnable化（Retrieverの利用）

VectorStoreはRunnableのサブクラスではないので、invokeを呼び出したりチェーンさせるにはひと工夫が必要になります。

デコレータで対応する方法

@chain
def retriever(query: str) -> list[Document]:
    return vector_store.similarity_search(query, k=1)


results = retriever.batch(
    [
        "How many distribution centers does Nike have in the US?",
        "When was Nike incorporated?",
    ],
)

for result in results:
    print(result)

as_retriever を使う方法

retriever = vector_store.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 1},
)

results = retriever.batch(
    [
        "How many distribution centers does Nike have in the US?",
        "When was Nike incorporated?",
    ],
)

FastAPIによるAPI化

公式のチュートリアルだと前節の内容で終わってしまうため、追加でFastAPIによるAPIを作成してみます。

※公式のチュートリアルからは外れた内容です。

ディレクトリ構造

src
└── semantic_search
    ├── api
    │   ├── document.py
    │   └── __init__.py
    ├── environment
    │   ├── env_loader.py
    │   └── __init__.py
    ├── example_data
    │   └── nke-10k-2023.pdf
    ├── __init__.py
    ├── llms
    │   ├── embedding.py
    │   └── __init__.py
    ├── lodaer
    │   ├── __init__.py
    │   └── pdf_loader.py
    ├── main.py
    ├── models
    │   ├── __init__.py
    │   └── message.py
    ├── retriever
    │   ├── __init__.py
    │   └── vector_retriever.py
    ├── splitter
    │   ├── __init__.py
    │   └── text_splitter.py
    └── vector_store
        ├── __init__.py
        └── in_memory_store.py
ディレクトリ 役割
api エンドポイント
environment 環境設定
example_data 文書のオリジナルを保存
llms 使用するモデルの定義
lodaer 読み込み用
main.py エントリーポイント
models APIで使用するデータモデル
retriever 検索を実行
splitter チャンク分割
vector_store ベクトルDB

主要モジュールについて

APIの定義と依存関係の解決

でリクエストを受け取ってRetrieverを実行して応答を返却するまでの流れを、Fast APIのDependsを使用して定義してみました。

普通にハンドラー関数でいろいろやっても良いと思うのですが、Fast API の復習も兼ねています。

src/semantic_search/api/document.py
from typing import Annotated

from fastapi import APIRouter, Depends

from semantic_search.models.message import LLMResponse
from semantic_search.retriever.vector_retriever import run_retriever

router = APIRouter(prefix="/documents", tags=["documents"])


@router.post("/")
def run_document_search(result: Annotated[list[str], Depends(run_retriever)]) -> LLMResponse:
    """ドキュメントの検索を実行する"""
    return LLMResponse(result=result)
src/semantic_search/retriever/vector_retriever.py
from typing import Annotated

from fastapi.params import Depends
from langchain_core.vectorstores import VectorStoreRetriever

from semantic_search.models.message import Message
from semantic_search.vector_store.in_memory_store import vector_store_instance


def get_vector_retriever() -> VectorStoreRetriever:
    """ベクトル検索用のインスタンスを取得する"""
    return vector_store_instance.vector_store.as_retriever(
        search_type="similarity",
        search_kwargs={"k": 1},
    )


def run_retriever(
    message: Message,
    retriever: Annotated[VectorStoreRetriever, Depends(get_vector_retriever)],
) -> list[str]:
    """ベクトル検索を実行する"""
    result = retriever.invoke(message.prompt)
    return [content.page_content for content in result]

依存関係を図示すると下記のようになります。

ベクトルDBの扱い

ベクトルDBの作成は一度だけで良いため、シングルトンで実装しました。

src/semantic_search/vector_store/in_memory_store.py
from langchain_core.vectorstores import InMemoryVectorStore as LangchainInMemoryVectorStore

from semantic_search.llms.embedding import create_embedding_model
from semantic_search.lodaer.pdf_loader import load_pdf
from semantic_search.splitter.text_splitter import create_text_splitter


class InMemoryVectorStore:
    """ベクトルDBのインスタンスを管理するクラス"""

    _instance = None

    def __new__(cls) -> "InMemoryVectorStore":
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._setup_vector_store()
        return cls._instance

    def _setup_vector_store(self) -> None:
        """ベクトルDBの初期化"""
        # ベクトルDBの初期化
        embedding = create_embedding_model()
        self._vector_store = LangchainInMemoryVectorStore(embedding=embedding)
        # ドキュメントの読み込み
        docs = load_pdf()
        # テキストの分割
        text_splitter = create_text_splitter()
        all_splits = text_splitter.split_documents(docs)
        # ドキュメントの追加
        self._vector_store.add_documents(all_splits)

    @property
    def vector_store(self) -> LangchainInMemoryVectorStore:
        """ベクトルDBのインスタンスを返す"""
        return self._vector_store


# シングルトンインスタンスを事前に作成
vector_store_instance = InMemoryVectorStore()

実装上の留意点としては下記が挙げられます。

  • __new__からベクトルDBを作成するメソッド（_setup_vector_store）を呼ぶ
    （__init__のタイミングだと、インスタンス化の度に呼ばれます）
  • 外から変更されないように、プライベート変数かつgetterのみでアクセス
  • モジュール読み込み時に作成

本来は、Fast API 起動時に作成するようにLifespan イベントを利用するべきかなと思ったのですが、インメモリであるため一旦この形に落ち着いています。

起動と確認

起動コマンド
uv run fastapi dev src/semantic_search/main.py

ベクトルDBを作成するため、初期起動に時間がかかります。
起動してから少し経って http://127.0.0.1:8000/documents/ にPOSTすると検索の結果が返ってくると思います。

また、http://127.0.0.1:8000/docs でドキュメントを閲覧でき、動作の確認もできます。

Retrieverの実行時は、LangSmithにログが残ります。

image.png

おわりに

基本中の基本みたいな内容ですが、他のベクトルDBを使ってみたり、RAGに挑戦してみたいと思います。

ここまでお読みいただきありがとうございました。ご参考になれば幸いです。

参考資料

1
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?