PaddleNLP Pipelines Answer Extractor 模块全解析：基于 UIE 的无监督问答数据自动构建

• 人工智能
• 大模型
• 预训练
• 微调
• LoRA
• RLHF
• 强化学习
• 分布式训练

本文围绕 PaddleNLP 中slm/pipelines子项目的answer_extractor节点模块展开，系统讲解其在无监督问答（Unsupervised QA）流水线中的定位、四个核心组件（预处理器、答案抽取器、问答对过滤器及其后处理器）的实现原理、参数配置与完整调用方式。读完本文，你将掌握如何基于通用信息抽取（UIE）模型自动从原始文本中生成"合成问答对"，并将其过滤、加工后用于构建 FAQ 检索库或训练数据。

模块定位：无监督问答流水线中的数据生产环节

在slm/pipelines项目中，answer_extractor是 QAGenerationPipeline（QA 生成流水线）的关键组成部分。该流水线用于"无监督问答"场景：不依赖人工标注，而是从一段段原始文本中自动抽取出答案、生成对应问题，最终产出可供检索系统使用的"上下文—问题—答案"三元组。

从 standard_pipelines.py 的源码可以看到流水线的三节点拓扑：

self.pipeline.add_node(component=answer_extractor, name="AnswerExtractor", inputs=["Query"]) self.pipeline.add_node(component=question_generator, name="QuestionGenerator", inputs=["AnswerExtractor"]) self.pipeline.add_node(component=qa_filter, name="QAFilter", inputs=["QuestionGenerator"])

• AnswerExtractor：输入原始段落文本，抽取候选答案（合成答案）；
• QuestionGenerator：基于候选答案生成合成问题（该节点不属于本文模块，位于同目录的 question_generator.md）；
• QAFilter：用独立训练的信息抽取模型过滤低质量的"问题—答案"对。

而answer_extractor节点模块本身（slm/pipelines/pipelines/nodes/answer_extractor/）共包含四个组件，全部继承自BaseComponent，并由init.py 统一导出：

AnswerExtractorPreprocessor：文本转换结果的预处理

AnswerExtractorPreprocessor（源码见 answer_extractor_preprocessor.py）是所有下游抽取工作的入口，其唯一职责是把上游FileConverter等节点产出的标准document字典列表，转换成后续节点需要的纯文本列表。

class AnswerExtractorPreprocessor(BaseComponent): def __init__(self, device="gpu"): paddle.set_device(device) def run(self, documents): results = {"meta": [document["content"] for document in documents]} return results, "output_1"

关键实现细节：

• 构造函数仅接收device（默认"gpu"，也可传"cpu"），其内部通过paddle.set_device(device)完成运行设备的初始化，其余无需任何模型加载，因此该节点非常轻量；
• run(documents)接收节点输入documents（元素为含"content"字段的文档字典），将其逐一取出并组装为{"meta": [段落1, 段落2, ...]}，随后通过return results, "output_1"将结果传递给output_1输出边；
• 与所有 pipeline 节点一致，它定义了outgoing_edges = 1（单输出边）以及return_no_answers、query_count、query_time等通用属性，用于与 Pipeline 框架的调度逻辑对接。

由此可以推断，在完整流水线中，AnswerExtractorPreprocessor承担的是"文本转换 → 段落抽取"之间的数据整形角色：它不改变内容语义，只负责统一下游模型输入的数据格式。

AnswerExtractor：基于 UIE 的合成答案抽取器

AnswerExtractor是本文模块的核心节点（源码见 answer_extractor.py），类注释明确说明其定位为 "Answer Extractor based on Universal Information Extraction"，即基于通用信息抽取（UIE）框架的答案抽取器。

构造参数与默认值

def __init__( self, model="uie-base-answer-extractor", schema=["答案"], task_path=None, device="gpu", batch_size=64, position_prob=0.01, max_answer_candidates=5, ):

资源文件自动下载与校验机制

