Merge pull request #12 from scuuy/main

haolpku · web-flow · commit 88d8bede2836 · 2025-04-07T21:16:56.000+08:00
add reasoners module in process, do necessary registry and add some demos.
diff --git a/.gitignore b/.gitignore
@@ -22,3 +22,4 @@ configs/process/experiments/
 **/data
 **/ckpt
 tmp.*
+configs/process/text_process_reasoner.yaml
diff --git a/configs/process/text_process.yaml b/configs/process/text_process.yaml
@@ -299,4 +299,12 @@ processors:
       device: 'cuda:0'
       model_name: 'hkust-nlp/deita-quality-scorer'
       max_length: 512
-  
+  MathProblemFilter:
+    min_score: 0
+    max_score: 1
+    api_args:
+      api_url: 'The URL of api, default using general method'
+      api_key: 'Your Key'
+      model_name: 'gpt-4o'
+      mode_test: True
+
diff --git a/configs/process/text_process_reasoner_ansfilter.yaml b/configs/process/text_process_reasoner_ansfilter.yaml
@@ -9,7 +9,7 @@ data:
     dataset_split: 'train'  
     name: 'default' 
     revision: null
-    data_path: 'demos/reasoning_process/math_5_samples.json'  # Local data path, supports json, jsonl, parquet formats
+    data_path: 'demos/text_process/reasoners/math_5_samples.json'  # Local data path, supports json, jsonl, parquet formats
     formatter: "TextFormatter" # Data loader type
     keys: 'answer' # Key name to be processed, for sft data, it can be specified as ['instruction','input','output']
 
diff --git a/dataflow/core/__init__.py b/dataflow/core/__init__.py
@@ -2,6 +2,7 @@
 from .process.filter import Filter, ImageFilter, ImageTextFilter, VideoFilter, TextFilter, VideoTextFilter
 from .process.refiner import Refiner, TextRefiner
 from .process.deduplicator import Deduplicator, TextDeduplicator, ImageDeduplicator
+from .process.reasoner import ReasonerFilter
 
 __all__  = [
     'Scorer',
@@ -20,5 +21,6 @@
     'Refiner',
     'TextRefiner',
     'Deduplicator',
-    'TextDeduplicator'
+    'TextDeduplicator',
+    'ReasonerFilter'
 ]
diff --git a/dataflow/core/process/reasoner.py b/dataflow/core/process/reasoner.py
@@ -0,0 +1,45 @@
+from dataflow.data import DataFlowDataset
+from dataflow.core import ScoreRecord
+from datasets import Dataset
+
+class Reasoner():
+    def __init__(self, args=None):
+        pass
+        
+    def reason_func(self, dataset):
+        pass
+    
+    def __call__(self, dataset: DataFlowDataset):
+        pass
+
+class ReasonerFilter(Reasoner):
+    def __init__(self, args=None):
+        super().__init__()
+        self.data_type = "text"
+        self.filter_name = "ReasonerFilter"
+        self.args = args
+        
+        api_args = args['api_args']
+        self.model_name = api_args['model_name']
+        self.api_url = api_args['api_url']
+        self.mode_test = api_args['mode_test']
+    def filter_func(self, dataset):
+        pass
+
+    def __call__(self, dataset: DataFlowDataset):
+        """Processes the dataset using the reasoner"""
+        init_len = len(dataset)
+        score_record = ScoreRecord()
+        dataset.set_score_record(score_record)
+        labels = self.filter_func(dataset)
+        
+        if isinstance(dataset.dataset, Dataset):
+            def filter_by_labels(example, index):
+                return labels[index] == 1
+            dataset.dataset = dataset.dataset.filter(filter_by_labels, with_indices=True)
+            filtered_dataset = dataset
+        else:
+            filtered_dataset = dataset.filter(labels)
+
+        print(f'Implemented {self.filter_name}. Data Number: {init_len} -> {len(filtered_dataset)}', flush=True)
+        return filtered_dataset
diff --git a/dataflow/process/text/__init__.py b/dataflow/process/text/__init__.py
@@ -1,4 +1,4 @@
 from .filters import *
 from .refiners import *
 from .deduplicators import *
