Data-Juicer 实体关系聚合算子 most_relevant_entities_aggregator 完全指南

本文以 Data-Juicer 开源仓库中的 aggregator 类算子 most_relevant_entities_aggregator 为核心，讲解如何借助语言模型 API，从一批相关文档（如事件描述）中提取并排序与指定实体最相关的实体列表。读者将掌握该算子的全部配置参数、数据流转方式、底层实现原理，以及如何在本地环境中复现官方效果演示与运行单元测试。

算子定位：聚合（Aggregator）类算子

在 Data-Juicer 的算子体系中，算子类型说明 将算子划分为 formatter、mapper、filter、deduplicator、grouper、aggregator、selector、pipeline 等类别。most_relevant_entities_aggregator 属于 aggregator（聚合） 类算子，官方标注的类型与标签为：

• Type 算子类型：aggregator
• Tags 标签：cpu, api

从源码角度看，它注册于 most_relevant_entities_aggregator.py，通过装饰器 @OPERATORS.register_module(OP_NAME) 注册，算子名为 most_relevant_entities_aggregator，类名为 MostRelevantEntitiesAggregator，继承自 base_op.py 中的 Aggregator 基类。

聚合算子通常工作在 Grouper（分组）算子之后：Grouper 先把数据按某种规则聚合成"批次样本"（batched sample），Aggregator 再对该批次内的多条子文档进行归纳、汇总，并把结果写回样本。Aggregator 基类的 run 方法 会保证每个样本都具备 __dj__batch_meta__（批处理元数据）字段，随后调用 process_single 完成聚合，这也是下文 most_relevant_entities_aggregator 输出结果所在字段的来源。

算子功能概述

该算子的核心能力是：从提供的文本中提取并排序与给定实体密切相关的实体。官方文档（中英文双语）给出的行为描述可以概括为以下几点：

1. 使用语言模型 API 识别并排序实体，过滤掉与给定实体类型相同的实体（例如给定实体是"人物"李莲花，则不再输出同为"人物"的实体本身）；
2. 排序后的列表按重要性降序排列（越重要的实体越靠前）；
3. 输入的多条子文档会被聚合后一起传给模型，可设置可选的 token 上限（max_token_num）以控制送入模型的文本量；
4. 模型输出通过正则表达式解析出实体列表；
5. 结果存储在批次元数据的 most_relevant_entities 键下（可通过 output_key 自定义）；
6. API 调用出错时，算子会最多重试指定次数（try_num）；
7. 系统提示（system prompt）、输入模板（input template）、输出模式（output pattern）全部可自定义。

从源码 docstring（most_relevant_entities_aggregator.py）可以逐条印证上述行为，且该类标注了 # TODO: LLM-based inference，表明其推理能力完全由外部语言模型 API 提供，算子本身只负责组装请求与解析输出。

参数配置详解

以下是官方文档提供的参数配置表，结合源码 init 方法 逐项补充了默认值语义与使用要点：

name 参数名	type 类型	default 默认值	desc 说明
api_model	<class 'str'>	'gpt-4o'	API 模型名称，需与所用 API 服务商支持的模型一致，官方演示使用 'qwen3.7-max'。
entity	<class 'str'>	None	给定的目标实体，必填。与 query_entity_type 任一为 None 时，初始化会直接抛出 ValueError（见源码 L94-L95 的校验逻辑）。
query_entity_type	<class 'str'>	None	要查询的相关实体类型，必填。例如给定实体为"李莲花"、类型为"人物"，则算子会从文档中找出与之最相关的人物。
input_key	<class 'str'>	'event_description'	批次样本 meta 字段中读取子文档的键名。默认读取每条子文档 meta 中的 event_description 字段（对应 constant.py 中 MetaKeys.event_description = "event_description"）。
output_key	<class 'str'>	'most_relevant_entities'	聚合结果写入批次元数据的键名，默认 most_relevant_entities（对应 constant.py 中 BatchMetaKeys.most_relevant_entities）。测试用例中可自定义为 'relevant_roles' 等。
max_token_num	typing.Optional[typing.Annotated[int, Gt(gt=0)]]	None	送入模型的全部子文档的 token 总数上限。为 None 时不加限制；设为正整数时，算子按子文档顺序累计 token 数，超过上限即截断（见下文实现原理）。
api_endpoint	typing.Optional[str]	None	API 服务 URL 端点，用于对接兼容 OpenAI 协议的服务（如阿里云 DashScope 的兼容模式）。
response_path	typing.Optional[str]	None	从 API 响应中提取内容的路径，默认指向 'choices.0.message.content'，即 OpenAI 风格响应的消息内容字段。
system_prompt_template	typing.Optional[str]	None	系统提示模板，内部用 format(entity=..., entity_type=...) 填充占位符；不传时使用内置默认模板。
input_template	typing.Optional[str]	None	用户输入模板，同样支持 entity、entity_type、sub_docs 占位符；不传时使用内置默认模板。
output_pattern	typing.Optional[str]	None	解析模型输出的正则表达式；不传时使用内置默认输出模式。
try_num	typing.Annotated[int, Gt(gt=0)]	3	API 调用出错或输出解析失败时的重试次数，最小为 1。
model_params	typing.Dict	{}	初始化 API 模型时的额外参数。
sampling_params	typing.Dict	{}	调用 API 时传入的采样参数，例如 {'temperature': 0.9, 'top_p': 0.95}；官方测试中还使用了 {'enable_thinking': False} 关闭模型思考链。
kwargs		''	透传给基类的其他关键字参数。

