Index

LlamaIndex对象。

ObjectIndex

Bases: Generic[OT]

对象索引。

Source code in llama_index/core/objects/base.py

class ObjectIndex(Generic[OT]):
    """对象索引。"""

    def __init__(
        self, index: BaseIndex, object_node_mapping: BaseObjectNodeMapping
    ) -> None:
        self._index = index
        self._object_node_mapping = object_node_mapping

    @property
    def index(self) -> BaseIndex:
        """索引。"""
        return self._index

    @property
    def object_node_mapping(self) -> BaseObjectNodeMapping:
        """对象节点映射。"""
        return self._object_node_mapping

    @classmethod
    def from_objects(
        cls,
        objects: Sequence[OT],
        object_mapping: Optional[BaseObjectNodeMapping] = None,
        from_node_fn: Optional[Callable[[TextNode], OT]] = None,
        to_node_fn: Optional[Callable[[OT], TextNode]] = None,
        index_cls: Type[BaseIndex] = VectorStoreIndex,
        **index_kwargs: Any,
    ) -> "ObjectIndex":
        from llama_index.core.objects.utils import get_object_mapping

        # pick the best mapping if not provided
        if object_mapping is None:
            object_mapping = get_object_mapping(
                objects,
                from_node_fn=from_node_fn,
                to_node_fn=to_node_fn,
            )

        nodes = object_mapping.to_nodes(objects)
        index = index_cls(nodes, **index_kwargs)
        return cls(index, object_mapping)

    @classmethod
    def from_objects_and_index(
        cls,
        objects: Sequence[OT],
        index: BaseIndex,
        object_mapping: Optional[BaseObjectNodeMapping] = None,
        from_node_fn: Optional[Callable[[TextNode], OT]] = None,
        to_node_fn: Optional[Callable[[OT], TextNode]] = None,
    ) -> "ObjectIndex":
        from llama_index.core.objects.utils import get_object_mapping

        # pick the best mapping if not provided
        if object_mapping is None:
            object_mapping = get_object_mapping(
                objects,
                from_node_fn=from_node_fn,
                to_node_fn=to_node_fn,
            )

        return cls(index, object_mapping)

    def insert_object(self, obj: Any) -> None:
        self._object_node_mapping.add_object(obj)
        node = self._object_node_mapping.to_node(obj)
        self._index.insert_nodes([node])

    def as_retriever(
        self,
        node_postprocessors: Optional[List[BaseNodePostprocessor]] = None,
        **kwargs: Any,
    ) -> ObjectRetriever:
        return ObjectRetriever(
            retriever=self._index.as_retriever(**kwargs),
            object_node_mapping=self._object_node_mapping,
            node_postprocessors=node_postprocessors,
        )

    def as_node_retriever(self, **kwargs: Any) -> BaseRetriever:
        return self._index.as_retriever(**kwargs)

    def persist(
        self,
        persist_dir: str = DEFAULT_PERSIST_DIR,
        obj_node_mapping_fname: str = DEFAULT_PERSIST_FNAME,
    ) -> None:
        # try to persist object node mapping
        try:
            self._object_node_mapping.persist(
                persist_dir=persist_dir, obj_node_mapping_fname=obj_node_mapping_fname
            )
        except (NotImplementedError, pickle.PickleError) as err:
            warnings.warn(
                (
                    "Unable to persist ObjectNodeMapping. You will need to "
                    "reconstruct the same object node mapping to build this ObjectIndex"
                ),
                stacklevel=2,
            )
        self._index._storage_context.persist(persist_dir=persist_dir)

    @classmethod
    def from_persist_dir(
        cls,
        persist_dir: str = DEFAULT_PERSIST_DIR,
        object_node_mapping: Optional[BaseObjectNodeMapping] = None,
    ) -> "ObjectIndex":
        from llama_index.core.indices import load_index_from_storage

        storage_context = StorageContext.from_defaults(persist_dir=persist_dir)
        index = load_index_from_storage(storage_context)
        if object_node_mapping:
            return cls(index=index, object_node_mapping=object_node_mapping)
        else:
            # try to load object_node_mapping
            # assume SimpleObjectNodeMapping for simplicity as its only subclass
            # that supports this method
            try:
                object_node_mapping = SimpleObjectNodeMapping.from_persist_dir(
                    persist_dir=persist_dir
                )
            except Exception as err:
                raise Exception(
                    "Unable to load from persist dir. The object_node_mapping cannot be loaded."
                ) from err
            else:
                return cls(index=index, object_node_mapping=object_node_mapping)

index property

index: BaseIndex

索引。

object_node_mapping property

object_node_mapping: BaseObjectNodeMapping

对象节点映射。

ObjectRetriever

Bases: ChainableMixin, Generic[OT]

对象检索器。

Source code in llama_index/core/objects/base.py

