Skip to content

Simple

SimpleKVStore #

Bases: BaseInMemoryKVStore

简单的内存键值存储。

Parameters:

Name Type Description Default
data 可选[数据类型]

用于初始化存储的数据

 None
Source code in llama_index/core/storage/kvstore/simple_kvstore.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class SimpleKVStore(BaseInMemoryKVStore):
    """简单的内存键值存储。

    Args:
        data (可选[数据类型]): 用于初始化存储的数据"""

    def __init__(
        self,
        data: Optional[DATA_TYPE] = None,
    ) -> None:
        """初始化一个SimpleKVStore。"""
        self._data: DATA_TYPE = data or {}

    def put(self, key: str, val: dict, collection: str = DEFAULT_COLLECTION) -> None:
        """将一个键值对放入存储中。"""
        if collection not in self._data:
            self._data[collection] = {}
        self._data[collection][key] = val.copy()

    async def aput(
        self, key: str, val: dict, collection: str = DEFAULT_COLLECTION
    ) -> None:
        """将一个键值对放入存储中。"""
        self.put(key, val, collection)

    def get(self, key: str, collection: str = DEFAULT_COLLECTION) -> Optional[dict]:
        """从商店获取一个值。"""
        collection_data = self._data.get(collection, None)
        if not collection_data:
            return None
        if key not in collection_data:
            return None
        return collection_data[key].copy()

    async def aget(
        self, key: str, collection: str = DEFAULT_COLLECTION
    ) -> Optional[dict]:
        """从商店获取一个值。"""
        return self.get(key, collection)

    def get_all(self, collection: str = DEFAULT_COLLECTION) -> Dict[str, dict]:
        """从商店获取所有的数值。"""
        return self._data.get(collection, {}).copy()

    async def aget_all(self, collection: str = DEFAULT_COLLECTION) -> Dict[str, dict]:
        """从商店获取所有的数值。"""
        return self.get_all(collection)

    def delete(self, key: str, collection: str = DEFAULT_COLLECTION) -> bool:
        """从存储中删除一个数值。"""
        try:
            self._data[collection].pop(key)
            return True
        except KeyError:
            return False

    async def adelete(self, key: str, collection: str = DEFAULT_COLLECTION) -> bool:
        """从存储中删除一个数值。"""
        return self.delete(key, collection)

    def persist(
        self, persist_path: str, fs: Optional[fsspec.AbstractFileSystem] = None
    ) -> None:
        """持久化存储。"""
        fs = fs or fsspec.filesystem("file")
        dirpath = os.path.dirname(persist_path)
        if not fs.exists(dirpath):
            fs.makedirs(dirpath)

        with fs.open(persist_path, "w") as f:
            f.write(json.dumps(self._data))

    @classmethod
    def from_persist_path(
        cls, persist_path: str, fs: Optional[fsspec.AbstractFileSystem] = None
    ) -> "SimpleKVStore":
        """从持久路径和文件系统加载一个SimpleKVStore。"""
        fs = fs or fsspec.filesystem("file")
        logger.debug(f"Loading {__name__} from {persist_path}.")
        with fs.open(persist_path, "rb") as f:
            data = json.load(f)
        return cls(data)

    def to_dict(self) -> dict:
        """将商店保存为字典。"""
        return self._data

    @classmethod
    def from_dict(cls, save_dict: dict) -> "SimpleKVStore":
        """从字典中加载一个SimpleKVStore。"""
        return cls(save_dict)

put #

put(
    key: str,
    val: dict,
    collection: str = DEFAULT_COLLECTION,
) -> None

将一个键值对放入存储中。

Source code in llama_index/core/storage/kvstore/simple_kvstore.py 
30
31
32
33
34
def put(self, key: str, val: dict, collection: str = DEFAULT_COLLECTION) -> None:
    """将一个键值对放入存储中。"""
    if collection not in self._data:
        self._data[collection] = {}
    self._data[collection][key] = val.copy()

aput async #

aput(
    key: str,
    val: dict,
    collection: str = DEFAULT_COLLECTION,
) -> None

将一个键值对放入存储中。

Source code in llama_index/core/storage/kvstore/simple_kvstore.py 
36
37
38
39
40
async def aput(
    self, key: str, val: dict, collection: str = DEFAULT_COLLECTION
) -> None:
    """将一个键值对放入存储中。"""
    self.put(key, val, collection)

get #

get(
    key: str, collection: str = DEFAULT_COLLECTION
) -> Optional[dict]

从商店获取一个值。