AnswerExtractor通过resource_files_names与resource_files_urls两张类属性表声明了五类模型资源（model_state.pdparams、model_config.json、vocab.txt、special_tokens_map.json、tokenizer_config.json），并在_check_task_files()中逐一检查：

• 若本地文件缺失，或经md5file（Paddle 内置 MD5 工具）校验与resource_files_urls中记录的 md5 不一致，则调用paddlenlp.taskflow.utils.download_file从bj.bcebos.com下载对应资源；
• 特别地，当model_state发生更新时会置位_param_updated标记；
• 该机制保证了uie-base-answer-extractor等模型在首次使用时的自动就绪，无需用户手工下载。

答案抽取的底层实现

AnswerExtractor在构造时通过Taskflow("information_extraction", ...)创建底层推理引擎（即 PaddleNLP 的信息抽取任务流）。真正的抽取逻辑集中在answer_generation_from_paragraphs()：

def answer_generation_from_paragraphs( self, paragraphs, batch_size=16, model=None, max_answer_candidates=5, schema=None, wf=None ): result = [] buffer = [] i = 0 len_paragraphs = len(paragraphs) for paragraph_tobe in tqdm(paragraphs): buffer.append(paragraph_tobe) if len(buffer) == batch_size or (i + 1) == len_paragraphs: predicts = model(buffer) paragraph_list = buffer buffer = [] for predict_dict, paragraph in zip(predicts, paragraph_list): answers = [] probabilitys = [] for prompt in schema: if prompt in predict_dict: answer_dicts = predict_dict[prompt] answers += [answer_dict["text"] for answer_dict in answer_dicts] probabilitys += [answer_dict["probability"] for answer_dict in answer_dicts] else: answers += [] probabilitys += [] candidates = sorted( list(set([(a, p) for a, p in zip(answers, probabilitys)])), key=lambda x: -x[1] ) if len(candidates) > max_answer_candidates: candidates = candidates[:max_answer_candidates] outdict = { "context": paragraph, "answer_candidates": candidates, } if wf: wf.write(json.dumps(outdict, ensure_ascii=False) + "\n") result.append(outdict) i += 1 return result

该方法的处理流程可归纳为五个步骤：

1. 分批缓冲：按batch_size累积段落，凑满一批或处理到末尾时统一送入model(buffer)推理，兼顾吞吐与显存占用；
2. 按 schema 汇总答案：遍历schema（默认["答案"]）中的每个提示词，从 UIE 输出字典predict_dict中取出"text"与"probability"字段，分别累积到答案列表与概率列表；
3. 去重排序：用set去掉(answer, probability)完全相同的重复项，再按概率降序排序，使置信度最高的答案排在最前；
4. 候选截断：超出max_answer_candidates时仅保留前 N 个候选；
5. 输出结构化结果：每个段落产出一个{"context": 段落原文, "answer_candidates": [(答案, 概率), ...]}字典；若提供了wf文件句柄，还会以 JSON 行（ensure_ascii=False，即保留中文原文）逐行写入文件，便于断点续跑或落盘。

run 方法与节点输出

def run(self, meta): print("creating synthetic answers...") synthetic_context_answer_pairs = self.answer_generation_from_paragraphs( meta, batch_size=self.batch_size, model=self.answer_generator, max_answer_candidates=self.max_answer_candidates, schema=self.schema, wf=None, ) results = {"ca_pairs": synthetic_context_answer_pairs} return results, "output_1"

run(meta)接收预处理节点传来的段落列表，调用上述抽取逻辑，最终将{"ca_pairs": [{"context": ..., "answer_candidates": [(答案, 概率), ...]}, ...]}输出到output_1边，供QuestionGenerator节点消费。注意此处传入的model参数正是构造函数中创建的self.answer_generator（Taskflow 信息抽取引擎）。

QAFilter：合成问答对的质量过滤器

