Load memory configurations with version history from JSON. Loads basic configuration only (instance is not saved, must be created via build()). Only the latest version will be instantiated by default if auto_initialize=True. Args: file_path: File
(self, file_path: Optional[str] = None, auto_initialize: bool = True)
| 859 | return contract_text |
| 860 | |
| 861 | async def load_from_json(self, file_path: Optional[str] = None, auto_initialize: bool = True) -> bool: |
| 862 | """Load memory configurations with version history from JSON. |
| 863 | |
| 864 | Loads basic configuration only (instance is not saved, must be created via build()). |
| 865 | Only the latest version will be instantiated by default if auto_initialize=True. |
| 866 | |
| 867 | Args: |
| 868 | file_path: File path to load from |
| 869 | auto_initialize: Whether to automatically create instance via build() after loading |
| 870 | |
| 871 | Returns: |
| 872 | True if loaded successfully, False otherwise |
| 873 | """ |
| 874 | |
| 875 | file_path = file_path if file_path is not None else self.save_path |
| 876 | |
| 877 | async with file_lock(file_path): |
| 878 | if not os.path.exists(file_path): |
| 879 | logger.warning(f"| ⚠️ Memory file not found: {file_path}") |
| 880 | return False |
| 881 | |
| 882 | try: |
| 883 | with open(file_path, "r", encoding="utf-8") as f: |
| 884 | load_data = json.load(f) |
| 885 | |
| 886 | memories_data = load_data.get("memory_systems", {}) |
| 887 | loaded_count = 0 |
| 888 | |
| 889 | for memory_name, memory_data in memories_data.items(): |
| 890 | try: |
| 891 | # Expected format: multiple versions stored as a dict {version_str: config_dict} |
| 892 | versions_data = memory_data.get("versions") |
| 893 | if not isinstance(versions_data, dict): |
| 894 | logger.warning(f"| ⚠️ Memory {memory_name} has invalid format for 'versions' (expected dict), skipping") |
| 895 | continue |
| 896 | |
| 897 | current_version_str = memory_data.get("current_version") |
| 898 | |
| 899 | # Load all versions |
| 900 | version_configs = [] |
| 901 | latest_config = None |
| 902 | latest_version = None |
| 903 | |
| 904 | for version_str, config_dict in versions_data.items(): |
| 905 | # Ensure version field is present |
| 906 | if "version" not in config_dict: |
| 907 | config_dict["version"] = version_str |
| 908 | |
| 909 | try: |
| 910 | memory_config = MemoryConfig.model_validate(config_dict) |
| 911 | version_configs.append(memory_config) |
| 912 | except Exception as e: |
| 913 | logger.warning(f"| ⚠️ Failed to load memory config for {memory_name}@{version_str}: {e}") |
| 914 | continue |
| 915 | |
| 916 | # Track latest version |
| 917 | if latest_config is None or ( |
| 918 | current_version_str and memory_config.version == current_version_str |
nothing calls this directly
no test coverage detected