简单

简单图存储索引。

SimpleGraphStore

基础：GraphStore

简单图存储。

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

参数

名称	类型	描述	默认值
simple_graph_store_data_dict	可选[dict]	包含三元组的数据字典。详见 SimpleGraphStoreData。	必需

源代码位于 llama-index-core/llama_index/core/graph_stores/simple.py

class SimpleGraphStore(GraphStore):
    """
    Simple Graph Store.

    In this graph store, triplets are stored within a simple, in-memory dictionary.

    Args:
        simple_graph_store_data_dict (Optional[dict]): data dict
            containing the triplets. See SimpleGraphStoreData
            for more details.

    """

    def __init__(
        self,
        data: Optional[SimpleGraphStoreData] = None,
        fs: Optional[fsspec.AbstractFileSystem] = None,
        **kwargs: Any,
    ) -> None:
        """Initialize params."""
        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":
        """Load from persist dir."""
        persist_path = os.path.join(persist_dir, DEFAULT_PERSIST_FNAME)
        return cls.from_persist_path(persist_path, fs=fs)

    @property
    def client(self) -> None:
        """
        Get client.
        Not applicable for this store.
        """
        return

    def get(self, subj: str) -> List[List[str]]:
        """Get triplets."""
        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]]]:
        """Get depth-aware rel map."""
        return self._data.get_rel_map(subjs=subjs, depth=depth, limit=limit)

    def upsert_triplet(self, subj: str, rel: str, obj: str) -> None:
        """Add triplet."""
        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:
        """Delete triplet."""
        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:
        """Persist the SimpleGraphStore to a directory."""
        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:
        """Get the schema of the Simple Graph store."""
        raise NotImplementedError("SimpleGraphStore does not support get_schema")

    def query(self, query: str, param_map: Optional[Dict[str, Any]] = {}) -> Any:
        """Query the Simple Graph store."""
        raise NotImplementedError("SimpleGraphStore does not support query")

    @classmethod
    def from_persist_path(
        cls, persist_path: str, fs: Optional[fsspec.AbstractFileSystem] = None
    ) -> "SimpleGraphStore":
        """Create a SimpleGraphStore from a persist directory."""
        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: None

获取客户端。不适用于此存储。

从持久化目录加载 类方法

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

从持久化目录加载。

源代码位于 llama-index-core/llama_index/core/graph_stores/simple.py

@classmethod
def from_persist_dir(
    cls,
    persist_dir: str = DEFAULT_PERSIST_DIR,
    fs: Optional[fsspec.AbstractFileSystem] = None,
) -> "SimpleGraphStore":
    """Load from persist dir."""
    persist_path = os.path.join(persist_dir, DEFAULT_PERSIST_FNAME)
    return cls.from_persist_path(persist_path, fs=fs)

获取

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

获取三元组。

源代码位于 llama-index-core/llama_index/core/graph_stores/simple.py

def get(self, subj: str) -> List[List[str]]:
    """Get triplets."""
    return self._data.graph_dict.get(subj, [])

获取关系图

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

获取深度感知关系图。

源代码位于 llama-index-core/llama_index/core/graph_stores/simple.py

def get_rel_map(
    self, subjs: Optional[List[str]] = None, depth: int = 2, limit: int = 30
) -> Dict[str, List[List[str]]]:
    """Get depth-aware rel map."""
    return self._data.get_rel_map(subjs=subjs, depth=depth, limit=limit)

插入或更新三元组

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

添加三元组。

源代码位于 llama-index-core/llama_index/core/graph_stores/simple.py

def upsert_triplet(self, subj: str, rel: str, obj: str) -> None:
    """Add triplet."""
    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(subj: str, rel: str, obj: str) -> None

删除三元组。

源代码位于 llama-index-core/llama_index/core/graph_stores/simple.py

def delete(self, subj: str, rel: str, obj: str) -> None:
    """Delete triplet."""
    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_path: str = join(DEFAULT_PERSIST_DIR, DEFAULT_PERSIST_FNAME), fs: Optional[AbstractFileSystem] = None) -> None

将 SimpleGraphStore 持久化到目录。

源代码位于 llama-index-core/llama_index/core/graph_stores/simple.py

def persist(
    self,
    persist_path: str = os.path.join(DEFAULT_PERSIST_DIR, DEFAULT_PERSIST_FNAME),
    fs: Optional[fsspec.AbstractFileSystem] = None,
) -> None:
    """Persist the SimpleGraphStore to a directory."""
    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(refresh: bool = False) -> str

获取 Simple Graph store 的模式。

源代码位于 llama-index-core/llama_index/core/graph_stores/simple.py

def get_schema(self, refresh: bool = False) -> str:
    """Get the schema of the Simple Graph store."""
    raise NotImplementedError("SimpleGraphStore does not support get_schema")

查询

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

查询 Simple Graph store。

源代码位于 llama-index-core/llama_index/core/graph_stores/simple.py

def query(self, query: str, param_map: Optional[Dict[str, Any]] = {}) -> Any:
    """Query the Simple Graph store."""
    raise NotImplementedError("SimpleGraphStore does not support query")

从持久化路径加载 类方法

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

从持久化目录创建 SimpleGraphStore。

源代码位于 llama-index-core/llama_index/core/graph_stores/simple.py

@classmethod
def from_persist_path(
    cls, persist_path: str, fs: Optional[fsspec.AbstractFileSystem] = None
) -> "SimpleGraphStore":
    """Create a SimpleGraphStore from a persist directory."""
    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)