Upload agentframe/memory/store.py with huggingface_hub

agentframe/memory/store.py

@@ -0,0 +1,53 @@
+"""持久化: 状态保存/加载 (JSON 快照)"""
+import json
+import os
+import numpy as np
+
+
+def _serialize(obj):
+    """递归序列化 (处理 numpy 类型)"""
+    if isinstance(obj, np.ndarray):
+        return {"__ndarray__": True, "shape": list(obj.shape),
+                "dtype": str(obj.dtype), "data": obj.tobytes().hex()}
+    if isinstance(obj, np.generic):
+        return obj.item()
+    if isinstance(obj, dict):
+        return {str(k): _serialize(v) for k, v in obj.items()}
+    if isinstance(obj, (list, tuple)):
+        return [_serialize(v) for v in obj]
+    if hasattr(obj, "__dict__"):
+        return _serialize(obj.__dict__)
+    return obj
+
+
+def _deserialize(obj):
+    """递归反序列化"""
+    if isinstance(obj, dict):
+        if obj.get("__ndarray__"):
+            arr = np.frombuffer(bytes.fromhex(obj["data"]), dtype=obj["dtype"])
+            return arr.reshape(obj["shape"])
+        return {k: _deserialize(v) for k, v in obj.items()}
+    if isinstance(obj, list):
+        return [_deserialize(v) for v in obj]
+    return obj
+
+
+class StateStore:
+    """JSON 快照存储: 保存引擎状态到文件, 可恢复"""
+
+    def __init__(self, path: str):
+        self.path = path
+
+    def save(self, state: dict):
+        os.makedirs(os.path.dirname(self.path) or ".", exist_ok=True)
+        with open(self.path, "w") as f:
+            json.dump(_serialize(state), f, ensure_ascii=False)
+
+    def load(self) -> dict:
+        if not os.path.exists(self.path):
+            return {}
+        with open(self.path) as f:
+            return _deserialize(json.load(f))
+
+    def exists(self) -> bool:
+        return os.path.exists(self.path)