尝试使用Redis缓存用户的问题，并考虑减少使用openapi的token

2 年 ago
新, 韵
3 minutes

晚上好！在本文中，我们将介绍使用Python和Redis将用户的问题缓存到Redis中，并对缓存的查询进行相似度搜索，以实现减少LLM令牌使用量和提高回答速度的方法。这是关于从问题到问题的相似度搜索的实现！

背景

在使用OpenAI的API时，有时会达到使用限制（$120），导致使用受限。同时，我也认为每次都使用OPENAPI来回答相似的问题是浪费的，所以我在LangChain上考虑实现缓存。然而，现有的实现并不足够理想，因此我决定自己使用Redis来实现一个问题相似性搜索系统。

环境的准备

Python版本是3.9.18，
Redis版本是4.6.0，
Docker版本是24.0.6（构建版本为ed223bc），
docker-compose版本是1.29.2（构建版本为5becea4c），
redis-om版本是0.2.1。

※关于从Docker构建的内容将在另一篇文章中发布。

建筑项目

    1. Redis配置和连接

 

    1. 创建索引

 

    1. 添加新的问题

 

    搜索相似问题

实现部分的详细解释

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.
質問を入力してください:

类似的问题已经存在。不添加到Redis。由此可见，由于得分为0.09或以下，新的缓存数据没有被注册。得分越接近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来生成回答，将其与用户的问题结合起来并进行缓存，这样就可以从缓存中获取类似的问题的答案！

✖
广告
将在 10 秒后关闭
bannerAds