はじめに

今回はLangChain + Guidance、Metaが開発したベクトル探索ライブラリのfaissを利用してChatBotを作る例を紹介したいと思います。

同内容を技術ブログにも書いています。

ベクトル探索とは？

ベクトル探索は、文章をベクトルで表現しベクトル間の距離や類似性を計算して類似性の高い文章を見つけることができる手法のことです。
簡潔に言えば、文章を数値に変換して、それらの類似性で文章を検索する手法です。

なぜベクトル探索？

以下に、ChatBotを作る際のベクトル探索の利点をまとめます。

トークン上限を回避: ベクトル探索では、長い文脈をベクトルで表現し、適切な情報を抽出します。これにより、言語モデルのトークン上限を回避しながら、長い文脈を処理できます。例えば、会話の継続や複数の質問に対する回答など、複雑な文脈を持つ応答を生成する際に有用です。
状態の保持: ベクトル探索では、各文脈をベクトルで表現することで、会話や対話の状態を保持することが可能です。このため、会話の流れや過去の情報を考慮しながら応答を生成することができます。例えば、ユーザの前回の質問や応答を参照して、より適切な回答を生成することができます。
高速な検索: ベクトル探索は、ベクトル間の距離や類似性を計算するため、高速な情報検索が可能です。これにより、過去の会話履歴や文脈に基づいて、適切な情報を迅速に取得することができます。また、複数の文脈の中から関連性の高い情報を抽出することも可能です。

上記で言う”文脈”は様々な意味に置き換えることが出来ると思います。ChatBotが保持する"状態"、記憶・知識・感情など...
ベクトル探索は上記の３つの利点も含めChatBotの"状態"を管理するために必須の手法であると言えると思います。

テキストをチャンクに切り分ける

テキストデータをベクトル化(埋め込み)する前に、適切な大きさでSplit（分割）する必要があります。（ベクトル化前の切り分けされたテキストデータをチャンク・テキストチャンクと呼びます）

チャンクへの切り分けは処理の目的や用途によって異なりますが、前述の状態の質に関わってくるので、とても重要な前処理になります。

切り分けの手法は改行や"、"などで切り分ける、token数で切り分けるなど色々ありますが、途切れのない意味のかたまりとして切り分けできるのが理想であると思っています。

ここではhtmlの文章を想定して、意味的な切り分けをどういう風にすればいいかを書いていきます。

まず、h2から始まるひとまとまり、section要素で囲まれた部分は意味のかたまりの区切りと言えるかもしれません。

<h2>タイトル</h2>
<div>説明</div>

<h3>タイトル</h3>
<div>説明</div>

<h4>タイトル</h4>
<div>説明</div>

<section>
    <h2>タイトル</h2>
    <div>説明</div>
<section>

あとは目次とかあればそれをベースに切り分けするのがいいかもしれないです。

Bingなどでは内部的におそらくh2ベースで切り分けをしているように思います。
(検索ランキングでh2タグの有無を評価しているようなので。あくまで予想ですが)

上記のようなhtmlのChunk切り分けはPythonであればBeautifulSoupで実現できます。

以下のような感じ。

from bs4 import BeautifulSoup, Tag
from typing import List 
 
def get_h2_part_elements(html_str: str) -> List[Tag]:
    """h2のパーツ要素を取得する(h2で区切られた1要素)
    Args:
        html_str (str): html text
    Returns:
        List[BeautifulSoup]: 記事のパーツ要素
    """
    results = []

    # パーツ要素の先頭のh2タグを取得してリストに追加
    soup = BeautifulSoup(html_str, "html.parser")
    h2s = soup.find_all("h2")
    
    # h2タグが見つからない場合は空リストを返す
    if h2s is None:
        return results

    # h2タグの次の要素から次のh2タグまでをリストに追加していく
    for h2 in h2s:
        result = []
        result.append(h2)
        
        # 次のh2タグまでをリストに追加していく
        target_tag = h2.find_next_sibling()
        while target_tag:
            if target_tag.name == "h2":
                break
            result.append(target_tag)
            target_tag = target_tag.find_next_sibling()
    return results

上記はtagだけをリストで返していますが、続けてTag要素に合わせてここから.textのテキスト要素だけ取ったり、画像であればaltのテキストを取得したり、色々処理する必要があります。

例えば、テーブルを文字に変換する処理とかも重要です。

以下のような感じで実装できます。

import re
from bs4 import BeautifulSoup, Tag
# srnなどの特殊文字を抽出する正規表現
re_srn = re.compile(r"[\s\r\n]")

def convert_table_to_matrix(table: BeautifulSoup) -> str:
    """tableタグを受け取り、表形式の文字列に変換する
    Args:
        table (str): tableタグ
    Returns:
        List[List[str]]: 表形式の2次元配列
    """
    # 2次元配列の初期化
    matrix = []
    for tr in table.find_all("tr"):
        matrix.append([])
        ths = tr.find_all("th")
        for th in ths:
            if th.text != "":
                matrix[-1].append(th.text)
            elif th.find("img"):
                matrix[-1].append(th.find("img").get("alt"))
        for td in tr.find_all("td"):
            if td.text != "":
                matrix[-1].append(td.text)
            elif td.find("img"):
                matrix[-1].append(td.find("img").get("alt"))
    str_matrix = ""
    for list in matrix:
        str_matrix += "| "
        for str_ in list:
            if str_ is None:
                str_ = ""
            str_matrix += re_srn.sub("", str_) + " | "
        str_matrix += "\n"
    return str_matrix

