From 805f337fc483467d3fa7c8270183195917db7d83 Mon Sep 17 00:00:00 2001 From: Linlang <30293408+SunsetWolf@users.noreply.github.com> Date: Fri, 6 Jun 2025 18:52:19 +0800 Subject: [PATCH] chore: add a rdagent server with UI & logger storage refinement(#553) * change_log_object * lint code * delete comments * change_log_object * change_log_object * fix import test error * update code * update code * fix bugs * skip mypy error * skip mypy error * skip mypy error * Start the flask server before running the demo. * achieve front and back interaction * fix github-advanced-security comments * fix github-advanced-security comments * tmp ignore * fix CI * move some logic * change format * adjust logic * log2json changes * tmp * fix * fix bug * refine log2json between 5 scenarios * fix * refine codes * fix logic * use localhost * add loop & all_duration param for old scenario startup * merge control logic * add README for server ui api * update README * reuse code in logger * add loop_n and all_duration param * fix upload * ui server now use port in setting * fix port setting * fix port setting * fix mypy check * refine logger and log storage * fix ruff error * fix CI * refine logger, loop, storage * bind one FileStorage with one logger * not truncate log storage * refine LoopBase.load(), use `checkout` instead of `output_path` and `do_truncate` * clear session folder when loading loop to run * move component info init step to ExpGen Class * Update rdagent/utils/workflow.py * move truncate_session function to LoopBase class * add checkout param for other scenarios * fix bug * move WebStorage to UI * change web_storage name * add randomname to requirements * add typer * fix requirements --------- Co-authored-by: WinstonLiyte <1957922024@qq.com> Co-authored-by: Bowen Xian Co-authored-by: you-n-g --- .gitignore | 4 + Makefile | 2 +- rdagent/app/cli.py | 8 + rdagent/app/data_mining/model.py | 6 +- rdagent/app/data_science/loop.py | 45 +-- rdagent/app/qlib_rd_loop/factor.py | 6 +- .../app/qlib_rd_loop/factor_from_report.py | 6 +- rdagent/app/qlib_rd_loop/model.py | 6 +- rdagent/app/qlib_rd_loop/quant.py | 6 +- rdagent/app/utils/ape.py | 4 +- rdagent/core/conf.py | 6 - rdagent/log/base.py | 16 +- rdagent/log/conf.py | 24 ++ rdagent/log/logger.py | 108 ++---- rdagent/log/mle_summary.py | 2 +- rdagent/log/server/README.md | 237 +++++++++++++ rdagent/log/server/app.py | 199 +++++++++++ rdagent/log/storage.py | 113 ++---- rdagent/log/ui/storage.py | 324 ++++++++++++++++++ rdagent/log/ui/utils.py | 2 +- rdagent/log/utils/__init__.py | 12 + rdagent/log/utils/folder.py | 2 +- rdagent/scenarios/data_science/loop.py | 7 +- .../data_science/proposal/exp_gen/proposal.py | 88 +++-- rdagent/utils/workflow.py | 78 +++-- requirements.txt | 4 + test/utils/test_import.py | 1 + 27 files changed, 1038 insertions(+), 278 deletions(-) create mode 100644 rdagent/log/conf.py create mode 100644 rdagent/log/server/README.md create mode 100644 rdagent/log/server/app.py create mode 100644 rdagent/log/ui/storage.py diff --git a/.gitignore b/.gitignore index 74500f0b..b6fa0225 100644 --- a/.gitignore +++ b/.gitignore @@ -175,3 +175,7 @@ mlruns/ /*.sh .aider* rdagent/app/benchmark/factor/example.json + +# UI Server resources +videos/ +static/ \ No newline at end of file diff --git a/Makefile b/Makefile index 45cbee6e..15c38391 100644 --- a/Makefile +++ b/Makefile @@ -92,7 +92,7 @@ isort: # First deal with the core folder, and then gradually increase the scope of detection, # and eventually realize the detection of the complete project. mypy: - $(PIPRUN) python -m mypy rdagent/core # --exclude rdagent/scripts,git_ignore_folder + $(PIPRUN) python -m mypy rdagent/core # Check lint with ruff. # First deal with the core folder, and then gradually increase the scope of detection, diff --git a/rdagent/app/cli.py b/rdagent/app/cli.py index ca58dfd1..5d6567a2 100644 --- a/rdagent/app/cli.py +++ b/rdagent/app/cli.py @@ -45,6 +45,13 @@ def ui(port=19899, log_dir="", debug=False): subprocess.run(cmds) +def server_ui(port=19899): + """ + start web app to show the log traces in real time + """ + subprocess.run(["python", "rdagent/log/server/app.py", f"--port={port}"]) + + def app(): fire.Fire( { @@ -58,5 +65,6 @@ def app(): "health_check": health_check, "collect_info": collect_info, "kaggle": kaggle, + "server_ui": server_ui, } ) diff --git a/rdagent/app/data_mining/model.py b/rdagent/app/data_mining/model.py index 8deaaf9c..d0ea970c 100644 --- a/rdagent/app/data_mining/model.py +++ b/rdagent/app/data_mining/model.py @@ -9,7 +9,7 @@ class ModelRDLoop(RDLoop): skip_loop_error = (ModelEmptyError,) -def main(path=None, step_n=None): +def main(path=None, step_n=None, loop_n=None, all_duration=None, checkout=True): """ Auto R&D Evolving loop for models in a medical scenario. @@ -23,8 +23,8 @@ def main(path=None, step_n=None): if path is None: model_loop = ModelRDLoop(MED_PROP_SETTING) else: - model_loop = ModelRDLoop.load(path) - model_loop.run(step_n=step_n) + model_loop = ModelRDLoop.load(path, checkout=checkout) + model_loop.run(step_n=step_n, loop_n=loop_n, all_duration=all_duration) if __name__ == "__main__": diff --git a/rdagent/app/data_science/loop.py b/rdagent/app/data_science/loop.py index d9f8d57f..18addef2 100644 --- a/rdagent/app/data_science/loop.py +++ b/rdagent/app/data_science/loop.py @@ -1,3 +1,5 @@ +from pathlib import Path + import fire from rdagent.app.data_science.conf import DS_RD_SETTING @@ -7,12 +9,11 @@ from rdagent.scenarios.data_science.loop import DataScienceRDLoop def main( - path=None, - output_path=None, - step_n=None, - loop_n=None, + path: str | None = None, + checkout: bool | str | Path = True, + step_n: int | None = None, + loop_n: int | None = None, competition="bms-molecular-translation", - do_truncate=True, timeout=None, replace_timer=True, exp_gen_cls: str | None = None, @@ -22,29 +23,33 @@ def main( Parameters ---------- path : - path like `$LOG_PATH/__session__/1/0_propose`. It indicates that we restore the state that after finish the step 0 in loop 1 - output_path : - path like `$LOG_PATH`. It indicates that where we want to save our session and log information. + A path like `$LOG_PATH/__session__/1/0_propose`. This indicates that we restore the state after finishing step 0 in loop 1. + checkout : + Used only when a path is provided. + Can be True, False, or a path. + Default is True. + - If True, the new loop will use the existing folder and clear logs for sessions after the one corresponding to the given path. + - If False, the new loop will use the existing folder but keep the logs for sessions after the one corresponding to the given path. + - If a path (or a str like Path) is provided, the new loop will be saved to that path, leaving the original path unchanged. step_n : - How many steps to run; if None, it will run forever until error or KeyboardInterrupt + Number of steps to run; if None, the process will run indefinitely until an error or KeyboardInterrupt occurs. loop_n : - How many loops to run; if None, it will run forever until error or KeyboardInterrupt - - if current loop is incomplete, it will be counted as the first loop for completion. - - if both step_n and loop_n are provided, the process will stop as soon as either condition is met. + Number of loops to run; if None, the process will run indefinitely until an error or KeyboardInterrupt occurs. + - If the current loop is incomplete, it will be counted as the first loop for completion. + - If both step_n and loop_n are provided, the process will stop as soon as either condition is met. competition : - do_truncate : - If set to True, the logger will truncate the future log messages by calling `logger.storage.truncate`. + Competition name. replace_timer : - If session is loaded, should we replace the timer with session.timer + If a session is loaded, determines whether to replace the timer with session.timer. exp_gen_cls : - When we have different stages, we can replace the exp_gen with the new proposal + When there are different stages, the exp_gen can be replaced with the new proposal. Auto R&D Evolving loop for models in a Kaggle scenario. - You can continue running session by + You can continue running a session by using the command: .. code-block:: bash - dotenv run -- python rdagent/app/data_science/loop.py [--competition titanic] $LOG_PATH/__session__/1/0_propose --step_n 1 # `step_n` is a optional parameter - rdagent kaggle --competition playground-series-s4e8 # You are encouraged to use this one. + dotenv run -- python rdagent/app/data_science/loop.py [--competition titanic] $LOG_PATH/__session__/1/0_propose --step_n 1 # `step_n` is an optional parameter + rdagent kaggle --competition playground-series-s4e8 # This command is recommended. """ if competition is not None: DS_RD_SETTING.competition = competition @@ -55,7 +60,7 @@ def main( if path is None: kaggle_loop = DataScienceRDLoop(DS_RD_SETTING) else: - kaggle_loop = DataScienceRDLoop.load(path, output_path, do_truncate, replace_timer) + kaggle_loop: DataScienceRDLoop = DataScienceRDLoop.load(path, checkout=checkout, replace_timer=replace_timer) # replace exp_gen if we have new class if exp_gen_cls is not None: diff --git a/rdagent/app/qlib_rd_loop/factor.py b/rdagent/app/qlib_rd_loop/factor.py index 8379b0f0..5c08a091 100755 --- a/rdagent/app/qlib_rd_loop/factor.py +++ b/rdagent/app/qlib_rd_loop/factor.py @@ -25,7 +25,7 @@ class FactorRDLoop(RDLoop): return exp -def main(path=None, step_n=None): +def main(path=None, step_n=None, loop_n=None, all_duration=None, checkout=True): """ Auto R&D Evolving loop for fintech factors. @@ -39,8 +39,8 @@ def main(path=None, step_n=None): if path is None: model_loop = FactorRDLoop(FACTOR_PROP_SETTING) else: - model_loop = FactorRDLoop.load(path) - model_loop.run(step_n=step_n) + model_loop = FactorRDLoop.load(path, checkout=checkout) + model_loop.run(step_n=step_n, loop_n=loop_n, all_duration=all_duration) if __name__ == "__main__": diff --git a/rdagent/app/qlib_rd_loop/factor_from_report.py b/rdagent/app/qlib_rd_loop/factor_from_report.py index 72e017f6..2cff1ca7 100644 --- a/rdagent/app/qlib_rd_loop/factor_from_report.py +++ b/rdagent/app/qlib_rd_loop/factor_from_report.py @@ -146,7 +146,7 @@ class FactorReportLoop(FactorRDLoop, metaclass=LoopMeta): return exp -def main(report_folder=None, path=None, step_n=None): +def main(report_folder=None, path=None, step_n=None, loop_n=None, all_duration=None, checkout=True): """ Auto R&D Evolving loop for fintech factors (the factors are extracted from finance reports). @@ -158,11 +158,11 @@ def main(report_folder=None, path=None, step_n=None): if path is None and report_folder is None: model_loop = FactorReportLoop() elif path is not None: - model_loop = FactorReportLoop.load(path) + model_loop = FactorReportLoop.load(path, checkout=checkout) else: model_loop = FactorReportLoop(report_folder=report_folder) - model_loop.run(step_n=step_n) + model_loop.run(step_n=step_n, loop_n=loop_n, all_duration=all_duration) if __name__ == "__main__": diff --git a/rdagent/app/qlib_rd_loop/model.py b/rdagent/app/qlib_rd_loop/model.py index 7a6dba56..40183360 100644 --- a/rdagent/app/qlib_rd_loop/model.py +++ b/rdagent/app/qlib_rd_loop/model.py @@ -13,7 +13,7 @@ class ModelRDLoop(RDLoop): skip_loop_error = (ModelEmptyError,) -def main(path=None, step_n=None): +def main(path=None, step_n=None, loop_n=None, all_duration=None, checkout=True): """ Auto R&D Evolving loop for fintech models @@ -27,8 +27,8 @@ def main(path=None, step_n=None): if path is None: model_loop = ModelRDLoop(MODEL_PROP_SETTING) else: - model_loop = ModelRDLoop.load(path) - model_loop.run(step_n=step_n) + model_loop = ModelRDLoop.load(path, checkout=checkout) + model_loop.run(step_n=step_n, loop_n=loop_n, all_duration=all_duration) if __name__ == "__main__": diff --git a/rdagent/app/qlib_rd_loop/quant.py b/rdagent/app/qlib_rd_loop/quant.py index 295446f6..c7778ed4 100644 --- a/rdagent/app/qlib_rd_loop/quant.py +++ b/rdagent/app/qlib_rd_loop/quant.py @@ -119,7 +119,7 @@ class QuantRDLoop(RDLoop): self.trace.hist.append((prev_out["running"], feedback)) -def main(path=None, step_n=None): +def main(path=None, step_n=None, loop_n=None, all_duration=None, checkout=True): """ Auto R&D Evolving loop for fintech factors. You can continue running session by @@ -129,8 +129,8 @@ def main(path=None, step_n=None): if path is None: quant_loop = QuantRDLoop(QUANT_PROP_SETTING) else: - quant_loop = QuantRDLoop.load(path) - quant_loop.run(step_n=step_n) + quant_loop = QuantRDLoop.load(path, checkout=checkout) + quant_loop.run(step_n=step_n, loop_n=loop_n, all_duration=all_duration) if __name__ == "__main__": diff --git a/rdagent/app/utils/ape.py b/rdagent/app/utils/ape.py index 474e9a2d..6511fdbb 100644 --- a/rdagent/app/utils/ape.py +++ b/rdagent/app/utils/ape.py @@ -5,7 +5,7 @@ This is the preliminary version of the APE (Automated Prompt Engineering) import pickle from pathlib import Path -from rdagent.core.conf import RD_AGENT_SETTINGS +from rdagent.log.conf import LOG_SETTINGS def get_llm_qa(file_path): @@ -21,7 +21,7 @@ def get_llm_qa(file_path): # Example usage # use -file_path = Path(RD_AGENT_SETTINGS.log_trace_path) / "debug_llm.pkl" +file_path = Path(LOG_SETTINGS.trace_path) / "debug_llm.pkl" llm_qa = get_llm_qa(file_path) print(len(llm_qa)) diff --git a/rdagent/core/conf.py b/rdagent/core/conf.py index 82e69cb7..417b0d24 100644 --- a/rdagent/core/conf.py +++ b/rdagent/core/conf.py @@ -1,6 +1,5 @@ from __future__ import annotations -# TODO: use pydantic for other modules in Qlib from pathlib import Path from typing import cast @@ -45,11 +44,6 @@ class ExtendedBaseSettings(BaseSettings): class RDAgentSettings(ExtendedBaseSettings): - # TODO: (xiao) I think LLMSetting may be a better name. - # TODO: (xiao) I think most of the config should be in oai.config - # Log configs - # TODO: (xiao) think it can be a separate config. - log_trace_path: str | None = None # azure document intelligence configs azure_document_intelligence_key: str = "" diff --git a/rdagent/log/base.py b/rdagent/log/base.py index 7ef274d6..7e0b96d8 100644 --- a/rdagent/log/base.py +++ b/rdagent/log/base.py @@ -5,7 +5,7 @@ from collections.abc import Generator from dataclasses import dataclass from datetime import datetime from pathlib import Path -from typing import Literal, Optional, Union +from typing import Literal, Optional @dataclass @@ -43,10 +43,8 @@ class Storage: def log( self, obj: object, - name: str = "", - save_type: Literal["json", "text", "pkl"] = "text", + tag: str = "", timestamp: datetime | None = None, - **kwargs: dict, ) -> str | Path: """ @@ -72,6 +70,16 @@ class Storage: """ ... + @abstractmethod + def truncate(self, time: datetime) -> None: + """ + Remove all log entries after the specified time. + """ + ... + + def __str__(self) -> str: + return self.__class__.__name__ + class View: """ diff --git a/rdagent/log/conf.py b/rdagent/log/conf.py new file mode 100644 index 00000000..72d0edf7 --- /dev/null +++ b/rdagent/log/conf.py @@ -0,0 +1,24 @@ +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from pydantic_settings import SettingsConfigDict + +from rdagent.core.conf import ExtendedBaseSettings + + +class LogSettings(ExtendedBaseSettings): + model_config = SettingsConfigDict(env_prefix="LOG_", protected_namespaces=()) + + trace_path: str = str(Path.cwd() / "log" / datetime.now(timezone.utc).strftime("%Y-%m-%d_%H-%M-%S-%f")) + + ui_server_port: int | None = None + + storages: dict[str, list[int | str]] = {} + + def model_post_init(self, _context: Any, /) -> None: + if self.ui_server_port is not None: + self.storages["rdagent.log.ui.storage.WebStorage"] = [self.ui_server_port, self.trace_path] + + +LOG_SETTINGS = LogSettings() diff --git a/rdagent/log/logger.py b/rdagent/log/logger.py index 3290a353..a8e9deca 100644 --- a/rdagent/log/logger.py +++ b/rdagent/log/logger.py @@ -1,28 +1,19 @@ -import json import os -import pickle import sys from contextlib import contextmanager -from datetime import datetime, timezone -from functools import partial -from logging import LogRecord -from multiprocessing import Pipe -from multiprocessing.connection import Connection +from datetime import datetime from pathlib import Path -from typing import TYPE_CHECKING, Any, Dict, Generator, Union +from typing import Generator from loguru import logger - -if TYPE_CHECKING: - from loguru import Record - from psutil import Process -from rdagent.core.conf import RD_AGENT_SETTINGS -from rdagent.core.utils import SingletonBaseClass +from rdagent.core.utils import SingletonBaseClass, import_class +from .base import Storage +from .conf import LOG_SETTINGS from .storage import FileStorage -from .utils import LogColors, get_caller_info +from .utils import get_caller_info class RDAgentLog(SingletonBaseClass): @@ -57,23 +48,15 @@ class RDAgentLog(SingletonBaseClass): # feedback = logger.get_reps() _tag: str = "" - def __init__(self, log_trace_path: Union[str, None] = RD_AGENT_SETTINGS.log_trace_path) -> None: - if log_trace_path is None: - timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d_%H-%M-%S-%f") - self.log_trace_path = Path.cwd() / "log" / timestamp - else: - self.log_trace_path = Path(log_trace_path) - - self.log_trace_path.mkdir(parents=True, exist_ok=True) - - self.storage = FileStorage(self.log_trace_path) + def __init__(self) -> None: + self.storage = FileStorage(LOG_SETTINGS.trace_path) + self.other_storages: list[Storage] = [] + for storage, args in LOG_SETTINGS.storages.items(): + storage_cls = import_class(storage) + self.other_storages.append(storage_cls(*args)) self.main_pid = os.getpid() - def set_trace_path(self, log_trace_path: str | Path) -> None: - self.log_trace_path = Path(log_trace_path) - self.storage = FileStorage(log_trace_path) - @contextmanager def tag(self, tag: str) -> Generator[None, None, None]: if tag.strip() == "": @@ -88,6 +71,15 @@ class RDAgentLog(SingletonBaseClass): finally: self._tag = self._tag[: -len(tag)] + def set_storages_path(self, path: str | Path) -> None: + for storage in [self.storage] + self.other_storages: + if hasattr(storage, "path"): + storage.path = path + + def truncate_storages(self, time: datetime) -> None: + for storage in [self.storage] + self.other_storages: + storage.truncate(time=time) + def get_pids(self) -> str: """ Returns a string of pids from the current process to the main process. @@ -103,65 +95,35 @@ class RDAgentLog(SingletonBaseClass): process = parent_process return pid_chain - def file_format(self, record: "Record", raw: bool = False) -> str: - # FIXME: the formmat is tightly coupled with the message reading in storage. - record["message"] = LogColors.remove_ansi_codes(record["message"]) - if raw: - return "{message}" - return "{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {name}:{function}:{line} - {message}\n" - def log_object(self, obj: object, *, tag: str = "") -> None: # TODO: I think we can merge the log_object function with other normal log methods to make the interface simpler. caller_info = get_caller_info() tag = f"{self._tag}.{tag}.{self.get_pids()}".strip(".") - logp = self.storage.log(obj, name=tag, save_type="pkl") - file_handler_id = logger.add( - self.log_trace_path / tag.replace(".", "/") / "common_logs.log", format=self.file_format - ) - logger.patch(lambda r: r.update(caller_info)).info(f"Logging object in {Path(logp).absolute()}") - logger.remove(file_handler_id) + for storage in [self.storage] + self.other_storages: + logp = storage.log(obj, tag=tag) + logger.patch(lambda r: r.update(caller_info)).info(f"Log object to [{storage}], uri: {logp}") - def info(self, msg: str, *, tag: str = "", raw: bool = False) -> None: - # TODO: too much duplicated. due to we have no logger with stream context; + def _log(self, level: str, msg: str, *, tag: str = "", raw: bool = False) -> None: caller_info = get_caller_info() + tag = f"{self._tag}.{tag}.{self.get_pids()}".strip(".") + if raw: logger.remove() logger.add(sys.stderr, format=lambda r: "{message}") - tag = f"{self._tag}.{tag}.{self.get_pids()}".strip(".") - log_file_path = self.log_trace_path / tag.replace(".", "/") / "common_logs.log" - if raw: - file_handler_id = logger.add(log_file_path, format=partial(self.file_format, raw=True)) - else: - file_handler_id = logger.add(log_file_path, format=self.file_format) - - logger.patch(lambda r: r.update(caller_info)).info(msg) - logger.remove(file_handler_id) + log_func = getattr(logger.patch(lambda r: r.update(caller_info)), level) + log_func(msg) if raw: logger.remove() logger.add(sys.stderr) - def warning(self, msg: str, *, tag: str = "") -> None: - # TODO: reuse code - # _log(self, msg: str, *, tag: str = "", level=Literal["warning", "error", ..]) -> None: - # getattr(logger.patch(lambda r: r.update(caller_info)), level)(msg) - caller_info = get_caller_info() + def info(self, msg: str, *, tag: str = "", raw: bool = False) -> None: + self._log("info", msg, tag=tag, raw=raw) - tag = f"{self._tag}.{tag}.{self.get_pids()}".strip(".") - file_handler_id = logger.add( - self.log_trace_path / tag.replace(".", "/") / "common_logs.log", format=self.file_format - ) - logger.patch(lambda r: r.update(caller_info)).warning(msg) - logger.remove(file_handler_id) + def warning(self, msg: str, *, tag: str = "", raw: bool = False) -> None: + self._log("warning", msg, tag=tag, raw=raw) - def error(self, msg: str, *, tag: str = "") -> None: - caller_info = get_caller_info() - - tag = f"{self._tag}.{tag}.{self.get_pids()}".strip(".") - file_handler_id = logger.add( - self.log_trace_path / tag.replace(".", "/") / "common_logs.log", format=self.file_format - ) - logger.patch(lambda r: r.update(caller_info)).error(msg) - logger.remove(file_handler_id) + def error(self, msg: str, *, tag: str = "", raw: bool = False) -> None: + self._log("error", msg, tag=tag, raw=raw) diff --git a/rdagent/log/mle_summary.py b/rdagent/log/mle_summary.py index 2f75e3d0..97aced02 100644 --- a/rdagent/log/mle_summary.py +++ b/rdagent/log/mle_summary.py @@ -37,7 +37,7 @@ def save_grade_info(log_trace_path: Path): try: mle_score_str = test_eval.eval(competition, msg.content.experiment_workspace) trace_storage.log( - mle_score_str, name=f"{msg.tag}.mle_score.pid", save_type="pkl", timestamp=msg.timestamp + mle_score_str, tag=f"{msg.tag}.mle_score.pid", save_type="pkl", timestamp=msg.timestamp ) except Exception as e: print(f"Error in {log_trace_path}: {e}") diff --git a/rdagent/log/server/README.md b/rdagent/log/server/README.md new file mode 100644 index 00000000..dc76e16d --- /dev/null +++ b/rdagent/log/server/README.md @@ -0,0 +1,237 @@ +# API + +## A. Controls + +### 1. /upload [POST] + +#### Request + +- "scenario": one of six values + 1. "Finance Data Building" + 2. "Finance Data Building (Reports)" + 3. "Finance Model Implementation" + 4. "General Model Implementation" + 5. "Medical Model Implementation" + 6. "Data Science" +- "files": **2** scenarios need this + 1. in "Finance Data Building (Reports)" Scenario, one or more pdf files. + 2. in "General Model Implementation" Scenario, one pdf file or one pdf link like `https://arxiv.org/pdf/2210.09789` +- "competition": **Data Science** Scenario need this, one of 75 competitions. +- "loops": Number of loops after which RD-Agent will automatically stop (optional; if not set, it will not stop automatically and must be stopped manually). +- "all_duration": Total duration (in hours) for which the RD-Agent should run before stopping automatically. If not set, the agent will continue running until stopped manually or by the "loops" parameter. + +#### Response + +- "id": a unique identifier string, such as `/home/rdagent_log/data_science/competition_A/trace_1` or `/home/rdagent_log/finance/trace_1`, used to mark the series of logs generated by this RD-Agent run. + +### 2. /control [POST] + +#### Request + +- "id": identifier +- "action": one of three values + 1. "pause" + 2. "resume" + 3. "stop" + +#### Response + +- "status": "success" / "error: ..." + +### 3. /trace [POST] + +Returns the sequence of Messages generated for the current id on the backend that **have not yet been returned to the frontend**. + +#### Request + +- "id": identifier +- "all": True / False. True means all Messages not yet provided to the frontend will be returned; False returns a random 1 to 10 Messages. In most cases, this should be True. +- "reset": True / False. Reset means the pointer for "not yet returned to the frontend" will be set back to the first Message generated for this id, i.e., return from the beginning. In most cases, this should be False. + +#### Response + +- a list of [Messages](#b-messages) + +## B. Messages + +### Research + +Only **2** Message in one loop + +1. hypothesis + +```json +{ + "tag": "research.hypothesis", + "timestamp": "", + "loop_id": "1", + "content": { + "hypothesis": "...", + "reason": "...", + "component": "...", // only exists in Data Science Scenario + "concise_reason": "...", + "concise_justification": "...", + "concise_observation": "...", + "concise_knowledge": "...", + } +} +``` + +2. tasks + +```json +{ + "tag": "research.tasks", + "timestamp": "", + "loop_id": "1", + "content": [ // list of tasks + { + "name": "...", + "description": "...", + "model_type": "...", // only exists in "Finance Model Implementation", "General Model Implementation", "Medical Model Implementation", or some tasks of "Data Science" + "architecture": "...", // same as above + "hyperparameters": "...", // same as above + }, + { + + } + //... same as above + ] +} +``` + +### evolving + +- 1 to 10 pairs of Messages (codes & feedbacks), each identified by an "evo_id" indicating the evolving round. +- In the **Data Science** scenario, each evolving round contains only **one task**, but the "codes" for that task may include **multiple code files**. +- In other scenarios, each evolving round may contain **multiple tasks**, but each task's "codes" will include only **one code file**. + +1. codes + +```json +{ + "tag": "evolving.codes", + "timestamp": "", + "loop_id": "1", + "evo_id": "0", + "content": [ // list of task_name & codes + { + "target_task_name": "task_1", + "codes": { // one or more codes + "a.py": "...", + "b.py": "...", + //... + } + }, + { + "target_task_name": "task_2", + "codes": { + "a.py": "...", + //... + } + } + //... same as above + ] +} + +{ + "tag": "evolving.codes", + "timestamp": "", + "loop_id": "1", + "evo_id": "1", + "content": [ + //... same as above + ] +} +``` + +2. feedbacks + +```json +{ + "tag": "evolving.feedbacks", + "timestamp": "", + "loop_id": "1", + "evo_id": "0", + "content": [ // list of feedbacks + { + "final_decision": "True", // True or False + "execution": "...", + "code": "...", + "return_checking": "..." + }, + //... same as above + ] +} + +{ + "tag": "evolving.codes", + "timestamp": "", + "loop_id": "1", + "evo_id": "1", + "content": [ + //... same as above + ] +} +``` + +### feedback + +Each tag below appears only once per loop. + +1. config (only exists in "Finance Data Building"/"Finance Data Building (Reports)"/"Finance Model Implementation") + +```json +{ + "tag": "feedback.config", + "timestamp": "", + "loop_id": "1", + "content": { + "config": "a markdown string", + } +} +``` + +2. return_chart (only exists in "Finance Data Building"/"Finance Data Building (Reports)"/"Finance Model Implementation") + +```json +{ + "tag": "feedback.return_chart", + "timestamp": "", + "loop_id": "1", + "content": { + "chart_html": "chart html codes string", + } +} +``` + +3. metric + +```json +{ + "tag": "feedback.metric", + "timestamp": "", + "loop_id": "1", + "content": { + "result": "{ \"\": , ... }" // A JSON string containing metric names and their corresponding values. + } +} +``` + +4. hypothesis_feedback + +```json +{ + "tag": "feedback.hypothesis_feedback", + "timestamp": "", + "loop_id": "1", + "content": { + "decision": "True", + "reason": "...", + "exception": "...", + "observations": "...", // may not exists + "hypothesis_evaluation": "...", // may not exists + "new_hypothesis": "...", // may not exists + } +} +``` \ No newline at end of file diff --git a/rdagent/log/server/app.py b/rdagent/log/server/app.py new file mode 100644 index 00000000..78ad2367 --- /dev/null +++ b/rdagent/log/server/app.py @@ -0,0 +1,199 @@ +import os +import random +import signal +import subprocess +import sys +from collections import defaultdict +from datetime import datetime, timezone +from pathlib import Path + +import randomname +import typer +from flask import Flask, jsonify, request, send_from_directory +from flask_cors import CORS + +msgs_for_frontend = defaultdict(list) + +app = Flask(__name__, static_folder="./docs/_static") +CORS(app) + +rdagent_processes = defaultdict() +server_port = 19899 + + +@app.route("/favicon.ico") +def favicon(): + return send_from_directory("./docs/_static", "favicon.ico", mimetype="image/vnd.microsoft.icon") + + +pointers = {id: 0 for id in msgs_for_frontend.keys()} + + +@app.route("/trace", methods=["POST"]) +def update_trace(): + data = request.get_json() + trace_id = data.get("id") + return_all = data.get("all") + reset = data.get("reset") + msg_num = random.randint(1, 10) + + if reset: + pointers[trace_id] = 0 + + end_pointer = pointers[trace_id] + msg_num + if end_pointer > len(msgs_for_frontend[trace_id]) or return_all: + end_pointer = len(msgs_for_frontend[trace_id]) + + print(f"trace_id: {trace_id}, start_pointer: {pointers[trace_id]}, end_pointer: {end_pointer}") + returned_msgs = msgs_for_frontend[trace_id][pointers[trace_id] : end_pointer] + + pointers[trace_id] = end_pointer + return jsonify(returned_msgs), 200 + + +@app.route("/upload", methods=["GET"]) +def upload_file(): + # 获取请求体中的字段 + global rdagent_processes, server_port + scenario = request.form.get("scenario") + files = request.files.getlist("files") + competition = request.form.get("competition") + loop_n = request.form.get("loops") + all_duration = request.form.get("all_duration") + + # scenario = "Data Science Loop" + trace_name = randomname.get_name() + log_folder_path = Path("./RD-Agent_server_trace").absolute() + log_trace_path = (log_folder_path / scenario / trace_name).absolute() + trace_files_path = log_folder_path / scenario / "uploads" / trace_name + + # save files + for file in files: + if file: + p = log_folder_path / scenario / "uploads" / trace_name + if not p.exists(): + p.mkdir(parents=True, exist_ok=True) + file.save(p / file.filename) + + if scenario == "Finance Data Building": + cmds = ["rdagent", "fin_factor"] + if scenario == "Finance Data Building (Reports)": + cmds = ["rdagent", "fin_factor_report", "--report_folder", str(trace_files_path)] + if scenario == "Finance Model Implementation": + cmds = ["rdagent", "fin_model"] + if scenario == "General Model Implementation": + if len(files) == 0: # files is one link + rfp = request.form.get("files")[0] + else: # one file is uploaded + rfp = str(trace_files_path / files[0].filename) + cmds = ["rdagent", "general_model", "--report_file_path", rfp] + if scenario == "Medical Model Implementation": + cmds = ["rdagent", "med_model"] + if scenario == "Data Science Loop": + cmds = ["rdagent", "kaggle", "--competition", competition] + + # time control parameters + if loop_n: + cmds += ["--loop_n", loop_n] + if all_duration: + cmds += ["--all_duration", all_duration] + + rdagent_processes[str(log_trace_path)] = subprocess.Popen( + cmds, + # stdout=subprocess.PIPE, + # stderr=subprocess.PIPE, + stdout=sys.stdout, + stderr=sys.stderr, + env={ + "LOG_TRACE_PATH": str(log_trace_path), + "UI_SERVER_PORT": server_port, + }, + ) + + return ( + jsonify( + { + "id": str(log_trace_path), + } + ), + 200, + ) + + +@app.route("/receive", methods=["POST"]) +def receive_msgs(): + try: + data = request.get_json() + if not data: + return jsonify({"error": "No JSON data received"}), 400 + except Exception as e: + return jsonify({"error": "Internal Server Error"}), 500 + + if isinstance(data, list): + for d in data: + msgs_for_frontend[d["id"]].append(d["msg"]) + else: + msgs_for_frontend[data["id"]].append(data["msg"]) + + return jsonify({"status": "success"}), 200 + + +@app.route("/control", methods=["POST"]) +def control_process(): + global rdagent_processes + data = request.get_json() + if not data or "id" not in data or "action" not in data: + return jsonify({"error": "Missing 'id' or 'action' in request"}), 400 + + id = data["id"] + action = data["action"] + + if id not in rdagent_processes or rdagent_processes[id] is None: + return jsonify({"error": "No running process for given id"}), 400 + + process = rdagent_processes[id] + + if process.poll() is not None: + msgs_for_frontend[id].append({"tag": "END", "timestamp": datetime.now(timezone.utc).isoformat(), "content": {}}) + return jsonify({"error": "Process has already terminated"}), 400 + + try: + if action == "pause": + os.kill(process.pid, signal.SIGSTOP) + return jsonify({"status": "paused"}), 200 + elif action == "resume": + os.kill(process.pid, signal.SIGCONT) + return jsonify({"status": "resumed"}), 200 + elif action == "stop": + process.terminate() + process.wait() + del rdagent_processes[id] + msgs_for_frontend[id].append( + {"tag": "END", "timestamp": datetime.now(timezone.utc).isoformat(), "content": {}} + ) + return jsonify({"status": "stopped"}), 200 + else: + return jsonify({"error": "Unknown action"}), 400 + except Exception as e: + return jsonify({"error": f"Failed to {action} process"}), 500 + + +@app.route("/", methods=["GET"]) +def index(): + # return 'Hello, World!' + return msgs_for_frontend + + +@app.route("/", methods=["GET"]) +def server_static_files(fn): + return send_from_directory(app.static_folder, fn) + + +def main(port: int = 19899): + global server_port + server_port = port + app.run(debug=False, host="0.0.0.0", port=port) + + +if __name__ == "__main__": + typer.run(main) diff --git a/rdagent/log/storage.py b/rdagent/log/storage.py index cfc5ff22..393e9056 100644 --- a/rdagent/log/storage.py +++ b/rdagent/log/storage.py @@ -3,13 +3,28 @@ import pickle import re from datetime import datetime, timezone from pathlib import Path -from typing import Any, Generator, Literal, Union, cast +from typing import Any, Generator, Literal from .base import Message, Storage +from .utils import gen_datetime LOG_LEVEL = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] +def _remove_empty_dir(path: Path) -> None: + """ + Recursively remove empty directories. + This function will remove the directory if it is empty after removing its subdirectories. + """ + if path.is_dir(): + sub_dirs = [sub for sub in path.iterdir() if sub.is_dir()] + for sub in sub_dirs: + _remove_empty_dir(sub) + + if not any(path.iterdir()): + path.rmdir() + + class FileStorage(Storage): """ The info are logginged to the file systems @@ -17,25 +32,21 @@ class FileStorage(Storage): TODO: describe the storage format """ - def __init__(self, path: str | Path = "./log/") -> None: + def __init__(self, path: str | Path) -> None: self.path = Path(path) - self.path.mkdir(parents=True, exist_ok=True) def log( self, obj: object, - name: str = "", - save_type: Literal["json", "text", "pkl"] = "text", + tag: str = "", timestamp: datetime | None = None, + save_type: Literal["json", "text", "pkl"] = "pkl", **kwargs: Any, - ) -> Union[str, Path]: + ) -> str | Path: # TODO: We can remove the timestamp after we implement PipeLog - if timestamp is None: - timestamp = datetime.now(timezone.utc) - else: - timestamp = timestamp.astimezone(timezone.utc) + timestamp = gen_datetime(timestamp) - cur_p = self.path / name.replace(".", "/") + cur_p = self.path / tag.replace(".", "/") cur_p.mkdir(parents=True, exist_ok=True) path = cur_p / f"{timestamp.strftime('%Y-%m-%d_%H-%M-%S-%f')}.log" @@ -65,49 +76,8 @@ class FileStorage(Storage): r"(?P.+:.+:\d+) - " ) - def iter_msg(self, common: bool = False, tag: str | None = None) -> Generator[Message, None, None]: + def iter_msg(self, tag: str | None = None) -> Generator[Message, None, None]: msg_l = [] - if common: # return string logs in common_logs.log - for file in self.path.glob("**/*.log"): - common_log_tag = ".".join(file.relative_to(self.path).as_posix().replace("/", ".").split(".")[:-3]) - - if tag is not None and tag not in common_log_tag: - continue - - pid = file.parent.name - - with file.open("r", encoding="utf-8") as f: - content = f.read() - - matches, next_matches = self.log_pattern.finditer(content), self.log_pattern.finditer(content) - next_match = next(next_matches, None) - # NOTE: the content will be the text between `match` and `next_match` - for match in matches: - next_match = next(next_matches, None) - - timestamp_str = match.group("timestamp") - timestamp = datetime.strptime(timestamp_str, "%Y-%m-%d %H:%M:%S.%f").replace(tzinfo=timezone.utc) - level: LOG_LEVEL = cast(LOG_LEVEL, match.group("level")) - caller = match.group("caller") - - # Extract the message content - message_start = match.end() - message_end = next_match.start() if next_match else len(content) - message_content = content[message_start:message_end].strip() - - if "Logging object in" in message_content: - continue - - m = Message( - tag=common_log_tag, - level=level, - timestamp=timestamp, - caller=caller, - pid_trace=pid, - content=message_content, - ) - - msg_l.append(m) pkl_files = "**/*.pkl" if tag is None else f"**/{tag.replace('.','/')}/**/*.pkl" for file in self.path.glob(pkl_files): @@ -130,35 +100,12 @@ class FileStorage(Storage): yield m def truncate(self, time: datetime) -> None: - # any message later than `time` will be removed - for file in self.path.glob("**/*.log"): - with file.open("r") as f: - content = f.read() + for file in self.path.glob("**/*.pkl"): + timestamp = datetime.strptime(file.stem, "%Y-%m-%d_%H-%M-%S-%f").replace(tzinfo=timezone.utc) + if timestamp > time: + file.unlink() - new_content = "" + _remove_empty_dir(self.path) - matches, next_matches = self.log_pattern.finditer(content), self.log_pattern.finditer(content) - - next_match = next(next_matches, None) - for match in matches: - next_match = next(next_matches, None) - timestamp_str = match.group("timestamp") - timestamp = datetime.strptime(timestamp_str, "%Y-%m-%d %H:%M:%S.%f").replace(tzinfo=timezone.utc) - - log_start = match.start() - log_end = next_match.start() if next_match else len(content) - msg = content[match.end() : log_end].strip() - - if timestamp > time: - if "Logging object in" in msg: - absolute_p = msg.split("Logging object in ")[1] - p = Path(absolute_p) - if p.exists(): - p.unlink() - else: - print(f"Missing pickle object: {p}.") - continue - - new_content += content[log_start:log_end] - with file.open("w") as f: - f.write(new_content) + def __str__(self) -> str: + return f"FileStorage({self.path})" diff --git a/rdagent/log/ui/storage.py b/rdagent/log/ui/storage.py new file mode 100644 index 00000000..97b58317 --- /dev/null +++ b/rdagent/log/ui/storage.py @@ -0,0 +1,324 @@ +from datetime import datetime +from pathlib import Path +from typing import Any, Generator + +import requests + +from rdagent.log.base import Message, Storage +from rdagent.log.utils import extract_evoid, extract_loopid_func_name, gen_datetime + + +class WebStorage(Storage): + """ + The storage for web app. + It is used to provide the data for the web app. + """ + + def __init__(self, port: int, path: str) -> None: + """ + Initializes the storage object with the specified port and identifier. + Args: + port (int): The port number to use for the storage service. + path (str): The unique identifier for local storage, the log path. + """ + self.url = f"http://localhost:{port}" + self.path = path + self.msgs = [] + + def __str__(self): + return f"WebStorage({self.url})" + + def log(self, obj: object, tag: str, timestamp: datetime | None = None, **kwargs: Any) -> str | Path: + timestamp = gen_datetime(timestamp) + + try: + data = self._obj_to_json(obj=obj, tag=tag, id=self.path, timestamp=timestamp.isoformat()) + if isinstance(data, list): + for d in data: + self.msgs.append(d) + else: + self.msgs.append(data) + headers = {"Content-Type": "application/json"} + requests.post(f"{self.url}/receive", json=data, headers=headers, timeout=1) + except (requests.ConnectionError, requests.Timeout) as e: + pass + + def truncate(self, time: datetime) -> None: + self.msgs = [m for m in self.msgs if datetime.fromisoformat(m["msg"]["timestamp"]) <= time] + + def iter_msg(self, **kwargs: Any) -> Generator[Message, None, None]: + for msg in self.msgs: + yield Message( + tag=msg["msg"]["tag"], + level="INFO", + timestamp=datetime.fromisoformat(msg["msg"]["timestamp"]), + content=msg, + ) + + def _obj_to_json( + obj: object, + tag: str, + id: str, + timestamp: str, + ) -> list[dict] | dict: + li, fn = extract_loopid_func_name(tag) + ei = extract_evoid(tag) + data = {} + if "hypothesis generation" in tag: + from rdagent.core.proposal import Hypothesis + + h: Hypothesis = obj + data = { + "id": id, + "msg": { + "tag": "research.hypothesis", + "timestamp": timestamp, + "loop_id": li, + "content": { + "hypothesis": h.hypothesis, + "reason": h.reason, + "concise_reason": h.concise_reason, + "concise_justification": h.concise_justification, + "concise_observation": h.concise_observation, + "concise_knowledge": h.concise_knowledge, + }, + }, + } + + elif "experiment generation" in tag: + from rdagent.components.coder.factor_coder.factor import FactorTask + from rdagent.components.coder.model_coder.model import ModelTask + from rdagent.core.experiment import Experiment + + tasks: list[FactorTask | ModelTask] = obj + if isinstance(tasks[0], FactorTask): + data = { + "id": id, + "msg": { + "tag": "research.tasks", + "timestamp": timestamp, + "loop_id": li, + "content": [ + { + "name": t.factor_name, + "description": t.factor_description, + "formulation": t.factor_formulation, + "variables": t.variables, + } + for t in tasks + ], + }, + } + elif isinstance(tasks[0], ModelTask): + data = { + "id": id, + "msg": { + "tag": "research.tasks", + "timestamp": timestamp, + "loop_id": li, + "content": [ + { + "name": t.name, + "description": t.description, + "model_type": t.model_type, + "formulation": t.formulation, + "variables": t.variables, + } + for t in tasks + ], + }, + } + elif "direct_exp_gen" in tag: + from rdagent.scenarios.data_science.experiment.experiment import ( + DSExperiment, + ) + + if isinstance(obj, DSExperiment): + from rdagent.scenarios.data_science.proposal.exp_gen.base import ( + DSHypothesis, + ) + + h: DSHypothesis = obj.hypothesis + tasks = [t[0] for t in obj.pending_tasks_list] + t = tasks[0] + data = [ + { + "id": id, + "msg": { + "tag": "research.hypothesis", + "old_tag": tag, + "timestamp": timestamp, + "loop_id": li, + "content": { + "name_map": { + "hypothesis": "RD-Agent proposes the hypothesis⬇️", + "concise_justification": "because the reason⬇️", + "concise_observation": "based on the observation⬇️", + "concise_knowledge": "Knowledge⬇️ gained after practice", + "no_hypothesis": f"No hypothesis available. Trying to construct the first runnable {h.component} component.", + }, + "hypothesis": h.hypothesis, + "reason": h.reason, + "component": h.component, + "concise_reason": h.concise_reason, + "concise_justification": h.concise_justification, + "concise_observation": h.concise_observation, + "concise_knowledge": h.concise_knowledge, + }, + }, + }, + { + "id": id, + "msg": { + "tag": "research.tasks", + "old_tag": tag, + "timestamp": timestamp, + "loop_id": li, + "content": [ + ( + { + "name": t.name, + "description": t.description, + } + if not hasattr(t, "architecture") + else { + "name": t.name, + "description": t.description, + "model_type": t.model_type, + "architecture": t.architecture, + "hyperparameters": t.hyperparameters, + } + ) + ], + }, + }, + ] + elif f"evo_loop_{ei}.evolving code" in tag and "coding" in tag: + from rdagent.components.coder.factor_coder.factor import FactorFBWorkspace + from rdagent.components.coder.model_coder.model import ( + ModelFBWorkspace, + ModelTask, + ) + from rdagent.core.experiment import FBWorkspace + + ws: list[FactorFBWorkspace | ModelFBWorkspace] = [i for i in obj] + if all(isinstance(item, FactorFBWorkspace) for item in ws) or all( + isinstance(item, ModelFBWorkspace) for item in ws + ): + data = { + "id": id, + "msg": { + "tag": "evolving.codes", + "timestamp": timestamp, + "loop_id": li, + "evo_id": ei, + "content": [ + { + "target_task_name": w.target_task.name, + "codes": w.file_dict, + } + for w in ws + if w + ], + }, + } + elif f"evo_loop_{ei}.evolving feedback" in tag and "coding" in tag: + from rdagent.components.coder.CoSTEER.evaluators import ( + CoSTEERSingleFeedback, + ) + + fl: list[CoSTEERSingleFeedback] = [i for i in obj] + data = { + "id": id, + "msg": { + "tag": "evolving.feedbacks", + "timestamp": timestamp, + "loop_id": li, + "evo_id": ei, + "content": [ + { + "final_decision": f.final_decision, + # "final_feedback": f.final_feedback, + "execution": f.execution, + "code": f.code, + "return_checking": f.return_checking, + } + for f in fl + if f + ], + }, + } + elif "scenario" in tag: + data = { + "id": id, + "msg": { + "tag": "feedback.config", + "timestamp": timestamp, + "loop_id": li, + "content": {"config": obj.experiment_setting}, + }, + } + + elif "Quantitative Backtesting Chart" in tag: + import plotly + + from rdagent.log.ui.qlib_report_figure import report_figure + + data = { + "id": id, + "msg": { + "tag": "feedback.return_chart", + "timestamp": timestamp, + "loop_id": li, + "content": {"chart_html": plotly.io.to_html(report_figure(obj))}, + }, + } + elif "running" in tag: + from rdagent.core.experiment import Experiment + + if isinstance(obj, Experiment): + if obj.result is not None: + result_str = obj.result.to_json() + data = { + "id": id, + "msg": { + "tag": "feedback.metric", + "old_tag": tag, + "timestamp": timestamp, + "loop_id": li, + "content": { + "result": result_str, + }, + }, + } + elif "feedback" in tag: + from rdagent.core.proposal import ExperimentFeedback, HypothesisFeedback + + ef: ExperimentFeedback = obj + content = ( + { + "observations": ef.observations, + "hypothesis_evaluation": ef.hypothesis_evaluation, + "new_hypothesis": ef.new_hypothesis, + "decision": ef.decision, + "reason": ef.reason, + "exception": ef.exception, + } + if isinstance(ef, HypothesisFeedback) + else { + "decision": ef.decision, + "reason": ef.reason, + "exception": ef.exception, + } + ) + data = { + "id": id, + "msg": { + "tag": "feedback.hypothesis_feedback", + "timestamp": timestamp, + "loop_id": li, + "content": content, + }, + } + + return data diff --git a/rdagent/log/ui/utils.py b/rdagent/log/ui/utils.py index c2029bfa..5bbf0504 100644 --- a/rdagent/log/ui/utils.py +++ b/rdagent/log/ui/utils.py @@ -197,7 +197,7 @@ def load_times(log_path: Path): max_step = max(int(p.name.split("_")[0]) for p in (session_path / str(max_li)).iterdir() if p.is_file()) rdloop_obj_p = next((session_path / str(max_li)).glob(f"{max_step}_*")) - rd_times = DataScienceRDLoop.load(rdloop_obj_p, do_truncate=False).loop_trace + rd_times = DataScienceRDLoop.load(rdloop_obj_p).loop_trace except Exception as e: rd_times = {} return rd_times diff --git a/rdagent/log/utils/__init__.py b/rdagent/log/utils/__init__.py index c6f68cdb..33a0236c 100644 --- a/rdagent/log/utils/__init__.py +++ b/rdagent/log/utils/__init__.py @@ -1,6 +1,7 @@ import inspect import json import re +from datetime import datetime, timezone from pathlib import Path from typing import Optional, TypedDict, cast @@ -100,3 +101,14 @@ def extract_json(log_content: str) -> dict | None: if match: return cast(dict, json.loads(match.group(0))) return None + + +def gen_datetime(dt: datetime | None = None) -> datetime: + """ + Generate a datetime object in UTC timezone. + - If `dt` is None, it will return the current time in UTC. + - If `dt` is provided, it will convert it to UTC timezone. + """ + if dt is None: + return datetime.now(timezone.utc) + return dt.astimezone(timezone.utc) diff --git a/rdagent/log/utils/folder.py b/rdagent/log/utils/folder.py index ec45c6a7..ea551223 100644 --- a/rdagent/log/utils/folder.py +++ b/rdagent/log/utils/folder.py @@ -51,7 +51,7 @@ def first_li_si_after_one_time(log_path: Path, hours: int = 12) -> tuple[int, in max_step = max(int(p.name.split("_")[0]) for p in (session_path / str(max_li)).iterdir() if p.is_file()) rdloop_obj_p = next((session_path / str(max_li)).glob(f"{max_step}_*")) - rdloop_obj = DataScienceRDLoop.load(rdloop_obj_p, do_truncate=False) + rdloop_obj = DataScienceRDLoop.load(rdloop_obj_p) loop_trace = rdloop_obj.loop_trace si2fn = rdloop_obj.steps diff --git a/rdagent/scenarios/data_science/loop.py b/rdagent/scenarios/data_science/loop.py index 0f46f4c6..d007be9c 100644 --- a/rdagent/scenarios/data_science/loop.py +++ b/rdagent/scenarios/data_science/loop.py @@ -270,12 +270,11 @@ class DataScienceRDLoop(RDLoop): @classmethod def load( cls, - path: Union[str, Path], - output_path: Optional[Union[str, Path]] = None, - do_truncate: bool = False, + path: str | Path, + checkout: bool | str | Path = False, replace_timer: bool = True, ) -> "LoopBase": - session = super().load(path, output_path, do_truncate, replace_timer) + session = super().load(path, checkout, replace_timer) logger.log_object(DS_RD_SETTING.competition, tag="competition") # NOTE: necessary to make mle_summary work. if DS_RD_SETTING.enable_knowledge_base and DS_RD_SETTING.knowledge_base_version == "v1": session.trace.knowledge_base = DSKnowledgeBase( diff --git a/rdagent/scenarios/data_science/proposal/exp_gen/proposal.py b/rdagent/scenarios/data_science/proposal/exp_gen/proposal.py index d8e5b63c..48feb0f3 100644 --- a/rdagent/scenarios/data_science/proposal/exp_gen/proposal.py +++ b/rdagent/scenarios/data_science/proposal/exp_gen/proposal.py @@ -1,5 +1,4 @@ import json -import pprint from enum import Enum from typing import Dict, List, Optional, Tuple @@ -222,45 +221,6 @@ class CodingSketch(BaseModel): ) -COMPONENT_TASK_MAPPING = { - "DataLoadSpec": { - "target_name": "Data loader and specification generation", - "spec_file": "spec/data_loader.md", - "task_output_format": T(".prompts:output_format.data_loader").r(), - "task_class": DataLoaderTask, - }, - "FeatureEng": { - "target_name": "Feature engineering", - "spec_file": "spec/feature.md", - "task_output_format": T(".prompts:output_format.feature").r(), - "task_class": FeatureTask, - }, - "Model": { - "target_name": "Model", - "spec_file": "spec/model.md", - "task_output_format": T(".prompts:output_format.model").r(), - "task_class": ModelTask, - }, - "Ensemble": { - "target_name": "Ensemble", - "spec_file": "spec/ensemble.md", - "task_output_format": T(".prompts:output_format.ensemble").r(), - "task_class": EnsembleTask, - }, - "Workflow": { - "target_name": "Workflow", - "spec_file": "spec/workflow.md", - "task_output_format": T(".prompts:output_format.workflow").r(), - "task_class": WorkflowTask, - }, - "Pipeline": { - "target_name": "Pipeline", - "task_output_format": T(".prompts:output_format.pipeline").r(), - "task_class": PipelineTask, - }, -} - - def draft_exp_in_decomposition(scen: Scenario, trace: DSTrace) -> None | DSDraftExpGen: next_missing_component = trace.next_incomplete_component() if next_missing_component is not None: @@ -273,6 +233,44 @@ def draft_exp_in_decomposition(scen: Scenario, trace: DSTrace) -> None | DSDraft class DSProposalV1ExpGen(ExpGen): + COMPONENT_TASK_MAPPING = { + "DataLoadSpec": { + "target_name": "Data loader and specification generation", + "spec_file": "spec/data_loader.md", + "task_output_format": T(".prompts:output_format.data_loader").r(), + "task_class": DataLoaderTask, + }, + "FeatureEng": { + "target_name": "Feature engineering", + "spec_file": "spec/feature.md", + "task_output_format": T(".prompts:output_format.feature").r(), + "task_class": FeatureTask, + }, + "Model": { + "target_name": "Model", + "spec_file": "spec/model.md", + "task_output_format": T(".prompts:output_format.model").r(), + "task_class": ModelTask, + }, + "Ensemble": { + "target_name": "Ensemble", + "spec_file": "spec/ensemble.md", + "task_output_format": T(".prompts:output_format.ensemble").r(), + "task_class": EnsembleTask, + }, + "Workflow": { + "target_name": "Workflow", + "spec_file": "spec/workflow.md", + "task_output_format": T(".prompts:output_format.workflow").r(), + "task_class": WorkflowTask, + }, + "Pipeline": { + "target_name": "Pipeline", + "task_output_format": T(".prompts:output_format.pipeline").r(), + "task_class": PipelineTask, + }, + } + def gen(self, trace: DSTrace) -> DSExperiment: # Drafting Stage if draft_exp := draft_exp_in_decomposition(self.scen, trace): @@ -353,7 +351,7 @@ class DSProposalV1ExpGen(ExpGen): # - after we know the selected component, we can use RAG. # Step 2: Generate the rest of the hypothesis & task - component_info = COMPONENT_TASK_MAPPING.get(component) + component_info = self.COMPONENT_TASK_MAPPING.get(component) if component_info: if DS_RD_SETTING.spec_enabled: @@ -658,9 +656,9 @@ class DSProposalV2ExpGen(ExpGen): failed_exp_feedback_list_desc: str, ) -> DSExperiment: if pipeline: - component_info = COMPONENT_TASK_MAPPING["Pipeline"] + component_info = self.COMPONENT_TASK_MAPPING["Pipeline"] else: - component_info = COMPONENT_TASK_MAPPING.get(hypothesis.component) + component_info = self.COMPONENT_TASK_MAPPING.get(hypothesis.component) if pipeline: task_spec = T(f"scenarios.data_science.share:component_spec.Pipeline").r() elif DS_RD_SETTING.spec_enabled and sota_exp is not None: @@ -950,9 +948,9 @@ class DSProposalV3ExpGen(DSProposalV2ExpGen): failed_exp_feedback_list_desc: str, ) -> DSExperiment: if pipeline: - component_info = COMPONENT_TASK_MAPPING["Pipeline"] + component_info = self.COMPONENT_TASK_MAPPING["Pipeline"] else: - component_info = COMPONENT_TASK_MAPPING.get(hypotheses[0].component) + component_info = self.COMPONENT_TASK_MAPPING.get(hypotheses[0].component) data_folder_info = self.scen.processed_data_folder_description sys_prompt = T(".prompts_v3:task_gen.system").r( # targets=component_info["target_name"], diff --git a/rdagent/utils/workflow.py b/rdagent/utils/workflow.py index 5f9fa215..c246dbb1 100644 --- a/rdagent/utils/workflow.py +++ b/rdagent/utils/workflow.py @@ -9,19 +9,19 @@ Postscripts: """ import datetime -import os import pickle import time from collections import defaultdict from dataclasses import dataclass from pathlib import Path -from typing import Any, Callable, Optional, TypeVar, Union, cast +from typing import Any, Callable, TypeVar, cast import pytz from tqdm.auto import tqdm from rdagent.core.conf import RD_AGENT_SETTINGS from rdagent.log import rdagent_logger as logger +from rdagent.log.conf import LOG_SETTINGS from rdagent.log.timer import RD_Agent_TIMER_wrapper, RDAgentTimer if RD_AGENT_SETTINGS.enable_mlflow: @@ -99,7 +99,7 @@ class LoopBase: self.step_idx = 0 # the index of next step to be run self.loop_prev_out: dict[str, Any] = {} # the step results of current loop self.loop_trace = defaultdict(list[LoopTrace]) # the key is the number of loop - self.session_folder = logger.log_trace_path / "__session__" + self.session_folder = Path(LOG_SETTINGS.trace_path) / "__session__" self.timer: RDAgentTimer = RD_Agent_TIMER_wrapper.timer def run(self, step_n: int | None = None, loop_n: int | None = None, all_duration: str | None = None) -> None: @@ -225,8 +225,7 @@ class LoopBase: if prev_path: loaded = type(self).load( prev_path, - output_path=self.session_folder.parent, - do_truncate=False, + checkout=True, replace_timer=True, ) logger.info(f"Load previous session from {prev_path}") @@ -244,34 +243,69 @@ class LoopBase: with path.open("wb") as f: pickle.dump(self, f) + def truncate_session_folder(self, li: int, si: int) -> None: + """ + Clear the session folder by removing all session objects after the given loop index (li) and step index (si). + """ + # clear session folders after the li + for sf in self.session_folder.iterdir(): + if sf.is_dir() and int(sf.name) > li: + for file in sf.iterdir(): + file.unlink() + sf.rmdir() + + # clear step session objects in the li + final_loop_session_folder = self.session_folder / str(li) + for step_session in final_loop_session_folder.glob("*_*"): + if step_session.is_file(): + step_id = int(step_session.name.split("_", 1)[0]) + if step_id > si: + step_session.unlink() + @classmethod def load( cls, - path: Union[str, Path], - output_path: Optional[Union[str, Path]] = None, - do_truncate: bool = False, + path: str | Path, + checkout: bool | Path | str = False, replace_timer: bool = True, ) -> "LoopBase": + """ + Load a session from a given path. + Parameters + ---------- + path : str | Path + The path to the session file. + checkout : bool | Path | str + If True, the new loop will use the existing folder and clear logs for sessions after the one corresponding to the given path. + If False, the new loop will use the existing folder but keep the logs for sessions after the one corresponding to the given path. + If a path (or a str like Path) is provided, the new loop will be saved to that path, leaving the original path unchanged. + replace_timer : bool + If a session is loaded, determines whether to replace the timer with session.timer. + Default is True, which means the session timer will be replaced with the current timer. + If False, the session timer will not be replaced. + Returns + ------- + LoopBase + An instance of LoopBase with the loaded session. + """ path = Path(path) with path.open("rb") as f: session = cast(LoopBase, pickle.load(f)) # set session folder - # - P1: if output_path explicitly specified. - # - P2: RD_AGENT_SETTINGS.log_trace_path - output_path_value = output_path if output_path is not None else RD_AGENT_SETTINGS.log_trace_path - if output_path_value is not None: - output_path_path = Path(output_path_value) - output_path_path.mkdir(parents=True, exist_ok=True) - session.session_folder = output_path_path / "__session__" + if checkout: + if checkout is True: + logger.set_storages_path(session.session_folder.parent) + max_loop = max(session.loop_trace.keys()) - # set trace path - logger.set_trace_path(session.session_folder.parent) - - # truncate future message - if do_truncate: - max_loop = max(session.loop_trace.keys()) - logger.storage.truncate(time=session.loop_trace[max_loop][-1].end) + # truncate log storages after the max loop + session.truncate_session_folder(max_loop, len(session.loop_trace[max_loop]) - 1) + logger.truncate_storages(session.loop_trace[max_loop][-1].end) + else: + checkout = Path(checkout) + checkout.mkdir(parents=True, exist_ok=True) + session.session_folder = checkout / "__session__" + logger.set_storages_path(checkout) if session.timer.started: if replace_timer: diff --git a/requirements.txt b/requirements.txt index ebac1ec2..e2475dd2 100644 --- a/requirements.txt +++ b/requirements.txt @@ -13,6 +13,7 @@ azure.identity pyarrow rich tqdm +typer numpy # we use numpy as default data format. So we have to install numpy pandas # we use pandas as default data format. So we have to install pandas @@ -43,6 +44,9 @@ docker streamlit plotly st-theme +randomname +flask +flask-cors # kaggle crawler selenium diff --git a/test/utils/test_import.py b/test/utils/test_import.py index 4c755447..89c0b530 100644 --- a/test/utils/test_import.py +++ b/test/utils/test_import.py @@ -30,6 +30,7 @@ class TestRDAgentImports(unittest.TestCase): or fstr.endswith("rdagent/app/cli.py") or fstr.endswith("rdagent/app/CI/run.py") or fstr.endswith("rdagent/app/utils/ape.py") + or fstr.endswith("rdagent/log/ui/utils.py") ): # the entrance points continue