| from typing import Any, Dict, List, Optional, Literal |
| from collections import defaultdict |
| from datetime import datetime |
| from bw_utils import load_json_file, save_json_file, load_jsonl_file, save_jsonl_file |
| import os |
|
|
|
|
| class HistoryManager: |
| def __init__(self): |
| self.detailed_history = [] |
| self.status_history = [] |
| self.message_history = [] |
|
|
| def add_record(self, record): |
| """添加一个事件记录 |
| record = { |
| "cur_round":cur_round, |
| "role_code":role_code, |
| "detail":detail, |
| "type":act_type, |
| "initiator":initiator, |
| "actor":actor |
| "group":group, |
| "other_info":other_info, |
| "record_id":record_id |
| } |
| """ |
| self.detailed_history.append(record) |
|
|
| def add_status(self, status): |
| self.status_history.append(status) |
| |
| def add_message(self, message): |
| self.message_history.append(message) |
| |
| def modify_record(self, record_id: str, detail: str): |
| """修改特定记录""" |
| for record in self.detailed_history: |
| if record["record_id"] == record_id: |
| record["detail"] = detail |
| print(f"Record {record_id} has been modified.") |
| return record['group'] |
| |
| def search_record_detail(self, record_id: str): |
| for record in self.detailed_history[::-1]: |
| if record["record_id"] == record_id: |
| return record["detail"] |
| return None |
| |
| def get_recent_history(self, recent_k = 5): |
| return [record["detail"] for record in self.detailed_history[-recent_k:]] |
| |
| def get_subsequent_history(self,start_idx): |
| return [record["detail"] for record in self.detailed_history[start_idx:]] |
| |
| def get_complete_history(self,): |
| return [record["detail"] for record in self.detailed_history[:]] |
| |
| def __len__(self): |
| return len(self.detailed_history) |
| |
| def __getstate__(self): |
| states = {key: value for key, value in self.__dict__.items() \ |
| if isinstance(value, (str, int, list, dict, bool, type(None)))} |
| return states |
|
|
| def __setstate__(self, states): |
| self.__dict__.update(states) |
|
|
| def save_to_file(self, root_dir): |
| filename = os.path.join(root_dir, f"./simulation_history.json") |
| save_json_file(filename, self.__getstate__() ) |
|
|
| def load_from_file(self, root_dir): |
| filename = os.path.join(root_dir, f"./simulation_history.json") |
| states = load_json_file(filename) |
| self.__setstate__(states) |
|
|
|
|
|
|
