跳转至

HistoryManager（历史记录管理）

evotoolkit.core.HistoryManager

管理进化算法运行历史的保存和加载

Source code in src/evotoolkit/core/history_manager.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
class HistoryManager:
    """管理进化算法运行历史的保存和加载"""

    def __init__(self, output_path: str):
        self.output_path = output_path
        self.history_dir = os.path.join(output_path, "history")
        self.summary_dir = os.path.join(output_path, "summary")

        # 确保目录存在
        os.makedirs(self.history_dir, exist_ok=True)
        os.makedirs(self.summary_dir, exist_ok=True)

    # ========== By-Generation Methods ==========

    def save_generation_history(
        self,
        generation: int,
        solutions: List[Solution],
        usage: List[Dict],
        statistics: Optional[Dict] = None,
    ) -> None:
        """保存某一代的历史记录"""
        gen_file = os.path.join(self.history_dir, f"gen_{generation}.json")

        # 转换Solution对象为字典
        solutions_json = []
        for sol in solutions:
            sol_dict = {
                "sol_string": sol.sol_string,
                "other_info": sol.other_info,
                "evaluation_res": None,
            }
            if sol.evaluation_res:
                sol_dict["evaluation_res"] = {
                    "valid": sol.evaluation_res.valid,
                    "score": sol.evaluation_res.score,
                    "additional_info": sol.evaluation_res.additional_info,
                }
            solutions_json.append(sol_dict)

        data = {
            "generation": generation,
            "solutions": solutions_json,
            "usage": usage,
            "statistics": statistics or {},
        }

        with open(gen_file, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=2, ensure_ascii=False)

    def load_generation_history(self, generation: int) -> Optional[Dict]:
        """加载某一代的历史记录"""
        gen_file = os.path.join(self.history_dir, f"gen_{generation}.json")
        if not os.path.exists(gen_file):
            return None

        with open(gen_file, "r", encoding="utf-8") as f:
            return json.load(f)

    def get_all_generations(self) -> List[int]:
        """获取所有已保存的代数"""
        generations = []
        if not os.path.exists(self.history_dir):
            return generations

        for filename in os.listdir(self.history_dir):
            if filename.startswith("gen_") and filename.endswith(".json"):
                try:
                    gen = int(filename.replace("gen_", "").replace(".json", ""))
                    generations.append(gen)
                except ValueError:
                    continue

        return sorted(generations)

    # ========== By-Batch Methods ==========

    def save_batch_history(
        self,
        batch_id: int,
        sample_range: tuple,
        solutions: List[Solution],
        usage: List[Dict],
        metadata: Optional[Dict] = None,
    ) -> None:
        """保存批次历史记录（用于FunSearch等）"""
        batch_file = os.path.join(self.history_dir, f"batch_{batch_id:04d}.json")

        # 转换Solution对象为字典
        solutions_json = []
        for sol in solutions:
            sol_dict = {
                "sol_string": sol.sol_string,
                "other_info": sol.other_info,
                "evaluation_res": None,
            }
            if sol.evaluation_res:
                sol_dict["evaluation_res"] = {
                    "valid": sol.evaluation_res.valid,
                    "score": sol.evaluation_res.score,
                    "additional_info": sol.evaluation_res.additional_info,
                }
            solutions_json.append(sol_dict)

        data = {
            "batch_id": batch_id,
            "sample_range": sample_range,
            "solutions": solutions_json,
            "usage": usage,
            "metadata": metadata or {},
        }

        with open(batch_file, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=2, ensure_ascii=False)

    def load_batch_history(self, batch_id: int) -> Optional[Dict]:
        """加载批次历史记录"""
        batch_file = os.path.join(self.history_dir, f"batch_{batch_id:04d}.json")
        if not os.path.exists(batch_file):
            return None

        with open(batch_file, "r", encoding="utf-8") as f:
            return json.load(f)

    def get_all_batches(self) -> List[int]:
        """获取所有已保存的批次"""
        batches = []
        if not os.path.exists(self.history_dir):
            return batches

        for filename in os.listdir(self.history_dir):
            if filename.startswith("batch_") and filename.endswith(".json"):
                try:
                    batch = int(filename.replace("batch_", "").replace(".json", ""))
                    batches.append(batch)
                except ValueError:
                    continue

        return sorted(batches)

    # ========== Summary Methods ==========

    def save_usage_history(self, usage_history: Dict) -> None:
        """保存完整的usage历史"""
        usage_file = os.path.join(self.summary_dir, "usage_history.json")
        with open(usage_file, "w", encoding="utf-8") as f:
            json.dump(usage_history, f, indent=2, ensure_ascii=False)

    def load_usage_history(self) -> Dict:
        """加载usage历史"""
        usage_file = os.path.join(self.summary_dir, "usage_history.json")
        if not os.path.exists(usage_file):
            return {}

        with open(usage_file, "r", encoding="utf-8") as f:
            return json.load(f)

    def save_best_per_generation(self, best_solutions: List[Dict]) -> None:
        """保存每代最优解摘要"""
        best_file = os.path.join(self.summary_dir, "best_per_generation.json")
        with open(best_file, "w", encoding="utf-8") as f:
            json.dump(best_solutions, f, indent=2, ensure_ascii=False)

    def load_best_per_generation(self) -> List[Dict]:
        """加载每代最优解摘要"""
        best_file = os.path.join(self.summary_dir, "best_per_generation.json")
        if not os.path.exists(best_file):
            return []

        with open(best_file, "r", encoding="utf-8") as f:
            return json.load(f)

