1  """Load and save experiences from the log file."""
 2  
 3  import json
 4  from pathlib import Path
 5  
 6  from metagpt.exp_pool import get_exp_manager
 7  from metagpt.exp_pool.schema import LOG_NEW_EXPERIENCE_PREFIX, Experience
 8  from metagpt.logs import logger
 9  
10  
11  def load_exps(log_file_path: str) -> list[Experience]:
12      """Loads experiences from a log file.
13  
14      Args:
15          log_file_path (str): The path to the log file.
16  
17      Returns:
18          list[Experience]: A list of Experience objects loaded from the log file.
19      """
20  
21      if not Path(log_file_path).exists():
22          logger.warning(f"`load_exps` called with a non-existent log file path: {log_file_path}")
23          return
24  
25      exps = []
26      with open(log_file_path, "r") as log_file:
27          for line in log_file:
28              if LOG_NEW_EXPERIENCE_PREFIX in line:
29                  json_str = line.split(LOG_NEW_EXPERIENCE_PREFIX, 1)[1].strip()
30                  exp_data = json.loads(json_str)
31  
32                  exp = Experience(**exp_data)
33                  exps.append(exp)
34  
35      logger.info(f"Loaded {len(exps)} experiences from log file: {log_file_path}")
36  
37      return exps
38  
39  
40  def save_exps(exps: list[Experience]):
41      """Saves a list of experiences to the experience pool.
42  
43      Args:
44          exps (list[Experience]): The list of experiences to save.
45      """
46  
47      if not exps:
48          logger.warning("`save_exps` called with an empty list of experiences.")
49          return
50  
51      manager = get_exp_manager()
52      manager.is_writable = True
53  
54      manager.create_exps(exps)
55      logger.info(f"Saved {len(exps)} experiences.")
56  
57  
58  def get_log_file_path() -> str:
59      """Retrieves the path to the log file.
60  
61      Returns:
62          str: The path to the log file.
63  
64      Raises:
65          ValueError: If the log file path cannot be found.
66      """
67  
68      handlers = logger._core.handlers
69  
70      for handler in handlers.values():
71          if "log" in handler._name:
72              return handler._name[1:-1]
73  
74      raise ValueError("Log file not found")
75  
76  
77  def main():
78      log_file_path = get_log_file_path()
79  
80      exps = load_exps(log_file_path)
81      save_exps(exps)
82  
83  
84  if __name__ == "__main__":
85      main()