-from .reasoning import *
+from .reasoners import *
diff --git a/dataflow/process/text/reasoners/__init__.py b/dataflow/process/text/reasoners/__init__.py
@@ -0,0 +1,12 @@
+import sys
+from dataflow.utils.registry import LazyLoader
+
+_import_structure = {
+    "MathProblemFilter": ("dataflow/process/text/reasoners/math_problem_filter.py", "MathProblemFilter"),
+    "AnswerGroundTruthFilter": ("dataflow/process/text/reasoners/answer_ground_truth_filter.py", "AnswerGroundTruthFilter"),
+    "AnswerFormatterFilter": ("dataflow/process/text/reasoners/answer_formatter_filter.py", "AnswerFormatterFilter"),
+    "AnswerNgramFilter": ("dataflow/process/text/reasoners/answer_ngram_filter.py", "AnswerNgramFilter"),
+    "AnswerTokenLengthFilter": ("dataflow/process/text/reasoners/answer_token_length_filter.py", "AnswerTokenLengthFilter"),
+}
+
+sys.modules[__name__] = LazyLoader(__name__, "dataflow/process/text/reasoners", _import_structure)
diff --git a/dataflow/process/text/reasoners/answer_formatter_filter.py b/dataflow/process/text/reasoners/answer_formatter_filter.py
diff --git a/dataflow/process/text/reasoners/answer_ground_truth_filter.py b/dataflow/process/text/reasoners/answer_ground_truth_filter.py
diff --git a/dataflow/process/text/reasoners/answer_ngram_filter.py b/dataflow/process/text/reasoners/answer_ngram_filter.py
diff --git a/dataflow/process/text/reasoners/answer_token_length_filter.py b/dataflow/process/text/reasoners/answer_token_length_filter.py
diff --git a/dataflow/process/text/reasoners/math_problem_filter.py b/dataflow/process/text/reasoners/math_problem_filter.py
@@ -0,0 +1,123 @@
+"""
+A filter class for evaluating mathematical problems using an API.
+This class constructs prompts for mathematical problem evaluation,
+sends them to the API, and returns a list of 0s and 1s indicating
+whether each problem is correctly formatted and solvable.
+
+Features:
+- Constructs evaluation prompts with four progressive checks
+- Handles API call failures gracefully
+- Processes datasets in parallel for efficiency
+- Returns results as a list of 0s and 1s
+
+Usage:
+1. Initialize the filter with appropriate parameters
+2. Call filter_func with a dataset (list of JSONL lines)
+3. Get results as a list of 0s and 1s
+"""
+
+from dataflow.utils.api_utils import api_chat
+import json
+import time
+import re
+from dataflow.core import ReasonerFilter
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from tqdm import tqdm
+from dataflow.utils.registry import PROCESSOR_REGISTRY
+
+@PROCESSOR_REGISTRY.register()
+class MathProblemFilter(ReasonerFilter):
+    def __init__(self, args_dict: dict):
+        super().__init__(args_dict)
+        self.system_prompt = args_dict.get("system_prompt", 
+            "You are an expert in evaluating mathematical problems. Follow the user's instructions strictly and output your final judgment in the required JSON format."
+        )
+        # need to set api_key first
+        self.model = self.model_name
+        self.api_key = args_dict.get("api_key", "")
+        self.filter_name = 'MathProblemFilter'
+
+    def build_prompt(self, question):
+        """Constructs an evaluation prompt with four progressive checks"""
+        prompt = f"""You are given a mathematical problem. Follow these four steps in order and stop at the first failure:
+1. Check only for spelling, grammar, and LaTeX formatting correctness. Do not interpret semantic meaning.
+2. For each minimal condition stated in the problem (that cannot be further decomposed), check if it violates the mathematical domain or objective facts (for example, 'half a person' is incorrect). Note: Magical operations are acceptable if the necessary assumption is explicitly stated. Average values (e.g., 15.5 items per minute) are acceptable.
+3. Check whether the problem-solving process contains any contradictions. This includes any two minimal conditions contradicting each other or if the final solution would be unreasonable (including unsolvable).
+4. If steps 1-3 pass, check whether the problem is fully solvable by verifying that all necessary conditions to answer the question are provided. This check should be based solely on the question.
+    
+After performing these steps in sequence, output your final judgment in JSON format with exactly the following keys:
+{{
+    "judgement_test": true/false,
+    "error_type": "<error description or null>"
+}}
+You may include your chain-of-thought, but the final answer must be the JSON object above.
+    
+Here is the problem to evaluate:
+-------------------------------
+{question}
+-------------------------------
+"""
+        return prompt
+
+    def process_problem(self, problem):
+        """Processes a single problem by calling the API"""
+        full_prompt = self.build_prompt(problem)
+        try:
+            response = api_chat(
+                system_info=self.system_prompt,
+                messages=full_prompt,
+                model=self.model,
+                api_url=self.api_url,
+                api_key=self.api_key
+            )
+        except Exception as e:
+            # API call failed, return 0
+            print(f"API call failed for problem: {problem}. Error: {e}")
+            return 0
+        else:
+            try:
+                pattern = re.compile(r'"judgement_test"\s*:\s*(true|false)', re.IGNORECASE)
+                match = pattern.search(response)
+                # print("---match---",match)
+                if match:
+                    test_value = match.group(1).lower()
+                return 1 if test_value == 'true' else 0
+            except Exception as e:
+                # Response format error, return 0
+                print(f"Response format error for problem: {problem}. Error: {e}")
+                return 0
+
+    def filter_func(self, dataset):
+        """Main filtering function that processes the entire dataset and returns a list of 0s and 1s"""
+        results = []
+        max_workers = 1  # Adjust based on your needs
+
+        dataset = dataset.to_list()
+        with ThreadPoolExecutor(max_workers=max_workers) as executor:
+            futures = []
+            for record in dataset:
+                try:
+                    problem = record.get("problem", "")
+                    if problem:
+                        futures.append(executor.submit(self.process_problem, problem))
+                except json.JSONDecodeError:
+                    print(f"Invalid JSON format in line: {line}")
+                    results.append(0)
+
+            for future in tqdm(as_completed(futures), total=len(futures), desc="Processing"):
+                result = future.result()
+                results.append(result)
+                # Add slight delay
+                time.sleep(0.1)
+
+        return results
+
+    def handle_api_error(self, error_message):
+        """Handles API errors and provides guidance"""
+        print(f"API Error: {error_message}")
+        print("Possible reasons:")
+        print("1. Network connection issue. Please check your internet connection.")
+        print("2. Invalid API URL. Please verify the URL format and accessibility.")
+        print("3. API service unavailable. Please check if the service is running properly.")
+        print("4. API key issue. Please ensure your API key is valid and has proper permissions.")
+        print("Suggestion: Try again after checking the above issues.")
diff --git a/dataflow/process/text/reasoning/__init__.py b/dataflow/process/text/reasoning/__init__.py
diff --git a/dataflow/utils/api_utils.py b/dataflow/utils/api_utils.py
@@ -0,0 +1,65 @@
+from openai import OpenAI
+import json
+import requests
+import os
+
+def api_chat( 
+            system_info: str,
+            messages: str,
+            model: str,
+            api_url : str = "",
+            api_key : str = "",
+            finish_try: int = 3,
+            mode_test : bool = True
+            ):
+    if api_key == "":
+        api_key = os.environ.get("API_KEY")
+
+    if api_key is None:
+        raise ValueError("Lack of API_KEY")
+    
+    if mode_test is True:
+        try:
+            payload = json.dumps({
+                "model": model,
+                "messages": [
+                    {"role": "system", "content": system_info},
+                    {"role": "user", "content": messages}
+                ]
+            })
+
+            headers = {
+            'Authorization': f"Bearer {api_key}",
+            'Content-Type': 'application/json',
+            'User-Agent': 'Apifox/1.0.0 (https://apifox.com)'
+            }
+            
+            # request OpenAI API
+            response = requests.post(api_url, headers=headers, data=payload, timeout=1800)
+            
+            # API debug code
+            # print("response ", response)
+            # print("response.status_code", response.status_code)
+
+            if response.status_code == 200:
+                response_data = response.json()
+                return response_data['choices'][0]['message']['content']
+
+
+        except Exception as e:
+            print("Error:", e)
+            pass
+    
+    else :
+        client = OpenAI(api_key=api_key)
+        api_response = client.chat.completions.create(
+            model=model,
+            messages=[
+                {"role": "system", "content": system_info},
+                {"role": "user", "content": messages}
+            ]
+        )
+
+        response_content = api_response.choices[0].message.content.strip()
+        
+        return response_content
diff --git a/dataflow/utils/registry.py b/dataflow/utils/registry.py
@@ -74,7 +74,7 @@ def get(self, name):
                         raise e
                 raise KeyError(f"No object named '{name}' found in '{self._name}' registry!")
             elif self._name == "processor":
