fix: add async to direct_exp_gen avoid infinite loop (#992)

* refactor: convert direct_exp_gen to async and enforce parallel limit

* fix bug

* change coroutine function position

* fix fin_quant's direct_exp_gen

* format with isort

---------

Co-authored-by: Bowen Xian <[email protected]>
Co-authored-by: SunsetWolf <[email protected]>

Author: 3 people

rdagent/app/qlib_rd_loop/factor_from_report.py

@@ -11,6 +11,7 @@
     extract_first_page_screenshot_from_pdf,
     load_and_process_pdfs_by_langchain,
 )
+from rdagent.core.conf import RD_AGENT_SETTINGS
 from rdagent.core.proposal import Hypothesis
 from rdagent.log import rdagent_logger as logger
 from rdagent.oai.llm_utils import APIBackend
@@ -105,21 +106,23 @@ def __init__(self, report_folder: str = None):
 
         self.loop_n = min(len(self.judge_pdf_data_items), FACTOR_FROM_REPORT_PROP_SETTING.report_limit)
 
-    def direct_exp_gen(self, prev_out: dict[str, Any]):
+    async def direct_exp_gen(self, prev_out: dict[str, Any]):
         while True:
-            report_file_path = self.judge_pdf_data_items[self.loop_idx]
-            logger.info(f"Processing number {self.loop_idx} report: {report_file_path}")
-            exp = extract_hypothesis_and_exp_from_reports(str(report_file_path))
-            if exp is None:
-                continue
-            exp.based_experiments = [QlibFactorExperiment(sub_tasks=[], hypothesis=exp.hypothesis)] + [
-                t[0] for t in self.trace.hist if t[1]
-            ]
-            exp.sub_workspace_list = exp.sub_workspace_list[: FACTOR_FROM_REPORT_PROP_SETTING.max_factors_per_exp]
-            exp.sub_tasks = exp.sub_tasks[: FACTOR_FROM_REPORT_PROP_SETTING.max_factors_per_exp]
-            logger.log_object(exp.hypothesis, tag="hypothesis generation")
-            logger.log_object(exp.sub_tasks, tag="experiment generation")
-            return exp
+            if self.get_unfinished_loop_cnt(self.loop_idx) < RD_AGENT_SETTINGS.get_max_parallel():
+                report_file_path = self.judge_pdf_data_items[self.loop_idx]
+                logger.info(f"Processing number {self.loop_idx} report: {report_file_path}")
+                exp = extract_hypothesis_and_exp_from_reports(str(report_file_path))
+                if exp is None:
+                    continue
+                exp.based_experiments = [QlibFactorExperiment(sub_tasks=[], hypothesis=exp.hypothesis)] + [
+                    t[0] for t in self.trace.hist if t[1]
+                ]
+                exp.sub_workspace_list = exp.sub_workspace_list[: FACTOR_FROM_REPORT_PROP_SETTING.max_factors_per_exp]
+                exp.sub_tasks = exp.sub_tasks[: FACTOR_FROM_REPORT_PROP_SETTING.max_factors_per_exp]
+                logger.log_object(exp.hypothesis, tag="hypothesis generation")
+                logger.log_object(exp.sub_tasks, tag="experiment generation")
+                return exp
+            await asyncio.sleep(1)
 
     def coding(self, prev_out: dict[str, Any]):
         exp = self.coder.develop(prev_out["direct_exp_gen"])

rdagent/app/qlib_rd_loop/quant.py

@@ -10,6 +10,7 @@
 from rdagent.app.qlib_rd_loop.conf import QUANT_PROP_SETTING
 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.developer import Developer
 from rdagent.core.exception import FactorEmptyError, ModelEmptyError
 from rdagent.core.proposal import (
@@ -64,15 +65,18 @@ def __init__(self, PROP_SETTING: BasePropSetting):
         self.trace = QuantTrace(scen=scen)
         super(RDLoop, self).__init__()
 
-    def direct_exp_gen(self, prev_out: dict[str, Any]):
-        hypo = self._propose()
-        assert hypo.action in ["factor", "model"]
-        if hypo.action == "factor":
-            exp = self.factor_hypothesis2experiment.convert(hypo, self.trace)
-        else:
-            exp = self.model_hypothesis2experiment.convert(hypo, self.trace)
-        logger.log_object(exp.sub_tasks, tag="experiment generation")
-        return {"propose": hypo, "exp_gen": exp}
+    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 == "factor":
+                    exp = self.factor_hypothesis2experiment.convert(hypo, self.trace)
+                else:
+                    exp = self.model_hypothesis2experiment.convert(hypo, self.trace)
+                logger.log_object(exp.sub_tasks, tag="experiment generation")
+                return {"propose": hypo, "exp_gen": exp}
+            await asyncio.sleep(1)
 
     def coding(self, prev_out: dict[str, Any]):
         if prev_out["direct_exp_gen"]["propose"].action == "factor":

rdagent/components/workflow/rd_loop.py

@@ -3,9 +3,11 @@
 It is from `rdagent/app/qlib_rd_loop/model.py` and try to replace `rdagent/app/qlib_rd_loop/RDAgent.py`
 """
 
+import asyncio
 from typing import Any
 
 from rdagent.components.workflow.conf import BasePropSetting
+from rdagent.core.conf import RD_AGENT_SETTINGS
 from rdagent.core.developer import Developer
 from rdagent.core.proposal import (
     Experiment2Feedback,
@@ -55,10 +57,13 @@ def _exp_gen(self, hypothesis: Hypothesis):
         return exp
 
     # included steps
-    def direct_exp_gen(self, prev_out: dict[str, Any]):
-        hypo = self._propose()
-        exp = self._exp_gen(hypo)
-        return {"propose": hypo, "exp_gen": exp}
+    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()
+                exp = self._exp_gen(hypo)
+                return {"propose": hypo, "exp_gen": exp}
+            await asyncio.sleep(1)
 
     def coding(self, prev_out: dict[str, Any]):
         exp = self.coder.develop(prev_out["direct_exp_gen"]["exp_gen"])

rdagent/scenarios/data_science/loop.py

@@ -156,9 +156,6 @@ async def direct_exp_gen(self, prev_out: dict[str, Any]):
         exp = await self.exp_gen.async_gen(self.trace, self)
 
         logger.log_object(exp)
-
-        # FIXME: this is for LLM debug webapp, remove this when the debugging is done.
-        logger.log_object(exp, tag="debug_exp_gen")
         return exp
 
     def coding(self, prev_out: dict[str, Any]):