📘 概述

ReasoningQuestionGenerator 是一个推理型问题生成算子，用于基于现有的问题调用大语言模型（LLM）生成新的、相似但多样化的问题。该算子可与多种 Prompt 模板（数学、通用、DIY）和 LLM 服务模块配合使用，实现问题的自动化扩增。

__init__函数

@prompt_restrict(
    MathQuestionSynthesisPrompt,
    GeneralQuestionSynthesisPrompt,
    DiyQuestionSynthesisPrompt
)

@OPERATOR_REGISTRY.register()
class ReasoningQuestionGenerator(OperatorABC):
    def __init__(self,
                num_prompts: int = 1,
                llm_serving: LLMServingABC = None,
                prompt_template = MathQuestionSynthesisPrompt | GeneralQuestionSynthesisPrompt | DiyQuestionSynthesisPrompt | DIYPromptABC
                ):

init参数说明

参数名	类型	默认值	说明
num_prompts	int	1	每个输入问题要生成的新问题的数量，范围为1到5（含）。
llm_serving	LLMServingABC	必需	大语言模型服务实例，用于执行问题的生成。
prompt_template	PromptABC	默认支持MathQuestionSynthesisPrompt, GeneralQuestionSynthesisPrompt, DiyQuestionSynthesisPrompt或者集成DIYPromptABC自定义prompt	提示词模板对象，用于构建问题生成所用的提示词。支持数学、通用与自定义模板。

Prompt模板说明

Prompt 模板名称	主要用途	适用场景	特点说明
MathQuestionSynthesisPrompt	数学问题生成	数学相关问题的扩增	基于数学公式和定理生成新问题，支持单步、多步计算问题。
GeneralQuestionSynthesisPrompt	通用问题生成	通用知识问题的扩增	基于通用知识生成新问题，不依赖特定领域知识。
DiyQuestionSynthesisPrompt	自定义问题生成	自定义问题的扩增	基于用户自定义的问题模板生成新问题。

run函数

def run(self, 
        storage: DataFlowStorage, 
        input_key: str,
        output_synth_or_input_flag: str = "Synth_or_Input"
        )

参数

名称	类型	默认值	说明
storage	DataFlowStorage	必需	数据流存储实例，负责读取与写入数据。
input_key	str	必需	输入列名，对应原始问题字段。
output_synth_or_input_flag	str	"Synth_or_Input"	输出列名，用于标识问题来源。'input'表示原始问题，'synth'表示生成的新问题。

🧠 示例用法

from dataflow.operators.reasoning import ReasoningQuestionGenerator
from dataflow.utils.storage import FileStorage
from dataflow.core import LLMServingABC
from dataflow.serving import APILLMServing_request
from dataflow.prompts.reasoning.math import MathQuestionSynthesisPrompt

class ReasoningQuestionGeneratorTest():
    def __init__(self, llm_serving: LLMServingABC = None):
        
        self.storage = FileStorage(
            first_entry_file_name="example.json",
            cache_path="./cache_local",
            file_name_prefix="dataflow_cache_step",
            cache_type="jsonl",
        )
        
        # use API server as LLM serving
        self.llm_serving = APILLMServing_request(
                    api_url="",
                    model_name="gpt-4o",
                    max_workers=30
        )
        
        self.operator = ReasoningQuestionGenerator(
            llm_serving = self.llm_serving,
            prompt_template = MathQuestionSynthesisPrompt()
        )
        
    def forward(self):
        self.operator.run(
            storage = self.storage.step(),
            input_key = "instruction",
            output_synth_or_input_flag = "Synth_or_Input"
        )

if __name__ == "__main__":
    pl = ReasoningQuestionGeneratorTest()
    pl.forward()

🧾 默认输出格式（Output Format）

字段	类型	说明
[input_key]	str	输入的问题文本，此列将包含原始问题和新生成的问题。
Synth_or_Input	str	标识该行问题是原始输入（'input'）还是模型生成的（'synth'）。

示例输入：

{
"instruction":"法国的首都是哪里？"
}

示例输出（假设 num_prompts=1）：

{
"instruction":"法国的首都是哪里？",
"Synth_or_Input":"input"
}
{
"instruction":"哪个城市是法国的首都？",
"Synth_or_Input":"synth"
}