必填参数校验

需要特别注意：entity 与 query_entity_type 是必填项。源码在初始化阶段执行了显式校验：

if entity is None or query_entity_type is None:
    raise ValueError("The entity and query_entity_type cannot be None!")

对应单元测试 test_validates_required_fields 验证了 query_entity_type=None 时会抛出 ValueError，配置算子前务必确保两者都已给出。

内置默认模板

若不自定义 system_prompt_template、input_template、output_pattern，算子使用源码中定义的默认值（most_relevant_entities_aggregator.py）：

• 默认系统提示（中文）：要求模型"给定与{entity}相关的一些文档，总结一些与{entity}最为相关的{entity_type}"，并明确三条约束——不包含与给定实体为同一类型的实体、按重要性降序排序（越重要越靠前）、输出格式为"## 分析"（关联度分析）+ "## 列表"（逗号分隔的实体列表）。
• 默认输入模板："{entity}"的相关文档：\n{sub_docs}\n\n与"{entity}"最相关的一些"{entity_type}"：\n。
• 默认输出模式：r"\#\#\s*列表\s*(.*?)\Z"，即匹配"## 列表"标题之后直到文本结尾的内容。

输入输出数据结构与数据流

该算子处理的是经过 Grouper 聚合后的"批次样本"，其输入输出结构以 __dj__meta__（meta 字段）与 __dj__batch_meta__（批处理元数据字段）为核心。process_single（源码 L167-L183）的执行逻辑如下：

1. 幂等检查：若批次元数据中已存在 output_key，直接返回原样本，不重复调用 API；
2. 字段检查：若样本没有 meta 字段，或第一条 meta 中缺少 input_key，则记录 logger.warning("The input key does not exist in the sample!") 并原样返回；
3. 收集子文档：从每条 meta 记录中取出 input_key 对应的文本，构成 sub_docs 列表；
4. 类型检查：使用 is_string_list 确认 sub_docs 是字符串列表（即确为批次样本），否则原样返回；
5. 执行聚合：调用 query_most_relevant_entities(sub_docs, rank=rank)，把结果写入 sample[Fields.batch_meta][self.output_key]。

因此，一个典型的数据流是：

Grouper 算子（按实体分组）→ 批次样本（meta 中含多条 event_description）
→ MostRelevantEntitiesAggregator → 批次元数据 most_relevant_entities 键新增实体列表

效果演示（官方示例）

官方文档给出两个基于电视剧《莲花楼》人物关系语料的演示，均可通过对应测试用例复现。

示例一：test_default_aggregator（默认配置）

构造算子：

MostRelevantEntitiesAggregator(api_model='qwen3.7-max', entity='李莲花', query_entity_type='人物')

输入数据：一个批次样本，其 __dj__meta__ 下包含 5 条 event_description 子文档：

1. "十年前，李相夷十五岁战胜西域天魔成为天下第一高手，十七岁建立四顾门，二十岁问鼎武林盟主，成为传奇人物。"
2. "有人视李相夷为中原武林的希望，但也有人以战胜他为目标，包括魔教金鸳盟盟主笛飞声。笛飞声设计加害李相夷的师兄单孤刀，引得李相夷与之一战。"
3. "在东海的一艘船上，李相夷独自一人对抗金鸳盟的高手，最终击败了大部分敌人。笛飞声突然出现，两人激战，李相夷在战斗中中毒，最终被笛飞声重伤，船只爆炸，李相夷沉入大海。"
4. "十年后，李莲花在一个寒酸的莲花楼内醒来，表现出与李相夷截然不同的性格。他以神医的身份在小镇上行医，但生活贫困。"
5. "小镇上的皮影戏摊讲述李相夷和笛飞声的故事，孩子们争论谁赢了。风火堂管事带着人来找李莲花，要求他救治一个"死人"。"

输出数据：__dj__batch_meta__ 下新增键 most_relevant_entities，值为：

['笛飞声', '单孤刀', '风火堂管事']

解释：这是算子的典型用法。模型从一组事件描述中识别并排列与"李莲花"（李相夷的化名）最相关的人物，输出"笛飞声"、"单孤刀"、"风火堂管事"，且按重要性降序排列（笛飞声是贯穿始终的主要对手，排在首位），结果写入批次元数据的 most_relevant_entities 键。

示例二：test_max_token_num（token 上限）

构造算子：

