diff options
| author | tanaka takashi <isazaka@gmail.com> | 2026-09-12 18:04:56 +0900 |
|---|---|---|
| committer | tanaka takashi <isazaka@gmail.com> | 2026-09-12 18:04:56 +0900 |
| commit | cba15a62829a0fe57a03c8031f981e8f790e2efc (patch) | |
| tree | 3bc0ef1a243f356f55c2fa97f63c3bfd5c77e8fc /src/mcp_sd15/pipeline_factory.py | |
feat: Stable Diffusion 1.x MCPサーバを実装
仕様書・設計書に基づき、ローカルCUDA GPU上のStable Diffusion 1.x互換モデルをMCP経由で利用するサーバを新規実装。
実装内容:
- generate_image、generate_guided_image、inpaint_imageの3ツールをMCP SDK v2で公開
- TOMLからcheckpoint、VAE、複数LoRAと重み、ControlNet、生成設定を読み込み
- モデル種別ごとの固定ディレクトリ解決と起動時の設定・ファイル存在検証を追加
- Depth/MiDaS、OpenPose body/full、ControlNet Inpaintの前処理とパイプライン切替を実装
- FP16/CUDA、1枚固定生成、ControlNetオンデマンドロード、単一GPU直列実行を実装
- Base64 PNG/JPEG入力の形式・サイズ検証と任意のdata URI入力に対応
- seed、サイズ、モード、時刻、処理時間をPNG metadataへ記録
- 任意のローカルPNG保存、生成エラー分類、CUDA OOM後のキャッシュ回復を追加
- --configと--versionを備えたstdio CLIエントリポイントを追加
開発・運用:
- Python 3.11およびuvによる依存・lock管理を構成
- config.example.tomlとセットアップ・運用手順をREADMEへ追加
- Ruff、Pyright、pytestの設定を追加
- 設定、パス解決、schema、画像検証、サービスルーティング、排他制御、MCP画像返却をテスト
検証結果:
- pytest: 30 passed
- Ruff lint/format: passed
- Pyright: 0 errors, 0 warnings
- CLI versionおよびMCPクライアント経由のPNG返却を確認
- 実GPU smoke testはCUDA GPUとモデルファイルがないため対象外
Diffstat (limited to 'src/mcp_sd15/pipeline_factory.py')
| -rw-r--r-- | src/mcp_sd15/pipeline_factory.py | 133 |
1 files changed, 133 insertions, 0 deletions
diff --git a/src/mcp_sd15/pipeline_factory.py b/src/mcp_sd15/pipeline_factory.py new file mode 100644 index 0000000..cd92690 --- /dev/null +++ b/src/mcp_sd15/pipeline_factory.py @@ -0,0 +1,133 @@ +"""Lazy Diffusers construction kept outside the MCP layer.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from .config import AppConfig +from .errors import ConfigError + + +class PipelineFactory: + """Build SD1.x pipelines in FP16 without network access.""" + + def __init__(self, config: AppConfig) -> None: + self.config = config + + @staticmethod + def _runtime() -> tuple[Any, Any]: + try: + import diffusers + import torch + except ImportError as exc: + raise ConfigError(f"inference dependency is unavailable: {exc}") from exc + if not torch.cuda.is_available(): + raise ConfigError("CUDA GPU is required but torch.cuda.is_available() is false") + return torch, diffusers + + def create_base(self) -> Any: + torch, diffusers = self._runtime() + try: + pipeline = diffusers.StableDiffusionPipeline.from_single_file( + str(self.config.model.checkpoint_path()), + torch_dtype=torch.float16, + local_files_only=True, + ) + vae_path = self.config.model.vae_path() + if vae_path is not None: + pipeline.vae = diffusers.AutoencoderKL.from_single_file( + str(vae_path), + torch_dtype=torch.float16, + local_files_only=True, + ) + adapter_names: list[str] = [] + adapter_weights: list[float] = [] + for index, (item, path) in enumerate( + zip(self.config.model.loras, self.config.model.lora_paths(), strict=True) + ): + name = f"configured_lora_{index}" + pipeline.load_lora_weights( + str(path.parent), + weight_name=path.name, + adapter_name=name, + local_files_only=True, + ) + adapter_names.append(name) + adapter_weights.append(item.weight) + if adapter_names: + pipeline.set_adapters(adapter_names, adapter_weights=adapter_weights) + pipeline.scheduler = self._scheduler(pipeline.scheduler, diffusers) + pipeline.set_progress_bar_config(disable=True) + return pipeline.to("cuda") + except ConfigError: + raise + except Exception as exc: + raise ConfigError(f"failed to load SD1.x model configuration: {exc}") from exc + + def create_controlnet(self, path: Path) -> Any: + torch, diffusers = self._runtime() + try: + return diffusers.ControlNetModel.from_single_file( + str(path), torch_dtype=torch.float16, local_files_only=True + ).to("cuda") + except Exception as exc: + raise ConfigError(f"failed to load ControlNet model {path}: {exc}") from exc + + @staticmethod + def create_guided_pipeline(base: Any, controlnet: Any) -> Any: + from diffusers.pipelines.controlnet.pipeline_controlnet import ( + StableDiffusionControlNetPipeline, + ) + + components: dict[str, Any] = dict( + vae=base.vae, + text_encoder=base.text_encoder, + tokenizer=base.tokenizer, + unet=base.unet, + controlnet=controlnet, + scheduler=base.scheduler, + safety_checker=base.safety_checker, + feature_extractor=base.feature_extractor, + requires_safety_checker=getattr(base, "requires_safety_checker", False), + ) + if getattr(base, "image_encoder", None) is not None: + components["image_encoder"] = base.image_encoder + pipeline = StableDiffusionControlNetPipeline(**components) + pipeline.set_progress_bar_config(disable=True) + return pipeline.to("cuda") + + @staticmethod + def create_inpaint_pipeline(base: Any, controlnet: Any) -> Any: + from diffusers.pipelines.controlnet.pipeline_controlnet_inpaint import ( + StableDiffusionControlNetInpaintPipeline, + ) + + components: dict[str, Any] = dict( + vae=base.vae, + text_encoder=base.text_encoder, + tokenizer=base.tokenizer, + unet=base.unet, + controlnet=controlnet, + scheduler=base.scheduler, + safety_checker=base.safety_checker, + feature_extractor=base.feature_extractor, + requires_safety_checker=getattr(base, "requires_safety_checker", False), + ) + if getattr(base, "image_encoder", None) is not None: + components["image_encoder"] = base.image_encoder + pipeline = StableDiffusionControlNetInpaintPipeline(**components) + pipeline.set_progress_bar_config(disable=True) + return pipeline.to("cuda") + + def _scheduler(self, scheduler: Any, diffusers: Any) -> Any: + classes = { + "dpmpp_2m": diffusers.DPMSolverMultistepScheduler, + "ddim": diffusers.DDIMScheduler, + "euler": diffusers.EulerDiscreteScheduler, + "euler_a": diffusers.EulerAncestralDiscreteScheduler, + } + options = dict(scheduler.config) + if self.config.generation.scheduler == "dpmpp_2m": + options.update(algorithm_type="dpmsolver++", solver_order=2) + return classes[self.config.generation.scheduler].from_config(options) |