class ObjectRetriever(ChainableMixin, Generic[OT]):
    """对象检索器。"""

    def __init__(
        self,
        retriever: BaseRetriever,
        object_node_mapping: BaseObjectNodeMapping[OT],
        node_postprocessors: Optional[List[BaseNodePostprocessor]] = None,
    ):
        self._retriever = retriever
        self._object_node_mapping = object_node_mapping
        self._node_postprocessors = node_postprocessors or []

    @property
    def retriever(self) -> BaseRetriever:
        """1. 检索器。"""
        return self._retriever

    @property
    def object_node_mapping(self) -> BaseObjectNodeMapping[OT]:
        """对象节点映射。"""
        return self._object_node_mapping

    @property
    def node_postprocessors(self) -> List[BaseNodePostprocessor]:
        """节点后处理器。"""
        return self._node_postprocessors

    def retrieve(self, str_or_query_bundle: QueryType) -> List[OT]:
        if isinstance(str_or_query_bundle, str):
            query_bundle = QueryBundle(query_str=str_or_query_bundle)
        else:
            query_bundle = str_or_query_bundle

        nodes = self._retriever.retrieve(query_bundle)
        for node_postprocessor in self._node_postprocessors:
            nodes = node_postprocessor.postprocess_nodes(
                nodes, query_bundle=query_bundle
            )

        return [self._object_node_mapping.from_node(node.node) for node in nodes]

    async def aretrieve(self, str_or_query_bundle: QueryType) -> List[OT]:
        if isinstance(str_or_query_bundle, str):
            query_bundle = QueryBundle(query_str=str_or_query_bundle)
        else:
            query_bundle = str_or_query_bundle

        nodes = await self._retriever.aretrieve(query_bundle)
        for node_postprocessor in self._node_postprocessors:
            nodes = node_postprocessor.postprocess_nodes(
                nodes, query_bundle=query_bundle
            )

        return [self._object_node_mapping.from_node(node.node) for node in nodes]

    def _as_query_component(self, **kwargs: Any) -> QueryComponent:
        """作为查询组件。"""
        return ObjectRetrieverComponent(retriever=self)

retriever property

retriever: BaseRetriever

1. 检索器。

object_node_mapping property

object_node_mapping: BaseObjectNodeMapping[OT]

对象节点映射。

node_postprocessors property

node_postprocessors: List[BaseNodePostprocessor]

节点后处理器。

SimpleObjectNodeMapping

Bases: BaseObjectNodeMapping[Any]

通用节点映射，适用于任何对象。

更具体地说，适用于任何具有有意义的字符串表示的对象。

Source code in llama_index/core/objects/base_node_mapping.py

class SimpleObjectNodeMapping(BaseObjectNodeMapping[Any]):
    """通用节点映射，适用于任何对象。

更具体地说，适用于任何具有有意义的字符串表示的对象。"""

    def __init__(self, objs: Optional[Sequence[Any]] = None) -> None:
        objs = objs or []
        for obj in objs:
            self.validate_object(obj)
        self._objs = {hash(str(obj)): obj for obj in objs}

    @classmethod
    def from_objects(
        cls, objs: Sequence[Any], *args: Any, **kwargs: Any
    ) -> "SimpleObjectNodeMapping":
        return cls(objs)

    @property
    def obj_node_mapping(self) -> Dict[int, Any]:
        return self._objs

    @obj_node_mapping.setter
    def obj_node_mapping(self, mapping: Dict[int, Any]) -> None:
        self._objs = mapping

    def _add_object(self, obj: Any) -> None:
        self._objs[hash(str(obj))] = obj

    def to_node(self, obj: Any) -> TextNode:
        return TextNode(id_=str(hash(str(obj))), text=str(obj))

    def _from_node(self, node: BaseNode) -> Any:
        return self._objs[hash(node.get_content(metadata_mode=MetadataMode.NONE))]

    def persist(
        self,
        persist_dir: str = DEFAULT_PERSIST_DIR,
        obj_node_mapping_fname: str = DEFAULT_PERSIST_FNAME,
    ) -> None:
        """持久化对象节点映射。

注意：这可能会失败，取决于对象类型是否可被pickle化。
"""
        if not os.path.exists(persist_dir):
            os.makedirs(persist_dir)
        obj_node_mapping_path = concat_dirs(persist_dir, obj_node_mapping_fname)
        try:
            with open(obj_node_mapping_path, "wb") as f:
                pickle.dump(self, f)
        except pickle.PickleError as err:
            raise ValueError("Objs is not pickleable") from err

    @classmethod
    def from_persist_dir(
        cls,
        persist_dir: str = DEFAULT_PERSIST_DIR,
        obj_node_mapping_fname: str = DEFAULT_PERSIST_FNAME,
    ) -> "SimpleObjectNodeMapping":
        obj_node_mapping_path = concat_dirs(persist_dir, obj_node_mapping_fname)
        try:
            with open(obj_node_mapping_path, "rb") as f:
                simple_object_node_mapping = pickle.load(f)
        except pickle.PickleError as err:
            raise ValueError("Objs cannot be loaded.") from err
        return simple_object_node_mapping

