Skip to content

Simple

简单的图形存储索引。

SimpleGraphStore #

Bases: GraphStore

简单图存储。

在这个图存储中，三元组被存储在一个简单的内存字典中。

Args: simple_graph_store_data_dict（可选[字典]）：包含三元组的数据字典。有关更多详细信息，请参阅SimpleGraphStoreData。

Source code in llama_index/core/graph_stores/simple.py 
 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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
class SimpleGraphStore(GraphStore):
    """简单图存储。

在这个图存储中，三元组被存储在一个简单的内存字典中。

Args:
simple_graph_store_data_dict（可选[字典]）：包含三元组的数据字典。有关更多详细信息，请参阅SimpleGraphStoreData。"""

    def __init__(
        self,
        data: Optional[SimpleGraphStoreData] = None,
        fs: Optional[fsspec.AbstractFileSystem] = None,
        **kwargs: Any,
    ) -> None:
        """初始化参数。"""
        self._data = data or SimpleGraphStoreData()
        self._fs = fs or fsspec.filesystem("file")

    @classmethod
    def from_persist_dir(
        cls,
        persist_dir: str = DEFAULT_PERSIST_DIR,
        fs: Optional[fsspec.AbstractFileSystem] = None,
    ) -> "SimpleGraphStore":
        """从持久化目录加载。"""
        persist_path = os.path.join(persist_dir, DEFAULT_PERSIST_FNAME)
        return cls.from_persist_path(persist_path, fs=fs)

    @property
    def client(self) -> None:
        """获取客户端。
不适用于此商店。
"""
        return

    def get(self, subj: str) -> List[List[str]]:
        """获取三元组。"""
        return self._data.graph_dict.get(subj, [])

    def get_rel_map(
        self, subjs: Optional[List[str]] = None, depth: int = 2, limit: int = 30
    ) -> Dict[str, List[List[str]]]:
        """获取深度感知的相对地图。"""
        return self._data.get_rel_map(subjs=subjs, depth=depth, limit=limit)

    def upsert_triplet(self, subj: str, rel: str, obj: str) -> None:
        """添加三元组。"""
        if subj not in self._data.graph_dict:
            self._data.graph_dict[subj] = []
        if (rel, obj) not in self._data.graph_dict[subj]:
            self._data.graph_dict[subj].append([rel, obj])

    def delete(self, subj: str, rel: str, obj: str) -> None:
        """删除三元组。"""
        if subj in self._data.graph_dict:
            if (rel, obj) in self._data.graph_dict[subj]:
                self._data.graph_dict[subj].remove([rel, obj])
                if len(self._data.graph_dict[subj]) == 0:
                    del self._data.graph_dict[subj]

    def persist(
        self,
        persist_path: str = os.path.join(DEFAULT_PERSIST_DIR, DEFAULT_PERSIST_FNAME),
        fs: Optional[fsspec.AbstractFileSystem] = None,
    ) -> None:
        """将SimpleGraphStore持久化到一个目录中。"""
        fs = fs or self._fs
        dirpath = os.path.dirname(persist_path)
        if not fs.exists(dirpath):
            fs.makedirs(dirpath)

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

    def get_schema(self, refresh: bool = False) -> str:
        """获取Simple Graph存储的模式。"""
        raise NotImplementedError("SimpleGraphStore does not support get_schema")

    def query(self, query: str, param_map: Optional[Dict[str, Any]] = {}) -> Any:
        """查询简单图形存储。"""
        raise NotImplementedError("SimpleGraphStore does not support query")

    @classmethod
    def from_persist_path(
        cls, persist_path: str, fs: Optional[fsspec.AbstractFileSystem] = None
    ) -> "SimpleGraphStore":
        """从持久化目录创建一个SimpleGraphStore。"""
        fs = fs or fsspec.filesystem("file")
        if not fs.exists(persist_path):
            logger.warning(
                f"No existing {__name__} found at {persist_path}. "
                "Initializing a new graph_store from scratch. "
            )
            return cls()

        logger.debug(f"Loading {__name__} from {persist_path}.")
        with fs.open(persist_path, "rb") as f:
            data_dict = json.load(f)
            data = SimpleGraphStoreData.from_dict(data_dict)
        return cls(data)

    @classmethod
    def from_dict(cls, save_dict: dict) -> "SimpleGraphStore":
        data = SimpleGraphStoreData.from_dict(save_dict)
        return cls(data)

    def to_dict(self) -> dict:
        return self._data.to_dict()

client property #

client: None

获取客户端。 不适用于此商店。

from_persist_dir classmethod #

from_persist_dir(
    persist_dir: str = DEFAULT_PERSIST_DIR,
    fs: Optional[AbstractFileSystem] = None,
) -> SimpleGraphStore

从持久化目录加载。