MostRelevantEntitiesAggregator(api_model='qwen3.7-max', entity='李莲花', query_entity_type='人物', max_token_num=40)

输入数据：与示例一完全相同的 5 条 event_description。

输出数据：most_relevant_entities 的值为：

['西域天魔', '四顾门成员', '武林盟主']

解释：该示例展示了一个重要的边缘情况。max_token_num=40 限制了语言模型 API 可以处理的文本总量，由于输入被截断（只保留累计 token 数不超过 40 的前若干条子文档），模型可依据的信息变少，相关实体的排名列表与默认配置时显著不同——此处输出为"西域天魔"、"四顾门成员"、"武林盟主"。这说明 token 上限会直接影响聚合结果的召回范围，在长文档场景下需要结合实际上下文长度权衡取舍。

底层实现原理：从源码看执行细节

1. 模型调用链

算子在初始化时通过 model_utils.py 的 prepare_model(model_type="api", ...) 注册 API 模型（api_endpoint、response_path、model_params 均在此透传），返回一个模型 key；运行时再通过 get_model 从 MODEL_ZOO 全局缓存中取出模型与 tokenizer。因此：

• api_model 决定使用哪个模型；
• api_endpoint 用于对接非默认的 OpenAI 兼容服务；
• response_path 决定从响应 JSON 的哪条路径取正文（默认 choices.0.message.content）。

2. token 计数与截断逻辑

query_most_relevant_entities（源码 L131-L165）先对每条子文档用 tokenizer 计算 token 数：

token_nums = [len(tokenizer.encode(sub_doc)) for sub_doc in sub_docs]

当 max_token_num 为 None 时全部子文档直接保留；否则按顺序累加 token 数，一旦累计超过上限就停止追加：

if self.max_token_num is None:
    final_docs = sub_docs
else:
    final_docs = []
    total_num = 0
    for token_num, doc in zip(token_nums, sub_docs):
        total_num += token_num
        if total_num > self.max_token_num:
            break
        final_docs.append(doc)

最终所有保留的子文档以 "\n\n" 连接，填充进输入模板后，连同系统提示组装成 OpenAI 风格的 messages 列表发给模型。

3. 输出解析：正则 + 标点切分

模型返回后，parse_output（源码 L120-L129）用 output_pattern 的正则（re.VERBOSE | re.DOTALL）提取"## 列表"之后的正文，随后调用 helper_func.py 中的 split_text_by_punctuation，按中英文标点把实体列表切成单个实体并去空。对应的纯逻辑单元测试 test_parse_output_extracts_entity_list 验证了 "## 列表\nrolea, roleb" 会被解析为 ["rolea", "roleb"]。

4. 重试机制

算子最多执行 try_num 次尝试：每次调用模型并解析输出，只要解析结果非空即中断重试；API 调用异常或解析为空则记录 logger.warning(f"Exception: {e}") 后继续下一次。默认 try_num=3，可提高对不稳定 API 服务的容错性。

本地运行测试与实战配置

仓库在 tests/ops/aggregator/test_most_relevant_entities_aggregator.py 中提供了两类测试：

• 纯逻辑测试（无需外部 API）：MostRelevantEntitiesAggregatorUnitTest 验证输出解析与必填参数校验，可直接运行。
• API 测试（需要真实模型服务）：MostRelevantEntitiesAggregatorTest 覆盖 test_default_aggregator、test_input_output（自定义 input_key='events'、output_key='relevant_roles'）、test_max_token_num 三个场景。测试注释明确要求先配置环境变量：

export OPENAI_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1/
export OPENAI_API_KEY=your_dashscope_key

即通过 OPENAI_BASE_URL 指向 OpenAI 兼容协议的服务端点，OPENAI_API_KEY 提供密钥。该测试类还带有 @skip_if_from_fork 装饰器，从 fork 仓库运行时会被自动跳过（以免占用外部 API 配额）。

参考测试的写法，在自有数据上使用该算子的最小调用方式为：

from data_juicer.ops.aggregator import MostRelevantEntitiesAggregator

op = MostRelevantEntitiesAggregator(
    api_model='qwen3.7-max',
    entity='李莲花',
    query_entity_type='人物',
    sampling_params={'enable_thinking': False},
)
dataset = op.run(dataset)   # dataset 需为经 Grouper 聚合后的批次数据集

小结

most_relevant_entities_aggregator 是 Data-Juicer 中面向"实体关系归纳"场景的开箱即用聚合算子：它把 LLM 的开放抽取能力封装成可配置、可重试、可解析的标准化算子，适合人物关系图谱构建、事件主角识别、领域实体相关性排序等数据加工任务。实际使用时，牢记三点：entity 与 query_entity_type 必填；max_token_num 会显著影响结果；输出键默认落在批次元数据的 most_relevant_entities 字段，可通过 output_key 自由调整。

相关资源

• 算子源代码
• 算子单元测试
• Aggregator 基类实现
• 标点切分工具函数
• 模型初始化与获取工具
• 返回算子列表