feat: add pipeline coder (#742)

* init commit

* limit problem numbers

* ensemble lower case

* add runtime and spec to coder

* submission check notice

* sub EDA in sample execution

* avoid lightgbm

* add time limit to scenario

* rephrase the submission check

* give positive feedback when facing warning in check

* ENABLE FEEDBACK

* fix feedback bug

---------

Co-authored-by: Xu Yang <[email protected]>

Author: RolandMinrui

rdagent/app/data_science/conf.py

@@ -25,6 +25,7 @@ class DataScienceBasePropSetting(KaggleBasePropSetting):
     spec_enabled: bool = True
 
     proposal_version: str = "v1"
+    coder_on_whole_pipeline: bool = False
 
     coder_max_loop: int = 10
     runner_max_loop: int = 3

rdagent/app/data_science/loop.py

@@ -10,6 +10,8 @@
 from rdagent.components.coder.data_science.feature.exp import FeatureTask
 from rdagent.components.coder.data_science.model import ModelCoSTEER
 from rdagent.components.coder.data_science.model.exp import ModelTask
+from rdagent.components.coder.data_science.pipeline import PipelineCoSTEER
+from rdagent.components.coder.data_science.pipeline.exp import PipelineTask
 from rdagent.components.coder.data_science.raw_data_loader import DataLoaderCoSTEER
 from rdagent.components.coder.data_science.raw_data_loader.exp import DataLoaderTask
 from rdagent.components.coder.data_science.workflow import WorkflowCoSTEER
@@ -54,6 +56,8 @@ def __init__(self, PROP_SETTING: BasePropSetting):
         self.ensemble_coder = EnsembleCoSTEER(scen)
         self.workflow_coder = WorkflowCoSTEER(scen)
 
+        self.pipeline_coder = PipelineCoSTEER(scen)
+
         self.runner = DSCoSTEERRunner(scen)
         # self.summarizer: Experiment2Feedback = import_class(PROP_SETTING.summarizer)(scen)
         # logger.log_object(self.summarizer, tag="summarizer")
@@ -86,6 +90,8 @@ def coding(self, prev_out: dict[str, Any]):
                     exp = self.ensemble_coder.develop(exp)
                 elif isinstance(exp.sub_tasks[0], WorkflowTask):
                     exp = self.workflow_coder.develop(exp)
+                elif isinstance(exp.sub_tasks[0], PipelineTask):
+                    exp = self.pipeline_coder.develop(exp)
                 else:
                     raise NotImplementedError(f"Unsupported component in DataScienceRDLoop: {exp.hypothesis.component}")
             exp.sub_tasks = []
@@ -106,7 +112,7 @@ def feedback(self, prev_out: dict[str, Any]) -> ExperimentFeedback:
         - If we come to feedback phase, the previous development steps are successful.
         """
         exp: DSExperiment = prev_out["running"]
-        if self.trace.next_incomplete_component() is None:
+        if self.trace.next_incomplete_component() is None or DS_RD_SETTING.coder_on_whole_pipeline:
             # we have alreadly completed components in previous trace. So current loop is focusing on a new proposed idea.
             # So we need feedback for the proposal.
             feedback = self.summarizer.generate_feedback(exp, self.trace)
@@ -130,7 +136,11 @@ def record(self, prev_out: dict[str, Any]):
                     ExperimentFeedback.from_exception(e),
                 )
             )
-            if self.trace.sota_experiment() is None and len(self.trace.hist) >= DS_RD_SETTING.consecutive_errors:
+            if (
+                self.trace.sota_experiment() is None
+                and len(self.trace.hist) >= DS_RD_SETTING.consecutive_errors
+                and not DS_RD_SETTING.coder_on_whole_pipeline
+            ):
                 # if {in inital/drafting stage} and {tried enough times}
                 for _, fb in self.trace.hist[-DS_RD_SETTING.consecutive_errors :]:
                     if fb:

rdagent/components/coder/data_science/pipeline/__init__.py

@@ -0,0 +1,163 @@
+"""
+
+Loop should not large change exclude
+- Action Choice[current data loader & spec]
+- other should share
+    - Propose[choice] => Task[Choice] => CoSTEER =>
+        -
+
+Extra feature:
+- cache
+
+
+File structure
+- ___init__.py: the entrance/agent of coder
+- evaluator.py
+- conf.py
+- exp.py: everything under the experiment, e.g.
+    - Task
+    - Experiment
+    - Workspace
+- test.py
+    - Each coder could be tested.
+"""
+
+import json
+import re
+from pathlib import Path
+from typing import Dict
+
+from rdagent.app.data_science.conf import DS_RD_SETTING
+from rdagent.components.coder.CoSTEER import CoSTEER
+from rdagent.components.coder.CoSTEER.evaluators import (
+    CoSTEERMultiEvaluator,
+    CoSTEERSingleFeedback,
+)
+from rdagent.components.coder.CoSTEER.evolving_strategy import (
+    MultiProcessEvolvingStrategy,
+)
+from rdagent.components.coder.CoSTEER.knowledge_management import (
+    CoSTEERQueriedKnowledge,
+)
+from rdagent.components.coder.data_science.conf import (
+    DSCoderCoSTEERSettings,
+    get_ds_env,
+)
+from rdagent.components.coder.data_science.pipeline.eval import PipelineCoSTEEREvaluator
+from rdagent.components.coder.data_science.raw_data_loader.eval import (
+    DataLoaderCoSTEEREvaluator,
+)
+from rdagent.components.coder.data_science.raw_data_loader.exp import DataLoaderTask
+from rdagent.core.exception import CoderError
+from rdagent.core.experiment import FBWorkspace
+from rdagent.core.scenario import Scenario
+from rdagent.oai.llm_utils import APIBackend
+from rdagent.utils.agent.ret import PythonAgentOut
+from rdagent.utils.agent.tpl import T
+
+DIRNAME = Path(__file__).absolute().resolve().parent
+
+
+class PipelineMultiProcessEvolvingStrategy(MultiProcessEvolvingStrategy):
+    def implement_one_task(
+        self,
+        target_task: DataLoaderTask,
+        queried_knowledge: CoSTEERQueriedKnowledge | None = None,
+        workspace: FBWorkspace | None = None,
+        prev_task_feedback: CoSTEERSingleFeedback | None = None,
+    ) -> dict[str, str]:
+        competition_info = self.scen.get_scenario_all_desc()
+        runtime_environment = self.scen.get_runtime_environment()
+        data_folder_info = self.scen.processed_data_folder_description
+        pipeline_task_info = target_task.get_task_information()
+
+        queried_similar_successful_knowledge = (
+            queried_knowledge.task_to_similar_task_successful_knowledge[pipeline_task_info]
+            if queried_knowledge is not None
+            else []
+        )
+        queried_former_failed_knowledge = (
+            queried_knowledge.task_to_former_failed_traces[pipeline_task_info] if queried_knowledge is not None else []
+        )
+        queried_former_failed_knowledge = (
+            [
+                knowledge
+                for knowledge in queried_former_failed_knowledge[0]
+                if knowledge.implementation.file_dict.get("main.py") != workspace.file_dict.get("main.py")
+            ],
+            queried_former_failed_knowledge[1],
+        )
+
+        system_prompt = T(".prompts:pipeline_coder.system").r(
+            task_desc=pipeline_task_info,
+            queried_similar_successful_knowledge=queried_similar_successful_knowledge,
+            queried_former_failed_knowledge=queried_former_failed_knowledge[0],
+            out_spec=PythonAgentOut.get_spec(),
+            runtime_environment=runtime_environment,
+            spec=T("scenarios.data_science.share:component_spec.Pipeline").r(),
+        )
+        user_prompt = T(".prompts:pipeline_coder.user").r(
+            competition_info=competition_info,
+            folder_spec=data_folder_info,
+            latest_code=workspace.file_dict.get("main.py"),
+            latest_code_feedback=prev_task_feedback,
+        )
+
+        for _ in range(5):
+            pipeline_code = PythonAgentOut.extract_output(
+                APIBackend().build_messages_and_create_chat_completion(
+                    user_prompt=user_prompt,
+                    system_prompt=system_prompt,
+                )
+            )
+            if pipeline_code != workspace.file_dict.get("main.py"):
+                break
+            else:
+                user_prompt = user_prompt + "\nPlease avoid generating same code to former code!"
+        else:
+            raise CoderError("Failed to generate a new pipeline code.")
+
+        return {
+            "main.py": pipeline_code,
+        }
+
+    def assign_code_list_to_evo(self, code_list: list[dict[str, str]], evo):
+        """
+        Assign the code list to the evolving item.
+
+        The code list is aligned with the evolving item's sub-tasks.
+        If a task is not implemented, put a None in the list.
+        """
+        for index in range(len(evo.sub_tasks)):
+            if code_list[index] is None:
+                continue
+            if evo.sub_workspace_list[index] is None:
+                # evo.sub_workspace_list[index] = FBWorkspace(target_task=evo.sub_tasks[index])
+                evo.sub_workspace_list[index] = evo.experiment_workspace
+            evo.sub_workspace_list[index].inject_files(**code_list[index])
+        return evo
+
+
+class PipelineCoSTEER(CoSTEER):
+    def __init__(
+        self,
+        scen: Scenario,
+        *args,
+        **kwargs,
+    ) -> None:
+        settings = DSCoderCoSTEERSettings()
+        eva = CoSTEERMultiEvaluator(
+            PipelineCoSTEEREvaluator(scen=scen), scen=scen
+        )  # Please specify whether you agree running your eva in parallel or not
+        es = PipelineMultiProcessEvolvingStrategy(scen=scen, settings=settings)
+
+        super().__init__(
+            *args,
+            settings=settings,
+            eva=eva,
+            es=es,
+            evolving_version=2,
+            scen=scen,
+            max_loop=DS_RD_SETTING.coder_max_loop,
+            **kwargs,
+        )

rdagent/components/coder/data_science/pipeline/eval.py

@@ -0,0 +1,119 @@
+# tess successfully running.
+# (GPT) if it aligns with the spec & rationality of the spec.
+import json
+import re
+from pathlib import Path
+
+import pandas as pd
+
+from rdagent.app.data_science.conf import DS_RD_SETTING
+from rdagent.components.coder.CoSTEER import CoSTEERMultiFeedback
+from rdagent.components.coder.CoSTEER.evaluators import (
+    CoSTEEREvaluator,
+    CoSTEERSingleFeedback,
+)
+from rdagent.components.coder.CoSTEER.knowledge_management import (
+    CoSTEERQueriedKnowledgeV2,
+)
+from rdagent.components.coder.data_science.conf import get_ds_env
+from rdagent.core.experiment import FBWorkspace, Task
+from rdagent.utils.agent.tpl import T
+from rdagent.utils.agent.workflow import build_cls_from_json_with_retry
+
+DIRNAME = Path(__file__).absolute().resolve().parent
+
+PipelineSingleFeedback = CoSTEERSingleFeedback
+PipelineMultiFeedback = CoSTEERMultiFeedback
+
+
+class PipelineCoSTEEREvaluator(CoSTEEREvaluator):
+
+    def evaluate(
+        self,
+        target_task: Task,
+        implementation: FBWorkspace,
+        gt_implementation: FBWorkspace,
+        queried_knowledge: CoSTEERQueriedKnowledgeV2 = None,
+        **kwargs,
+    ) -> PipelineSingleFeedback:
+
+        target_task_information = target_task.get_task_information()
+        if (
+            queried_knowledge is not None
+            and target_task_information in queried_knowledge.success_task_to_knowledge_dict
+        ):
+            return queried_knowledge.success_task_to_knowledge_dict[target_task_information].feedback
+        elif queried_knowledge is not None and target_task_information in queried_knowledge.failed_task_info_set:
+            return PipelineSingleFeedback(
+                execution="This task has failed too many times, skip implementation.",
+                return_checking="This task has failed too many times, skip implementation.",
+                code="This task has failed too many times, skip implementation.",
+                final_decision=False,
+            )
+
+        env = get_ds_env()
+        env.conf.extra_volumes = {f"{DS_RD_SETTING.local_data_path}/sample/{self.scen.competition}": "/kaggle/input"}
+
+        # Clean the scores.csv & submission.csv.
+        implementation.execute(env=env, entry=f"rm submission.csv scores.csv")
+        stdout = implementation.execute(env=env, entry=f"python main.py")
+        stdout = re.sub(r"=== Start of EDA part ===(.*)=== End of EDA part ===", "", stdout)
+
+        score_fp = implementation.workspace_path / "scores.csv"
+        score_ret_code = 0
+        score_check_text = ""
+        if not score_fp.exists():
+            score_check_text = "[Error] Metrics file (scores.csv) is not generated!"
+            score_ret_code = 1
+        else:
+            try:
+                score_df = pd.read_csv(score_fp, index_col=0)
+                model_set_in_scores = set(score_df.index)
+
+                # Check model names (index)
+                if "ensemble" not in model_set_in_scores:
+                    score_check_text += (
+                        f"\n[Error] The score dataframe doesn't contain the ensemble model.\nscore_df is:\n{score_df}"
+                    )
+                    score_ret_code = 1
+
+                # Check metric name (columns)
+                if score_df.columns.tolist() != [self.scen.metric_name]:
+                    score_check_text += f"\n[Error] The scores dataframe does not contain the correct column names.\nCorrect columns is: ['{self.scen.metric_name}']\nBut got: {score_df.columns.tolist()}"
+                    score_ret_code = 1
+
+            except Exception as e:
+                score_check_text += f"\n[Error] in checking the scores.csv file: {e}\nscores.csv's content:\n-----\n{score_fp.read_text()}\n-----"
+                score_ret_code = 1
+
+        # Check submission file
+        base_check_code = (DIRNAME / "eval_tests" / "submission_format_test.txt").read_text()
+        implementation.inject_files(**{"test/submission_format_test.py": base_check_code})
+        # stdout += "----Submission Check 1-----\n"
+        submission_check_out, submission_ret_code = implementation.execute_ret_code(
+            env=env, entry="python test/submission_format_test.py"
+        )
+        stdout += "\n" + submission_check_out
+
+        system_prompt = T(".prompts:pipeline_eval.system").r(
+            scenario=self.scen.get_scenario_all_desc(),
+            task_desc=target_task.get_task_information(),
+            spec=T("scenarios.data_science.share:component_spec.Pipeline").r(),
+        )
+        user_prompt = T(".prompts:pipeline_eval.user").r(
+            stdout=stdout.strip(),
+            code=implementation.file_dict["main.py"],
+        )
+        wfb = build_cls_from_json_with_retry(
+            PipelineSingleFeedback,
+            system_prompt=system_prompt,
+            user_prompt=user_prompt,
+            init_kwargs_update_func=PipelineSingleFeedback.val_and_update_init_dict,
+        )
+        if score_ret_code != 0:
+            wfb.final_decision = False
+            wfb.return_checking += "\n" + score_check_text
+        if submission_ret_code != 0:
+            wfb.final_decision = False
+            wfb.return_checking += "\nSubmission file check failed."
+        return wfb