fix(security): replace remaining assert statements with proper error handling

TPTBusiness · TPTBusiness · commit 3a1a3d5f77ba · 2026-05-01T13:49:58.000+02:00
Replaced 53 assert statements across 22 files with proper
if/raise patterns (TypeError, ValueError, AssertionError)
to resolve Bandit B101 alerts.
diff --git a/rdagent/app/data_science/conf.py b/rdagent/app/data_science/conf.py
@@ -201,6 +201,5 @@ class DataScienceBasePropSetting(KaggleBasePropSetting):
 DS_RD_SETTING = DataScienceBasePropSetting()
 
 # enable_cross_trace_diversity and llm_select_hypothesis should not be true at the same time
-assert not (
-    DS_RD_SETTING.enable_cross_trace_diversity and DS_RD_SETTING.llm_select_hypothesis
-), "enable_cross_trace_diversity and llm_select_hypothesis cannot be true at the same time"
+if DS_RD_SETTING.enable_cross_trace_diversity and DS_RD_SETTING.llm_select_hypothesis:
+    raise ValueError("enable_cross_trace_diversity and llm_select_hypothesis cannot be true at the same time")
diff --git a/rdagent/app/finetune/llm/loop.py b/rdagent/app/finetune/llm/loop.py
@@ -58,18 +58,18 @@ def main(
 
     if user_target_scenario:
         FT_RD_SETTING.user_target_scenario = user_target_scenario
-    assert (
-        FT_RD_SETTING.user_target_scenario is None
-    ), "user_target_scenario is not yet supported, please specify via benchmark and benchmark_description"
+    if FT_RD_SETTING.user_target_scenario is not None:
+        raise ValueError("user_target_scenario is not yet supported, please specify via benchmark and benchmark_description")
     if upper_data_size_limit:
         FT_RD_SETTING.upper_data_size_limit = upper_data_size_limit
         logger.info(f"Set upper_data_size_limit to {FT_RD_SETTING.upper_data_size_limit}")
     if benchmark and benchmark_description:
         FT_RD_SETTING.target_benchmark = benchmark
         FT_RD_SETTING.benchmark_description = benchmark_description
-    assert FT_RD_SETTING.user_target_scenario or (
-        FT_RD_SETTING.target_benchmark and FT_RD_SETTING.benchmark_description
-    ), "Either user_target_scenario or target_benchmark must be specified for LLM fine-tuning."
+    if not (
+        FT_RD_SETTING.user_target_scenario or (FT_RD_SETTING.target_benchmark and FT_RD_SETTING.benchmark_description)
+    ):
+        raise ValueError("Either user_target_scenario or target_benchmark must be specified for LLM fine-tuning.")
 
     # Update configuration with provided parameters
     if dataset:
@@ -82,9 +82,8 @@ def main(
     model_target = FT_RD_SETTING.base_model if FT_RD_SETTING.base_model else "auto selected model"
 
     # Temporary assertion until auto-selection is implemented
-    assert (
-        FT_RD_SETTING.base_model is not None
-    ), "Base model auto selection not yet supported, please specify via --base-model"
+    if FT_RD_SETTING.base_model is None:
+        raise ValueError("Base model auto selection not yet supported, please specify via --base-model")
 
     logger.info(f"Starting LLM fine-tuning on dataset='{data_set_target}' with model='{model_target}'")
 
diff --git a/rdagent/app/qlib_rd_loop/quant.py b/rdagent/app/qlib_rd_loop/quant.py
@@ -78,7 +78,8 @@ async def direct_exp_gen(self, prev_out: dict[str, Any]):
         while True:
             if self.get_unfinished_loop_cnt(self.loop_idx) < RD_AGENT_SETTINGS.get_max_parallel():
                 hypo = self._propose()
-                assert hypo.action in ["factor", "model"]
+                if hypo.action not in ["factor", "model"]:
+                    raise ValueError(f"hypo.action must be 'factor' or 'model', got {hypo.action!r}")
                 if hypo.action == "factor":
                     exp = self.factor_hypothesis2experiment.convert(hypo, self.trace)
                 else:
diff --git a/rdagent/components/coder/CoSTEER/__init__.py b/rdagent/components/coder/CoSTEER/__init__.py
@@ -75,8 +75,10 @@ def get_develop_max_seconds(self) -> int | None:
 
     def _get_last_fb(self) -> CoSTEERMultiFeedback:
         fb = self.evolve_agent.evolving_trace[-1].feedback
-        assert fb is not None, "feedback is None"
-        assert isinstance(fb, CoSTEERMultiFeedback), "feedback must be of type CoSTEERMultiFeedback"
+        if fb is None:
+            raise AssertionError("feedback is None")
+        if not isinstance(fb, CoSTEERMultiFeedback):
+            raise TypeError("feedback must be of type CoSTEERMultiFeedback")
         return fb
 
     def should_use_new_evo(self, base_fb: CoSTEERMultiFeedback | None, new_fb: CoSTEERMultiFeedback) -> bool:
@@ -121,7 +123,8 @@ def develop(self, exp: Experiment) -> Experiment:
 
         for evo_exp in self.evolve_agent.multistep_evolve(evo_exp, self.evaluator):
             iteration_count += 1
-            assert isinstance(evo_exp, Experiment)  # multiple inheritance
+            if not isinstance(evo_exp, Experiment):
+                raise TypeError("evo_exp must be an instance of Experiment")
             evo_fb = self._get_last_fb()
             update_fallback = self.should_use_new_evo(
                 base_fb=fallback_evo_fb,
@@ -154,7 +157,8 @@ def develop(self, exp: Experiment) -> Experiment:
                 evo_exp = fallback_evo_exp
                 evo_exp.recover_ws_ckp()
                 evo_fb = fallback_evo_fb
-            assert evo_fb is not None  # multistep_evolve should run at least once
+            if evo_fb is None:
+                raise AssertionError("multistep_evolve should run at least once")
             evo_exp = self._exp_postprocess_by_feedback(evo_exp, evo_fb)
         except CoderError as e:
             e.caused_by_timeout = reached_max_seconds
@@ -264,9 +268,12 @@ def _exp_postprocess_by_feedback(self, evo: Experiment, feedback: CoSTEERMultiFe
         - Raise Error if it failed to handle the develop task
         -
         """
-        assert isinstance(evo, Experiment)
-        assert isinstance(feedback, CoSTEERMultiFeedback)
-        assert len(evo.sub_workspace_list) == len(feedback)
+        if not isinstance(evo, Experiment):
+            raise TypeError("evo must be an instance of Experiment")
+        if not isinstance(feedback, CoSTEERMultiFeedback):
+            raise TypeError("feedback must be an instance of CoSTEERMultiFeedback")
+        if len(evo.sub_workspace_list) != len(feedback):
+            raise ValueError("Length of sub_workspace_list must match length of feedback")
 
         # FIXME: when whould the feedback be None?
         failed_feedbacks = [
diff --git a/rdagent/components/coder/CoSTEER/evolving_strategy.py b/rdagent/components/coder/CoSTEER/evolving_strategy.py
@@ -122,7 +122,8 @@ def evolve_iter(
         last_feedback = None
         if len(evolving_trace) > 0:
             last_feedback = evolving_trace[-1].feedback
-            assert isinstance(last_feedback, CoSTEERMultiFeedback)
+            if not isinstance(last_feedback, CoSTEERMultiFeedback):
+                raise TypeError("last_feedback must be of type CoSTEERMultiFeedback")
 
         # 1.找出需要evolve的task
         to_be_finished_task_index: list[int] = []
diff --git a/rdagent/components/coder/CoSTEER/knowledge_management.py b/rdagent/components/coder/CoSTEER/knowledge_management.py
@@ -1028,7 +1028,8 @@ def graph_query_by_intersection(
 
         """
         node_count = len(nodes)
-        assert node_count >= 2, "nodes length must >=2"
+        if node_count < 2:
+            raise ValueError("nodes length must >=2")
         intersection_node_list = []
         if output_intersection_origin:
             origin_list = []
diff --git a/rdagent/components/coder/model_coder/eva_utils.py b/rdagent/components/coder/model_coder/eva_utils.py
@@ -58,10 +58,12 @@ def evaluate(
         model_execution_feedback: str = "",
         model_value_feedback: str = "",
     ):
-        assert isinstance(target_task, ModelTask)
-        assert isinstance(implementation, ModelFBWorkspace)
-        if gt_implementation is not None:
-            assert isinstance(gt_implementation, ModelFBWorkspace)
+        if not isinstance(target_task, ModelTask):
+            raise TypeError("target_task must be of type ModelTask")
+        if not isinstance(implementation, ModelFBWorkspace):
+            raise TypeError("implementation must be of type ModelFBWorkspace")
+        if gt_implementation is not None and not isinstance(gt_implementation, ModelFBWorkspace):
+            raise TypeError("gt_implementation must be of type ModelFBWorkspace")
 
         model_task_information = target_task.get_task_information()
         code = implementation.all_codes
@@ -113,10 +115,12 @@ def evaluate(
         model_value_feedback: str,
         model_code_feedback: str,
     ):
-        assert isinstance(target_task, ModelTask)
-        assert isinstance(implementation, ModelFBWorkspace)
-        if gt_implementation is not None:
-            assert isinstance(gt_implementation, ModelFBWorkspace)
+        if not isinstance(target_task, ModelTask):
+            raise TypeError("target_task must be of type ModelTask")
+        if not isinstance(implementation, ModelFBWorkspace):
+            raise TypeError("implementation must be of type ModelFBWorkspace")
+        if gt_implementation is not None and not isinstance(gt_implementation, ModelFBWorkspace):
+            raise TypeError("gt_implementation must be of type ModelFBWorkspace")
 
         system_prompt = T(".prompts:evaluator_final_feedback.system").r(
             scenario=(
diff --git a/rdagent/components/document_reader/document_reader.py b/rdagent/components/document_reader/document_reader.py
@@ -85,13 +85,16 @@ def load_and_process_one_pdf_by_azure_document_intelligence(
 
 
 def load_and_process_pdfs_by_azure_document_intelligence(path: Path) -> dict[str, str]:
-    assert RD_AGENT_SETTINGS.azure_document_intelligence_key is not None
-    assert RD_AGENT_SETTINGS.azure_document_intelligence_endpoint is not None
+    if RD_AGENT_SETTINGS.azure_document_intelligence_key is None:
+        raise AssertionError("azure_document_intelligence_key must be set")
+    if RD_AGENT_SETTINGS.azure_document_intelligence_endpoint is None:
+        raise AssertionError("azure_document_intelligence_endpoint must be set")
 
     content_dict = {}
     ab_path = path.resolve()
     if ab_path.is_file():
-        assert ".pdf" in ab_path.suffixes, "The file must be a PDF file."
+        if ".pdf" not in ab_path.suffixes:
+            raise ValueError("The file must be a PDF file.")
         proc = load_and_process_one_pdf_by_azure_document_intelligence
         content_dict[str(ab_path)] = proc(
             ab_path,
diff --git a/rdagent/components/loader/task_loader.py b/rdagent/components/loader/task_loader.py
@@ -87,7 +87,8 @@ def __init__(self, path: Path) -> None:
         self.path = Path(path)
 
     def load(self, task: ModelTask) -> ModelFBWorkspace:
-        assert task.name is not None
+        if task.name is None:
+            raise AssertionError("task.name should not be None")
         mti = ModelFBWorkspace(task)
         mti.prepare()
         with open(self.path / f"{task.name}.py", "r") as f:
diff --git a/rdagent/oai/backend/base.py b/rdagent/oai/backend/base.py
@@ -541,7 +541,8 @@ def _try_create_chat_completion_or_embedding(  # type: ignore[no-untyped-def]
         **kwargs,
     ) -> str | list[list[float]]:
         """This function to share operation between embedding and chat completion"""
-        assert not (chat_completion and embedding), "chat_completion and embedding cannot be True at the same time"
+        if chat_completion and embedding:
+            raise ValueError("chat_completion and embedding cannot be True at the same time")
         max_retry = LLM_SETTINGS.max_retry if LLM_SETTINGS.max_retry is not None else max_retry
         timeout_count = 0
         violation_count = 0
diff --git a/rdagent/scenarios/data_science/debug/data.py b/rdagent/scenarios/data_science/debug/data.py
@@ -268,7 +268,8 @@ def reduce(self, data: dict) -> dict:
                 parent[key] = sampled  # type: ignore  # parent 是 list，key 是 index, list.__setitem__(key, sampled)
             self.sampled_files.extend([self.extract_filename(i) for i in sampled])
             break
-        assert len(self.sampled_files) > 0
+        if len(self.sampled_files) <= 0:
+            raise AssertionError("sampled_files must contain at least one file")
         return data
 
     def _find_all_lists(
diff --git a/rdagent/scenarios/data_science/example/source_data/arf-12-hours-prediction-task/prepare.py b/rdagent/scenarios/data_science/example/source_data/arf-12-hours-prediction-task/prepare.py
@@ -56,14 +56,17 @@
 sparse.save_npz(public / "train" / "X.npz", X_train)
 df_train.to_csv(public / "train" / "ARF_12h.csv", index=False)
 
-assert (
-    X_train.shape[0] == df_train.shape[0]
-), f"Mismatch: X_train rows ({X_train.shape[0]}) != df_train rows ({df_train.shape[0]})"
-assert (
-    X_test.shape[0] == df_test.shape[0]
-), f"Mismatch: X_test rows ({X_test.shape[0]}) != df_test rows ({df_test.shape[0]})"
-assert df_test.shape[1] == 2, "Public test set should have 2 columns"
-assert df_train.shape[1] == 3, "Public train set should have 3 columns"
-assert len(df_train) + len(df_test) == len(
-    df_label
-), "Length of new_train and new_test should equal length of old_train"
+if X_train.shape[0] != df_train.shape[0]:
+    raise ValueError(
+        f"Mismatch: X_train rows ({X_train.shape[0]}) != df_train rows ({df_train.shape[0]})"
+    )
+if X_test.shape[0] != df_test.shape[0]:
+    raise ValueError(
+        f"Mismatch: X_test rows ({X_test.shape[0]}) != df_test rows ({df_test.shape[0]})"
+    )
+if df_test.shape[1] != 2:
+    raise ValueError("Public test set should have 2 columns")
+if df_train.shape[1] != 3:
+    raise ValueError("Public train set should have 3 columns")
+if len(df_train) + len(df_test) != len(df_label):
+    raise ValueError("Length of new_train and new_test should equal length of old_train")
diff --git a/rdagent/scenarios/data_science/loop.py b/rdagent/scenarios/data_science/loop.py
@@ -320,7 +320,8 @@ def record(self, prev_out: dict[str, Any]):
             # only clean current workspace without affecting other loops.
             for k in "direct_exp_gen", "coding", "running":
                 if k in prev_out and prev_out[k] is not None:
-                    assert isinstance(prev_out[k], DSExperiment)
+                    if not isinstance(prev_out[k], DSExperiment):
+                        raise TypeError(f"prev_out[{k!r}] must be an instance of DSExperiment")
                     clean_workspace(prev_out[k].experiment_workspace.workspace_path)
 
             # Backup the workspace (only necessary files are included)
diff --git a/rdagent/scenarios/data_science/proposal/exp_gen/base.py b/rdagent/scenarios/data_science/proposal/exp_gen/base.py
@@ -213,7 +213,8 @@ def has_component(
         self, component: COMPONENT, search_list: list[tuple[DSExperiment, ExperimentFeedback]] = []
     ) -> bool:
         for exp, fb in search_list:
-            assert isinstance(exp.hypothesis, DSHypothesis), "Hypothesis should be DSHypothesis (and not None)"
+            if not isinstance(exp.hypothesis, DSHypothesis):
+                raise TypeError("Hypothesis should be DSHypothesis (and not None)")
             if exp.hypothesis.component == component and fb:
                 return True
         return False
diff --git a/rdagent/scenarios/data_science/proposal/exp_gen/merge.py b/rdagent/scenarios/data_science/proposal/exp_gen/merge.py
@@ -377,7 +377,8 @@ def gen(
             if DS_RD_SETTING.enable_multi_version_exp_gen:
                 exp_gen_version_list = DS_RD_SETTING.exp_gen_version_list.split(",")
                 for version in exp_gen_version_list:
-                    assert version in ["v3", "v2", "v1"]
+                    if version not in ["v3", "v2", "v1"]:
+                        raise ValueError(f"version must be 'v1', 'v2', or 'v3', got {version!r}")
 
                 if len(trace.hist) == 0:
                     # set the proposal version for the first sub-trace
diff --git a/rdagent/scenarios/data_science/proposal/exp_gen/proposal.py b/rdagent/scenarios/data_science/proposal/exp_gen/proposal.py
@@ -339,7 +339,8 @@ def gen(
             eda_output = sota_exp.experiment_workspace.file_dict.get("EDA.md", None)
         scenario_desc = trace.scen.get_scenario_all_desc(eda_output=eda_output)
 
-        assert sota_exp is not None, "SOTA experiment is not provided."
+        if sota_exp is None:
+            raise ValueError("SOTA experiment is not provided.")
         last_exp = trace.last_exp()
         # exp_and_feedback = trace.hist[-1]
         # last_exp = exp_and_feedback[0]
@@ -445,8 +446,10 @@ def _f(user_prompt):
                         json_target_type=dict[str, dict[str, str | dict] | str],
                     )
                 )
-                assert "hypothesis_proposal" in resp_dict, "Hypothesis proposal not provided."
-                assert "task_design" in resp_dict, "Task design not provided."
+                if "hypothesis_proposal" not in resp_dict:
+                    raise ValueError("Hypothesis proposal not provided.")
+                if "task_design" not in resp_dict:
+                    raise ValueError("Task design not provided.")
                 task_class = component_info["task_class"]
                 hypothesis_proposal = resp_dict.get("hypothesis_proposal", {})
                 hypothesis = DSHypothesis(
@@ -1149,8 +1152,10 @@ def hypothesis_select_with_llm(
         )
 
         response_dict = json.loads(response)
-        assert response_dict.get("component") in HypothesisComponent.__members__, f"Invalid component"
-        assert response_dict.get("hypothesis") is not None, f"Invalid hypothesis"
+        if response_dict.get("component") not in HypothesisComponent.__members__:
+            raise ValueError(f"Invalid component: {response_dict.get('component')}")
+        if response_dict.get("hypothesis") is None:
+            raise ValueError("Invalid hypothesis")
         return response_dict
 
     # END: for support llm-based hypothesis selection  -----
@@ -1253,7 +1258,8 @@ def task_gen(
             description=task_desc,
         )
 
-        assert isinstance(task, PipelineTask), f"Task {task_name} is not a PipelineTask, got {type(task)}"
+        if not isinstance(task, PipelineTask):
+            raise TypeError(f"Task {task_name} is not a PipelineTask, got {type(task)}")
         # only for llm with response schema.(TODO: support for non-schema llm?)
         # If the LLM provides a "packages" field (list[str]), compute runtime environment now and cache it for subsequent prompts in later loops.
         if isinstance(task_dict, dict) and "packages" in task_dict and isinstance(task_dict["packages"], list):
diff --git a/rdagent/scenarios/kaggle/experiment/scenario.py b/rdagent/scenarios/kaggle/experiment/scenario.py
@@ -165,7 +165,8 @@ def source_data(self) -> str:
         return data_info
 
     def output_format(self, tag=None) -> str:
-        assert tag in [None, "feature", "model"]
+        if tag not in [None, "feature", "model"]:
+            raise ValueError(f"tag must be None, 'feature', or 'model', got {tag!r}")
         feature_output_format = f"""The feature code should output following the format:
 {T(".prompts:kg_feature_output_format").r()}"""
         model_output_format = f"""The model code should output following the format:\n""" + T(
@@ -180,7 +181,8 @@ def output_format(self, tag=None) -> str:
             return model_output_format
 
     def interface(self, tag=None) -> str:
-        assert tag in [None, "feature", "XGBoost", "RandomForest", "LightGBM", "NN"]
+        if tag not in [None, "feature", "XGBoost", "RandomForest", "LightGBM", "NN"]:
+            raise ValueError(f"tag must be None, 'feature', 'XGBoost', 'RandomForest', 'LightGBM', or 'NN', got {tag!r}")
         feature_interface = f"""The feature code should follow the interface:
 {T(".prompts:kg_feature_interface").r()}"""
         if tag == "feature":
@@ -195,7 +197,8 @@ def interface(self, tag=None) -> str:
             return model_interface
 
     def simulator(self, tag=None) -> str:
-        assert tag in [None, "feature", "model"]
+        if tag not in [None, "feature", "model"]:
+            raise ValueError(f"tag must be None, 'feature', or 'model', got {tag!r}")
 
         kg_feature_simulator = (
             "The feature code will be sent to the simulator:\n" + T(".prompts:kg_feature_simulator").r()
diff --git a/rdagent/scenarios/kaggle/kaggle_crawler.py b/rdagent/scenarios/kaggle/kaggle_crawler.py
@@ -87,7 +87,12 @@ def kaggle_description_css_selectors() -> tuple[str, str]:
         content = e.get_attribute("innerHTML")
         contents.append(content)
 
-    assert len(subtitles) == len(contents) + 1 and subtitles[-1] == "Citation"
+    if not (len(subtitles) == len(contents) + 1 and subtitles[-1] == "Citation"):
+        raise AssertionError(
+            f"Expected len(contents)+1 == len(subtitles) and last subtitle == 'Citation', "
+            f"got len(subtitles)={len(subtitles)}, len(contents)={len(contents)}, "
+            f"last subtitle={subtitles[-1]!r}"
+        )
     for i in range(len(subtitles) - 1):
         descriptions[subtitles[i]] = contents[i]
 
diff --git a/rdagent/scenarios/kaggle/proposal/proposal.py b/rdagent/scenarios/kaggle/proposal/proposal.py
@@ -307,7 +307,8 @@ def convert_response(self, response: str) -> Hypothesis:
 class KGHypothesis2Experiment(FactorAndModelHypothesis2Experiment):
     def prepare_context(self, hypothesis: Hypothesis, trace: Trace) -> Tuple[dict, bool]:
         scenario = trace.scen.get_scenario_all_desc(filtered_tag="hypothesis_and_experiment")
-        assert isinstance(hypothesis, KGHypothesis)
+        if not isinstance(hypothesis, KGHypothesis):
+            raise TypeError("hypothesis must be an instance of KGHypothesis")
         experiment_output_format = (
             T("scenarios.kaggle.prompts:feature_experiment_output_format").r()
             if hypothesis.action in [KG_ACTION_FEATURE_ENGINEERING, KG_ACTION_FEATURE_PROCESSING]
diff --git a/rdagent/scenarios/qlib/experiment/quant_experiment.py b/rdagent/scenarios/qlib/experiment/quant_experiment.py
diff --git a/rdagent/utils/env.py b/rdagent/utils/env.py
diff --git a/rdagent/utils/workflow/tracking.py b/rdagent/utils/workflow/tracking.py