save_generation_history 

save_generation_history(
    generation: int,
    solutions: List[Solution],
    usage: List[Dict],
    statistics: Optional[Dict] = None,
) -> None

保存某一代的历史记录

Source code in src/evotoolkit/core/history_manager.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def save_generation_history(
    self,
    generation: int,
    solutions: List[Solution],
    usage: List[Dict],
    statistics: Optional[Dict] = None,
) -> None:
    """保存某一代的历史记录"""
    gen_file = os.path.join(self.history_dir, f"gen_{generation}.json")

    # 转换Solution对象为字典
    solutions_json = []
    for sol in solutions:
        sol_dict = {
            "sol_string": sol.sol_string,
            "other_info": sol.other_info,
            "evaluation_res": None,
        }
        if sol.evaluation_res:
            sol_dict["evaluation_res"] = {
                "valid": sol.evaluation_res.valid,
                "score": sol.evaluation_res.score,
                "additional_info": sol.evaluation_res.additional_info,
            }
        solutions_json.append(sol_dict)

    data = {
        "generation": generation,
        "solutions": solutions_json,
        "usage": usage,
        "statistics": statistics or {},
    }

    with open(gen_file, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)

load_generation_history 

load_generation_history(generation: int) -> Optional[Dict]

加载某一代的历史记录

Source code in src/evotoolkit/core/history_manager.py 
62
63
64
65
66
67
68
69
def load_generation_history(self, generation: int) -> Optional[Dict]:
    """加载某一代的历史记录"""
    gen_file = os.path.join(self.history_dir, f"gen_{generation}.json")
    if not os.path.exists(gen_file):
        return None

    with open(gen_file, "r", encoding="utf-8") as f:
        return json.load(f)

get_all_generations 

get_all_generations() -> List[int]

获取所有已保存的代数

Source code in src/evotoolkit/core/history_manager.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def get_all_generations(self) -> List[int]:
    """获取所有已保存的代数"""
    generations = []
    if not os.path.exists(self.history_dir):
        return generations

    for filename in os.listdir(self.history_dir):
        if filename.startswith("gen_") and filename.endswith(".json"):
            try:
                gen = int(filename.replace("gen_", "").replace(".json", ""))
                generations.append(gen)
            except ValueError:
                continue

    return sorted(generations)

save_batch_history 

save_batch_history(
    batch_id: int,
    sample_range: tuple,
    solutions: List[Solution],
    usage: List[Dict],
    metadata: Optional[Dict] = None,
) -> None

保存批次历史记录（用于FunSearch等）

