feat: add coder check and give more time (#1127)

* feat: introduce max_seconds_multiplier for timeout management across components

* avoid using assert in test

* hot fix a very small bug

* runner multiply twice

* revert feedback change

* add a switch to longer timeout
This commit is contained in:
Xu Yang
2025-07-29 14:59:38 +08:00
committed by GitHub
parent 53dd98abf4
commit dd2e06ade6
29 changed files with 102 additions and 83 deletions
+3
View File
@@ -118,6 +118,9 @@ class DataScienceBasePropSetting(KaggleBasePropSetting):
enable_planner: bool = False
model_architecture_suggestion_time_percent: float = 0.75
allow_longer_timeout: bool = False
coder_longer_timeout_multiplier: int = 3
runner_longer_timeout_multiplier: int = 2
#### hypothesis critique and rewrite
enable_hypo_critique_rewrite: bool = True
+3 -2
View File
@@ -28,6 +28,7 @@ class CoSTEER(Developer[Experiment]):
es: EvolvingStrategy,
evolving_version: int,
*args,
max_seconds: int | None = None,
with_knowledge: bool = True,
with_feedback: bool = True,
knowledge_self_gen: bool = True,
@@ -37,7 +38,7 @@ class CoSTEER(Developer[Experiment]):
) -> None:
super().__init__(*args, **kwargs)
self.max_loop = settings.max_loop if max_loop is None else max_loop
self.max_seconds = settings.max_seconds
self.max_seconds = max_seconds
self.knowledge_base_path = (
Path(settings.knowledge_base_path) if settings.knowledge_base_path is not None else None
)
@@ -105,7 +106,7 @@ class CoSTEER(Developer[Experiment]):
logger.log_object(evo_exp.sub_workspace_list, tag="evolving code")
for sw in evo_exp.sub_workspace_list:
logger.info(f"evolving workspace: {sw}")
if (datetime.now() - start_datetime).seconds > self.max_seconds:
if self.max_seconds is not None and (datetime.now() - start_datetime).seconds > self.max_seconds:
logger.info(f"Reached max time limit {self.max_seconds} seconds, stop evolving")
break
if RD_Agent_TIMER_wrapper.timer.started and RD_Agent_TIMER_wrapper.timer.is_timeout():
+1 -1
View File
@@ -33,7 +33,7 @@ class CoSTEERSettings(ExtendedBaseSettings):
new_knowledge_base_path: Union[str, None] = None
"""Path to the new knowledge base"""
max_seconds: int = 10**6
max_seconds_multiplier: int = 10**6
CoSTEER_SETTINGS = CoSTEERSettings()
@@ -19,7 +19,7 @@ class DSCoderCoSTEERSettings(CoSTEERSettings):
class Config:
env_prefix = "DS_Coder_CoSTEER_"
max_seconds: int = DS_RD_SETTING.debug_timeout * 4
max_seconds_multiplier: int = 4
env_type: str = "docker"
# TODO: extract a function for env and conf.
@@ -28,7 +28,7 @@ class DSCoderCoSTEERSettings(CoSTEERSettings):
Extra evaluators
The evaluator follows the following assumptions:
- It runs after previous evaluator (So the running results are alreadly there)
- It runs after previous evaluator (So the running results are already there)
It is not a complete feature due to it is only implemented in DS Pipeline & Coder.
@@ -160,5 +160,6 @@ class EnsembleCoSTEER(CoSTEER):
evolving_version=2,
scen=scen,
max_loop=DS_RD_SETTING.coder_max_loop,
max_seconds=scen.real_debug_timeout() * settings.max_seconds_multiplier,
**kwargs,
)
@@ -47,7 +47,10 @@ class EnsembleCoSTEEREvaluator(CoSTEEREvaluator):
final_decision=False,
)
env = get_ds_env(extra_volumes={self.scen.debug_path: T("scenarios.data_science.share:scen.input_path").r()})
env = get_ds_env(
extra_volumes={self.scen.debug_path: T("scenarios.data_science.share:scen.input_path").r()},
running_timeout_period=self.scen.real_debug_timeout(),
)
fname = "test/ensemble_test.txt"
test_code = (DIRNAME / "eval_tests" / "ensemble_test.txt").read_text()
@@ -138,5 +138,6 @@ class FeatureCoSTEER(CoSTEER):
evolving_version=2,
scen=scen,
max_loop=DS_RD_SETTING.coder_max_loop,
max_seconds=scen.real_debug_timeout() * settings.max_seconds_multiplier,
**kwargs,
)
@@ -43,7 +43,10 @@ class FeatureCoSTEEREvaluator(CoSTEEREvaluator):
final_decision=False,
)
env = get_ds_env(extra_volumes={self.scen.debug_path: T("scenarios.data_science.share:scen.input_path").r()})
env = get_ds_env(
extra_volumes={self.scen.debug_path: T("scenarios.data_science.share:scen.input_path").r()},
running_timeout_period=self.scen.real_debug_timeout(),
)
# TODO: do we need to clean the generated temporary content?
fname = "test/feature_test.py"
@@ -170,5 +170,6 @@ class ModelCoSTEER(CoSTEER):
evolving_version=2,
scen=scen,
max_loop=DS_RD_SETTING.coder_max_loop,
max_seconds=scen.real_debug_timeout() * settings.max_seconds_multiplier,
**kwargs,
)
@@ -57,7 +57,10 @@ class ModelGeneralCaseSpecEvaluator(CoSTEEREvaluator):
final_decision=False,
)
env = get_ds_env(extra_volumes={self.scen.debug_path: T("scenarios.data_science.share:scen.input_path").r()})
env = get_ds_env(
extra_volumes={self.scen.debug_path: T("scenarios.data_science.share:scen.input_path").r()},
running_timeout_period=self.scen.real_debug_timeout(),
)
if_model_removed = False
@@ -161,5 +161,6 @@ class PipelineCoSTEER(CoSTEER):
evolving_version=2,
scen=scen,
max_loop=DS_RD_SETTING.coder_max_loop,
max_seconds=scen.real_debug_timeout() * settings.max_seconds_multiplier,
**kwargs,
)
@@ -53,7 +53,10 @@ class PipelineCoSTEEREvaluator(CoSTEEREvaluator):
final_decision=False,
)
env = get_ds_env(extra_volumes={self.scen.debug_path: T("scenarios.data_science.share:scen.input_path").r()})
env = get_ds_env(
extra_volumes={self.scen.debug_path: T("scenarios.data_science.share:scen.input_path").r()},
running_timeout_period=self.scen.real_debug_timeout(),
)
stdout = ""
implementation.execute(env=env, entry=get_clear_ws_cmd())
@@ -97,6 +100,11 @@ class PipelineCoSTEEREvaluator(CoSTEEREvaluator):
else:
stdout += "Debug mode did not provide debug_time or estimated_time, it's a buggy implementation.\n"
test_eval = get_test_eval()
if test_eval.enabled(self.scen.competition):
submission_check_out, submission_ret_code = test_eval.valid(self.scen.competition, implementation)
stdout += f"\n### Submission check:\n{submission_check_out}\nIf Submission check returns a 'Submission is valid' or similar message, despite some warning messages, you should still consider the submission as valid and give a positive final decision. "
score_fp = implementation.workspace_path / "scores.csv"
score_ret_code = 0
score_check_text = ""
@@ -153,6 +161,7 @@ class PipelineCoSTEEREvaluator(CoSTEEREvaluator):
system_prompt = T(".prompts:pipeline_eval.system").r(
is_sub_enabled=test_eval.is_sub_enabled(self.scen.competition),
debug_mode=DS_RD_SETTING.sample_data_by_LLM,
mle_check=(DS_RD_SETTING.sample_data_by_LLM and test_eval.is_sub_enabled(self.scen.competition)),
)
user_prompt = T(".prompts:pipeline_eval.user").r(
scenario=self.scen.get_scenario_all_desc(eda_output=eda_output),
@@ -259,7 +259,14 @@ pipeline_eval:
- Data sampling should only be applied in debug mode. Always use the full data in the full run.
- The label classes number should be the same as the full run even in debug mode.
- If the code passes this step: Finalize evaluation.
- If the code does not pass this step: Clearly document the debug mode compliance issues and reject the implementation.
- If the code does not pass this step: Clearly document the debug mode compliance issues and reject the implementation.{% endif %}
{% if mle_check %}
### Step 5: Test format check
- The user has done a format check for your submission. Since you didn't sample any test data, your debug mode output should be the same format as the full run.
- The user will put the check result in the "Submission check" section of the execution output.
- If the submission check returns a 'Submission is valid' or similar message, despite some warning messages, you should give the conclusion that the code executed successfully. If no other code related issues are found, set the "final_decision" to true.
- If the submission check returns an error message, you should set the "final_decision" to false and clearly document the issues in the "return_checking" field.
{% endif %}
## Output Format
@@ -218,6 +218,7 @@ class DataLoaderCoSTEER(CoSTEER):
evolving_version=2,
scen=scen,
max_loop=DS_RD_SETTING.coder_max_loop,
max_seconds=scen.real_debug_timeout() * settings.max_seconds_multiplier,
**kwargs,
)
@@ -230,7 +231,7 @@ class DataLoaderCoSTEER(CoSTEER):
"scenarios.data_science.share:scen.input_path"
).r()
},
running_timeout_period=DS_RD_SETTING.full_timeout,
running_timeout_period=self.scen.real_full_timeout(),
)
stdout = new_exp.experiment_workspace.execute(env=env, entry=f"python test/data_loader_test.py")
@@ -46,7 +46,10 @@ class DataLoaderCoSTEEREvaluator(CoSTEEREvaluator):
final_decision=False,
)
env = get_ds_env(extra_volumes={self.scen.debug_path: T("scenarios.data_science.share:scen.input_path").r()})
env = get_ds_env(
extra_volumes={self.scen.debug_path: T("scenarios.data_science.share:scen.input_path").r()},
running_timeout_period=self.scen.real_debug_timeout(),
)
# TODO: do we need to clean the generated temporary content?
fname = "test/data_loader_test.py"
@@ -55,7 +55,7 @@ class ModelDumpEvaluator(CoSTEEREvaluator):
env = get_ds_env(
extra_volumes={data_source_path: T("scenarios.data_science.share:scen.input_path").r()},
running_timeout_period=(
DS_RD_SETTING.full_timeout if self.data_type == "full" else DS_RD_SETTING.debug_timeout
self.scen.real_full_timeout() if self.data_type == "full" else self.scen.real_debug_timeout()
),
)
@@ -131,5 +131,6 @@ class WorkflowCoSTEER(CoSTEER):
evolving_version=2,
scen=scen,
max_loop=DS_RD_SETTING.coder_max_loop,
max_seconds=scen.real_debug_timeout() * settings.max_seconds_multiplier,
**kwargs,
)
@@ -55,7 +55,10 @@ class WorkflowGeneralCaseSpecEvaluator(CoSTEEREvaluator):
final_decision=False,
)
env = get_ds_env(extra_volumes={self.scen.debug_path: T("scenarios.data_science.share:scen.input_path").r()})
env = get_ds_env(
extra_volumes={self.scen.debug_path: T("scenarios.data_science.share:scen.input_path").r()},
running_timeout_period=self.scen.real_debug_timeout(),
)
# # DockerEnv for MLEBench submission validation
# mle_de_conf = MLEBDockerConf()
@@ -55,34 +55,6 @@ class DSExperiment2Feedback(Experiment2Feedback):
else:
diff_edition = []
sota_exp_and_feedback = trace.experiment_and_feedback_list_after_init(return_type="sota", with_index=True)
sota_valid_score_df = pd.DataFrame(
index=[index for index, exp, fb in sota_exp_and_feedback], columns=["score", "improvement"]
)
for row_index, (index, sota_exp, sota_fb) in enumerate(sota_exp_and_feedback):
sota_valid_score_df.loc[index, "score"] = (
sota_exp.result.loc["ensemble"].iloc[0] if sota_exp.result is not None else "N/A"
)
if row_index > 0:
improve_percent = (
(
((sota_valid_score_df.loc[index, "score"] - sota_valid_score_df.iloc[row_index - 1]["score"]))
/ sota_valid_score_df.iloc[row_index - 1]["score"]
* 100
)
if self.scen.metric_direction
else (
((sota_valid_score_df.iloc[row_index - 1]["score"] - sota_valid_score_df.loc[index, "score"]))
/ sota_valid_score_df.iloc[row_index - 1]["score"]
* 100
)
)
sota_valid_score_df.loc[index, "improvement"] = f"{improve_percent:.2f}%"
sota_valid_score_df["improvement"] = sota_valid_score_df["score"].diff()
all_sota_results_desc = T("scenarios.data_science.share:describe.sota_results").r(
sota_valid_score_dict=sota_valid_score_df.to_dict(orient="index"),
)
# assumption:
# The feedback should focus on experiment **improving**.
# Assume that all the the sota exp is based on the previous sota experiment
@@ -106,7 +78,6 @@ class DSExperiment2Feedback(Experiment2Feedback):
diff_edition=diff_edition,
feedback_desc=feedback_desc,
cur_vs_sota_score=cur_vs_sota_score,
all_sota_results_desc=all_sota_results_desc,
)
resp_dict = json.loads(
@@ -40,23 +40,15 @@ exp_feedback:
- Directly compare the current `ensemble` validation score to the SOTA `ensemble` validation score. Do not focus on individual models unless anomalies are significant.
- Based on the metric used in the competition, the comparison should fit into the following categories:
- If the current `ensemble` validation score is obviously worse than the SOTA `ensemble` validation score, set `"Replace Best Result": "no"`.
- If the current `ensemble` validation score is obviously better than the SOTA `ensemble` validation score, proceed to Step 4.
- If the current `ensemble` validation score is similar to the SOTA `ensemble` validation score or both reach the ceiling performance, proceed to Step 5.
- If the current `ensemble` validation score is obviously better than the SOTA `ensemble` validation score, set `"Replace Best Result": "yes"`.
- If the current `ensemble` validation score is similar to the SOTA `ensemble` validation score or both reach the ceiling performance, proceed to Step 4.
- Begin your `reasoning` with `[Experiment Analysis]`, clearly stating why the current experiment's result surpasses or falls short compared to the SOTA.
- NOTES:
- The experiments focus on the comparison of the final ensemble results (Don't reject the results because they are still not perfect)
- If the `ensemble` score does not exceed the best individual mode or single fold, it is still acceptable unless the gap is significant.
Step 4: Identify validation score improvement source
- The user also provides you all the ensemble validation scores of the SOTA experiments you can get the improve pattern of the validation score numbers.
- Since the current validation score is obviously better than the SOTA `ensemble` validation score, you should analyze whether the current number based on all the history numbers following the patterns:
- If the sota validation score has been stabilized or joggling for several rounds, you should be very careful to accept the current score as a valid improvement because it is likely that this validation score improvement is the cause of over-fit.
- If the code manipulate the validation dataset or modified the validation process. please set `"Replace Best Result": "no"` and explain in `"Reasoning"` starting with `[Validation Score Manipulation]`.
- If the improvement is based a big innovated code change, please proceed to Step 5 to check the code quality and give decision based on both the validation score and code quality.
- If no much sota validation score is provided, you should proceed to Step 5 to check the code quality and give decision based on both the validation score and code quality.
Step 5: Analyze Code To get Better validation results
- If the current `ensemble` validation score is similar to the SOTA `ensemble` validation score or suspicious validation improvement after several trials, give the decision based on the comparison between the current experiment and SOTA.
Step 4: Analyze Code With Similar validation Results
- If the current `ensemble` validation score is similar to the SOTA `ensemble` validation score, give the decision based on the comparison between the current experiment and SOTA.
- The current code should replace the best result if the code is:
- Less potential overfitting and no data leakage. The code should not modify the validation and test set distributions.
- Using best practices and modeling techniques. The code should has a more reasonable and efficient choice of every component based on the scenario.
@@ -31,7 +31,7 @@ class DSRunnerCoSTEERSettings(CoSTEERSettings):
class Config:
env_prefix = "DS_Runner_CoSTEER_"
max_seconds: int = DS_RD_SETTING.full_timeout
max_seconds_multiplier: int = 1
env_type: str = "docker"
diff_mode: bool = False
# TODO: extract a function for env and conf.
@@ -147,6 +147,7 @@ class DSCoSTEERRunner(CoSTEER):
evolving_version=2,
scen=scen,
max_loop=DS_RD_SETTING.runner_max_loop,
max_seconds=scen.real_full_timeout() * settings.max_seconds_multiplier,
**kwargs,
)
@@ -57,7 +57,7 @@ class DSCoSTEERCoSTEEREvaluator(CoSTEEREvaluator):
"scenarios.data_science.share:scen.input_path"
).r()
},
running_timeout_period=DS_RD_SETTING.full_timeout,
running_timeout_period=self.scen.real_full_timeout(),
)
stdout = implementation.execute(
@@ -4,15 +4,18 @@ from mlebench.grade import validate_submission
from mlebench.registry import registry
# Check if our submission file exists
assert Path("submission.csv").exists(), "Error: submission.csv not found"
if Path("submission.csv").exists():
print("Submission file found, proceeding with validation...")
COMPETITION_ID = "<competition_id>"
new_registry = registry.set_data_dir(Path("/mle/data"))
competition = new_registry.get_competition(COMPETITION_ID)
COMPETITION_ID = "<competition_id>"
new_registry = registry.set_data_dir(Path("/mle/data"))
competition = new_registry.get_competition(COMPETITION_ID)
is_valid, message = validate_submission(Path("submission.csv"), competition)
is_valid, message = validate_submission(Path("submission.csv"), competition)
print(message)
print(message)
if not is_valid:
raise AssertionError("Submission is invalid")
else:
if not is_valid:
raise AssertionError("Submission is invalid")
print("Error: submission.csv not found. Seems code execution failed in some step.")
@@ -171,7 +171,6 @@ class DSTrace(Trace[DataScienceScen, KnowledgeBase]):
search_type: Literal["all", "ancestors"] = "ancestors",
selection: tuple[int, ...] | None = None,
max_retrieve_num: int | None = None,
with_index: bool = False,
) -> list[tuple[DSExperiment, ExperimentFeedback]]:
"""
Retrieve a list of experiments and feedbacks based on the return_type.
@@ -181,13 +180,13 @@ class DSTrace(Trace[DataScienceScen, KnowledgeBase]):
has_final_component = True if DS_RD_SETTING.coder_on_whole_pipeline else False
SOTA_exp_and_feedback_list = []
failed_exp_and_feedback_list_after_sota = []
for index, (exp, fb) in enumerate(search_list):
for exp, fb in search_list:
if has_final_component:
if fb.decision:
SOTA_exp_and_feedback_list.append((index, exp, fb) if with_index else (exp, fb))
SOTA_exp_and_feedback_list.append((exp, fb))
failed_exp_and_feedback_list_after_sota = []
else:
failed_exp_and_feedback_list_after_sota.append((index, exp, fb) if with_index else (exp, fb))
failed_exp_and_feedback_list_after_sota.append((exp, fb))
if exp.hypothesis.component == final_component and fb:
has_final_component = True
if max_retrieve_num is not None and (SOTA_exp_and_feedback_list or failed_exp_and_feedback_list_after_sota):
@@ -71,13 +71,13 @@ class ParallelMultiTraceExpGen(ExpGen):
while True:
if loop.get_unfinished_loop_cnt(loop.loop_idx) < RD_AGENT_SETTINGS.get_max_parallel():
# set trace current selection
leaves: list[int] = trace.get_leaves()
if not timer.started or timer.remain_time() >= timedelta(hours=DS_RD_SETTING.merge_hours):
local_selection = await self.trace_scheduler.next(trace)
# set the local selection as the global current selection for the trace
trace.set_current_selection(local_selection)
else:
leaves: list[int] = trace.get_leaves()
if len(leaves) < 2:
local_selection = (-1,)
trace.set_current_selection(selection=local_selection)
@@ -111,6 +111,23 @@ class DataScienceScen(Scenario):
)
self.metric_name = response_json_analysis.get("Metric Name", "custom_metric")
self.metric_direction_guess = response_json_analysis.get("Metric Direction", True)
self.longer_time_limit_required = response_json_analysis.get(
"Longer time limit required", False
) # True or False, whether the competition scenario requires a longer time limit to the code.
def real_debug_timeout(self):
return (
DS_RD_SETTING.debug_timeout * DS_RD_SETTING.coder_longer_timeout_multiplier
if self.longer_time_limit_required and DS_RD_SETTING.allow_longer_timeout
else DS_RD_SETTING.debug_timeout
)
def real_full_timeout(self):
return (
DS_RD_SETTING.full_timeout * DS_RD_SETTING.runner_longer_timeout_multiplier
if self.longer_time_limit_required and DS_RD_SETTING.allow_longer_timeout
else DS_RD_SETTING.full_timeout
)
@property
def background(self) -> str:
@@ -160,10 +177,10 @@ class DataScienceScen(Scenario):
metric_direction=self.metric_direction,
raw_description=self.raw_description,
use_raw_description=DS_RD_SETTING.use_raw_description,
time_limit=f"{DS_RD_SETTING.full_timeout / 60 / 60 : .2f} hours",
time_limit=f"{self.real_full_timeout() / 60 / 60 : .2f} hours",
eda_output=eda_output,
sample_data_by_LLM=DS_RD_SETTING.sample_data_by_LLM,
debug_time_limit=f"{DS_RD_SETTING.debug_timeout / 60 / 60 : .2f} hours",
debug_time_limit=f"{self.real_debug_timeout() / 60 / 60 : .2f} hours",
runtime_environment=self.get_runtime_environment(),
)
@@ -66,6 +66,7 @@ competition_description_template:
"Metric Evaluation Description": "A precise explanation of how the submissions are scored in this competition, including how the metric is calculated and any specific considerations.",
"Metric Name": "The name of the metric which this competition use for scoring the submission."
"Metric Direction": True or False as True means bigger metric number is better, False means smaller is better.
"Longer time limit required": "True or False, whether the competition scenario requires a longer time limit to the code. Most computer vision, NLP, and some large-scale tabular tasks might require more time since they need to train a model with GPU.",
}
user: |-
Competition Description:
@@ -65,12 +65,6 @@ describe: # some template to describe some object
{% endif %}
{% endfor %}
{% endif %}
sota_results: |-
All validation scores of the SOTA experiments:
{% for index, data in sota_valid_score_dict.items() %}
## Trial index: {{ index }}, validation score: {{ data.score }}, validation score improvement: {{ data.improvement }}
{% endfor %}
scen: # customizable
role: |-
+5 -5
View File
@@ -29,14 +29,14 @@ class ConfUtils(unittest.TestCase):
assert QlibDockerConf().enable_cache is True
def test_ds_costeer_conf(self):
os.environ["DS_CODER_COSTEER_MAX_SECONDS"] = "1000"
os.environ["DS_CODER_COSTEER_MAX_SECONDS_MULTIPLIER"] = "1000"
coder_conf = DSCoderCoSTEERSettings()
runner_conf = DSRunnerCoSTEERSettings()
print(coder_conf.max_seconds)
print(runner_conf.max_seconds)
assert coder_conf.max_seconds == 1000
print(coder_conf.max_seconds_multiplier)
print(runner_conf.max_seconds_multiplier)
assert coder_conf.max_seconds_multiplier == 1000
# NOTE: coder's config should not affect runner's config
assert runner_conf.max_seconds == DS_RD_SETTING.full_timeout
assert runner_conf.max_seconds_multiplier == 1
os.environ["DS_RUNNER_COSTEER_MAX_SECONDS"] = "2000"
assert DSRunnerCoSTEERSettings().max_seconds == 2000