企业官网建设流程全解析

• 人工智能
• AI Agent
• Agent 框架
• 大模型
• 工具调用
• RAG
• 提示工程
• 强化学习

本篇文章聚焦 openJiuwen agent-core 检索模块中的Extractor抽象基类（位于openjiuwen.core.retrieval.indexing.processor.extractor包），它定义了索引管线中"从文本块抽取信息（如 RDF 三元组）"的统一接口。通过阅读本文，你将掌握Extractor的类继承关系、extract抽象方法签名与语义、上下游关键数据类型（TextChunk/Triple），并能基于仓库内TripleExtractor、OntologyTripleExtractor两个生产级实现，理解如何编写并接入自己的抽取器，为知识图谱检索与图存储索引提供三元组数据源。

Extractor 在检索索引管线中的定位

openJiuwen agent-core 的检索模块以"文档 → 文本块 → 结构化信息"的流水线方式为向量库与图存储准备数据。在 processor/base.py 中定义了所有处理器的统一抽象Processor，其模块注释明确指出：

Base class for all processors (Parser, Chunker, Extractor).

即检索索引阶段的三类核心处理器——Parser（文档解析）、Chunker（文本切块）、Extractor（信息抽取）——都继承自同一个Processor抽象基类。Extractor正是其中负责"语义结构化"的一环：它把Chunker产生的纯文本块（TextChunk）转换成可供知识图谱存储与图检索使用的三元组（Triple）列表，是openjiuwen.core.retrieval.indexing.processor.extractor包对外暴露的抽象根接口。

Extractor 抽象基类核心 API

类定义与继承关系

依据 base.md 与对应源码 base.py，Extractor的完整定义为：

class Extractor(Processor): """Extractor abstract base class (inherits from Processor, used for extracting triples, etc.)"""

• 继承自Processor：因此所有Extractor子类天然具备process方法契约，可以统一被索引管线调用。
• 抽象基类：本身不能实例化，只定义接口骨架，具体抽取逻辑由子类实现。
• 职责定位：抽取三元组（triples）及其他信息，是知识图谱数据入口的统一抽象。

抽象方法 extract

这是Extractor对外唯一的抽象方法，其签名（来自 base.md 与 base.py）：

@abstractmethod async def extract( self, chunks: List[TextChunk], **kwargs, ) -> List[Triple]:

参数说明：

返回值：

• 类型为List[Triple]，即抽取结果列表，例如[Triple(...), Triple(...)]。
• 语义上，每个Triple表示一条 RDF 风格的主语—谓语—宾语关系，可直接落入图存储。

process 方法的默认实现

作为对Processor抽象方法的落实，Extractor还在基类层面给出了process的默认实现（base.py）：

async def process(self, chunks: List[TextChunk], **kwargs) -> List[Triple]: return await self.extract(chunks, **kwargs)

也就是说，process只是对extract的薄封装。索引管线既可以直接调用extract，也可以通过统一的process入口触发抽取，二者行为等价，保证了Parser/Chunker/Extractor在流水线中的调用方式一致。

上下游关键数据类型

为了正确使用Extractor，需要理解它的输入与输出类型，二者均定义在检索模块的 common 目录下。

输入类型 TextChunk

TextChunk是 Pydantic 数据模型（document.py），字段如下：

同时提供类方法TextChunk.from_document(doc, chunk_text, id_)，可直接从Document生成文本块（未指定id_时自动生成 UUID）。

输出类型 Triple

Triple同样是 Pydantic 数据模型（triple.py），字段如下：

在仓库的具体实现中，metadata通常携带doc_id与chunk_id，用于追溯每条三元组的来源文本块，是图检索与溯源的关键索引信息。

生产级实现一：TripleExtractor（通用 OpenIE 三元组抽取）

TripleExtractor（triple_extractor.py）是Extractor最直接的落地实现，通过 LLM 完成开放信息抽取（OpenIE），并支持可选的三元组二次校验。构造函数参数如下：

