こんばんは！！本記事ではPythonとRedisを使ってユーザからの質問をredis内にキャッシュし、キャッシュした問合せに対して類似性検索を行い、LLMへのtoken利用数の削減や回答速度の向上を計る方法について記載します。QからQへの類似性検索の実装となります！
背景
OpenAIのAPIを使用する際、usage limit($120)に達してしまうことがあり、利用が制限されることがありました。またユーザから同じような質問を毎回OPENAPIを使うのは無駄と思い何とかならないかと思っていました。
LangChainでキャッシュの実装を検討していたのですが、既存の実装では十分とは言えなかったので、自前で質問の類似性検索システムをRedisを使って実装しました。
環境の準備
Python 3.9.18
redis 4.6.0
Docker version 24.0.6, build ed223bc
docker-compose version 1.29.2, build 5becea4c
redis-om==0.2.1
※dockerからの構築は別記事にて投稿します。
工程
- Redisの設定と接続
- インデックスの作成
- 新しい質問の追加
- 類似する質問の検索
実装部分の詳細解説
1. Redisの設定と接続
まず、Redisサーバーに接続する必要があります。以下のコードを使用して、Redisに接続します。
import redis
# Redis接続設定
r = redis.Redis(host="pj_config_ai_redis_1", port=6379)
2. インデックスの作成
次に、Redis内にインデックスを作成します。このインデックスは、質問の埋め込みベクトルを保存するために使用されます。
from redis.commands.search.field import TagField, VectorField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
# 定数の設定
INDEX_NAME = "index" # Vector Index Name
VECTOR_DIMENSIONS = 1536 # Vector Dimensions
def create_index(vector_dimensions: int):
try:
# check to see if index exists
r.ft(INDEX_NAME).info()
print("Index already exists!")
except:
# schema
schema = (
TagField("tag"), # Tag Field Name
VectorField("vector", # Vector Field Name
"FLAT", { # Vector Index Type: FLAT or HNSW
"TYPE": "FLOAT32", # FLOAT32 or FLOAT64
"DIM": vector_dimensions, # Number of Vector Dimensions
"DISTANCE_METRIC": "COSINE", # Vector Search Distance Metric
}
),
)
# index Definition
definition = IndexDefinition(prefix=[DOC_PREFIX], index_type=IndexType.HASH)
# create Index
r.ft(INDEX_NAME).create_index(fields=schema, definition=definition)
3. 新しい質問の追加
ユーザーからの新しい質問を受け取り、それを埋め込みベクトルとしてRedisに保存します。
import numpy as np
import openai
import hashlib
DOC_PREFIX = "doc:" # RediSearch Key Prefix for the Index
SIMILARITY_THRESHOLD = 0.09 # 類似度の閾値
def create_document_key(question):
# 質問を20文字に制限
truncated_question = question[:20]
# ユーザーの質問をハッシュ化
question_hash = hashlib.sha256(question.encode('utf-8', errors='replace')).hexdigest() # 最初の10文字のハッシュを使用
# ユニークなインデックスキーを作成
document_key = f"{truncated_question}_{question_hash}"
return document_key
def add_new_question(new_question, new_answer):
# 新しい質問のベクトルを作成
response = openai.Embedding.create(input=[new_question], engine="text-embedding-ada-002")
new_question_embedding = np.array([r["embedding"] for r in response["data"]], dtype=np.float32)[0]
# 既存の質問と類似度を計算
query = (
Query("(@tag:{ openai })=>[KNN 1 @vector $vec as score]")
.sort_by("score")
.return_fields("score")
.paging(0, 1)
.dialect(2)
)
query_params = {"vec": new_question_embedding.tobytes()}
result = r.ft(INDEX_NAME).search(query, query_params).docs
print(f"add_new_qu: {result}")
# 類似度スコアが閾値以上であるかチェック
if result and float(result[0].score) <= SIMILARITY_THRESHOLD:
print("Similar question already exists. Not adding to Redis.")
else:
# 類似度スコアが閾値以下の場合、新しい質問をRedisに追加
doc_key = create_document_key(new_question)
r.hset(f"doc:{doc_key}", mapping={
"vector": new_question_embedding.tobytes(),
"answer": new_answer,
"tag": "openai"
})
print(f"Added new question: {new_question}")
4. 類似する質問の検索
ユーザーからの新しい質問に対して、類似する質問を検索します。
from redis.commands.search.query import Query
def search_similar_questions(query_question):
# クエリのベクトルを作成
response = openai.Embedding.create(input=[query_question], engine="text-embedding-ada-002")
query_embedding = np.array([r["embedding"] for r in response["data"]], dtype=np.float32)[0]
# 類似する質問を検索
query = (
Query("(@tag:{ openai })=>[KNN 3 @vector $vec as score]")
.sort_by("score")
.return_fields("answer", "score")
.paging(0, 5)
.dialect(2)
)
query_params = {"vec": query_embedding.t
obytes()}
result = r.ft(INDEX_NAME).search(query, query_params).docs
if result:
for doc in result:
print(f"Answer: {doc.answer}, Score: {doc.score}")
else:
print("No similar questions found.") # scoreでresultを絞った場合有効
動作検証
上記のコードを実行すると、ユーザーからの新しい質問に対して類似する質問を検索し、その回答と類似度スコアを表示します。
以下でredisにアクセスします。
pero0125@hpdesktop:~/pj_config_ai$ docker exec -it pj_config_ai_redis_1 redis-cli --raw
あらかじめデータセットを用意しておきます。
127.0.0.1:6379> keys *
doc:朝の挨拶は？_1addbec461763ef8b4049aa8ef82e82d312d462fcf996eb0a187776e3bf12fc3
doc:近所の犬はどうですか？_a4329822fc61481fd433ca0e9951210bda5be739156d84d06988b3b00a462e2f
doc:日本の挨拶を教えて_60a934ccb7b415fd23a82edbff4635ff1c2d235d4fea9c08733106e2295cb0db
doc:あなたの猫はどうですか？_0e1747975d38d91d50e2519667756e6e89482fefca0bb565d6783f5dae84483d
doc:猫について教えて_56c04baa15a49eaecf31ac51fd2bda114767a194cf78e65f81129bf4522866ff
上記のコードを実行して質問を作成します。
root@a20c6bbbdee0:/app/call_langchain# python3 redisearch.py
Index already exists!
新しい質問を入力してください: 猫の調子はどう？
その質問の回答を入力してください: 元気だよ
add_new_qu: [Document {'id': 'doc:あなたの猫はどうですか？_0e1747975d38d91d50e2519667756e6e89482fefca0bb565d6783f5dae84483d', 'payload': None, 'score': '0.0652241706848'}]
Similar question already exists. Not adding to Redis.
質問を入力してください:
Similar question already exists. Not adding to Redis. となり、scoreが0.09以下のため新たにキャッシュデータに登録はしていないことがわかります。scoreが0に近いほど意味が等しいことになります。
質問を入力してください: ペットの調子はどう？
Answer: My cats escaped and got out before I could close the door., Score: 0.092055439949
Answer: The dog next door barks really loudly., Score: 0.0969498157501
Answer: ねこ, Score: 0.134126365185
root@a20c6bbbdee0:/app/call_langchain#
ペットの調子はどう？ と質問をするとキャッシュ内のデータからスコア順にピックアップしてくれます。
以上、これでLLMで回答を生成してもらって、ユーザの質問と合わせてキャッシュしておけば、類似の質問はキャッシュから回答を得られるようになります！