🚀
フルスタックの RAG を利用したレストラン メニューを構築する
llama-indexとWeaveというベクターストアを使って、レストランのメニューに関するあらゆる質問に答えることができるフルスタックチャットボットの構築を作ったようです。
備忘録用にまとめました。引用元にあった説明を省いてるので、ある程度RAGの知識がある方向けです。
やりたいこと
「ヘルシーなものを食べたい」「今日は魚の気分」といったあいまいな指示でレストランのメニューを決めたい。
それはまるで高級レストランのウェイターに聞くように。
LlamaIndexとW&B Weaveの助けを借りて、私たちは今日それを構築します。
アプリの主な機能
-
意味検索機能
単に単語を検索するのではなく、文脈や意味に基づいてメニュー項目を検索できるようにします。 -
ユーザーフレンドリーなインターフェース
メニューのPDF画面とチャット画面 -
アプリのデータ分析
ユーザーのインタラクションや嗜好を分析する機能もつけます
技術概要
以下のステップでやっていきます
- GPT-4によるデータの標準化：メニューを構造化され標準化されたJSONにします
- ベクトル・インデックスの作成：それをベクトルに変換します。ベクトルは単語の意味を含んでおり、意味検索ができるようになります。
- ベクトル・インデックスのクエリ：クエリ検索をするにはクエリもベクトル化しなくてはなりません。
- GPT-3.5による結果のフィルタリング：ベクトルインデックスだけだと関連してない結果を返すことがあるので絞り込みのためにGPT-3.5を使います。
なぜベクトル・インデックスが必要なのか？
このアプリを実装する立場として、あなたはこう考えたでしょう。
「メニューをループして、ユーザのクエリと各ページをモデルに渡して、どの結果が関連するのかをモデルに尋ねるだけでいいじゃないか」と。
しかしそれではAPI料金がバカ高くなります。すべてのメニュー検索に対してAPI通信を行わなくてはならないからです。応答時間の遅れも問題です。
そのためまずはベクトル検索を行うことで、料金の節約をはかるのです。
GPT-4によるデータの標準化：基礎固め
ベクトルインデックスを作成する前に、メニューデータを簡単にアクセスできる形式にする必要があります。そのためには、PDFのメニューをJSON形式に変換する必要があります。
GPT-4はこのような作業に最適なツールです。モデルと特別に作られたプロンプトを使って、ドキュメントをモデルに渡し、メニューをJSON形式に変換することができます。
そのスクリプトは割愛しますが、基本的にメニューの各ページをループし、その内容をGPT-4に渡し、GPT-4がメニューをJSON形式に変換します。動かないときもありますが都度調整。
with open(pdf_path, 'rb') as file：
reader = PyPDF2.PdfReader(file)
for page_num in range(len(reader.pages))：
page_text = reader.pages[page_num].extract_text()
text_chunks = split_text(page_text)
for chunk in text_chunks：
prompt_text = ("JSONオブジェクト(リスト)に変換してほしいテキストを渡します")
"各項目にはタイトルと説明があります。例えば、テキストがレストランの様々な料理について説明している場合、"
"JSON出力は次のようになります：[{'タイトル'：'title': 'Chicken Special', 'description': 'ハーブとスパイスで味付けされた美味しいチキン料理です。', 'keywords': 'chicken'}, " "のようになります。
"{'title'：'title': 'シーフードプラッター', 'description': 'エビ、ホタテ、ロブスターなどの新鮮なシーフードの盛り合わせ。', 'keywords': 'シーフード、エビ、プラッター'}""
"注意：アイテムのタイトルは、アイテムのカテゴリではなく、特定の料理/料理/前菜/ドリンクなどです。-> あくまで特定のアイテムであり、カテゴリーではありません。
"注意：キーワードは、アイテムを説明する2-4のカテゴリ/一般的な検索用語であるべきです。例えば（ただし、これらに限定されない）→デザート、サイド、チキン、サンドイッチ、カクテル、ドリンク、ハンバーガー、サラダなど→複数のカテゴリ/説明語を使用するようにしてください"
"このデータはチャンクされているので、（タイトルや説明の）最初か最後のデータが、他のエントリーと比べて不完全に見える場合は、それをスキップしてください"
"jsonオブジェクトだけで応答してください。完全なデータを変換してください。切り捨てたり、早めに止めたりしないでください。変換してほしいデータはこちらです： " + chunk)
response = client.chat.completions.create(
model="gpt-4-1106-preview",
messages=[
{"role"："system", "content"：「あなたは親切なアシスタントです、｝
{"role"："user", "content": prompt_text}.
]
)
これは秘蔵のプロンプトですが特別に公開します。
メニューからモデルに一度に渡すのは4000文字程度です。
さらにモデルに渡されるセクションの間にオーバーラップ(テキストやデータをモデルに入力する際に、一部の情報が重複するようにする方法のこと)を使用することで、関連する項目をスキップすることを防いでいます。
以下は、出来上がったファイルのサンプルです：
{
"title"："ローストハーフチキン"
"description"："自家製ハーブバターでじっくりロースト。お好きなサイドメニューを2つお選びください。"、
"キーワード"："チキン、ロースト、ハーブバター"、
"ページ"：1
},
{
"title"："パルメザンチキンパスタ"、
"説明"："リングイネにパルメザンチキンの胸肉とモッツァレラチーズ、マリナーラソースをかけたもの、
"キーワード"："チキン、パスタ、パルメザンチーズ"、
"ページ"：1
},
スクリプトは一回実行するだけなので、コストの増加を抑えられます。
ベクトルインデックスの作成
メニューのjsonができたらそれをベクトルインデックスにします。
from flask import Flask, render_template, request, jsonify
import json
import os
from llama_index import VectorStoreIndex, ServiceContext, StorageContext, load_index_from_storage
from llama_index.schema import Document
from llama_index.embeddings import HuggingFaceEmbedding
from llama_index.retrievers import VectorIndexRetriever
from llama_index.query_engine import RetrieverQueryEngine
from llama_index.postprocessor import SimilarityPostprocessor
from llama_index import (VectorStoreIndex, get_response_synthesizer)
def create_document_from_json_item(json_item):
ti = json_item['title']
des = json_item['description']
keys = json_item['keywords']
# if des: # menu/embedding modelによっては役に立ちます
# ti += des
if keys:
ti += "keywords:" + keys
document = Document(text=ti, metadata=json_item)
return document
def generate_embeddings_for_document(document, model_name="BAAI/bge-small-en-v1.5"):
embed_model = HuggingFaceEmbedding(model_name=model_name)
embeddings = embed_model.get_text_embedding(document.text)
return embeddings
file_path = "./gpt4_menu_data_v2.json"
index = None
if not os.path.exists("./index"):
with open(file_path, 'r', encoding='utf-8') as file:
json_data = json.load(file)
documents = []
for item in json_data:
document = create_document_from_json_item(item)
document_embeddings = generate_embeddings_for_document(document)
document.embedding = document_embeddings
documents.append(document)
service_context = ServiceContext.from_defaults(llm=None, embed_model='local')
index = VectorStoreIndex.from_documents(documents, service_context=service_context)
index.storage_context.persist(persist_dir="./index")
else:
storage_context = StorageContext.from_defaults(persist_dir="./index")
service_context = ServiceContext.from_defaults(llm=None, embed_model='local')
index = load_index_from_storage(storage_context, service_context=service_context)
ベクトル・インデックスが作成されたので、ユーザ・クエリからインデックスをクエリするコードを書く準備ができました。
retriever = VectorIndexRetriever(index=index, similarity_top_k=10)
service_context = ServiceContext.from_defaults(llm=None, embed_model='local')
response_synthesizer = get_response_synthesizer(service_context=service_context)
query_engine = RetrieverQueryEngine(retriever=retriever, response_synthesizer=response_synthesizer, node_postprocessors=[SimilarityPostprocessor(similarity_cutoff=0.6)])
def parse_response_to_json(response_str)：
items = response_str.split("title: ")[1:] # レスポンスを分割し、最初の空のチャンクは無視する。
json_list = [].
for item in items：
lines = item.strip().split('\n')
item_json = { { "title": lines[0].
"title": lines[0].strip()、
"description": lines[1].replace("description: ", "").strip()、
"keywords": lines[2].replace("keywords: ", "").strip()、
"page": int(行[3].replace("ページ: ", "").strip()))。
}
json_list.append(item_json)
return json_list
def query_index(query)：
response = query_engine.query(query)
# 元の結果はJSONにパースされる
return parse_response_to_json(str(response))
GPT-3.5で検索の関連性を強化
検索結果の中から最も関連性の高いアイテムだけ提示されるよう、GPT-3.5を使う。
本番環境では、GPT-3.5のようなモデルから始めて、データを収集しより小さなオープンソースモデルを使うこのも良いでしょう。
W&B Weaveでアプリの使用状況をログする
このアプリのログには価値があります。ログを残すことで企業は人気のある機能、一般的なクエリ、ユーザーの嗜好を特定することができます。
以下のコードではflaskを使ってWebアプリを作成しています。
# W&Bへのログイン
wandb.login()
# StreamTable の定数を定義する。
WB_ENTITY = "" # ここに W&B のエンティティ名を設定する、または現在のログインエンティティを使用するために空のままにする
WB_PROJECT = "ai_menu" # ここにW&Bエンティティ名を設定します。
STREAM_TABLE_NAME = "usage_data" (ストリームテーブル名)
# ストリームテーブルの定義
st = StreamTable(f"{WB_ENTITY}/{WB_PROJECT}/{STREAM_TABLE_NAME}")
app = Flask(__name__)
client = openai.OpenAI(api_key='sk-YOUR_API_KEY')
app.route('/chat')
def index()：
return render_template('chat.html')
app.route('/')
def menu()：
return render_template('index.html')
def describe_items(json_list)：
description_str = "あなたが興味を持ちそうなアイテムには、以下のようなものがあります:<br><br>"
for item in json_list：
description_str += f"<strong>{item['title']}</strong> - {item['description']}<br><br>"
return description_str
def generate_response_gpt(query, original_res)：
# GPTのプロンプトを生成する
prompt = f "これはレストランで注文する商品を探しているユーザーです。次のユーザークエリ'{query}'に対する初期結果{original_res}が与えられたら、レスポンスとして含めるのが理にかなっている項目のJSONオブジェクトを返します（たとえば、query='{query}'にまったく関係のない項目だけを削除します）。結果をjson形式で返す必要があります。"
response = client.chat.completions.create(
model="gpt-3.5-turbo-1106",
messages=[
{"role"："system", "content"：「あなたは親切なアシスタントです、｝
{"role"："user", "content": プロンプト}。
]
)
if response.choices：
reply = response.choices[0].message.content
filtered_res_json_str = re.search(r"``json(.+?)```", reply, re.DOTALL)
print(filtered_res_json_str)
if filtered_res_json_str：
filtered_res_json = json.loads(filtered_res_json_str.group(1))
if filtered_res_json_str:
filtered_res_json = json.loads(filtered_res_json_str.group(1))
if not len(filtered_res_json):
return original_res
else:
filtered_res_json = original_res
return filtered_res_json
else:
return original_res
app.route('/search', methods=['POST'])
def search()：
query = request.form.get('query')
original_res = query_index(query)
filtered_res_json = generate_response_gpt(query, original_res)
st.log({"query": query, "results": describe_items(filtered_res_json)})
return jsonify({'res': describe_items(filtered_res_json)})
if __name__ == '__main__'：
app.run(host='0.0.0.0', port=5000, debug=True)
Discussion