Skip to content

Index

节点解析器接口。

BaseExtractor #

Bases: TransformComponent

元数据提取器。

Source code in llama_index/core/extractors/interface.py 
 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
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
class BaseExtractor(TransformComponent):
    """元数据提取器。"""

    is_text_node_only: bool = True

    show_progress: bool = Field(default=True, description="Whether to show progress.")

    metadata_mode: MetadataMode = Field(
        default=MetadataMode.ALL, description="Metadata mode to use when reading nodes."
    )

    node_text_template: str = Field(
        default=DEFAULT_NODE_TEXT_TEMPLATE,
        description="Template to represent how node text is mixed with metadata text.",
    )
    disable_template_rewrite: bool = Field(
        default=False, description="Disable the node template rewrite."
    )

    in_place: bool = Field(
        default=True, description="Whether to process nodes in place."
    )

    num_workers: int = Field(
        default=4,
        description="Number of workers to use for concurrent async processing.",
    )

    @classmethod
    def from_dict(cls, data: Dict[str, Any], **kwargs: Any) -> Self:  # type: ignore
        if isinstance(kwargs, dict):
            data.update(kwargs)

        data.pop("class_name", None)

        llm_predictor = data.get("llm_predictor", None)
        if llm_predictor:
            from llama_index.core.llm_predictor.loading import load_predictor

            llm_predictor = load_predictor(llm_predictor)
            data["llm_predictor"] = llm_predictor

        llm = data.get("llm", None)
        if llm:
            from llama_index.core.llms.loading import load_llm

            llm = load_llm(llm)
            data["llm"] = llm

        return cls(**data)

    @classmethod
    def class_name(cls) -> str:
        """获取类名。"""
        return "MetadataExtractor"

    @abstractmethod
    async def aextract(self, nodes: Sequence[BaseNode]) -> List[Dict]:
        """提取节点序列的元数据，返回与每个节点对应的元数据字典列表。

Args:
    nodes（Sequence[Document]）：要提取元数据的节点
"""

    def extract(self, nodes: Sequence[BaseNode]) -> List[Dict]:
        """提取节点序列的元数据，返回与每个节点对应的元数据字典列表。

Args:
    nodes（Sequence[Document]）：要提取元数据的节点
"""
        return asyncio_run(self.aextract(nodes))

    async def aprocess_nodes(
        self,
        nodes: List[BaseNode],
        excluded_embed_metadata_keys: Optional[List[str]] = None,
        excluded_llm_metadata_keys: Optional[List[str]] = None,
        **kwargs: Any,
    ) -> List[BaseNode]:
        """后处理从文档解析的节点。

允许链接提取器进行链接。

Args:
    nodes（List[BaseNode]）：需要后处理的节点
    excluded_embed_metadata_keys（Optional[List[str]）：
        要从嵌入元数据中排除的键
    excluded_llm_metadata_keys（Optional[List[str]）：
        要从llm元数据中排除的键
"""
        if self.in_place:
            new_nodes = nodes
        else:
            new_nodes = [deepcopy(node) for node in nodes]

        cur_metadata_list = await self.aextract(new_nodes)
        for idx, node in enumerate(new_nodes):
            node.metadata.update(cur_metadata_list[idx])

        for idx, node in enumerate(new_nodes):
            if excluded_embed_metadata_keys is not None:
                node.excluded_embed_metadata_keys.extend(excluded_embed_metadata_keys)
            if excluded_llm_metadata_keys is not None:
                node.excluded_llm_metadata_keys.extend(excluded_llm_metadata_keys)
            if not self.disable_template_rewrite:
                if isinstance(node, TextNode):
                    cast(TextNode, node).text_template = self.node_text_template

        return new_nodes

    def process_nodes(
        self,
        nodes: List[BaseNode],
        excluded_embed_metadata_keys: Optional[List[str]] = None,
        excluded_llm_metadata_keys: Optional[List[str]] = None,
        **kwargs: Any,
    ) -> List[BaseNode]:
        return asyncio_run(
            self.aprocess_nodes(
                nodes,
                excluded_embed_metadata_keys=excluded_embed_metadata_keys,
                excluded_llm_metadata_keys=excluded_llm_metadata_keys,
                **kwargs,
            )
        )

    def __call__(self, nodes: List[BaseNode], **kwargs: Any) -> List[BaseNode]:
        """后处理从文档中解析出的节点。

允许链接提取器进行链接。

Args:
    nodes（List[BaseNode]）：需要后处理的节点
"""
        return self.process_nodes(nodes, **kwargs)

    async def acall(self, nodes: List[BaseNode], **kwargs: Any) -> List[BaseNode]:
        """后处理从文档中解析出的节点。

允许链接提取器进行链接。

Args:
    nodes（List[BaseNode]）：需要后处理的节点
"""
        return await self.aprocess_nodes(nodes, **kwargs)

