perplexity 算子，在计算中文数据集时，都特别大

[nuaabuaa07]

Before Asking 在提问之前

• I have read the README carefully. 我已经仔细阅读了 README 上的操作指引。

• I have pulled the latest code of main branch to run again and the problem still existed. 我已经拉取了主分支上最新的代码，重新运行之后，问题仍不能解决。

Search before asking 先搜索，再提问

• I have searched the Data-Juicer issues and found no similar questions. 我已经在 issue列表 中搜索但是没有发现类似的问题。

Question

perplexity 算子，在处理中文数据集时，得分特别大超出预期，请问是否合理

{"text":"支付完成","bad_type_opdev":"words_num","source_opdev":"badcase","language_opdev":"cn","score":null,"source":null,"dj__stats":{"alnum_ratio":1.0,"char_rep_ratio":0.0,"num_words":1,"perplexity":225308.5,"special_char_ratio":0.0,"text_len":4,"word_rep_ratio":0.0}}

如这个例子，仅仅“支付完成”这个文本，给出了225308.5 复杂度的。请问，这个是什么问题呢

Additional 额外信息

No response

[nuaabuaa07]

配置是，

• perplexity_filter:
  lang:'zh'
  max_ppl:2000

[nuaabuaa07]

发现这个现象出现在1.4.4 版本，使用1.4.3版本就正常了

[HYLcool]

嗨，我们这边试了一下，没法复现出你遇到的问题，1.4.3和1.4.4版本下都能得到合理的结果并且两者是一致的，用dj-process --config xxx.yaml走完整处理流程也是一样的结果。可能还是得你那边再check下会不会是其他原因导致的。如还有其他问题可随时与我们交流~

[nuaabuaa07]

我发现了 perplexity_filter 算子处理中文时“得分特别大”原因了。确实不是版本问题，而是 子进程共用一个模型实例冲突问题导致的。

问题根源

当 np > 1 时，HuggingFace datasets.map() 会使用多个进程并行处理数据。这导致两个问题：

1. 模型共享冲突：多个进程可能同时访问同一个 KenLM 模型实例
2. Fork/Spawn 问题：在 macOS 上，如果使用 fork 方法，子进程会共享父进程的内存空间，但 KenLM 的 C++ 对象可能处于不一致状态。

核心就是，KenLM 模型不支持多进程共用，可能里面kenlm_model.score 方法不是“线程安全”。
修复思路，独立使用。参考代码：

# Some code here has been modified from:
# https://huggingface.co/spaces/huggingface/text-data-filtering
# --------------------------------------------------------

import multiprocess as mp

from data_juicer.utils.constant import Fields, InterVars, StatsKeys
from data_juicer.utils.model_utils import get_model, prepare_model

from ..base_op import OPERATORS, Filter
from ..common import get_words_from_document
from ..op_fusion import INTER_WORDS

OP_NAME = "perplexity_filter"


@OPERATORS.register_module(OP_NAME)
@INTER_WORDS.register_module(OP_NAME)
class PerplexityFilter(Filter):
    """Filter to keep samples with perplexity score in a specified range.

    This operator computes the perplexity of text samples using a Hugging Face tokenizer and
    a KenLM language model. It keeps samples with perplexity scores within the specified
    minimum and maximum values. The perplexity is calculated character-based by default. If
    the perplexity is already computed, it will be reused from the 'perplexity' field in the
    sample's stats. The operator supports batched operations for efficiency."""

    _batched_op = True

    def __init__(self, lang: str = "en", min_ppl: float = 0, max_ppl: float = 1500, *args, **kwargs):
        """
        Initialization method.

        :param lang: Compute perplexity for samples in which language.
        :param min_ppl: The min filter perplexity in this op.
        :param max_ppl: The max filter perplexity in this op.
        :param args: extra args
        :param kwargs: extra args
        """
        super().__init__(*args, **kwargs)
        self.min_ppl = min_ppl
        self.max_ppl = max_ppl
        self.lang = lang
        self.sp_model_key = prepare_model(model_type="sentencepiece", lang=lang)
        self.kl_model_key = prepare_model(model_type="kenlm", lang=lang)

        # Process-local model instances to avoid conflicts in multiprocessing
        self._process_local_models = {}

    def _get_process_local_model(self, model_key):
        """Get or create a process-local model instance.

        This ensures each process has its own model instance, avoiding conflicts
        when using multiprocessing with num_proc > 1.
        """
        current_process = mp.current_process().pid

        if current_process not in self._process_local_models:
            self._process_local_models[current_process] = {}

        if model_key not in self._process_local_models[current_process]:
            self._process_local_models[current_process][model_key] = get_model(model_key)

        return self._process_local_models[current_process][model_key]

    def compute_stats_batched(self, samples, context=False):
        print('enter perplexity compute_stats_batched')
        samples_list = samples[self.text_key]
        samples_stats = samples[Fields.stats]
        words_key = f"{InterVars.words}-{self.sp_model_key}"

        # Use process-local model instance instead of global shared model
        tokenizer = self._get_process_local_model(self.sp_model_key)

        for idx, stat in enumerate(samples_stats):
            # check if it's computed already
            if StatsKeys.perplexity in stat:
                continue
            # tokenization
            if context and words_key in samples[Fields.context][idx]:
                words = samples[Fields.context][idx][words_key]
            else:
                words = get_words_from_document(
                    samples_list[idx], token_func=tokenizer.encode_as_pieces if tokenizer else None
                )
                if context:
                    samples[Fields.context][idx][words_key] = words
            text = " ".join(words)
            # compute perplexity
            logits, length = 0, 0
            # Use process-local model instance
            kenlm_model = self._get_process_local_model(self.kl_model_key)
            for line in text.splitlines():
                logits += kenlm_model.score(line)
                length += len(line.split()) + 1
            ppl = (10.0 ** (-logits / length)) if length != 0 else 0.0
            samples_stats[idx][StatsKeys.perplexity] = round(ppl, 1)

        return samples

    def process_batched(self, samples):
        assert isinstance(samples[Fields.stats], list)
        return map(
            lambda stat: self.get_keep_boolean(stat[StatsKeys.perplexity], self.min_ppl, self.max_ppl),
            samples[Fields.stats],
        )

[HYLcool]

嗨 @nuaabuaa07

感谢你的反馈！如果是因为这个原因的话，还有一个可能可行的简单的方法，即在这个算子中引入UNFORKABLE装饰器，并给这个算子装饰上：

# 导入该装饰器
from ..base_op import UNFORKABLE
...
@UNFORKABLE.register_module(OP_NAME)  # 新增该装饰器
@OPERATORS.register_module(OP_NAME)
@INTER_WORDS.register_module(OP_NAME)
class PerplexityFilter(Filter):
...

这样该算子会强制使用非fork的多进程启动方法（如forkserver或者spawn），你可以尝试一下~