QAFilter（源码见 qa_filter.py）负责把QuestionGenerator生成的"上下文—合成问题—合成答案"三元组做二次校验，剔除答非所问的低质量样本，类注释将其定位为 "Question Answer Pairs Filter based on Universal Information Extraction"。

构造参数与资源机制

def __init__( self, model="uie-base-qa-filter", schema=["答案"], task_path=None, device="gpu", batch_size=64, position_prob=0.1, ):

与AnswerExtractor高度对称：

• 默认模型为"uie-base-qa-filter"，同样支持task_path自定义路径、"uie-base"走 Taskflow 内置模型的两种旁路；
• 默认position_prob=0.1，比抽取器（0.01）更严格，符合"过滤"场景对置信度更高的要求；
• resource_files_urls中维护了uie-base-qa-filter-v1的五类资源文件及其 md5 值，_check_task_files()的下载/校验逻辑与AnswerExtractor完全一致。

过滤逻辑核心：filtration

def filtration(self, paragraphs, batch_size=16, model=None, schema=None, wf=None, wf_debug=None): ... for d in buffer: context = d["context"] synthetic_question = d["synthetic_question"] prefix = "问题：" + synthetic_question + "上下文：" content = prefix + context model_inputs.append(content) predicts = model(model_inputs) ... if synthetic_answer in candidates: # 视为有效对，写入 wf 并追加到 result valid_num += 1 else: # 视为无效对，写入 wf_debug invalid_num += 1

过滤思路非常巧妙：过滤器先把QuestionGenerator合成的"问题"与原始"上下文"拼接为"问题：<question>上下文：<context>"的提示文本，交给 UIE 模型抽取答案；若抽取结果中包含合成的答案文本（synthetic_answer in candidates），说明问题与上下文确实相关、答案确实可从上下文找到，则该三元组判定为有效并保留；否则判定为无效并丢弃。

• 有效结果会通过wf写盘（JSON 行格式），无效结果写入wf_debug，方便后续人工抽查；
• 方法末尾会打印valid synthetic question-answer pairs number与invalid ... number，给出本次过滤的有效/无效统计；
• 有效对保留了context、synthetic_answer、synthetic_answer_probability、synthetic_question、synthetic_question_probability五个字段，完整承载三元组信息。

run 方法：可开关的过滤

def run(self, cqa_triples, is_filter=True): if is_filter: print("filtering synthetic question-answer pairs...") filtered_cqa_triples = self.filtration( cqa_triples, batch_size=self.batch_size, model=self.filtration_model, schema=self.schema ) print("filter synthetic question-answer pairs successfully!") else: filtered_cqa_triples = cqa_triples results = {"filtered_cqa_triples": filtered_cqa_triples} return results, "output_1"

run()通过is_filter开关控制是否执行过滤：为True时走完整过滤流程；为False时原样透传，便于在需要保留全部合成数据的场景下跳过过滤。节点输出键为filtered_cqa_triples——这也正是 run_pipelines_example.py 中从prediction里直接取用的键名。

QAFilterPostprocessor：问答三元组的文档化后处理

QAFilterPostprocessor（源码见 qa_filter_postprocessor.py）是过滤环节的收尾节点，将过滤后的三元组转换为下游文档库（Document Store）可直接写入的标准文档结构：

def run(self, filtered_cqa_triples): results = { "documents": [ { "content": triple["synthetic_question"], "content_type": "text", "meta": {"answer": triple["synthetic_answer"], "_split_id": 0}, } for triple in filtered_cqa_triples ] } return results, "output_1"

其转换规则可以概括为：

• 每个三元组的synthetic_question（合成问题）被写入content，即文档正文——这与 FAQ 检索场景中"用户问的是问题，检索的是问题文本"的语义一致；
• 每个三元组的synthetic_answer（合成答案）被放入meta.answer，作为问题的伴随元数据；
• content_type固定为"text"，_split_id固定为0，用于满足文档库对分片与类型字段的约定。

