Zenn
Cykinso's Tech BlogPublicationへの投稿
🦜

🦜🔗 LangChain v0.2.5 + OpenAI での RAG を用いた ChatBot 実装例

yamasaKit
2024/06/20に公開
chatbot
OpenAI
LangChain
LLM
RAG
tech

背景

LangChain は OpenAI API を利用し自分たちがやりたいことを実現することに非常に便利なライブラリですがバージョンアップによってクラス名やサブライブラリ名の変更がやや多く少し古い Web 記事を参考にしてもうまくワークしないことがあります。
この記事は 2024/6/20 現在の LangChain (バージョン 0.2.5) で OpenAI API や Azure OpenAI API を動かす例として残しておきます。
同じようなことをしようとして私のように苦戦している方の助けになれば幸いです。

ソフトウェアのバージョンなど

pyproject.toml
python = ">=3.12,<3.13"
python-dotenv = "^1.0.1"
chromadb = "0.5.2"
langchain = "0.2.5"
langchain-cli = "0.0.25"
langchain-openai = "0.1.8"
langchain-community = "0.2.5"
langchain-chroma = "0.1.1"
langchainhub = "0.1.20"
streamlit = "1.35.0"

💡 Poetry で Python のバージョンを指定する時に ^3.12 とすると lancghain-chroma がインストールできなくなるので >=3.12,<3.13 としました。
(langchain-chroma は >=3.12,<3.13 という指定があります)

方法

OpenAI API を使う場合と AzureOpenAI API を使う場合は基本同じことをするのでまず OpenAI API を使う場合を説明し、記事が長くなってしまったので、後日別記事にて AzureOpenAI API を使う場合はどの部分をアップデートしたらよいのかを説明したいと思います。

6/28 追記) AzureOpenAI API 版の記事も書きましたのでよければぜひどうぞ

https://zenn.dev/cykinso/articles/b055e33734d06b

.env

以下のように .env を用意します。

OPENAI_API_KEY=sk-XXXXXXXXXXXXXXXX
OPENAI_API_VERSION=2024-02-01

もし OPENAI_API_KEY をまだ取得していない場合は以下の方法で取得してください。

OPENAI_API_KEY

OPENAI_API_KEY は OpenAI の Dashboard で作成できます。
(課金対象なのでご自身の責任のもとご利用ください)

「+ Create new secret key」 を押すとモーダルが開くので後から区別できるような名前をつけて 「Create secret key」 を押します。

表示される API キーを .env にメモしておきます。 「Done」 を押すともう表示できません。

データ

OpenAI がまだ学習していなさそうなデータの例として弊社 Cykinso のブログ記事の「会社のビジョンを話しているページ」を今回は用いたいと思います。

https://note.com/cykinso/n/n432d5ea70783

💡 プライベートで実装する場合は、好きなマンガなどの詳細をテキストにまとめてデータとするとモチベーションも上がると思います。

ざっと文章をコピーして以下のように整形しました。

note.txt
細菌叢からの新たな気付きを通じて、新ビジョンを策定しました！
2023年11月にサイキンソーは10期目に入りました。高齢化社会による社会保障への不安が募る現在、病気を未然に防ぐ0次予防の重要性が高まっています。「細菌叢で人々を健康に」というミッションに向けて、サイキンソーは新たなビジョンを制定しました。
新しいビジョンについて
ービジョン変更で具体的にどの個所が変更したかをまずご紹介します。
こちらがサイキンソーのミッション（MISSION）、ビジョン(VISION)、バリュー(VALUE)になります。
私たちが目指し続ける「細菌叢で人々を健康に」というミッションはそのままに、ビジョンを新しく変更致しました。

＜これまでのビジョン＞
菌叢データから「次世代のライフスタイル」を提供するプラットフォームになる

＜新しいビジョン＞
細菌叢からの新たな気付きを通じて
ヒト、社会、地球環境を健康にするエコシステムを実現する
新しいビジョンでは、サイキンソーが影響を与えていきたい範囲もこれまでよりさらに大きくなったことが分かります。

