如何基于Pandas DataFrame批量生成Snorkel标注函数并收集至列表

核心问题解析 #

循环生成标注函数时容易触发变量延迟绑定问题，导致所有生成的函数最终都使用循环最后一轮的词汇参数，这是你之前尝试失败的核心原因。下面给出两种可行的解决方式，均能批量生成独立的标注函数并存入列表。

代码实现 #

1. 前置准备 #

先导入依赖并定义标签常量：

import pandas as pd
import re
from functools import partial
from snorkel.labeling import LabelingFunction, PandasLFApplier

# 定义标签：ABSTAIN为Snorkel默认无标注值，FOERD为自定义匹配标签
ABSTAIN = 0
FOERD = 1

2. 闭包方式生成标注函数 #

通过嵌套函数捕获每一行的词汇对，确保每个标注函数绑定独立的参数：

def build_lf(word1, word2):
    def labeling_func(x):
        # 转义正则特殊字符，匹配全词且忽略大小写
        pattern1 = re.compile(rf'\b{re.escape(word1)}\b', re.IGNORECASE)
        pattern2 = re.compile(rf'\b{re.escape(word2)}\b', re.IGNORECASE)
        if pattern1.search(x.text) and pattern2.search(x.text):
            return FOERD
        return ABSTAIN
    return labeling_func

# 批量生成并收集标注函数
lf_collection = []
for _, row in df_wordlist.iterrows():
    # 提取当前行的词汇和函数名
    w1, w2, func_name = row['Column_1'], row['Column_2'], row['Column_3']
    # 创建标注函数并设置名称（方便后续调试）
    lf = build_lf(w1, w2)
    lf.__name__ = func_name
    lf_collection.append(lf)

3. functools.partial替代方案 #

如果更偏好偏函数的实现方式，可以用以下代码：

def check_text_match(x, word1, word2):
    pattern1 = re.compile(rf'\b{re.escape(word1)}\b', re.IGNORECASE)
    pattern2 = re.compile(rf'\b{re.escape(word2)}\b', re.IGNORECASE)
    return FOERD if pattern1.search(x.text) and pattern2.search(x.text) else ABSTAIN

lf_collection = []
for _, row in df_wordlist.iterrows():
    w1, w2, func_name = row['Column_1'], row['Column_2'], row['Column_3']
    # 固定word1和word2参数，生成标注函数
    lf = partial(check_text_match, word1=w1, word2=w2)
    lf.__name__ = func_name
    lf_collection.append(lf)

4. 标注函数的使用 #

生成后的列表可直接用于Snorkel的标注流程：

applier = PandasLFApplier(lfs=lf_collection)
label_matrix = applier.apply(df=df)

关键注意事项 #

• 用re.escape()处理词汇中的特殊字符（如., *等正则元字符），避免匹配逻辑出错。
• 正则中的\b用于匹配单词边界，防止误匹配包含目标词汇的更长单词（比如避免用"free"匹配"freedom"）。
• 为每个标注函数设置__name__，方便后续查看标注统计和调试。

内容的提问来源于stack exchange，提问作者Xuefei X