经过这一步，过滤后的问答数据即可交给FAISSDocumentStore等存储组件建立索引，供语义检索使用。

实战示例：完整无监督问答数据构建流程

slm/applications/question_answering/unsupervised_qa/run_pipelines_example.py提供了本文模块的端到端用法，其中的qa_generation_pipeline()函数给出了标准装配方式：

answer_extractor = AnswerExtractor( model="uie-base-answer-extractor", device=args.device, schema=["答案"], max_answer_candidates=3, position_prob=0.01, batch_size=1, ) question_generator = QuestionGenerator( model="unimo-text-1.0-question-generation", device=args.device, num_return_sequences=2, ) qa_filter = QAFilter( model="uie-base-qa-filter", device=args.device, schema=["答案"], position_prob=0.1, ) pipe = QAGenerationPipeline( answer_extractor=answer_extractor, question_generator=question_generator, qa_filter=qa_filter ) pipeline_params = {"QAFilter": {"is_filter": True}} meta = [ "世界上最早的电影院是美国洛杉矶的“电气剧场”，建于1902年。", "以脸书为例，2020年时，54%的成年人表示，他们从该平台获取新闻。而现在，这个数字下降到了44%。……", ] prediction = pipe.run(meta=meta, params=pipeline_params) prediction = prediction["filtered_cqa_triples"] pprint(prediction)

该示例同时支持两种输入方式：

1. 列表输入：直接把多段原始文本作为meta传入pipe.run()，结果打印到屏幕；
2. 文件输入：通过--source_file指定原始文本文件（每行一段），脚本逐行读入后运行流水线，最终将过滤结果按问题\t答案的格式写入--doc_dir/generated_qa_pairs.txt，可直接作为 FAQ 检索库或问答训练数据使用。

示例中几个参数值得注意：batch_size=1便于小规模演示与调试；max_answer_candidates=3限制每段候选答案数量；QAGenerationPipeline构造时依次传入抽取器、问题生成器与过滤器，内部按AnswerExtractor → QuestionGenerator → QAFilter的顺序组网。

此外，qa_generation_pipeline()之后紧跟着dense_faq_pipeline()，后者用 FAISS 构建 FAQ 向量索引并完成检索——两者组合即构成完整的"无监督 FAQ 数据生产 + 语义检索"闭环。

设计要点总结

• 两级 UIE 模型分工：AnswerExtractor（uie-base-answer-extractor）负责从段落中抽取候选答案，QAFilter（uie-base-qa-filter）负责校验合成问答对的质量，二者都基于 PaddleNLP 的Taskflow("information_extraction", ...)引擎，通过schema指定抽取目标、position_prob控制置信阈值；
• 全程无人工标注：从原始文本到可用问答对，所有环节均由模型自动完成，这正是"无监督问答"的核心价值；
• 资源自动管理：两个模型节点均内置了五类资源文件的 md5 校验与自动下载逻辑，首次使用即自动就绪，无需手工准备模型文件；
• 标准节点协议：四个组件均继承BaseComponent，通过run()返回(results, "output_1")二元组与outgoing_edges = 1定义单输出边，可无缝嵌入Pipeline/QAGenerationPipeline等标准流水线框架；
• 数据流清晰可追踪：模块内数据的流转路径为documents → meta 段落列表 → ca_pairs（上下文+候选答案）→ cqa_triples（问题+答案+上下文）→ filtered_cqa_triples → documents（问题为 content、答案为 meta），每一环的输出键名都是后续环节的输入键名，便于逐节点调试。

如需深入理解相邻节点，可继续阅读同目录下的 question_generator.md（问题生成器）与 retriever.md（检索器）等文档，或直接在slm/pipelines/tests/中查找对应测试用例验证各节点的输入输出契约。

• 人工智能
• 大模型
• 预训练
• 微调
• LoRA
• RLHF
• 强化学习
• 分布式训练