-                for x in ['text.refiners', 'text.filters', 'text.deduplicators', 'text.reasoning','image.filters', 'image.deduplicators', 'video.filters']:
+                for x in ['text.refiners', 'text.filters', 'text.deduplicators', 'text.reasoners', 'image.filters', 'image.deduplicators', 'video.filters']:
                 # for x in ['image.filters', 'image.refiners']:
                     module_path = "dataflow.process." + x
                     try:
diff --git a/demos/demos_result/processed.jsonl b/demos/demos_result/processed.jsonl
@@ -0,0 +1,2 @@
+{"problem": "A rectangle has a length of 10 cm and a width of 5 cm. What is its area?"}
+{"problem": "If a car travels at 60 km/h for 2 hours, how far does it go?"}
diff --git a/demos/text_process/reasoners/math_5_samples.json b/demos/text_process/reasoners/math_5_samples.json
diff --git a/demos/text_process/reasoners/text_process_mathproblem.jsonl b/demos/text_process/reasoners/text_process_mathproblem.jsonl
@@ -0,0 +1,3 @@
+{"problem": "A rectangle has a length of 10 cm and a width of 5 cm. What is its area?"}
+{"problem": "If a car travels at 60 km/h for 2 hours, how far does it go?"}
+{"problem": "Solveaaaaaa for x: 2x + 5 = 15"}
diff --git a/processed.jsonl b/processed.jsonl
@@ -1 +1,2 @@
-{"answer": "Solution:\n        1. Find critical points:\n        \u2192 f'(x) = 3x\u00b2 - 6x\n        \u2192 Set derivative to zero: 3x(x-2) = 0 \u21d2 x=0, x=2\n    \n        2. Evaluate function at critical points and endpoints:\n        \u2192 f(-1) = (-1)^3 - 3(-1)^2 + 4 = -1 -3 +4 = 0.0000\n        \u2192 f(0) = 0\u00b3 - 3(0)\u00b2 +4 = 4.0000\n        \u2192 f(2) = 8 - 12 +4 = 0.0000\n        \u2192 f(3) = 27 - 27 +4 = 4.0000\n    \n        3. Compare values:\n        \u2192 Minimum occurs at x=-1 and x=2\n    \n        Verification:\n        \u2192 Second derivative test: f''(x) = 6x-6\n        \u2192 f''(-1) = -12 < 0 (local max)\n        \u2192 f''(2) = 6 > 0 (local min)\n    \n        \\boxed{0}", "ground_truth_answer": "0", "language": null}
+{"problem": "A rectangle has a length of 10 cm and a width of 5 cm. What is its area?"}
+{"problem": "If a car travels at 60 km/h for 2 hours, how far does it go?"}

-Original file line number
+Diff line change
 **/data
 **/ckpt
 tmp.*
 +configs/process/text_process_reasoner.yaml