class_name classmethod #

class_name() -> str

获取类名。

Source code in llama_index/core/extractors/interface.py 
72
73
74
75
@classmethod
def class_name(cls) -> str:
    """获取类名。"""
    return "MetadataExtractor"

aextract abstractmethod async #

aextract(nodes: Sequence[BaseNode]) -> List[Dict]

提取节点序列的元数据，返回与每个节点对应的元数据字典列表。

Source code in llama_index/core/extractors/interface.py 
77
78
79
80
81
82
83
    @abstractmethod
    async def aextract(self, nodes: Sequence[BaseNode]) -> List[Dict]:
        """提取节点序列的元数据，返回与每个节点对应的元数据字典列表。

Args:
    nodes（Sequence[Document]）：要提取元数据的节点
"""

extract #

extract(nodes: Sequence[BaseNode]) -> List[Dict]

提取节点序列的元数据，返回与每个节点对应的元数据字典列表。

Source code in llama_index/core/extractors/interface.py 
85
86
87
88
89
90
91
    def extract(self, nodes: Sequence[BaseNode]) -> List[Dict]:
        """提取节点序列的元数据，返回与每个节点对应的元数据字典列表。

Args:
    nodes（Sequence[Document]）：要提取元数据的节点
"""
        return asyncio_run(self.aextract(nodes))

aprocess_nodes async #

aprocess_nodes(
    nodes: List[BaseNode],
    excluded_embed_metadata_keys: Optional[
        List[str]
    ] = None,
    excluded_llm_metadata_keys: Optional[List[str]] = None,
    **kwargs: Any
) -> List[BaseNode]

后处理从文档解析的节点。

允许链接提取器进行链接。

Source code in llama_index/core/extractors/interface.py 
 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
    async def aprocess_nodes(
        self,
        nodes: List[BaseNode],
        excluded_embed_metadata_keys: Optional[List[str]] = None,
        excluded_llm_metadata_keys: Optional[List[str]] = None,
        **kwargs: Any,
    ) -> List[BaseNode]:
        """后处理从文档解析的节点。

允许链接提取器进行链接。

Args:
    nodes（List[BaseNode]）：需要后处理的节点
    excluded_embed_metadata_keys（Optional[List[str]）：
        要从嵌入元数据中排除的键
    excluded_llm_metadata_keys（Optional[List[str]）：
        要从llm元数据中排除的键
"""
        if self.in_place:
            new_nodes = nodes
        else:
            new_nodes = [deepcopy(node) for node in nodes]

        cur_metadata_list = await self.aextract(new_nodes)
        for idx, node in enumerate(new_nodes):
            node.metadata.update(cur_metadata_list[idx])

        for idx, node in enumerate(new_nodes):
            if excluded_embed_metadata_keys is not None:
                node.excluded_embed_metadata_keys.extend(excluded_embed_metadata_keys)
            if excluded_llm_metadata_keys is not None:
                node.excluded_llm_metadata_keys.extend(excluded_llm_metadata_keys)
            if not self.disable_template_rewrite:
                if isinstance(node, TextNode):
                    cast(TextNode, node).text_template = self.node_text_template

        return new_nodes

acall async #

acall(
    nodes: List[BaseNode], **kwargs: Any
) -> List[BaseNode]

后处理从文档中解析出的节点。

允许链接提取器进行链接。

Source code in llama_index/core/extractors/interface.py 
157
158
159
160
161
162
163
164
165
    async def acall(self, nodes: List[BaseNode], **kwargs: Any) -> List[BaseNode]:
        """后处理从文档中解析出的节点。

允许链接提取器进行链接。

Args:
    nodes（List[BaseNode]）：需要后处理的节点
"""
        return await self.aprocess_nodes(nodes, **kwargs)