persist

persist(
    persist_dir: str = DEFAULT_PERSIST_DIR,
    obj_node_mapping_fname: str = DEFAULT_PERSIST_FNAME,
) -> None

持久化对象节点映射。

注意：这可能会失败，取决于对象类型是否可被pickle化。

Source code in llama_index/core/objects/base_node_mapping.py

    def persist(
        self,
        persist_dir: str = DEFAULT_PERSIST_DIR,
        obj_node_mapping_fname: str = DEFAULT_PERSIST_FNAME,
    ) -> None:
        """持久化对象节点映射。

注意：这可能会失败，取决于对象类型是否可被pickle化。
"""
        if not os.path.exists(persist_dir):
            os.makedirs(persist_dir)
        obj_node_mapping_path = concat_dirs(persist_dir, obj_node_mapping_fname)
        try:
            with open(obj_node_mapping_path, "wb") as f:
                pickle.dump(self, f)
        except pickle.PickleError as err:
            raise ValueError("Objs is not pickleable") from err

SQLTableNodeMapping

Bases: BaseObjectNodeMapping[SQLTableSchema]

SQL表节点映射。

Source code in llama_index/core/objects/table_node_mapping.py

class SQLTableNodeMapping(BaseObjectNodeMapping[SQLTableSchema]):
    """SQL表节点映射。"""

    def __init__(self, sql_database: SQLDatabase) -> None:
        self._sql_database = sql_database

    @classmethod
    def from_objects(
        cls,
        objs: Sequence[SQLTableSchema],
        *args: Any,
        sql_database: Optional[SQLDatabase] = None,
        **kwargs: Any,
    ) -> "BaseObjectNodeMapping":
        """初始化节点映射。"""
        if sql_database is None:
            raise ValueError("Must provide sql_database")
        # ignore objs, since we are building from sql_database
        return cls(sql_database)

    def _add_object(self, obj: SQLTableSchema) -> None:
        raise NotImplementedError

    def to_node(self, obj: SQLTableSchema) -> TextNode:
        """到节点。"""
        # taken from existing schema logic
        table_text = (
            f"Schema of table {obj.table_name}:\n"
            f"{self._sql_database.get_single_table_info(obj.table_name)}\n"
        )

        metadata = {"name": obj.table_name}

        if obj.context_str is not None:
            table_text += f"Context of table {obj.table_name}:\n"
            table_text += obj.context_str
            metadata["context"] = obj.context_str

        table_identity = f"{obj.table_name}{obj.context_str}"

        return TextNode(
            id_=str(hash(table_identity)),
            text=table_text,
            metadata=metadata,
            excluded_embed_metadata_keys=["name", "context"],
            excluded_llm_metadata_keys=["name", "context"],
        )

    def _from_node(self, node: BaseNode) -> SQLTableSchema:
        """来自节点。"""
        if node.metadata is None:
            raise ValueError("Metadata must be set")
        return SQLTableSchema(
            table_name=node.metadata["name"], context_str=node.metadata.get("context")
        )

    @property
    def obj_node_mapping(self) -> Dict[int, Any]:
        """节点和对象之间的映射数据结构。"""
        raise NotImplementedError("Subclasses should implement this!")

    def persist(
        self, persist_dir: str = ..., obj_node_mapping_fname: str = ...
    ) -> None:
        """持久化对象。"""
        raise NotImplementedError("Subclasses should implement this!")

    @classmethod
    def from_persist_dir(
        cls,
        persist_dir: str = DEFAULT_PERSIST_DIR,
        obj_node_mapping_fname: str = DEFAULT_PERSIST_FNAME,
    ) -> "SQLTableNodeMapping":
        raise NotImplementedError(
            "This object node mapping does not support persist method."
        )

obj_node_mapping property

obj_node_mapping: Dict[int, Any]

节点和对象之间的映射数据结构。

from_objects classmethod

from_objects(
    objs: Sequence[SQLTableSchema],
    *args: Any,
    sql_database: Optional[SQLDatabase] = None,
    **kwargs: Any
) -> BaseObjectNodeMapping

初始化节点映射。

Source code in llama_index/core/objects/table_node_mapping.py

@classmethod
def from_objects(
    cls,
    objs: Sequence[SQLTableSchema],
    *args: Any,
    sql_database: Optional[SQLDatabase] = None,
    **kwargs: Any,
) -> "BaseObjectNodeMapping":
    """初始化节点映射。"""
    if sql_database is None:
        raise ValueError("Must provide sql_database")
    # ignore objs, since we are building from sql_database
    return cls(sql_database)

