Redisを使ってユーザからの質問をキャッシュして、openapiのtoken利用削減を検討してみる。

こんばんは！！本記事では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からの構築は別記事にて投稿します。

## 工程
1. **Redisの設定と接続**
2. **インデックスの作成**
3. **新しい質問の追加**
4. **類似する質問の検索**

### 実装部分の詳細解説

#### 1. Redisの設定と接続
まず、Redisサーバーに接続する必要があります。以下のコードを使用して、Redisに接続します。

```python
import redis

# Redis接続設定
r = redis.Redis(host="pj_config_ai_redis_1", port=6379)
```

#### 2. インデックスの作成
次に、Redis内にインデックスを作成します。このインデックスは、質問の埋め込みベクトルを保存するために使用されます。

```python
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に保存します。

```python
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. 類似する質問の検索
ユーザーからの新しい質問に対して、類似する質問を検索します。

```python
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にアクセスします。
```python
pero0125@hpdesktop:~/pj_config_ai$ docker exec -it pj_config_ai_redis_1 redis-cli --raw
```

あらかじめデータセットを用意しておきます。
```python
127.0.0.1:6379> keys *
doc:朝の挨拶は？_1addbec461763ef8b4049aa8ef82e82d312d462fcf996eb0a187776e3bf12fc3
doc:近所の犬はどうですか？_a4329822fc61481fd433ca0e9951210bda5be739156d84d06988b3b00a462e2f
doc:日本の挨拶を教えて_60a934ccb7b415fd23a82edbff4635ff1c2d235d4fea9c08733106e2295cb0db
doc:あなたの猫はどうですか？_0e1747975d38d91d50e2519667756e6e89482fefca0bb565d6783f5dae84483d
doc:猫について教えて_56c04baa15a49eaecf31ac51fd2bda114767a194cf78e65f81129bf4522866ff
```

上記のコードを実行して質問を作成します。
```python
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に近いほど意味が等しいことになります。
```python
質問を入力してください: ペットの調子はどう？
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で回答を生成してもらって、ユーザの質問と合わせてキャッシュしておけば、類似の質問はキャッシュから回答を得られるようになります！