File size: 1,478 Bytes
8ab53f1 | 1 2 3 4 5 6 7 8 9 10 11 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 | """
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]}...") |