Migrate 4 operators written by @yunwenkai2003 from DataFlow421 (OpenDCAI#202)

Author: yuwenkai2003

dataflow/operators/generate/AgenticRAG/auto_prompt_generator.py

@@ -1 +1,79 @@
 from dataflow.prompts.agenticrag import AutoPromptGeneratorPrompt
+import pandas as pd
+from dataflow.utils.registry import OPERATOR_REGISTRY
+from dataflow import get_logger
+
+from dataflow.utils.storage import DataFlowStorage  
+from dataflow.core import OperatorABC
+from dataflow.core import LLMServingABC
+
+@OPERATOR_REGISTRY.register()
+class AutoPromptGenerator(OperatorABC):
+    '''
+    AutoPromptGenerator is a class that generates prompts for given document fragments to generate seed QA pairs.
+    '''
+    def __init__(self, llm_serving: LLMServingABC):
+        self.logger = get_logger()
+        self.prompts = AutoPromptGeneratorPrompt()    
+        self.llm_serving = llm_serving
+    
+    @staticmethod
+    def get_desc(lang: str = "zh"):
+        if lang == "zh":
+            return (
+                "该算子用于为给的的文档片段生成提示词，用于生成种子QA对\n\n"
+                "输入参数：\n"
+                "- input_key: 包含文档片段的字段名\n"
+                "- output_key: 包含提示词的字段名\n"
+            )
+        elif lang == "en":
+            return (
+                "This operator generates prompts for given document fragments to generate seed QA pairs.\n\n"
+                "Input Parameters:\n"
+                "- input_key: Field name containing the content\n"
+                "- output_key: Field name containing the generated prompt\n"
+            )
+        else:
+            return "AutoPromptGenerator generates prompts for given document fragments to generate seed QA pairs."
+
+    def _validate_dataframe(self, dataframe: pd.DataFrame):
+        required_keys = [self.input_key]
+        forbidden_keys = [self.output_key]
+
+        missing = [k for k in required_keys if k not in dataframe.columns]
+        conflict = [k for k in forbidden_keys if k in dataframe.columns]
+
+        if missing:
+            raise ValueError(f"Missing required column(s): {missing}")
+        if conflict:
+            raise ValueError(f"The following column(s) already exist and would be overwritten: {conflict}")
+
+    def _reformat_prompt(self, dataframe):
+        """
+        Reformat the prompts in the dataframe to generate questions.
+        """
+        questions = dataframe[self.input_key].tolist()
+        inputs = [self.prompts.auto_prompt_generator_prompt(question) for question in questions]
+
+        return inputs
+
+    def run(
+        self, 
+        storage: DataFlowStorage, 
+        input_key:str = "text", 
+        output_key:str = "generated_prompt"
+        ):
+        '''
+        Runs the answer generation process, reading from the input file and saving results to output.
+        '''
+        self.input_key, self.output_key = input_key, output_key
+        dataframe = storage.read("dataframe")
+        self._validate_dataframe(dataframe)
+        formatted_prompts = self._reformat_prompt(dataframe)
+        answers = self.llm_serving.generate_from_input(user_inputs=formatted_prompts, system_prompt="")
+
+        dataframe[self.output_key] = answers
+        output_file = storage.write(dataframe)
+        self.logger.info(f"Results saved to {output_file}")
+
+        return [output_key]

dataflow/operators/generate/AgenticRAG/qa_generator.py

@@ -1 +1,93 @@
 import pandas as pd
+from dataflow.utils.registry import OPERATOR_REGISTRY
+from dataflow import get_logger
+
+from dataflow.utils.storage import DataFlowStorage
+from dataflow.core import OperatorABC
+from dataflow.core import LLMServingABC
+
+@OPERATOR_REGISTRY.register()
+class QAGenerator:
+    '''
+    SeedQAGenerator is a class that uses LLMs to generate QA pairs based on seed input.
+    '''
+
+    def __init__(self, llm_serving: LLMServingABC):
+        self.logger = get_logger()
+        self.llm_serving = llm_serving
+    
+    @staticmethod
+    def get_desc(lang: str = "zh"):
+        if lang == "zh":
+            return (
+                "该算子用于生成对应文档片段的QA对。\n\n"
+                "输入参数：\n"
+                "- input_key: 包含文档片段的字段名\n"
+                "- prompt_key: 包含提示词的字段名\n"
+                "- output_quesion_key: 包含生成问题的字段名\n"
+                "- output_answer_key: 包含生成答案的字段名\n"
+            )
+        elif lang == "en":
+            return (
+                "This operator generates QA pairs for given document fragments.\n\n"
+                "Input Parameters:\n"
+                "- input_key: Field name containing the content\n"
+                "- prompt_key: Field name containing the generated prompt\n"
+                "- output_quesion_key: Field name containing the generated question\n"
+                "- output_answer_key: Field name containing the generated answer\n"
+            )
+        else:
+            return "QAGenerator generates QA pairs for given document fragments."
+
+    def _validate_dataframe(self, dataframe: pd.DataFrame):
+        required_keys = [self.input_key]
+        forbidden_keys = [self.output_question_key, self.output_answer_key]
+
+        missing = [k for k in required_keys if k not in dataframe.columns]
+        conflict = [k for k in forbidden_keys if k in dataframe.columns]
+
+        if missing:
+            raise ValueError(f"Missing required column(s): {missing}")
+        if conflict:
+            raise ValueError(f"The following column(s) already exist and would be overwritten: {conflict}")
+
+    def _build_prompt(self, df):
+        prompts = []
+        for index, row in df.iterrows():
+            prompts.append(row[self.prompt_key] + "Format:\nQ: ...\nA: ..." + "\nSeed data:\n" + row[self.input_key])
+        return prompts
+
+    def _parse_qa(self, response: str) -> tuple:
+        lines = response.strip().split('\n')
+        q = next((line[2:].strip() for line in lines if line.lower().startswith("q:")), "")
+        a = next((line[2:].strip() for line in lines if line.lower().startswith("a:")), "")
+        return q, a
+
+    def run(
+        self, 
+        storage: DataFlowStorage, 
+        input_key:str = "text", 
+        output_prompt_key:str = "generated_prompt",
+        output_quesion_key:str = "generated_question",
+        output_answer_key:str = "generated_answer"
+        ):
+        '''
+        Runs the answer generation process, reading from the input file and saving results to output.
+        '''
+
+        self.input_key, self.prompt_key, self.output_question_key, self.output_answer_key = input_key, output_prompt_key, output_quesion_key, output_answer_key
+
+        dataframe = storage.read("dataframe")
+        self._validate_dataframe(dataframe)
+        formatted_prompts = self._build_prompt(dataframe)
+        responses = self.llm_serving.generate_from_input(user_inputs=formatted_prompts, system_prompt="")
+
+        questions, answers = zip(*[self._parse_qa(r) for r in responses])
+
+        dataframe[self.output_question_key] = questions
+        dataframe[self.output_answer_key] = answers
+
+        output_file = storage.write(dataframe)
+        self.logger.info(f"Results saved to {output_file}")
+
+        return [self.output_question_key, self.output_answer_key]

dataflow/operators/generate/AgenticRAG/qa_scorer.py

@@ -1 +1,155 @@
 from dataflow.prompts.agenticrag import QAScorerPrompt
+import pandas as pd
+from dataflow.utils.registry import OPERATOR_REGISTRY
+from dataflow import get_logger
+import re
+from dataflow.utils.storage import DataFlowStorage
+from dataflow.core import OperatorABC
+from dataflow.core import LLMServingABC
+
+@OPERATOR_REGISTRY.register()
+class QAScorer(OperatorABC):
+    '''
+    Answer Generator is a class that generates answers for given questions.
+    '''
+    def __init__(self, llm_serving: LLMServingABC):
+        self.logger = get_logger()
+        self.prompts = QAScorerPrompt()    
+        self.llm_serving = llm_serving
+    
+    @staticmethod
+    def get_desc(lang: str = "zh"):
+        if lang == "zh":
+            return (
+                "该算子用于为给的的文档片段生成种子QA对打分\n\n"
+                "输入参数：\n"
+                "- input_question_key: Field name containing the generated question\n"
+                "- input_answer_key: Field name containing the generated answer\n"
+                "- output_question_quality_key: Field name containing the question quality grade\n"
+                "- output_question_quality_feedback_key: Field name containing the question quality feedback\n"
+                "- output_answer_alignment_key: Field name containing the answer alignment grade\n"
+                "- output_answer_alignment_feedback_key: Field name containing the answer alignment feedback\n"
+                "- output_answer_verifiability_key: Field name containing the answer verifiability grade\n"
+                "- output_downstream_value_key: Field name containing the downstream value grade\n"
+                "- output_downstream_value_feedback_key: Field name containing the downstream value feedback\n"
+            )
+        elif lang == "en":
+            return (
+                "This operator generates prompts for given document fragments to generate seed QA pairs.\n\n"
+                "Input Parameters:\n"
+                "- input_question_key: Field name containing the generated question\n"
+                "- input_answer_key: Field name containing the generated answer\n"
+                "- output_question_quality_key: Field name containing the question quality grade\n"
+                "- output_question_quality_feedback_key: Field name containing the question quality feedback\n"
+                "- output_answer_alignment_key: Field name containing the answer alignment grade\n"
+                "- output_answer_alignment_feedback_key: Field name containing the answer alignment feedback\n"
+                "- output_answer_verifiability_key: Field name containing the answer verifiability grade\n"
+                "- output_downstream_value_key: Field name containing the downstream value grade\n"
+                "- output_downstream_value_feedback_key: Field name containing the downstream value feedback\n"
+            )
+        else:
+            return "QAScorer scores QA pairs for given document fragments."
+
+    def _validate_dataframe(self, dataframe: pd.DataFrame):
+        required_keys = [self.input_question_key, self.input_answer_key]
+        forbidden_keys = [self.output_question_quality_key, self.output_question_quality_feedback_key, self.output_answer_alignment_key, self.output_answer_alignment_feedback_key, self.output_answer_verifiability_key, self.output_answer_verifiability_feedback_key, self.output_downstream_value_key, self.output_downstream_value_feedback_key]
+
+        missing = [k for k in required_keys if k not in dataframe.columns]
+        conflict = [k for k in forbidden_keys if k in dataframe.columns]
+
+        if missing:
+            raise ValueError(f"Missing required column(s): {missing}")
+        if conflict:
+            raise ValueError(f"The following column(s) already exist and would be overwritten: {conflict}")
+
+    def _build_prompts(self, dataframe):
+        """
+        Reformat the prompts in the dataframe to generate questions.
+        """
+        question_quality_inputs = []
+        question_quality_prompt = self.prompts.question_quality_prompt()
+        answer_alignment_inputs = []
+        answer_alignment_prompt = self.prompts.answer_alignment_prompt()
+        answer_verifiability_inputs = []
+        answer_verifiability_prompt = self.prompts.answer_verifiability_prompt()
+        downstream_value_inputs = []
+        downstream_value_prompt = self.prompts.downstream_value_prompt()
+
+        for index, row in dataframe.iterrows():
+            question_quality_content = question_quality_prompt + "Question: " + row[self.input_question_key] + "\n" + "Answer: " + row[self.input_answer_key]
+            question_quality_inputs.append(question_quality_content)
+            answer_alignment_content = answer_alignment_prompt + "Question: " + row[self.input_question_key] + "\n" + "Answer: " + row[self.input_answer_key]
+            answer_alignment_inputs.append(answer_alignment_content)
+            answer_verifiability_content = answer_verifiability_prompt + "Question: " + row[self.input_question_key] + "\n" + "Answer: " + row[self.input_answer_key]
+            answer_verifiability_inputs.append(answer_verifiability_content)
+            downstream_value_content = downstream_value_prompt + "Question: " + row[self.input_question_key] + "\n" + "Answer: " + row[self.input_answer_key]
+            downstream_value_inputs.append(downstream_value_content)
+
+        return question_quality_inputs, answer_alignment_inputs, answer_verifiability_inputs, downstream_value_inputs
+
+    def _parse_grade_and_feedback(self, response: str) -> tuple:
+        grading_match = re.search(r"\*\*Grading\*\*:\s*(\d+)", response)
+        feedback_match = re.search(r"\*\*Feedback\*\*:\s*(.+)", response, re.DOTALL)
+        grading = float(grading_match.group(1)) if grading_match else 0
+        feedback = feedback_match.group(1).strip() if feedback_match else ''
+
+        return grading, feedback
+
+    def run(
+        self,
+        storage: DataFlowStorage,
+        input_question_key: str = "generated_question",
+        input_answer_key: str = "generated_answer",
+        output_question_quality_key: str = "question_quality_grades",
+        output_question_quality_feedback_key: str = "question_quality_feedbacks",
+        output_answer_alignment_key: str = "answer_alignment_grades",
+        output_answer_alignment_feedback_key: str = "answer_alignment_feedbacks",
+        output_answer_verifiability_key: str = "answer_verifiability_grades",
+        output_answer_verifiability_feedback_key: str = "answer_verifiability_feedbacks",
+        output_downstream_value_key: str = "downstream_value_grades",
+        output_downstream_value_feedback_key: str = "downstream_value_feedbacks"
+    ):
+        self.input_question_key, self.input_answer_key, self.output_question_quality_key, self.output_question_quality_feedback_key, self.output_answer_alignment_key, self.output_answer_alignment_feedback_key, self.output_answer_verifiability_key, self.output_answer_verifiability_feedback_key, self.output_downstream_value_key, self.output_downstream_value_feedback_key = input_question_key, input_answer_key, output_question_quality_key, output_question_quality_feedback_key, output_answer_alignment_key, output_answer_alignment_feedback_key, output_answer_verifiability_key, output_answer_verifiability_feedback_key, output_downstream_value_key, output_downstream_value_feedback_key
+
+        dataframe = storage.read("dataframe")
+        self._validate_dataframe(dataframe)
+
+        # 构建prompt
+        q_inputs, a_inputs, v_inputs, d_inputs = self._build_prompts(dataframe)
+
+        # 生成四类分数和反馈
+        self.logger.info("Scoring question quality...")
+        q_scores = self.llm_serving.generate_from_input(user_inputs=q_inputs, system_prompt="")
+        q_grades, q_feedbacks = zip(*[self._parse_grade_and_feedback(r) for r in q_scores])
+
+        self.logger.info("Scoring answer alignment...")
+        a_scores = self.llm_serving.generate_from_input(user_inputs=a_inputs, system_prompt="")
+        a_grades, a_feedbacks = zip(*[self._parse_grade_and_feedback(r) for r in a_scores])
+
+        self.logger.info("Scoring answer verifiability...")
+        v_scores = self.llm_serving.generate_from_input(user_inputs=v_inputs, system_prompt="")
+        v_grades, v_feedbacks = zip(*[self._parse_grade_and_feedback(r) for r in v_scores])
+
+        self.logger.info("Scoring downstream value...")
+        d_scores = self.llm_serving.generate_from_input(user_inputs=d_inputs, system_prompt="")
+        d_grades, d_feedbacks = zip(*[self._parse_grade_and_feedback(r) for r in d_scores])
+
+        # 写回结果
+        dataframe[self.output_question_quality_key] = q_grades
+        dataframe[self.output_question_quality_feedback_key] = q_feedbacks
+        dataframe[self.output_answer_alignment_key] = a_grades
+        dataframe[self.output_answer_alignment_feedback_key] = a_feedbacks
+        dataframe[self.output_answer_verifiability_key] = v_grades
+        dataframe[self.output_answer_verifiability_feedback_key] = v_feedbacks
+        dataframe[self.output_downstream_value_key] = d_grades
+        dataframe[self.output_downstream_value_feedback_key] = d_feedbacks
+
+        output_file = storage.write(dataframe)
+        self.logger.info(f"Results saved to {output_file}")
+
+        return [
+            output_question_quality_key, output_question_quality_feedback_key,
+            output_answer_alignment_key, output_answer_alignment_feedback_key,
+            output_answer_verifiability_key, output_answer_verifiability_feedback_key,
+            output_downstream_value_key, output_downstream_value_feedback_key
+        ]