Source code in llama_index/core/graph_stores/simple.py 
87
88
89
90
91
92
93
94
95
@classmethod
def from_persist_dir(
    cls,
    persist_dir: str = DEFAULT_PERSIST_DIR,
    fs: Optional[fsspec.AbstractFileSystem] = None,
) -> "SimpleGraphStore":
    """从持久化目录加载。"""
    persist_path = os.path.join(persist_dir, DEFAULT_PERSIST_FNAME)
    return cls.from_persist_path(persist_path, fs=fs)

get #

get(subj: str) -> List[List[str]]

获取三元组。

Source code in llama_index/core/graph_stores/simple.py 
104
105
106
def get(self, subj: str) -> List[List[str]]:
    """获取三元组。"""
    return self._data.graph_dict.get(subj, [])

get_rel_map #

get_rel_map(
    subjs: Optional[List[str]] = None,
    depth: int = 2,
    limit: int = 30,
) -> Dict[str, List[List[str]]]

获取深度感知的相对地图。

Source code in llama_index/core/graph_stores/simple.py 
108
109
110
111
112
def get_rel_map(
    self, subjs: Optional[List[str]] = None, depth: int = 2, limit: int = 30
) -> Dict[str, List[List[str]]]:
    """获取深度感知的相对地图。"""
    return self._data.get_rel_map(subjs=subjs, depth=depth, limit=limit)

upsert_triplet #

upsert_triplet(subj: str, rel: str, obj: str) -> None

添加三元组。

Source code in llama_index/core/graph_stores/simple.py 
114
115
116
117
118
119
def upsert_triplet(self, subj: str, rel: str, obj: str) -> None:
    """添加三元组。"""
    if subj not in self._data.graph_dict:
        self._data.graph_dict[subj] = []
    if (rel, obj) not in self._data.graph_dict[subj]:
        self._data.graph_dict[subj].append([rel, obj])

delete #

delete(subj: str, rel: str, obj: str) -> None

删除三元组。

Source code in llama_index/core/graph_stores/simple.py 
121
122
123
124
125
126
127
def delete(self, subj: str, rel: str, obj: str) -> None:
    """删除三元组。"""
    if subj in self._data.graph_dict:
        if (rel, obj) in self._data.graph_dict[subj]:
            self._data.graph_dict[subj].remove([rel, obj])
            if len(self._data.graph_dict[subj]) == 0:
                del self._data.graph_dict[subj]

persist #

persist(
    persist_path: str = os.path.join(
        DEFAULT_PERSIST_DIR, DEFAULT_PERSIST_FNAME
    ),
    fs: Optional[AbstractFileSystem] = None,
) -> None

将SimpleGraphStore持久化到一个目录中。

Source code in llama_index/core/graph_stores/simple.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
def persist(
    self,
    persist_path: str = os.path.join(DEFAULT_PERSIST_DIR, DEFAULT_PERSIST_FNAME),
    fs: Optional[fsspec.AbstractFileSystem] = None,
) -> None:
    """将SimpleGraphStore持久化到一个目录中。"""
    fs = fs or self._fs
    dirpath = os.path.dirname(persist_path)
    if not fs.exists(dirpath):
        fs.makedirs(dirpath)

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

get_schema #

get_schema(refresh: bool = False) -> str

获取Simple Graph存储的模式。

Source code in llama_index/core/graph_stores/simple.py 
143
144
145
def get_schema(self, refresh: bool = False) -> str:
    """获取Simple Graph存储的模式。"""
    raise NotImplementedError("SimpleGraphStore does not support get_schema")

query #

query(
    query: str, param_map: Optional[Dict[str, Any]] = {}
) -> Any

查询简单图形存储。

Source code in llama_index/core/graph_stores/simple.py 
147
148
149
def query(self, query: str, param_map: Optional[Dict[str, Any]] = {}) -> Any:
    """查询简单图形存储。"""
    raise NotImplementedError("SimpleGraphStore does not support query")

from_persist_path classmethod #

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

从持久化目录创建一个SimpleGraphStore。

Source code in llama_index/core/graph_stores/simple.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
@classmethod
def from_persist_path(
    cls, persist_path: str, fs: Optional[fsspec.AbstractFileSystem] = None
) -> "SimpleGraphStore":
    """从持久化目录创建一个SimpleGraphStore。"""
    fs = fs or fsspec.filesystem("file")
    if not fs.exists(persist_path):
        logger.warning(
            f"No existing {__name__} found at {persist_path}. "
            "Initializing a new graph_store from scratch. "
        )
        return cls()

    logger.debug(f"Loading {__name__} from {persist_path}.")
    with fs.open(persist_path, "rb") as f:
        data_dict = json.load(f)
        data = SimpleGraphStoreData.from_dict(data_dict)
    return cls(data)