Zenn
🙆‍♂️

RAGのTutorialやってみた part3

2024/12/20に公開
tutorial
LangChain
LangSmith
RAG
LangGraph
tech

前回したこと

  • LangChain, LangSmith, LangGraphの説明
  • chatModel, embeddings, vectorStoreのインスタンス化

今回すること

実際にブログのポストをvector_storeに保存して、そのブログ用のRAGを作リマス。
Lilian Wengさんという方の投稿されたブログを使用します。
こちらがそのブログです。
https://lilianweng.github.io/posts/2023-06-23-agent/
では、やっていくだべよ~ｵﾋｵﾋ ｵﾋｲｲｲｲｲ

インデックス貼りとRAGの一連の流れの実装

前回の続きをしていくので、chatModelとembeddingsとvectore_storeのインスタンスはある前提で話を進めていきます。
対象となるwebページはこれです

実装する前に以下のコマンドを実行し、langgraphをインストールします。
(rag_tutorial)$ pip install -U langgraph langsmith

では実際コードを実装していきます

import bs4
from langchain import hub
from langchain_community.document_loaders import WebBaseLoader
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langgraph.graph import START, StateGraph
from typing_extensions import List, TypedDict

loader = WebBaseLoader(
    web_paths=("https://lilianweng.github.io/posts/2023-06-23-agent/",),
    bs_kwargs=dict(
        parse_only=bs4.SoupStrainer(
            class_=("post-content", "post-title", "post-header")
        )
    ),
)
docs = loader.load()

ここでは、必要なライブラリをインストールして、WebBaseLoaderのインスタンスを作成してload()メソッドでdocsを取得しています。
WebBaseLoaderというのは具体的に裏で何をやっているのはソースコードを読まないと分かりませんが、webpathsで与えられた引数のurlのpageから、bs_kawrgsの"post-content", "post-title", "post-header"このあたりのclassに属するタグ(divタグとか)から文章内容を取得する用のクラスなのでしょう。(気になる方は公式ドキュメントを読んでください)
docs = loader.load()ここのdocsはList[Document]型をしていて、Documentというはlangchainが提供しているオブジェクトです。このままではdocsの中身がよく分からないので実際に中身を見てみましょう。
docsは長さ１のリストでその中身はこのようになっています。

どうやらpage_contentというのを持っているそうでここが今回使用したいところなのでしょう。
docs[0].page_contentとすればブログの記事の内容が取得できそうです。

次にいきます。

text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
all_splits = text_splitter.split_documents(docs)

# Index chunks
_ = vector_store.add_documents(documents=all_splits)
prompt = hub.pull("rlm/rag-prompt")

chunk_sizeというのはほぼ文字数と見て問題ないです。これが1000なので文章を1000文字程度の文章に分割しているのでしょう。さらにchunk_over_lap=200なので200文字の重複をさせながら分割しているようです。
all_splitsの中身はこのようになっています。

中身はDocumentオブジェクトのリストとなっているようです。

_ = vector_store.add_documents(documents=all_splits)

ここでは、all_splitsをvector_storeに保存しています。
prompt = hub.pull("rlm/rag-prompt")この部分ではどのような形式で入力するかなどの情報を取得しています。promptインスタンスの中身を見ると

You are an assistant for question-answering tasks. Use the following pieces of retrieved context to answer the question. If you don't know the answer, just say that you don't know. Use three sentences maximum and keep the answer concise.\nQuestion: {question} \nContext: {context} \nAnswer:"

このようなテンプレートとなっているようです。文章から察するにこれはrag用のテンプレートなのでしょう。

LangGraphの使用

LangGraphを使用するメリット

  • アプリケーションロジックを一度定義するだけで複数の呼び出しモードをサポート
    • ストリーミングや非同期処理（async）、バッチ処理など、さまざまな実行モードを自動的にサポートします。
    • retrieve関数とgeneration関数などを渡して自らコードを書かずにロジック(処理の流れ)を自動的に実行してくれます。ユーザーはこの処理の流れを簡単に実装できます
  • LangGraph Platformを活用した簡易なデプロイ
    • アプリケーションのデプロイ作業がスムーズになります。
  • LangSmithでアプリケーションのステップをトレース
    • アプリケーション内の各ステップ（プロセスの流れ）をLangSmithを使って自動的に追跡できます。
    • LangSmithでプロセス過程を追跡可能