今回のビジョン変更を通して、サイキンソーがどんな価値発揮を目指していくのか、それに伴い事業面ではどんな挑戦をしていくのかを、代表取締役CEOの沢井さんに聞いてきました！

...(以下略)

コード

続けてコードを実装します。
今回は RAG として外部の情報を参照しつつ回答する ChatBot を実装してみます。
インターフェースとして streamlit を用います。

先にコード全体を示すと以下のようになります。
(streamlit のコードのベースとして以下の記事を参考にさせていただきました。ありがとうございます)

https://tech-lab.sios.jp/archives/41574

chatbot.py
from pathlib import Path

import streamlit as st
from dotenv import load_dotenv
from langchain import hub
from langchain.schema import AIMessage, HumanMessage
from langchain_chroma import Chroma
from langchain_community.document_loaders import TextLoader
from langchain_core.runnables import RunnablePassthrough, RunnableSequence
from langchain_core.vectorstores import VectorStoreRetriever
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

load_dotenv()


def initialize_vector_store() -> Chroma:
    """VectorStoreの初期化."""
    embeddings = OpenAIEmbeddings()

    vector_store_path = "./resources/note.db"
    if Path(vector_store_path).exists():
        vector_store = Chroma(embedding_function=embeddings, persist_directory=vector_store_path)
    else:
        loader = TextLoader("resources/note.txt")
        docs = loader.load()

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

        vector_store = Chroma.from_documents(
            documents=splits, embedding=embeddings, persist_directory=vector_store_path
        )

    return vector_store


def initialize_retriever() -> VectorStoreRetriever:
    """Retrieverの初期化."""
    vector_store = initialize_vector_store()
    return vector_store.as_retriever()


def initialize_chain() -> RunnableSequence:
    """Langchainの初期化."""
    prompt = hub.pull("rlm/rag-prompt")
    llm = ChatOpenAI()
    retriever = initialize_retriever()
    chain = (
        {"context": retriever, "question": RunnablePassthrough()} | prompt | llm
    )
    return chain


def main() -> None:
    """ChatGPTを使ったチャットボットのメイン関数."""
    chain = initialize_chain()

    # ページの設定
    st.set_page_config(page_title="RAG ChatGPT")
    st.header("RAG ChatGPT")

    # チャット履歴の初期化
    if "messages" not in st.session_state:
        st.session_state.messages = []

    # ユーザーの入力を監視
    if user_input := st.chat_input("聞きたいことを入力してね！"):
        st.session_state.messages.append(HumanMessage(content=user_input))
        with st.spinner("GPT is typing ..."):
            response = chain.invoke(user_input)
        st.session_state.messages.append(AIMessage(content=response.content))

    # チャット履歴の表示
    messages = st.session_state.get("messages", [])
    for message in messages:
        if isinstance(message, AIMessage):
            with st.chat_message("assistant"):
                st.markdown(message.content)
        elif isinstance(message, HumanMessage):
            with st.chat_message("user"):
                st.markdown(message.content)
        else:
            st.write(f"System message: {message.content}")


if __name__ == "__main__":
    main()

コードの各部分を説明していきます。

initialize_vector_store

def initialize_vector_store() -> Chroma:
    """VectorStoreの初期化."""
    embeddings = OpenAIEmbeddings()

    vector_store_path = "./resources/note.db"
    if Path(vector_store_path).exists():
        vector_store = Chroma(embedding_function=embeddings, persist_directory=vector_store_path)
    else:
        loader = TextLoader("resources/note.txt")
        docs = loader.load()

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

        vector_store = Chroma.from_documents(
            documents=splits, embedding=embeddings, persist_directory=vector_store_path
        )

    return vector_store

Vector store とは情報として読み込ませているテキストファイルを適切な長さに分割し(チャンクと呼ばれます) すぐに取り出せるように保存するデータベースのようなものです。
Embeddings と呼ばれるモデルでテキストは数値情報のベクトルに保存されます。その作業を行うために OpenAI API が必要になっているため関数の最初で OpenAIEmbeddings を呼び出しています。