to_node

to_node(obj: SQLTableSchema) -> TextNode

到节点。

Source code in llama_index/core/objects/table_node_mapping.py

def to_node(self, obj: SQLTableSchema) -> TextNode:
    """到节点。"""
    # taken from existing schema logic
    table_text = (
        f"Schema of table {obj.table_name}:\n"
        f"{self._sql_database.get_single_table_info(obj.table_name)}\n"
    )

    metadata = {"name": obj.table_name}

    if obj.context_str is not None:
        table_text += f"Context of table {obj.table_name}:\n"
        table_text += obj.context_str
        metadata["context"] = obj.context_str

    table_identity = f"{obj.table_name}{obj.context_str}"

    return TextNode(
        id_=str(hash(table_identity)),
        text=table_text,
        metadata=metadata,
        excluded_embed_metadata_keys=["name", "context"],
        excluded_llm_metadata_keys=["name", "context"],
    )

persist

persist(
    persist_dir: str = ...,
    obj_node_mapping_fname: str = ...,
) -> None

持久化对象。

Source code in llama_index/core/objects/table_node_mapping.py

def persist(
    self, persist_dir: str = ..., obj_node_mapping_fname: str = ...
) -> None:
    """持久化对象。"""
    raise NotImplementedError("Subclasses should implement this!")

SQLTableSchema

Bases: BaseModel

SQL表的轻量级表示。

Source code in llama_index/core/objects/table_node_mapping.py

class SQLTableSchema(BaseModel):
    """SQL表的轻量级表示。"""

    table_name: str
    context_str: Optional[str] = None

SimpleQueryToolNodeMapping

Bases: BaseQueryToolNodeMapping

简单的查询工具映射。

Source code in llama_index/core/objects/tool_node_mapping.py

class SimpleQueryToolNodeMapping(BaseQueryToolNodeMapping):
    """简单的查询工具映射。"""

    def __init__(self, objs: Optional[Sequence[QueryEngineTool]] = None) -> None:
        objs = objs or []
        self._tools = {tool.metadata.name: tool for tool in objs}

    def validate_object(self, obj: QueryEngineTool) -> None:
        if not isinstance(obj, QueryEngineTool):
            raise ValueError(f"Object must be of type {QueryEngineTool}")

    @classmethod
    def from_objects(
        cls, objs: Sequence[QueryEngineTool], *args: Any, **kwargs: Any
    ) -> "BaseObjectNodeMapping":
        return cls(objs)

    def _add_object(self, tool: QueryEngineTool) -> None:
        if tool.metadata.name is None:
            raise ValueError("Tool name must be set")
        self._tools[tool.metadata.name] = tool

    def to_node(self, obj: QueryEngineTool) -> TextNode:
        """到节点。"""
        return convert_tool_to_node(obj)

    def _from_node(self, node: BaseNode) -> QueryEngineTool:
        """来自节点。"""
        if node.metadata is None:
            raise ValueError("Metadata must be set")
        return self._tools[node.metadata["name"]]

to_node

to_node(obj: QueryEngineTool) -> TextNode

到节点。

Source code in llama_index/core/objects/tool_node_mapping.py

def to_node(self, obj: QueryEngineTool) -> TextNode:
    """到节点。"""
    return convert_tool_to_node(obj)

SimpleToolNodeMapping

Bases: BaseToolNodeMapping

简单的工具映射。

在这个设置中，我们假设工具名称是唯一的，并且所有工具的列表都存储在内存中。

Source code in llama_index/core/objects/tool_node_mapping.py

class SimpleToolNodeMapping(BaseToolNodeMapping):
    """简单的工具映射。

在这个设置中，我们假设工具名称是唯一的，并且所有工具的列表都存储在内存中。"""

    def __init__(self, objs: Optional[Sequence[BaseTool]] = None) -> None:
        objs = objs or []
        self._tools = {tool.metadata.name: tool for tool in objs}

    @classmethod
    def from_objects(
        cls, objs: Sequence[BaseTool], *args: Any, **kwargs: Any
    ) -> "BaseObjectNodeMapping":
        return cls(objs)

    def _add_object(self, tool: BaseTool) -> None:
        self._tools[tool.metadata.name] = tool

    def to_node(self, tool: BaseTool) -> TextNode:
        """到节点。"""
        return convert_tool_to_node(tool)

    def _from_node(self, node: BaseNode) -> BaseTool:
        """来自节点。"""
        if node.metadata is None:
            raise ValueError("Metadata must be set")
        return self._tools[node.metadata["name"]]

to_node

to_node(tool: BaseTool) -> TextNode

到节点。

Source code in llama_index/core/objects/tool_node_mapping.py

def to_node(self, tool: BaseTool) -> TextNode:
    """到节点。"""
    return convert_tool_to_node(tool)