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 <xianbowen@outlook.com> Co-authored-by: you-n-g <you-n-g@users.noreply.github.com>
2026-07-27 23:47:46 +00:00 · 2025-06-06 18:52:19 +08:00
parent bdaa376569
commit 805f337fc4
27 changed files with 1038 additions and 278 deletions
@@ -175,3 +175,7 @@ mlruns/
 /*.sh
 .aider*
 rdagent/app/benchmark/factor/example.json
+
+# UI Server resources
+videos/
+static/
@@ -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,
@@ -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,
        }
    )
@@ -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__":
@@ -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:
@@ -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__":
@@ -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__":
@@ -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__":
@@ -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__":
@@ -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))

@@ -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 = ""
@@ -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:
    """
@@ -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()
@@ -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)
@@ -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}")
@@ -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": "<isoformat>",
+    "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": "<isoformat>",
+    "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": "<isoformat>",
+    "loop_id": "1",
+    "evo_id": "0",
+    "content": [ // list of task_name & codes
+        {
+            "target_task_name": "task_1",
+            "codes": { // one or more codes
+                "a.py": "...<python codes>",
+                "b.py": "...<python codes>",
+                //...
+            }
+        },
+        {
+            "target_task_name": "task_2",
+            "codes": {
+                "a.py": "...<python codes>",
+                //...
+            }
+        }
+        //... same as above
+    ]
+}
+
+{
+    "tag": "evolving.codes",
+    "timestamp": "<isoformat>",
+    "loop_id": "1",
+    "evo_id": "1",
+    "content": [
+        //... same as above
+    ]
+}
+```
+
+2. feedbacks
+
+```json
+{
+    "tag": "evolving.feedbacks",
+    "timestamp": "<isoformat>",
+    "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": "<isoformat>",
+    "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": "<isoformat>",
+    "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": "<isoformat>",
+    "loop_id": "1",
+    "content": {
+        "chart_html": "chart html codes string",
+    }
+}
+```
+
+3. metric
+
+```json
+{
+    "tag": "feedback.metric",
+    "timestamp": "<isoformat>",
+    "loop_id": "1",
+    "content": {
+        "result": "{ \"<metric_name>\": <value>, ... }" // A JSON string containing metric names and their corresponding values.
+    }
+}
+```
+
+4. hypothesis_feedback
+
+```json
+{
+    "tag": "feedback.hypothesis_feedback",
+    "timestamp": "<isoformat>",
+    "loop_id": "1",
+    "content": {
+        "decision": "True",
+        "reason": "...",
+        "exception": "...",
+        "observations": "...", // may not exists
+        "hypothesis_evaluation": "...", // may not exists
+        "new_hypothesis": "...", // may not exists
+    }
+}
+```
@@ -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("/<path:fn>", 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)
@@ -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<caller>.+:.+:\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})"
@@ -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
@@ -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
@@ -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)
@@ -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

@@ -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(
@@ -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"],
@@ -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:
@@ -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
@@ -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