LangGraphの重要な要素

  • 状態(State)
    • アプリケーションの入力データ、ステップ間のデータ、出力データを管理します。
    • RAGの場合は、入力された質問(question)と取得した関連情報(context)と出力(answer)
  • ノード(Nodes)
    • アプリケーションの各ステップを定義します。 例: 質問を処理するステップ、関連情報を検索するステップ、回答を生成するステップなど。
    • 今回はretrieveとgenerationの二つの処理を書きます
  • 制御フロー（Control Flow）
    • ステップの順序や実行の流れを定義します。 例: 「質問 → 情報取得 → 回答生成」という流れ。
    • ノードA->ノードBのような流れを作ってくれる。

では先ほどの続きにいきましょう

LangGraphの前に状態とノードを作成する

from langchain_core.documents import Document
from typing_extensions import List, TypedDict


class State(TypedDict):
    question: str
    context: List[Document]
    answer: str

ここでRAG用の状態を定義しています。
TypedDictというの型の決まった辞書です。
例

from typing_extensions import TypedDict  # Python 3.8 以降なら typing を使う

class User(TypedDict):
    id: int
    name: str
    age: int
#型が正しい
user: User = {"id": 1, "name": "Alice", "age": 30}

# 間違ったキーや型の例（型チェックエラーになる）
invalid_user: User = {"id": "1", "name": "Bob"}

要は辞書です。難しく取られなくて大丈夫です。これをノードの処理の中で使用していきます
今回の例ではquestionにstr型、contextにはList[Document]型、answerにはstr型な辞書を使用します。なぜ状態と呼ばれているかというと、ノード内でこのStateを処理して少しづつ更新していくからです。ステップAではStateのこの部分を更新してステップBへStateを渡し、ステップBではStateの別の部分を更新して次へ、とこのようにStateを更新・渡すを繰り返します。そのため、Stateという風にしているのでしょう。

次にvectore_storeからの文脈情報取得と回答生成の処理を書いていきましょう。いずれもStateオブジェクトを受け取らねばなりません

def retrieve(state: State):
    retrieved_docs = vector_store.similarity_search(state["question"])
    return {"context": retrieved_docs}


def generate(state: State):
    docs_content = "\n\n".join(doc.page_content for doc in state["context"])
    messages = prompt.invoke({"question": state["question"], "context": docs_content})
    response = llm.invoke(messages)
    return {"answer": response.content}

一つずつ見ていきましょう。

retireveはまず、state(型の厳しい辞書です)を受け取って、vectore_store.similarity_searchメソッドでstateの中のquestionに対応する値(str型の文字列)を受け取って、その内容に近いドキュメントを取ってきてくれます(retrieve)。そして、取得したList[Document]を{"context": retrieved_docs}で返します。(おそらくcontext: List[Document]に対応しているのでしょう。retrieveの処理が終わると、contextがあるのでStateを更新して次のノードのgenerateを実行します。)

generateも同様にstateを受け取り、stateのcontextのdocumentを一つずつ取り出し、一つ一つのcontentを改行して結合します。そのcontext情報と、質問内容を同時にprompt.invokeメソッドに渡してmessages(llmに送る用のテンプレートに当てはめたもの)を取得します。
最後に、これをllm.invokeに渡して、response(answer)を取得します。

実際にLangGraphを使ってみる

実際にLangGraphでグラフを作ってみましょう

from langgraph.graph import START, StateGraph

graph_builder = StateGraph(State).add_sequence([retrieve, generate])
graph_builder.add_edge(START, "retrieve")
graph = graph_builder.compile()

今回はStateGraphというのを使用します。
他にも別のグラフがあるんか？と気になったのでみてみると

この程度でしたので、基本的にはStateGraphを使用するっぽいです。
ここでは実際にノードの流れを作っています。[retrieve, generate]で処理のリストをadd_sequenceに渡して、左から順に処理するグラフを作成しています。受け渡しするオブジェクトの方をStateGraph(State)で指定しているのでしょう。そしてgraph_builderというインスタンスを作成し、.add_edge(START, "retrieve")でスタート地点を指定しています。おそらくENDなども指定できるような感じがします。
最後に.compileメソッドで今までの設定(型)などを総合的に読み込む処理をしています。ここでようやくretrieveからgenerateまでの処理の流れが確立したgraphを取得します。

今回はここまでで、次回はこのgraphを使用して、実際にRAGを体験してみましょう　ｵﾋ
part3
https://zenn.dev/kurutazoku/articles/ba04f270e7f3c8

Discussion