Source code in src/evotoolkit/core/history_manager.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def save_batch_history(
    self,
    batch_id: int,
    sample_range: tuple,
    solutions: List[Solution],
    usage: List[Dict],
    metadata: Optional[Dict] = None,
) -> None:
    """保存批次历史记录（用于FunSearch等）"""
    batch_file = os.path.join(self.history_dir, f"batch_{batch_id:04d}.json")

    # 转换Solution对象为字典
    solutions_json = []
    for sol in solutions:
        sol_dict = {
            "sol_string": sol.sol_string,
            "other_info": sol.other_info,
            "evaluation_res": None,
        }
        if sol.evaluation_res:
            sol_dict["evaluation_res"] = {
                "valid": sol.evaluation_res.valid,
                "score": sol.evaluation_res.score,
                "additional_info": sol.evaluation_res.additional_info,
            }
        solutions_json.append(sol_dict)

    data = {
        "batch_id": batch_id,
        "sample_range": sample_range,
        "solutions": solutions_json,
        "usage": usage,
        "metadata": metadata or {},
    }

    with open(batch_file, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)

load_batch_history 

load_batch_history(batch_id: int) -> Optional[Dict]

加载批次历史记录

Source code in src/evotoolkit/core/history_manager.py 
127
128
129
130
131
132
133
134
def load_batch_history(self, batch_id: int) -> Optional[Dict]:
    """加载批次历史记录"""
    batch_file = os.path.join(self.history_dir, f"batch_{batch_id:04d}.json")
    if not os.path.exists(batch_file):
        return None

    with open(batch_file, "r", encoding="utf-8") as f:
        return json.load(f)

get_all_batches 

get_all_batches() -> List[int]

获取所有已保存的批次

Source code in src/evotoolkit/core/history_manager.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def get_all_batches(self) -> List[int]:
    """获取所有已保存的批次"""
    batches = []
    if not os.path.exists(self.history_dir):
        return batches

    for filename in os.listdir(self.history_dir):
        if filename.startswith("batch_") and filename.endswith(".json"):
            try:
                batch = int(filename.replace("batch_", "").replace(".json", ""))
                batches.append(batch)
            except ValueError:
                continue

    return sorted(batches)

save_usage_history 

save_usage_history(usage_history: Dict) -> None

保存完整的usage历史

Source code in src/evotoolkit/core/history_manager.py 
154
155
156
157
158
def save_usage_history(self, usage_history: Dict) -> None:
    """保存完整的usage历史"""
    usage_file = os.path.join(self.summary_dir, "usage_history.json")
    with open(usage_file, "w", encoding="utf-8") as f:
        json.dump(usage_history, f, indent=2, ensure_ascii=False)

load_usage_history 

load_usage_history() -> Dict

加载usage历史

Source code in src/evotoolkit/core/history_manager.py 
160
161
162
163
164
165
166
167
def load_usage_history(self) -> Dict:
    """加载usage历史"""
    usage_file = os.path.join(self.summary_dir, "usage_history.json")
    if not os.path.exists(usage_file):
        return {}

    with open(usage_file, "r", encoding="utf-8") as f:
        return json.load(f)

save_best_per_generation 

save_best_per_generation(
    best_solutions: List[Dict],
) -> None

保存每代最优解摘要

Source code in src/evotoolkit/core/history_manager.py 
169
170
171
172
173
def save_best_per_generation(self, best_solutions: List[Dict]) -> None:
    """保存每代最优解摘要"""
    best_file = os.path.join(self.summary_dir, "best_per_generation.json")
    with open(best_file, "w", encoding="utf-8") as f:
        json.dump(best_solutions, f, indent=2, ensure_ascii=False)

load_best_per_generation 

load_best_per_generation() -> List[Dict]

加载每代最优解摘要

Source code in src/evotoolkit/core/history_manager.py 
175
176
177
178
179
180
181
182
def load_best_per_generation(self) -> List[Dict]:
    """加载每代最优解摘要"""
    best_file = os.path.join(self.summary_dir, "best_per_generation.json")
    if not os.path.exists(best_file):
        return []

    with open(best_file, "r", encoding="utf-8") as f:
        return json.load(f)