続けて Vector store として保存されているデータがすでに存在していないかを調べています。基本的にデータやモデルがアップデートされない限り Vector store のデータはまったく同じになるため、毎回実行してしまうと時間も API の利用料金も無駄になってしまいます。
そこで

  • まだ存在していない場合： 新規作成
  • すでに存在している場合： 保存しているデータベースを読み込む

としています。

なお存在している場合 Chroma クラスの引数 embedding_function に embeddings を指定していますが、これはクエリとしてあたえられるユーザの質問と Vector store に保存されているデータとの間の関連性を調べるために、クエリも Embeddings でベクトルに変換する必要があるからです。

initialize_retriver

def initialize_retriever() -> VectorStoreRetriever:
    """Retrieverの初期化."""
    vector_store = initialize_vector_store()
    return vector_store.as_retriever()

こちらでは、先ほど作成した(あるいはすでにあるものを読み込んだ) vector_store を retriever に変換しています。こちらの retriever この後、 Vectore store から情報を取り出すのに利用されます。

initialize_chain

def initialize_chain() -> RunnableSequence:
    """Langchainの初期化."""
    prompt = hub.pull("rlm/rag-prompt")
    llm = ChatOpenAI()
    retriever = initialize_retriever()
    chain = (
        {"context": retriever, "question": RunnablePassthrough()} | prompt | llm
    )
    return chain

こちらでは retriever の情報を LLM で利用できるように chain と呼ばれる概念を利用しております。

よくメソッドチェーンという言葉がプログラミング言語では使われています。
例えば Python の Pandas では

import pandas as pd

# サンプルデータフレーム
data = {
    'Name': ['Alice', 'Bob', 'Charlie', 'David', 'Edward', 'Frank'],
    'Age': [24, 27, 22, 32, 29, 24],
    'City': ['New York', 'Los Angeles', 'New York', 'Chicago', 'Los Angeles', 'New York'],
    'Score': [85, 90, 88, 92, 95, 70]
}

df = pd.DataFrame(data)

# メソッドチェーンを使ったデータ変換
result = (df
          .query('Age > 25')                     # 年齢が25以上の行をフィルタリング
          .groupby('City')                       # 都市ごとにグループ化
          .agg({'Score': 'mean'})                # スコアの平均を計算
          .rename(columns={'Score': 'Average Score'})  # 列名を変更
          .sort_values(by='Average Score', ascending=False)  # 平均スコアで並べ替え
          .reset_index()                         # インデックスをリセット
         )

print(result)

というようにメソッドの返り値を次のメソッドへとバケツリレーのようにつないで行き処理させることがあります。これをメソッドチェーンといいます。

LangChain でもこのバケツリレーを行い、どのようにユーザの質問に答えるかのルールを決めることができます。
LangChain の場合は昔のバージョンでは関数の返り値をさらに関数の引数にすると言うことを繰り返してバケツリレーを行っていましたが、今のバージョン 0.2.5 では以下のように | を利用しチェーンを示すことが推奨されています。

    chain = (
        {"context": retriever, "question": RunnablePassthrough()} | prompt | llm
    )

今回の場合

  1. {"context": retriever, "question": RunnablePassthrough()}
  2. prompt
  3. llm

という順番でチェーンがつながっています。

チェーンの先頭がなぜ辞書であるかは、2番目の prompt の説明を聞いてもらえればわかると思います。

    prompt = hub.pull("rlm/rag-prompt")

prompt は LLM にどのような質問や依頼をするのかを決める部分です。今回はプロンプト(変数名を指していない場合カタカナ表記とします)を有志の方がアップロードし他の人が利用できるようにしてくれているサイト LangChain Hub から ⭐ が多いものをお借りしてきました。もちろん自作してもらってもOKです。

https://smith.langchain.com/hub/rlm/rag-prompt

お借りした rag-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.
Question: {question} 
Context: {context} 
Answer:

