# 從 invokeai/app/services/invocation_services.py 擷取 class InvocationServices: """Services that can be used by invocations""" def __init__(self, board_images: "BoardImagesServiceABC", boards: "BoardServiceABC", configuration: "InvokeAIAppConfig", events: "EventServiceBase", images: "ImageServiceABC", model_manager: "ModelManagerServiceBase", session_queue: "SessionQueueBase", # ... 更多服務 ): self.board_images = board_images self.boards = boards self.configuration = configuration self.events = events self.images = images self.model_manager = model_manager self.session_queue = session_queue

# 從 invokeai/app/services/shared/graph.py 擷取 class Edge(BaseModel): source: EdgeConnection = Field(description="The connection for the edge's from node and field") destination: EdgeConnection = Field(description="The connection for the edge's to node and field") class Graph(BaseModel): id: str = Field(description="The id of this graph") nodes: dict[str, BaseInvocation] = Field(description="The nodes in this graph") edges: list[Edge] = Field(description="The edges in this graph") def execute(self, context: InvocationContext) -> Iterator[BaseInvocationOutput]: # 使用 NetworkX 進行拓撲排序 g = nx.DiGraph() g.add_edges_from([(e.source.node_id, e.destination.node_id) for e in self.edges]) for node_id in nx.topological_sort(g): node = self.nodes[node_id] output = node.invoke(context) yield output

# 從 invokeai/backend/model_manager/load/model_cache/model_cache.py 擷取 class ModelCache: """A cache for managing models in memory. The cache is based on two levels of model storage: - execution_device: The device where models are executed (typically "cuda", "mps", or "cpu"). - storage_device: The device where models are offloaded when not in active use (typically "cpu"). """ def __init__(self, max_cache_size: int, execution_device: torch.device): self._max_cache_size = max_cache_size self._execution_device = execution_device self._storage_device = torch.device("cpu") self._cache_records: Dict[str, CacheRecord] = {} self._lock = threading.Lock() @synchronized def get_model(self, model_key: str) -> CacheRecord: # 檢查快取命中 if model_key in self._cache_records: self._move_to_execution_device(model_key) return self._cache_records[model_key] # 快取未命中，需要載入模型 self._make_room_for_model(model_size) loaded_model = self._load_model_from_disk(model_key) self._cache_records[model_key] = CacheRecord(loaded_model) return self._cache_records[model_key]

# 從 invokeai/backend/stable_diffusion/extensions_manager.py 擷取 class ExtensionsManager: def __init__(self, is_canceled: Callable[[], bool]): self._is_canceled = is_canceled self._extensions: List[ExtensionBase] = [] self._ordered_callbacks: Dict[ExtensionCallbackType, List[CallbackFunctionWithMetadata]] = {} def add_extension(self, extension: ExtensionBase): """Register an extension and its callbacks.""" self._extensions.append(extension) # 收集並排序回調函數 for callback_type in ExtensionCallbackType: if hasattr(extension, callback_type.value): callback_fn = getattr(extension, callback_type.value) self._register_callback(callback_type, callback_fn, extension.priority) def run_callback(self, callback_type: ExtensionCallbackType, ctx: DenoiseContext): """Execute all registered callbacks for a given type.""" for callback in self._ordered_callbacks.get(callback_type, []): if self._is_canceled(): break callback.function(ctx)

# 從 invokeai/backend/stable_diffusion/diffusion_backend.py 擷取 class StableDiffusionBackend(ABC): """Abstract base class for all diffusion model backends.""" @abstractmethod def denoise(self, context: DenoiseContext, latents: torch.Tensor, timesteps: List[int], noise: torch.Tensor) -> torch.Tensor: """Run the denoising process.""" pass # SD1.5/2.0 適配器 class StableDiffusion15Backend(StableDiffusionBackend): def denoise(self, context, latents, timesteps, noise): # SD 1.5 特定的去噪實現 for t in timesteps: latents = self.unet(latents, t, context.conditioning) return latents # SDXL 適配器 class SDXLBackend(StableDiffusionBackend): def denoise(self, context, latents, timesteps, noise): # SDXL 需要額外的條件處理 add_time_ids = self._get_add_time_ids(context) for t in timesteps: latents = self.unet(latents, t, context.conditioning, add_time_ids) return latents # FLUX 適配器 class FluxBackend(StableDiffusionBackend): def denoise(self, context, latents, timesteps, noise): # FLUX 使用不同的架構 packed_latents = self._pack_latents(latents) for t in timesteps: packed_latents = self.transformer(packed_latents, t, context) return self._unpack_latents(packed_latents)