解决方案概述
好的,我会从多个方向来解决问题。这里有一些很好的建议,如果我是你,我会使用这些方法的组合(例如,在二元情况下,通过多数表决预测标签,该标签由50%以上分类器同意)。
我考虑以下方法:
这样,三个中的两个将必须同意某个概念是医学概念,从而进一步减小错误的可能性。
在此期间,我反对@ananand_v.singh在this answer中提出的方法,因为:
- 距离度量不应该是欧几里得度量,余弦相似度是更好的度量方式(例如,spaCy使用的方式),因为它不考虑向量的大小(而且不应该,这就是word2vec或GloVe的训练方式)。
- 如果我理解正确,将会创建许多人工聚类,而我们只需要两个:医学和非医学。此外,药物的质心不是集中在药物本身上。这带来了额外的问题,例如质心远离药物,并且其他词语(如
computer
或human
或任何你认为不适合于医学的词语)可能进入聚类。
- 很难评估结果,更何况这个问题是严格主观的。此外,单词向量很难可视化和理解(使用PCA / TSNE /类似方法将它们投射到较低的维度[2D / 3D],对于您的更长数据集,PCA获得约5%的解释方差,真的非常低)。
基于上述问题,我提出了使用
主动学习的解决方案,这是一种相当被遗忘的方法。
主动学习方法中,当我们难以想出一个精确的算法(比如什么是“医疗”类别的术语),我们会向专家(实际上不必是专家)寻求答案。
正如
anand_v.singh所指出的那样,词向量是最有前途的方法之一,我在这里也将使用它(虽然方式不同,但我认为更加清晰易懂)。
不要使用当前可用的最先进的上下文化词嵌入(例如
BERT)。检查你的概念中有多少个没有表示(例如被表示为零向量)。应该进行检查(并且已经在我的代码中进行了检查,在讨论时间到来时会有进一步的讨论),可以使用其中大多数存在的嵌入。
使用
spaCy来测量相似度。
这个类测量药品编码为spaCy的GloVe词向量与其他概念之间的相似度。
class Similarity:
def __init__(self, centroid, nlp, n_threads: int, batch_size: int):
self.centroid = centroid
self.nlp = nlp
self.n_threads: int = n_threads
self.batch_size: int = batch_size
self.missing: typing.List[int] = []
def __call__(self, concepts):
concepts_similarity = []
for i, concept in enumerate(
self.nlp.pipe(
concepts, n_threads=self.n_threads, batch_size=self.batch_size
)
):
if concept.has_vector:
concepts_similarity.append(self.centroid.similarity(concept))
else:
concepts_similarity.append(-1)
self.missing.append(i)
return np.array(concepts_similarity)
这段代码将返回每个概念与质心相似程度的数字,并记录缺失表征的概念的索引。可以这样调用它:
import json
import typing
import numpy as np
import spacy
nlp = spacy.load("en_vectors_web_lg")
centroid = nlp("medicine")
concepts = json.load(open("concepts_new.txt"))
concepts_similarity = Similarity(centroid, nlp, n_threads=-1, batch_size=4096)(
concepts
)
您可以将数据替换为new_concepts.json
。
查看spacy.load并注意我使用了en_vectors_web_lg
。它包含 685,000个唯一的词向量(非常多),并且可能可直接用于您的情况。安装spaCy后,您必须单独下载它,更多信息在上面的链接中提供。
此外,您可能希望使用多个质心词,例如添加像disease
或health
这样的词,并平均它们的词向量。但我不确定这是否会对您的情况产生积极影响。
其他可能性是使用多个质心,并计算每个概念与多个质心之间的相似度。在这种情况下,我们可能有几个阈值,这很可能会消除一些false positives,但可能会错过一些被认为类似于medicine
的术语。此外,它还会使情况变得更加复杂,但如果您的结果不令人满意,则应考虑上述两个选项(仅在没有考虑之前不要轻易采用此方法)。
现在,我们有了概念相似度的粗略测量。但是,某个概念与医学的相似度为0.1是什么意思呢?它是应该归类为医学概念吗?还是说已经太远了?
询问专家
为了得到阈值(低于此阈值将被视为非医学术语),最简单的方法是请人类为我们分类一些概念(这就是主动学习的内容)。是的,我知道这是一种非常简单的主动学习形式,但我仍然认为它属于主动学习范畴。
我编写了一个类,具有类似于sklearn的接口,可以要求人类对概念进行分类,直到达到最优阈值(或最大迭代次数)。
class ActiveLearner:
def __init__(
self,
concepts,
concepts_similarity,
max_steps: int,
samples: int,
step: float = 0.05,
change_multiplier: float = 0.7,
):
sorting_indices = np.argsort(-concepts_similarity)
self.concepts = concepts[sorting_indices]
self.concepts_similarity = concepts_similarity[sorting_indices]
self.max_steps: int = max_steps
self.samples: int = samples
self.step: float = step
self.change_multiplier: float = change_multiplier
self._checked_concepts: typing.Set[int] = set()
self._min_threshold: float = -1
self._max_threshold: float = 1
self.threshold_: float = 1
samples
参数描述了每次迭代中专家将看到多少个示例(这是最大值,如果已经要求样本或者没有足够的样本显示,则会返回较少的数量)。
step
表示阈值下降的程度(我们从1开始,表示完美相似度,在每次迭代中逐步降低)。
change_multiplier
- 如果专家回答概念不相关(或者大多数不相关,因为返回了多个),则通过这个浮点数来乘以步长。它用于在每次迭代中精确定位阈值的确切位置。
- 概念根据它们的相似性排序(相似度越高,排名越靠前)。
以下函数向专家询问意见,并根据他的答案找到最佳阈值。
def _ask_expert(self, available_concepts_indices):
concepts_to_show = set(
np.random.choice(
available_concepts_indices, len(available_concepts_indices)
).tolist()
)
concepts_to_show = concepts_to_show - self._checked_concepts
self._checked_concepts.update(concepts_to_show)
if concepts_to_show:
print("\nAre those concepts related to medicine?\n")
print(
"\n".join(
f"{i}. {concept}"
for i, concept in enumerate(
self.concepts[list(concepts_to_show)[: self.samples]]
)
),
"\n",
)
return input("[y]es / [n]o / [any]quit ")
return "y"
示例问题如下:
Are those concepts related to medicine?
0. anesthetic drug
1. child and adolescent psychiatry
2. tertiary care center
3. sex therapy
4. drug design
5. pain disorder
6. psychiatric rehabilitation
7. combined oral contraceptive
8. family practitioner committee
9. cancer family syndrome
10. social psychology
11. drug sale
12. blood system
[y]es / [n]o / [any]quit y
...解析专家的答案:
def _parse_expert_decision(self, decision) -> bool:
if decision.lower() == "y":
self._max_threshold = self.threshold_
if self.threshold_ - self.step < self._min_threshold:
return False
self.threshold_ -= self.step
return True
if decision.lower() == "n":
self._min_threshold = self.threshold_
self.step *= self.change_multiplier
if self.threshold_ + self.step < self._max_threshold:
return False
self.threshold_ += self.step
return True
return False
最后,整个ActiveLearner
代码,根据专家找到相似性的最佳阈值:
class ActiveLearner:
def __init__(
self,
concepts,
concepts_similarity,
samples: int,
max_steps: int,
step: float = 0.05,
change_multiplier: float = 0.7,
):
sorting_indices = np.argsort(-concepts_similarity)
self.concepts = concepts[sorting_indices]
self.concepts_similarity = concepts_similarity[sorting_indices]
self.samples: int = samples
self.max_steps: int = max_steps
self.step: float = step
self.change_multiplier: float = change_multiplier
self._checked_concepts: typing.Set[int] = set()
self._min_threshold: float = -1
self._max_threshold: float = 1
self.threshold_: float = 1
def _ask_expert(self, available_concepts_indices):
concepts_to_show = set(
np.random.choice(
available_concepts_indices, len(available_concepts_indices)
).tolist()
)
concepts_to_show = concepts_to_show - self._checked_concepts
self._checked_concepts.update(concepts_to_show)
if concepts_to_show:
print("\nAre those concepts related to medicine?\n")
print(
"\n".join(
f"{i}. {concept}"
for i, concept in enumerate(
self.concepts[list(concepts_to_show)[: self.samples]]
)
),
"\n",
)
return input("[y]es / [n]o / [any]quit ")
return "y"
def _parse_expert_decision(self, decision) -> bool:
if decision.lower() == "y":
self._max_threshold = self.threshold_
if self.threshold_ - self.step < self._min_threshold:
return False
self.threshold_ -= self.step
return True
if decision.lower() == "n":
self._min_threshold = self.threshold_
self.step *= self.change_multiplier
if self.threshold_ + self.step < self._max_threshold:
return False
self.threshold_ += self.step
return True
return False
def fit(self):
for _ in range(self.max_steps):
available_concepts_indices = np.nonzero(
self.concepts_similarity >= self.threshold_
)[0]
if available_concepts_indices.size != 0:
decision = self._ask_expert(available_concepts_indices)
if not self._parse_expert_decision(decision):
break
else:
self.threshold_ -= self.step
return self
总的来说,你需要手动回答一些问题,但在我看来,这种方法更加准确。此外,您不必查看所有样本,只需查看其中的一小部分即可。您可以决定多少个样本构成一个医学术语(例如,是否应该仍将40个医学样本和10个非医学样本视为医学?),这使您可以根据自己的喜好进行微调。如果有异常值(比如说,在50个样本中有1个非医学样本),我认为阈值仍然有效。
再次强调:为了最大程度地减少错误分类的可能性,应该将此方法与其他方法混合使用。
分类器
当我们从专家那里获得阈值时,分类将是瞬间完成的。以下是一个简单的分类类:
class Classifier:
def __init__(self, centroid, threshold: float):
self.centroid = centroid
self.threshold: float = threshold
def predict(self, concepts_pipe):
predictions = []
for concept in concepts_pipe:
predictions.append(self.centroid.similarity(concept) > self.threshold)
return predictions
为了简洁起见,这里是最终的源代码:
import json
import typing
import numpy as np
import spacy
class Similarity:
def __init__(self, centroid, nlp, n_threads: int, batch_size: int):
self.centroid = centroid
self.nlp = nlp
self.n_threads: int = n_threads
self.batch_size: int = batch_size
self.missing: typing.List[int] = []
def __call__(self, concepts):
concepts_similarity = []
for i, concept in enumerate(
self.nlp.pipe(
concepts, n_threads=self.n_threads, batch_size=self.batch_size
)
):
if concept.has_vector:
concepts_similarity.append(self.centroid.similarity(concept))
else:
concepts_similarity.append(-1)
self.missing.append(i)
return np.array(concepts_similarity)
class ActiveLearner:
def __init__(
self,
concepts,
concepts_similarity,
samples: int,
max_steps: int,
step: float = 0.05,
change_multiplier: float = 0.7,
):
sorting_indices = np.argsort(-concepts_similarity)
self.concepts = concepts[sorting_indices]
self.concepts_similarity = concepts_similarity[sorting_indices]
self.samples: int = samples
self.max_steps: int = max_steps
self.step: float = step
self.change_multiplier: float = change_multiplier
self._checked_concepts: typing.Set[int] = set()
self._min_threshold: float = -1
self._max_threshold: float = 1
self.threshold_: float = 1
def _ask_expert(self, available_concepts_indices):
concepts_to_show = set(
np.random.choice(
available_concepts_indices, len(available_concepts_indices)
).tolist()
)
concepts_to_show = concepts_to_show - self._checked_concepts
self._checked_concepts.update(concepts_to_show)
if concepts_to_show:
print("\nAre those concepts related to medicine?\n")
print(
"\n".join(
f"{i}. {concept}"
for i, concept in enumerate(
self.concepts[list(concepts_to_show)[: self.samples]]
)
),
"\n",
)
return input("[y]es / [n]o / [any]quit ")
return "y"
def _parse_expert_decision(self, decision) -> bool:
if decision.lower() == "y":
self._max_threshold = self.threshold_
if self.threshold_ - self.step < self._min_threshold:
return False
self.threshold_ -= self.step
return True
if decision.lower() == "n":
self._min_threshold = self.threshold_
self.step *= self.change_multiplier
if self.threshold_ + self.step < self._max_threshold:
return False
self.threshold_ += self.step
return True
return False
def fit(self):
for _ in range(self.max_steps):
available_concepts_indices = np.nonzero(
self.concepts_similarity >= self.threshold_
)[0]
if available_concepts_indices.size != 0:
decision = self._ask_expert(available_concepts_indices)
if not self._parse_expert_decision(decision):
break
else:
self.threshold_ -= self.step
return self
class Classifier:
def __init__(self, centroid, threshold: float):
self.centroid = centroid
self.threshold: float = threshold
def predict(self, concepts_pipe):
predictions = []
for concept in concepts_pipe:
predictions.append(self.centroid.similarity(concept) > self.threshold)
return predictions
if __name__ == "__main__":
nlp = spacy.load("en_vectors_web_lg")
centroid = nlp("medicine")
concepts = json.load(open("concepts_new.txt"))
concepts_similarity = Similarity(centroid, nlp, n_threads=-1, batch_size=4096)(
concepts
)
learner = ActiveLearner(
np.array(concepts), concepts_similarity, samples=20, max_steps=50
).fit()
print(f"Found threshold {learner.threshold_}\n")
classifier = Classifier(centroid, learner.threshold_)
pipe = nlp.pipe(concepts, n_threads=-1, batch_size=4096)
predictions = classifier.predict(pipe)
print(
"\n".join(
f"{concept}: {label}"
for concept, label in zip(concepts[20:40], predictions[20:40])
)
)
回答了一些问题后,使用阈值0.1(介于[-1, 0.1)
之间的所有内容都被认为是非医学的,而[0.1, 1]
则被认为是医学的),我得到了以下结果:
kartagener s syndrome: True
summer season: True
taq: False
atypical neuroleptic: True
anterior cingulate: False
acute respiratory distress syndrome: True
circularity: False
mutase: False
adrenergic blocking drug: True
systematic desensitization: True
the turning point: True
9l: False
pyridazine: False
bisoprolol: False
trq: False
propylhexedrine: False
type 18: True
darpp 32: False
rickettsia conorii: False
sport shoe: True
如您所见,这种方法远非完美,因此最后一节描述了可能的改进:
可能的改进
如开头所提到的,使用我的方法与其他答案混合,可能会忽略像sport shoe
属于medicine
的想法,而主动学习方法在两个上述启发式之间的平局时将更具决定性的投票。
我们也可以创建一个主动学习集合。我们将使用多个阈值(增加或减少),而不是一个阈值,比如说0.1,让我们称其为0.1, 0.2, 0.3, 0.4, 0.5
。
假设 对于每个阈值,sport shoe
都有它相应的True/False
True True False False False
,
通过多数表决,我们将以3票对2票标记它为non-medical
。此外,如果低于它的阈值超过它(True / False
的情况看起来像这样:True True True False False
),那么过于严格的阈值也会得到缓解。
我想到的最后一个可能的改进: 在上面的代码中,我使用了
Doc
向量,它是单词向量的平均值,创建了这个概念。假设有一个单词缺失(向量由零组成),在这种情况下,它将被推离
medicine
质心更远。你可能不希望这样(因为一些小众医学术语[缩写如
gpv
或其他]可能会丢失他们的表现),在这种情况下,你可以只平均那些不同于零的向量。
我知道这篇文章相当冗长,如果你有任何问题,请在下面发表评论。
dbc:
是您提取的维基百科类别。我建议检查这些类别是否具有dbc:Medicine
作为祖先类别。如果超过一半的概念类别具有dbc:Medicine
作为祖先类别,则可以将此概念视为“医学”。 - Stanislav Kralin{m,n}
是 Virtuoso 特定的 SPARQL 1.1 属性路径扩展。您也可以尝试使用“未限定”的skos:broader+
或skos:broader+
。 - Stanislav Kralin" ASK { dbc:Lipid_metabolism_disorders skos:broader+ dbc:Medicine } "
- Stanislav Kralin+
表示一个或多个
,*
表示零个或多个
。{1,7}
表示从一到七跳
,但只有Virtuoso SPARQL终端支持。 - Stanislav Kralin