Source code in llama_index/core/storage/kvstore/simple_kvstore.py 
42
43
44
45
46
47
48
49
def get(self, key: str, collection: str = DEFAULT_COLLECTION) -> Optional[dict]:
    """从商店获取一个值。"""
    collection_data = self._data.get(collection, None)
    if not collection_data:
        return None
    if key not in collection_data:
        return None
    return collection_data[key].copy()

aget async #

aget(
    key: str, collection: str = DEFAULT_COLLECTION
) -> Optional[dict]

从商店获取一个值。

Source code in llama_index/core/storage/kvstore/simple_kvstore.py 
51
52
53
54
55
async def aget(
    self, key: str, collection: str = DEFAULT_COLLECTION
) -> Optional[dict]:
    """从商店获取一个值。"""
    return self.get(key, collection)

get_all #

get_all(
    collection: str = DEFAULT_COLLECTION,
) -> Dict[str, dict]

从商店获取所有的数值。

Source code in llama_index/core/storage/kvstore/simple_kvstore.py 
57
58
59
def get_all(self, collection: str = DEFAULT_COLLECTION) -> Dict[str, dict]:
    """从商店获取所有的数值。"""
    return self._data.get(collection, {}).copy()

aget_all async #

aget_all(
    collection: str = DEFAULT_COLLECTION,
) -> Dict[str, dict]

从商店获取所有的数值。

Source code in llama_index/core/storage/kvstore/simple_kvstore.py 
61
62
63
async def aget_all(self, collection: str = DEFAULT_COLLECTION) -> Dict[str, dict]:
    """从商店获取所有的数值。"""
    return self.get_all(collection)

delete #

delete(
    key: str, collection: str = DEFAULT_COLLECTION
) -> bool

从存储中删除一个数值。

Source code in llama_index/core/storage/kvstore/simple_kvstore.py 
65
66
67
68
69
70
71
def delete(self, key: str, collection: str = DEFAULT_COLLECTION) -> bool:
    """从存储中删除一个数值。"""
    try:
        self._data[collection].pop(key)
        return True
    except KeyError:
        return False

adelete async #

adelete(
    key: str, collection: str = DEFAULT_COLLECTION
) -> bool

从存储中删除一个数值。

Source code in llama_index/core/storage/kvstore/simple_kvstore.py 
73
74
75
async def adelete(self, key: str, collection: str = DEFAULT_COLLECTION) -> bool:
    """从存储中删除一个数值。"""
    return self.delete(key, collection)

persist #

persist(
    persist_path: str,
    fs: Optional[AbstractFileSystem] = None,
) -> None

持久化存储。

Source code in llama_index/core/storage/kvstore/simple_kvstore.py 
77
78
79
80
81
82
83
84
85
86
87
def persist(
    self, persist_path: str, fs: Optional[fsspec.AbstractFileSystem] = None
) -> None:
    """持久化存储。"""
    fs = fs or fsspec.filesystem("file")
    dirpath = os.path.dirname(persist_path)
    if not fs.exists(dirpath):
        fs.makedirs(dirpath)

    with fs.open(persist_path, "w") as f:
        f.write(json.dumps(self._data))

from_persist_path classmethod #

from_persist_path(
    persist_path: str,
    fs: Optional[AbstractFileSystem] = None,
) -> SimpleKVStore

从持久路径和文件系统加载一个SimpleKVStore。

Source code in llama_index/core/storage/kvstore/simple_kvstore.py 
89
90
91
92
93
94
95
96
97
98
@classmethod
def from_persist_path(
    cls, persist_path: str, fs: Optional[fsspec.AbstractFileSystem] = None
) -> "SimpleKVStore":
    """从持久路径和文件系统加载一个SimpleKVStore。"""
    fs = fs or fsspec.filesystem("file")
    logger.debug(f"Loading {__name__} from {persist_path}.")
    with fs.open(persist_path, "rb") as f:
        data = json.load(f)
    return cls(data)

to_dict #

to_dict() -> dict

将商店保存为字典。

Source code in llama_index/core/storage/kvstore/simple_kvstore.py 
100
101
102
def to_dict(self) -> dict:
    """将商店保存为字典。"""
    return self._data

from_dict classmethod #

from_dict(save_dict: dict) -> SimpleKVStore

从字典中加载一个SimpleKVStore。

Source code in llama_index/core/storage/kvstore/simple_kvstore.py 
104
105
106
107
@classmethod
def from_dict(cls, save_dict: dict) -> "SimpleKVStore":
    """从字典中加载一个SimpleKVStore。"""
    return cls(save_dict)