GPTCache
LangChainでOpenAI APIを利用したアプリケーションを開発していると、以下のような悩みが出てくる。
- OpenAI APIへの問い合わせを極力減らしたい...タダじゃないし
- OpenAI APIさん、実は
temperature=0でも出力がブレることがある（特に入力が長い場合）(1)
これを解決するには、OpenAI APIの回答をキャッシュしておいて、同じ質問が来たら再利用すれば一件落着👏なのだが、単純に
「同じクエリがきたらキャッシュを使うよ！」 (exact match cache)
という発想だと、一般的なチャットボット等を考えた場合実用性がない。ユーザが一字一句同じクエリを投げてくることは稀だろうし、全てのクエリに対する回答をいちいちキャッシュしてたらあっという間にメモリがいっぱいになってしまう。そこで、
「似たようなクエリが来たらキャッシュを使うよ！」 (similar match cache)
という発想でキャッシュを使用したいが、これを実現するためには、
ユーザクエリのembeddingを取る⇨既存のクエリのembeddingと類似度を比較する⇨閾値以上ならキャッシュされた回答を返す
という流れを実装する必要があり、結構骨が折れる。
GPTCacheというライブラリを使えば、exact match cacheもsimilar match cacheも簡単に実装できる。
https://github.com/zilliztech/GPTCache
仕組み
核となるのはVector storeとCache storage。
Vector storeにはユーザクエリの本文の埋め込みベクトルを保持し、
Cache storageにはユーザークエリの本文とChatGPTの回答、およびそのIDを保持する。
新しいクエリが入力されると、クエリの埋め込みベクトルを計算し、Vector storeに保存されている既存クエリの埋め込みベクトルの中から類似したベクトルを検索(topkは指定可能)。
閾値＜類似度 なら、既存クエリに対応する回答をCache Storageから取得。
閾値＞類似度 なら、OpenAI APIを利用する。結果はVector storeとCache storageに保存。
LLM Adapterを選ぶことで、OpenAI APIに限らずいろいろなLLMモデルやLangChainと合わせて使うことができる。
使い方
以下は、LangChainのRetrievalQAにsimilar match cacheを使用する例。GPTCacheのドキュメントで紹介している方法(2) に倣っています（ちなみにLangChainのドキュメントでもGPTCacheの使用例が紹介されているが、GPTCacheのドキュメントで紹介されている方法と異なり、GPTCacheのadapterを使わずlangchain.llm_cacheを使用するようになっている。どちらの方が良いかは未検証)。
まずはlangchainのchainオブジェクトを作成。このとき、使用するllmをGPTCacheのLangChainLLMsでラップするのがポイント。
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI
from langchain.chains import RetrievalQA
from gptcache.adapter.langchain_models import LangChainLLMs
prompt_template = """Use the following context, answer the question at the end.
If you don't know the answer, do not try to make up, just say that you don't know.
{context}
Question: {question}
answer:"""
prompt = PromptTemplate(
template=prompt_template, input_variables=["context", "question"]
)
llm = LangChainLLMs(
llm=OpenAI(
temperature=0.0,
model_name="gpt-3.5-turbo",
))
qa_engine = RetrievalQA.from_chain_type(
llm=llm,
chain_type_kwargs={"prompt": prompt},
retriever=tfidf_retriever,
return_source_documents=True,
)
次にcacheをinitializeする。
from gptcache.embedding import Huggingface
from gptcache.manager import manager_factory
from gptcache import cache
from gptcache.similarity_evaluation.distance import SearchDistanceEvaluation
# {context}まで含めたprompt全体ではなくユーザクエリだけでembeddingを取りたいので、promptからユーザクエリを抜き出す。
def get_content_func(data, **_):
return data.get("prompt").split("Question")[-1]
cache_embeddings = Huggingface("sentence-transformers/distiluse-base-multilingual-cased")
# data_managerでCacheBaseとVectorBaseにそれぞれ何を使うか定義する。ドキュメントの例のように get_data_manager を使って定義してもよい。
data_manager = manager_factory(
"sqlite,faiss", # CacheBaseにsqliteを、VectorBaseにfaissを指定
data_dir="./cache_dir", # キャッシュ置き場
vector_params={"dimension": cache_embeddings.dimension}
)
cache.init(
pre_embedding_func=get_content_func,
embedding_func=cache_embeddings.to_embeddings,
data_manager=data_manager,
similarity_evaluation=SearchDistanceEvaluation(), # 類似度の測り方
)
cache.config.auto_flush = 1
デフォルトではcache.config.auto_flush=20になっている。このままだとOpenAI APIからの回答が20件溜まるまでキャッシュファイルに書き込まれないので、開発しているアプリケーションの規模に合わせて変更。
その他、configで類似判定の閾値なども設定可能。
（ちなみにconfigのpromptsパラメータを使ったらpre_embedding_func不要になったりしないか気になっているが未調査）
キャッシュそのものについては、manager_factoryの引数で詳細設定可能。主なパラメータのデフォルト値は以下の通り。
max_size = 1000 (保持するキャッシュの最大件数)
clean_size = None (max_sizeを超えた時に削除するキャッシュの件数。指定しない場合 max_size * 0.2)
eviction = "LRU" (キャッシュを削除する際のポリシー。LRUは長い間参照されていないものから順に削除する方法）
以上。これだけでsimilar match cacheが実装できた🙌
Ref
(1) https://community.openai.com/t/a-question-on-determinism/8185
(2) https://gptcache.readthedocs.io/en/latest/bootcamp/langchain/question_answering.html#init-for-similar-match-cache