From 5c5508c8ccb5ed6e967af6f4ba4499327d0bafa7 Mon Sep 17 00:00:00 2001 From: you-n-g Date: Thu, 26 Jun 2025 22:10:52 +0800 Subject: [PATCH] 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 Co-authored-by: SunsetWolf --- .../app/qlib_rd_loop/factor_from_report.py | 31 ++++++++++--------- rdagent/app/qlib_rd_loop/quant.py | 22 +++++++------ rdagent/components/workflow/rd_loop.py | 13 +++++--- rdagent/scenarios/data_science/loop.py | 3 -- 4 files changed, 39 insertions(+), 30 deletions(-) diff --git a/rdagent/app/qlib_rd_loop/factor_from_report.py b/rdagent/app/qlib_rd_loop/factor_from_report.py index 5f596986..de60190c 100644 --- a/rdagent/app/qlib_rd_loop/factor_from_report.py +++ b/rdagent/app/qlib_rd_loop/factor_from_report.py @@ -11,6 +11,7 @@ from rdagent.components.document_reader.document_reader import ( 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 @@ class FactorReportLoop(FactorRDLoop, metaclass=LoopMeta): 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"]) diff --git a/rdagent/app/qlib_rd_loop/quant.py b/rdagent/app/qlib_rd_loop/quant.py index c956c940..fddf470d 100644 --- a/rdagent/app/qlib_rd_loop/quant.py +++ b/rdagent/app/qlib_rd_loop/quant.py @@ -10,6 +10,7 @@ import fire 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 @@ class QuantRDLoop(RDLoop): 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": diff --git a/rdagent/components/workflow/rd_loop.py b/rdagent/components/workflow/rd_loop.py index 004ceb37..49d689f1 100644 --- a/rdagent/components/workflow/rd_loop.py +++ b/rdagent/components/workflow/rd_loop.py @@ -3,9 +3,11 @@ Model workflow with session control 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 @@ class RDLoop(LoopBase, metaclass=LoopMeta): 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"]) diff --git a/rdagent/scenarios/data_science/loop.py b/rdagent/scenarios/data_science/loop.py index 77bec4a4..306b4e0d 100644 --- a/rdagent/scenarios/data_science/loop.py +++ b/rdagent/scenarios/data_science/loop.py @@ -156,9 +156,6 @@ class DataScienceRDLoop(RDLoop): 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]):