簡単に日本語に訳すと 「 context の情報のみを使って question に答えるように Answer を考えなさい。答えられないならわからないと答えなさい。」 となります。 context の情報のみを用いるよう指定することでハルシネーションを防ぐ効果があります (ただし１００％防ぐとは断言できないです)

こちらの question 部分にユーザのクエリが、 context 部分に retriever を指定して prompt を実行せよとしているのがチェーンの {"context": retriever, "question": RunnablePassthrough()} | prompt の部分です。

最後に完成した prompt を llm に渡しなさいと指定しているのが prompt | llm の部分です。
このメソッドチェーンをまとめた chain という変数は invoke メソッドを持っており、このメソッドに質問を投げるとそれが question に入りチェーンが前から順番に実行されます。

以上のように定義することで RAG として外部の情報を参照しつつ回答する ChatBot を実装できました。

main

def main() -> None:
    """ChatGPTを使ったチャットボットのメイン関数."""
    chain = initialize_chain()

    # ページの設定
    st.set_page_config(page_title="RAG ChatGPT")
    st.image(img, use_column_width=False)
    st.header("RAG ChatGPT")

    # チャット履歴の初期化
    if "messages" not in st.session_state:
        st.session_state.messages = []

    # ユーザーの入力を監視
    if user_input := st.chat_input("聞きたいことを入力してね！"):
        st.session_state.messages.append(HumanMessage(content=user_input))
        with st.spinner("GPT is typing ..."):
            response = chain.invoke(user_input)
        st.session_state.messages.append(AIMessage(content=response.content))

    # チャット履歴の表示
    messages = st.session_state.get("messages", [])
    for message in messages:
        if isinstance(message, AIMessage):
            with st.chat_message("assistant"):
                st.markdown(message.content)
        elif isinstance(message, HumanMessage):
            with st.chat_message("user"):
                st.markdown(message.content)
        else:
            st.write(f"System message: {message.content}")

最後に main 関数です。こちらは LangChain の実装というよりは Streamlit を利用したフロントエンド部分の実装になります。

キーとなる点だけ解説すると

  1. ユーザと ChatBot の会話は messages に保存されています。以下のコードを見るとわかるようにユーザからの質問と ChatBot の返答はどちらも messages に append されています。
    # チャット履歴の初期化
    if "messages" not in st.session_state:
        st.session_state.messages = []

    # ユーザーの入力を監視
    if user_input := st.chat_input("聞きたいことを入力してね！"):
        st.session_state.messages.append(HumanMessage(content=user_input))
        with st.spinner("GPT is typing ..."):
            response = chain.invoke(user_input)
        st.session_state.messages.append(AIMessage(content=response.content))
  1. ユーザからの質問へのレスポンスは先ほど定義した chain の invoke メソッドを用いて作られています。
        with st.spinner("GPT is typing ..."):
            response = chain.invoke(user_input)
        st.session_state.messages.append(AIMessage(content=response.content))

という点があげられます。

テスト

それでは実装したものを動かしてみましょう。

> streamlit run chatbot.py

Collecting usage statistics. To deactivate, set browser.gatherUsageStats to false.


  You can now view your Streamlit app in your browser.

  Local URL: http://localhost:8501
  Network URL: http://10.200.0.4:8501
  External URL: http://13.73.233.61:8501

http://localhost:8501 へブラウザからアクセスし、質問してみます。

現時点での note.txt の先頭の方に書いてあった新しいビジョンをちゃんと答えています。この情報は OpenAI にはないためきちんと RAG が働いていると言えそうです。

また「100年後に発売予定の新商品」という情報として含まないような質問をするとちゃんと「情報を持っていない」と返答してくれます。こちらも期待通りですね。

💡 まとめ

  • LangChain v0.2.5 時点での RAG を用いた ChatBot の実装を行いました
  • Streamlit を用いてブラウザからユーザがアクセスできるようにしました

ぜひ参考にしてみてください。

Cykinso's Tech Blog
Cykinso's Tech BlogPublication

「細菌叢の力で人々を健康に」をミッションに掲げるバイオテックスタートアップ「サイキンソー」の技術ブログ。

Discussion