| """ |
| Latent DAG 图节点增量缓存引擎 (ComfyUI Style) |
| Author: XiaoZhe (Commercial Contact: janejulius119@gmail.com / WeChat: julius119) |
| """ |
|
|
| import hashlib |
| import json |
| import logging |
| from typing import Dict, Any, Optional |
|
|
| logger = logging.getLogger("VerseFlow.GraphCache") |
|
|
| class LatentGraphCacheEngine: |
| """DAG 中间 Latent 特征缓存器，支持增量重绘与算力节省""" |
| def __init__(self): |
| self._cache_store: Dict[str, Dict[str, Any]] = {} |
|
|
| def compute_node_hash(self, node_id: str, inputs: Dict[str, Any], upstream_hashes: Dict[str, str]) -> str: |
| hasher = hashlib.sha256() |
| hasher.update(node_id.encode('utf-8')) |
| inputs_str = json.dumps(inputs, sort_keys=True, default=str) |
| hasher.update(inputs_str.encode('utf-8')) |
| |
| for up_id, up_hash in sorted(upstream_hashes.items()): |
| hasher.update(f"{up_id}:{up_hash}".encode('utf-8')) |
| |
| return hasher.hexdigest() |
|
|
| def get_cached_latent(self, node_hash: str) -> Optional[Dict[str, Any]]: |
| if node_hash in self._cache_store: |
| logger.info(f"⚡ [GraphCache Hit] 命中增量缓存 Hash: {node_hash[:12]}...") |
| return self._cache_store[node_hash] |
| return None |
|
|
| def store_latent(self, node_hash: str, latent_data: Dict[str, Any]): |
| self._cache_store[node_hash] = latent_data |
| logger.info(f"💾 [GraphCache Saved] 保存 Latent 特征至缓存 Hash: {node_hash[:12]}...") |