TripleExtractor( llm_client: Any, # LLM 客户端实例，需提供 invoke(messages, temperature) 能力 model_name: str, # 模型名称 temperature: float = 0.0, # 采样温度，默认 0.0 保证输出稳定 max_concurrent: int = 50, # 最大并发数，内部通过 asyncio.Semaphore 控制 validate: bool = False, # 是否启用 LLM 三元组校验 **kwargs, )

核心执行流程（extract方法）：

1. 并行抽取：对每个TextChunk创建 asyncio 任务调用 LLM，用asyncio.Semaphore(max_concurrent)限制并发，避免打爆模型服务。
2. 提示词构造：_build_prompt构造 OpenIE 提示词，要求模型返回严格 JSON——顶层对象必须包含named_entities与triples两个键，triples中每一项是恰好三个字符串的数组；提示词内置了 Magic Johnson、Elden Ring 两个完整演示样例（few-shot），并明确要求消解代词、保持实体/谓词与源语言一致、去除重复三元组。
3. 容错解析：_parse_triples先剥离可能的 markdown 代码围栏（```），再用json_repair库修复模型输出的残缺 JSON；结构化校验通过后组装为Triple对象，并写入metadata={"doc_id", "chunk_id"}；无效三元组（非数组、不足三元素、含嵌套结构或 None）被记录并跳过。
4. 可选校验：当validate=True时，调用_validate_internal按来源 chunk 分组，用_build_validation_prompt构造校验提示词，要求模型仅保留"文本直接支持"的三元组，删除依赖外部知识、日期/数字/地点不匹配或并非必然为真的三元组，允许修正谓词措辞——这相当于对抽取结果做一轮基于证据的降噪。

值得注意的工程细节：_gather_results统一处理并发任务的异常，按 chunk 顺序抛出遇到的第一个错误；BaseError原样重抛，普通异常则包装为RETRIEVAL_KB_TRIPLE_EXTRACTION_PROCESS_ERROR，保证调用方获得稳定的错误语义。

生产级实现二：OntologyTripleExtractor（本体约束抽取）

OntologyTripleExtractor（ontology_triple_extractor.py）是面向领域约束的抽取器：允许预先加载本体文件，让 LLM 只在本体定义的类与属性范围内抽取实体与关系。构造函数：

OntologyTripleExtractor( llm_client: Any, model_name: str, ontology_name: str, # 本体名称 ontology_path: str | None = None, # 本体文件路径，仅支持 .nt 或 .ttl constrain_ontology: bool = False, # 是否强制抽取结果遵守本体约束 temperature: float = 0.0, max_concurrent: int = 50, **kwargs, )

与通用抽取器的关键差异：

• 本体加载与校验：_load_ontology使用pyoxigraph的 RDF Store 加载 N-Triples / Turtle 文件，通过 SPARQL 查询收集本体中的rdfs:Class/owl:Class类、rdf:Property/owl:ObjectProperty/owl:DatatypeProperty属性，并预计算subClassOf子类映射。文件后缀不合法抛RETRIEVAL_INDEXING_FORMAT_NOT_SUPPORT，文件不存在抛RETRIEVAL_INDEXING_FILE_NOT_FOUND，加载失败抛RETRIEVAL_KB_ONTOLOGY_INVALID。
• 实体抽取（先实体后关系）：_extract_entities用_build_entity_prompt让模型从文本中识别实体，输出{uri, label, class}三元结构；在constrain_ontology=True且提供本体时，提示词会把允许的类及其subclassof、comment摘要注入上下文，限制模型只能使用这些类。
• 关系抽取与属性过滤：_extract_relations调用_get_valid_properties，基于已抽取实体的 class 与本体中属性的 domain/range 做兼容性推理（_is_compatible检查类精确匹配或存在超类传递闭包关系），动态筛出"当前文本可用的属性集合"注入提示词，实现属性级约束；数据属性（range 含 XMLSchema/Literal）与对象属性分别处理。
• 两阶段异步流水：每个 chunk 内部先抽取实体再抽取关系，多个 chunk 之间以asyncio.Semaphore并发执行，gather时同样按首个异常优先抛出。

可见，OntologyTripleExtractor把"本体文件 + LLM"结合，将抽取结果约束到领域模式之内，适合对图谱模式一致性有强要求的业务场景；而TripleExtractor无需任何前置本体，开箱即用，适合通用知识抽取。

自定义 Extractor 的实战指南

基于上述抽象与实现模式，在 openJiuwen agent-core 中扩展一个自定义抽取器只需三步：

1. 继承Extractor，实现extract(chunks: List[TextChunk], **kwargs) -> List[Triple]；若希望被流水线以process统一调用，可保持基类默认封装不变。
2. 复用Triple与TextChunk：输出时务必为每个Triple携带metadata（建议写入doc_id、chunk_id，参考 triple_extractor.py 的做法），保证图检索可溯源。
3. 对齐并发与容错约定：多 chunk 场景建议使用asyncio.Semaphore限流，并对解析失败统一抛出RETRIEVAL_KB_TRIPLE_EXTRACTION_PROCESS_ERROR（参考_gather_results的错误聚合策略），保持与既有实现一致的错误可观测性。

例如，一个最小实现骨架：

from typing import Any, List from openjiuwen.core.retrieval.common.document import TextChunk from openjiuwen.core.retrieval.common.triple import Triple from openjiuwen.core.retrieval.indexing.processor.extractor.base import Extractor class MyExtractor(Extractor): async def extract(self, chunks: List[TextChunk], **kwargs) -> List[Triple]: triples: List[Triple] = [] for chunk in chunks: # 在此接入你自己的抽取逻辑（规则、模型或混合方案） triples.append( Triple( subject="subject", predicate="predicate", object="object", metadata={"doc_id": chunk.doc_id, "chunk_id": chunk.id_}, ) ) return triples

小结

Extractor抽象基类是 openJiuwen agent-core 检索模块中知识图谱数据生产的统一接口：它向上继承Processor保持流水线调用一致，向下以extract(chunks, **kwargs) -> List[Triple]定义抽取契约。仓库中TripleExtractor与OntologyTripleExtractor分别展示了"通用 OpenIE + 可选校验"与"本体约束 + 两阶段抽取"两种生产级范式，涉及并发限流、JSON 容错解析、领域属性推理等工程细节。理解这一抽象层，即可在 agent-core 的图检索与索引链路中按需接入自己的三元组抽取实现。

• 人工智能
• AI Agent
• Agent 框架
• 大模型
• 工具调用
• RAG
• 提示工程
• 强化学习