chore: withdraw for policy error (microsoft#907)

* chore: withdraw for policy error

Co-authored-by: you-n-g <you-n-g@users.noreply.github.com>

* Apply suggestions from code review

Author: qew21

rdagent/core/exception.py

@@ -57,3 +57,9 @@ class KaggleError(Exception):
     """
     Exceptions raised when calling Kaggle API
     """
+
+
+class PolicyError(Exception):
+    """
+    Exceptions raised due to content management policy
+    """

rdagent/oai/backend/base.py

@@ -14,6 +14,7 @@
 import pytz
 from pydantic import TypeAdapter
 
+from rdagent.core.exception import PolicyError
 from rdagent.core.utils import LLM_CACHE_SEED_GEN, SingletonBaseClass
 from rdagent.log import LogColors
 from rdagent.log import rdagent_logger as logger
@@ -368,7 +369,7 @@ def _try_create_chat_completion_or_embedding(  # type: ignore[no-untyped-def]
                         violation_count += 1
                         if violation_count >= LLM_SETTINGS.violation_fail_limit:
                             logger.warning("Content policy violation detected.")
-                            raise e
+                            raise PolicyError(e)
 
                     if (
                         openai_imported

rdagent/scenarios/data_science/loop.py

@@ -21,7 +21,7 @@
 from rdagent.components.workflow.conf import BasePropSetting
 from rdagent.components.workflow.rd_loop import RDLoop
 from rdagent.core.conf import RD_AGENT_SETTINGS
-from rdagent.core.exception import CoderError, RunnerError
+from rdagent.core.exception import CoderError, PolicyError, RunnerError
 from rdagent.core.proposal import ExperimentFeedback, ExpGen
 from rdagent.core.scenario import Scenario
 from rdagent.core.utils import import_class
@@ -36,6 +36,7 @@
 class DataScienceRDLoop(RDLoop):
     # NOTE: we move the DataScienceRDLoop here to be easier to be imported
     skip_loop_error = (CoderError, RunnerError)
+    withdraw_loop_error = (PolicyError,)
 
     @staticmethod
     def _get_exp_gen(class_uri: str, scen: Scenario):

rdagent/utils/workflow.py

@@ -88,6 +88,9 @@ class LoopBase:
     loop_trace: dict[int, list[LoopTrace]]
 
     skip_loop_error: tuple[type[BaseException], ...] = ()  # you can define a list of error that will skip current loop
+    withdraw_loop_error: tuple[
+        type[BaseException], ...
+    ] = ()  # you can define a list of error that will withdraw current loop
 
     EXCEPTION_KEY = "_EXCEPTION"
 
@@ -183,6 +186,11 @@ def run(self, step_n: int | None = None, loop_n: int | None = None, all_duration
                             self.step_idx = len(self.steps) - 1  # directly jump to the last step.
                             self.loop_prev_out[self.EXCEPTION_KEY] = e
                             continue
+                        elif isinstance(e, self.withdraw_loop_error):
+                            logger.warning(f"Withdraw loop {li} due to {e}")
+                            # Back to previous loop
+                            self.step_backward(li - 1)
+                            continue
                         else:
                             raise
                     finally:
@@ -207,6 +215,27 @@ def run(self, step_n: int | None = None, loop_n: int | None = None, all_duration
 
                 self.dump(self.session_folder / f"{li}" / f"{si}_{name}")  # save a snapshot after the session
 
+    def step_backward(self, li: int) -> None:
+        prev_session_dir = self.session_folder / str(li)
+        prev_path = min(
+            (p for p in prev_session_dir.glob("*_*") if p.is_file()),
+            key=lambda item: int(item.name.split("_", 1)[0]),
+            default=None,
+        )
+        if prev_path:
+            loaded = type(self).load(
+                prev_path,
+                output_path=self.session_folder.parent,
+                do_truncate=False,
+                replace_timer=True,
+            )
+            logger.info(f"Load previous session from {prev_path}")
+            # Overwrite current instance state
+            self.__dict__ = loaded.__dict__
+        else:
+            logger.error(f"No previous dump found at {prev_session_dir}, cannot withdraw loop {li}")
+            raise
+
     def dump(self, path: str | Path) -> None:
         if RD_Agent_TIMER_wrapper.timer.started:
             RD_Agent_TIMER_wrapper.timer.update_remain_time()