その他にも色々ありますが、それらの処理のコードは非常に長くなるので、ここでは、割愛させていただきます。

他にも適切なSplitの方法や分割の大きさは、具体的なタスクやデータによって異なります。適切なSplitを行うことで、テキストデータの類似性などを効果的に探索できます。

テキストを埋め込みに変換

OpenAI Embeddings APIを埋め込みを利用します。

個別で切り分けでDBに入れることも出来ますが、ここではLangChainのStore機能を利用します。(個別でする場合は次元数を意識して格納していきます。OpenAI Embeddings APIは1536次元。)

上記で説明したh2の区切りをチャンクとして利用しますが、どうしても1000トークン以上のものがあるので、そちらは仕方ないので改行で切り分けするようにします。

# テキストファイルを読み込む
from langchain.docstore.document import Document
from langchain.document_loaders.csv_loader import CSVLoader
# Embedding用
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter
# Vector 格納 / FAISS
from langchain.vectorstores import FAISS

embeddings = OpenAIEmbeddings()
loader = CSVLoader(file_path="チャンクを書き込んだcsvのパス",
           csv_args={
                    "delimiter": ",",
                    "quotechar": '"',
           },
)
documents = loader.load()
text_splitter = CharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=1000, chunk_overlap=100, separator="\n"
)
docs = text_splitter.split_documents(documents)

db = FAISS.from_documents(docs, embeddings)
db.save_local("DB保存するパス")

これでChunkからベクトル探索できるようになります。

ベクトル探索DBを利用してChatBot構築

回答の生成にはGuidanceを利用します。
Guidanceについての詳しい情報は以前書いたこちらの記事を参照してください。

まずは初期化

import guidance
from guidance._program import Program
import tiktoken
import re

# Vector 格納 / FAISS
from langchain.vectorstores import FAISS
from langchain.embeddings.openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings() # type: ignore

# DBのロード
db = FAISS.load_local("DB保存したパス", embeddings)

# guidanceの設定
guidance.llm = guidance.llms.OpenAI("gpt-3.5-turbo")

Guidanceのtemplateは以下のような感じにします。
（LangChainのqaのstuff設定を移植しています）

# templateの設定
guidance_template = """
{{#system~}}
Use the following pieces of context to answer the users question. 
If you don't know the answer, just say that you don't know, don't try to make up an answer.
If following pieces of context does not relate to the question, just say that you don't know, don't try to make up an answer.
----------------
{{doc}}
{{~/system}}
{{#user~}}
{{query}}
{{~/user}}
{{#assistant~}}
{{gen 'answer' max_tokens=1000}}
{{~/assistant}}
"""

Guidanceによる回答生成はこんな感じ

def stuff_completion(query: str, ref_doc: str) -> str:
    """回答生成
    Args:
        query (str): 質問
        ref_doc (str): 参照元の文書
    Returns:
        str: 回答
    """
    bot: Program = guidance(
        template=guidance_template
    )  # type: ignore
    completion = bot(query=query, doc=ref_doc)
    return completion["answer"]

参考にする文献のトークン数を計算する必要があるので、tittokenというライブラリを利用して計算します。

def calc_tokens_length(text: str)->int:
    """トークン数の計算
    Args:
        text (str): 文章
    Returns:
        int: トークン数
    """
    encoding = tiktoken.encoding_for_model("gpt-3.5-turbo")
    tokens = encoding.encode(text)
    return len(tokens)

上記で用意した関数を使ってベクトル探索を利用した回答生成を作ります。

def qa_by_vector_search(query)->str:
    """ベクトル検索を利用した回答生成
    Args:
        query (str): 質問
    Returns:
        str: 回答
    """
    # 類似文書検索
    # queryはベクトル化して類似度を計算する文章、kは取得数
    docs_and_scores = db.similarity_search_with_score(query=query, k=5)  
    ref_url = []  # 参照元のURL
    ref_doc = []  # 参照元の文書
    all_tokens = 0  # 参照元のトークン数

    for doc_score in docs_and_scores:
        doc = doc_score[0]
        score = doc_score[1]
        all_tokens += calc_tokens_length(str(doc))
        if (all_tokens < 2000 and score <= 0.365):
            ref_doc.append(doc)

    doc_str = ""
    for doc in ref_doc:
        text = str(doc.page_content)
        text = re.sub(r"text: ", "\n:", text)
        doc_str += text
            
    return stuff_completion(query, doc_str)

これで全部基本的なことは出来上がりです。

あとはslackbotやapiで問い合わせするようにすれば出来上がりです。

一応ローカルで実行するものを置いておきます。

while True:
    text = input("質問：")
    if not text:
        break
    print("回答:" + qa_by_vector_search(text))

最後に

上記のやり方をベースに記憶とか感情持つChatBotも応用で作れると思います。

興味湧いた人は色々調べて作ってみるとよいかと思います。

また、言語モデルもChatGPTだけではないので、色々な言語モデルをベースに試してみると世界が広がるかもしれないですね！(GuidanceもLangChainもどちらも様々な言語モデルを利用できるように設計されていますので)

GameWithでは現在エンジニアを絶賛募集中です！

サーバーエンジニアやフロントエンジニアの方、AIに興味がある方や、Unityでの開発に興味がある方もお気軽にカジュアル面談をお申し込みください！