Class Introduction

Function

Provides cache update, cache search, and cache flushing functions for user questions and answers.

Prototype

from mx_rag.cache import MxRAGCache

MxRAGCache(cache_name, config)

Parameters

Example

import json
import getpass
from paddle.base import libpaddle
from pymilvus import MilvusClient
from mx_rag.cache import CacheConfig, SimilarityCacheConfig
from mx_rag.cache import EvictPolicy
from mx_rag.cache import MxRAGCache
from mx_rag.utils import ClientParam

dim = 1024

cache_config = CacheConfig(
    cache_size=100,
    eviction_policy=EvictPolicy.LRU,
    data_save_folder="path_to_cache_save_folder"
)
cache = MxRAGCache("memory_cache", cache_config)
# Check whether the cache instance is successfully initialized.
cache_obj = cache.get_obj()
if cache_obj is None:
    print(f"cache init failed")
similarity_config = SimilarityCacheConfig(
    vector_config={
        "vector_type": "milvus_db",
        "x_dim": dim,
        "client": MilvusClient("https://x.x.x.x:port", user="xxx", password=getpass.getpass(), secure=True,   client_pem_path="path_to/client.pem", client_key_path="path_to/client.key", ca_pem_path="path_to/ca.pem", server_name="localhost")
        "collection_name": "mxrag_cache_123",  # Label of milvus_db
        "use_http": False,
        "param": None
    },
    cache_config="sqlite",
    emb_config={
        "embedding_type": "tei_embedding",
        "url": "https://<ip>:<port>/embed",  # IP address and listening port of the tei_embedding service
        "client_param": ClientParam(ca_file="/path/to/ca.crt")
    },
    similarity_config={
        "similarity_type": "tei_reranker",
        "url": "https://<ip>:<port>/rerank",  # IP address and listening port of the tei_reranker service
        "client_param": ClientParam(ca_file="/path/to/ca.crt")
    },
    retrieval_top_k=1,
    cache_size=100,
    auto_flush=100,
    similarity_threshold=0.70,
    data_save_folder="path_to_cache_save_folder",
    disable_report=True,
    eviction_policy=EvictPolicy.FIFO
)
similarity_cache = MxRAGCache("similarity_cache", similarity_config)
# Set cache cascading.
cache.join(similarity_cache)
# Set the maximum number of characters in each cache record to 4000.
cache.set_cache_limit(4000)
# Set whether to display the cache process in detail.
cache.set_verbose(False)
# Manually update the cache.
cache.update("Who is Xiao Ming's father?", json.dumps({"Who is Xiao Ming's father?": "Xiao Ming's father is Da Ming."}))
# Exact match result
res = cache.search("Who is Xiao Ming's father?")
print(f"memory match res: {res}")
# Semantic similarity matching result
res = cache.search("What's the name of Xiao Ming's father?")
print(f"similarity match res: {res}")
# Manually call flush to flush the cache to drives. The cache can also be automatically flushed to drives based on the auto_flush configuration.
cache.flush()
# Delete the files and data that have been flushed to drives.
cache.clear()