From ece52e651ca91fcc62916bc89ae8378cfc0050d7 Mon Sep 17 00:00:00 2001 From: silveroxides Date: Fri, 14 Aug 2026 19:04:13 +0200 Subject: [PATCH] feat(analysis): add filtered API workflow runner --- nodes/model_analysis.py | 33 ++++++++++- run_model_analysis_api_workflow.py | 73 +++++++++++++++++++++++++ tests/test_model_analysis.py | 37 +++++++++++++ tests/test_model_analysis_api_runner.py | 56 +++++++++++++++++++ 4 files changed, 198 insertions(+), 1 deletion(-) create mode 100644 run_model_analysis_api_workflow.py create mode 100644 tests/test_model_analysis_api_runner.py diff --git a/nodes/model_analysis.py b/nodes/model_analysis.py index e6d409b..d0052d8 100644 --- a/nodes/model_analysis.py +++ b/nodes/model_analysis.py @@ -25,7 +25,7 @@ from .device_utils import ( estimate_model_size, prepare_for_large_operation, ) -from .merger import load_documentation_from_file +from .merger import _compile_patterns, _matches_any_pattern, load_documentation_from_file from .lora_resize import layer_tensor_keys, parse_lora_layers, validate_canonical_blocks from .quantization_guard import inspect_low_bit_input @@ -618,6 +618,8 @@ def _render_comparison( f"Shape mismatches: {len(inventory['shape'])}", f"Non-floating shared tensors: {len(inventory['nonfloat'])}", f"Unsupported low-bit tensors: {len(inventory['low_bit'])}", + f"Excluded A tensors: {len(inventory['excluded_a'])}", + f"Excluded B tensors: {len(inventory['excluded_b'])}", f"CUDA OOM CPU fallbacks: {len(fallback_keys)}", "", "PARAMETER-WEIGHTED GLOBAL METRICS", @@ -671,6 +673,7 @@ def _render_comparison( for label, name in ( ("A ONLY", "a_only"), ("B ONLY", "b_only"), ("SHAPE MISMATCH", "shape"), ("NON-FLOATING", "nonfloat"), ("LOW-BIT UNSUPPORTED", "low_bit"), + ("EXCLUDED A", "excluded_a"), ("EXCLUDED B", "excluded_b"), ): lines.append(f"[{label}] ({len(inventory[name])})") lines.extend(f"- {value}" for value in inventory[name]) @@ -773,9 +776,26 @@ class ModelAnalysisLogic: similarity_alignment = bool(params.get("cwb_similarity_alignment", False)) keys_a = set(handlers[0].keys()) keys_b = set(handlers[1].keys()) + glob_mode = bool(params.get("glob_patterns", False)) + excluded_patterns = _compile_patterns( + params.get("exclude_patterns", ""), + glob_mode=glob_mode, + ) + excluded_a = { + key for key in keys_a + if _matches_any_pattern(key, excluded_patterns, glob_mode=glob_mode) + } + excluded_b = { + key for key in keys_b + if _matches_any_pattern(key, excluded_patterns, glob_mode=glob_mode) + } + keys_a.difference_update(excluded_a) + keys_b.difference_update(excluded_b) inventory = { "a_only": [], "b_only": [], "shape": [], "nonfloat": [], "low_bit": [], + "excluded_a": sorted(excluded_a), + "excluded_b": sorted(excluded_b), } units: list[tuple[str, str, str]] = [] lora_pairs: list[tuple[str, dict[str, str], dict[str, str]]] = [] @@ -1068,6 +1088,17 @@ def _common_inputs(model_type: str, *, alignment_control: bool): default=True, tooltip="Run garbage collection and clear the CUDA allocator cache after every analyzed work unit. Saves retained memory but slows analysis.", ), + io.String.Input( + "exclude_patterns", + default="", + multiline=True, + tooltip="One pattern per line. Matching tensors are excluded from all comparison metrics and topology counts. Uses regex unless Glob Patterns is enabled.", + ), + io.Boolean.Input( + "glob_patterns", + default=False, + tooltip="Interpret exclusion entries as shell-style glob patterns instead of regular expressions.", + ), ]) return inputs diff --git a/run_model_analysis_api_workflow.py b/run_model_analysis_api_workflow.py new file mode 100644 index 0000000..2fa644a --- /dev/null +++ b/run_model_analysis_api_workflow.py @@ -0,0 +1,73 @@ +"""Queue a model-analysis API workflow and capture both report outputs.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +import run_cwb_api_workflow as api + + +DEFAULT_WORKFLOW = Path("ANALYZETESTCANDIDATE_API.json") +ANALYSIS_CLASS = "LoRAModelAnalysis" + + +def _linked_previews(workflow: dict, analysis_id: str) -> dict[int, str]: + linked = {} + for node_id, node in workflow.items(): + if node.get("class_type") != "PreviewAny": + continue + source = node.get("inputs", {}).get("source") + if isinstance(source, list) and len(source) == 2 and str(source[0]) == analysis_id: + linked[int(source[1])] = node_id + missing = [index for index in (0, 1) if index not in linked] + if missing: + raise RuntimeError(f"Missing PreviewAny capture for analysis output(s): {missing}") + return linked + + +def run_analysis(workflow: dict, server: str, poll_seconds: float) -> dict: + analysis_id, _ = api._single_node(workflow, ANALYSIS_CLASS) + linked = _linked_previews(workflow, analysis_id) + queued = api._json_request( + f"{server.rstrip('/')}/prompt", + api._prompt_payload(workflow), + ) + prompt_id = queued["prompt_id"] + history = api._wait_for_history(server.rstrip("/"), prompt_id, poll_seconds) + status = history.get("status", {}) + if status.get("status_str") == "error": + raise RuntimeError(f"Analysis failed: {json.dumps(status, indent=2)}") + outputs = history.get("outputs", {}) + return { + "prompt_id": prompt_id, + "comparison_report": api._text_output(outputs, linked[0]), + "cwb_report": api._text_output(outputs, linked[1]), + } + + +def main() -> int: + parser = argparse.ArgumentParser(description="Run a model-analysis API workflow.") + parser.add_argument("--workflow", type=Path, default=DEFAULT_WORKFLOW) + parser.add_argument("--server", default="http://127.0.0.1:8188") + parser.add_argument("--poll-seconds", type=float, default=2.0) + parser.add_argument("--result-json", type=Path) + args = parser.parse_args() + + workflow = json.loads(args.workflow.read_text(encoding="utf-8")) + result = run_analysis(workflow, args.server, args.poll_seconds) + if args.result_json: + args.result_json.parent.mkdir(parents=True, exist_ok=True) + args.result_json.write_text( + json.dumps(result, indent=2, ensure_ascii=False) + "\n", + encoding="utf-8", + ) + sys.stdout.write(f"Prompt: {result['prompt_id']}\n") + sys.stdout.write("Analysis reports captured\n") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_model_analysis.py b/tests/test_model_analysis.py index 3facc98..145f86b 100644 --- a/tests/test_model_analysis.py +++ b/tests/test_model_analysis.py @@ -32,6 +32,8 @@ def _params(**overrides): "top_weight_differences": 3, "process_device": "cpu", "force_clear_cache": False, + "exclude_patterns": "", + "glob_patterns": False, } params.update(overrides) return params @@ -225,6 +227,41 @@ def test_lora_cross_format_keys_compare_as_one_logical_pair(monkeypatch, tmp_pat assert all(len(call[1]) == 2 for call in async_calls) +def test_excluded_lora_blocks_do_not_enter_metrics(monkeypatch, tmp_path, analysis): + path_a = tmp_path / "lora_a.safetensors" + path_b = tmp_path / "lora_b.safetensors" + tensors_a = {} + tensors_b = {} + for block, value_a, value_b in ((3, 1.0, 2.0), (4, 10.0, 20.0)): + prefix = f"diffusion_model.blocks.{block}.attn" + tensors_a[f"{prefix}.lora_A.weight"] = torch.full((1, 2), value_a) + tensors_a[f"{prefix}.lora_B.weight"] = torch.full((2, 1), value_a) + tensors_b[f"{prefix}.lora_A.weight"] = torch.full((1, 2), value_b) + tensors_b[f"{prefix}.lora_B.weight"] = torch.full((2, 1), value_b) + save_file(tensors_a, str(path_a)) + save_file(tensors_b, str(path_b)) + paths = {"a": str(path_a), "b": str(path_b)} + monkeypatch.setattr(analysis.folder_paths, "get_full_path", lambda _, name: paths[name]) + monkeypatch.setattr(analysis, "prepare_for_large_operation", lambda *args, **kwargs: None) + monkeypatch.setattr(analysis, "cleanup_after_operation", lambda: None) + + comparison, cwb_report = analysis.ModelAnalysisLogic.execute( + "a", "b", "loras", + _params(exclude_patterns=r"^diffusion_model\.blocks\.4\."), + lora_mode=True, + ) + + assert "Comparable floating tensors: 2" in comparison + assert "Excluded A tensors: 2" in comparison + assert "Excluded B tensors: 2" in comparison + layerwise = comparison.split("LAYERWISE TENSOR METRICS", 1)[1].split( + "TOP SCALAR WEIGHT DIFFERENCES", 1 + )[0] + assert "blocks_3_" in layerwise + assert "blocks_4_" not in layerwise + assert "blocks_4_" not in cwb_report + + def test_embedding_similarity_alignment_is_explicit_opt_in(analysis): a = torch.tensor([[1.0, 0.0], [0.0, 1.0]]) b = torch.tensor([[0.0, 1.0], [1.0, 0.0]]) diff --git a/tests/test_model_analysis_api_runner.py b/tests/test_model_analysis_api_runner.py new file mode 100644 index 0000000..7ea0c5b --- /dev/null +++ b/tests/test_model_analysis_api_runner.py @@ -0,0 +1,56 @@ +import importlib.util +from pathlib import Path + +import pytest + + +REPO_ROOT = Path(__file__).resolve().parents[1] + + +@pytest.fixture(scope="module") +def runner(): + spec = importlib.util.spec_from_file_location( + "model_analysis_api_runner_test", + REPO_ROOT / "run_model_analysis_api_workflow.py", + ) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_linked_previews_require_both_reports(runner): + workflow = { + "1": {"class_type": "LoRAModelAnalysis", "inputs": {}}, + "2": {"class_type": "PreviewAny", "inputs": {"source": ["1", 0]}}, + "3": {"class_type": "PreviewAny", "inputs": {"source": ["1", 1]}}, + } + assert runner._linked_previews(workflow, "1") == {0: "2", 1: "3"} + + +def test_linked_previews_reject_missing_report(runner): + workflow = { + "1": {"class_type": "LoRAModelAnalysis", "inputs": {}}, + "2": {"class_type": "PreviewAny", "inputs": {"source": ["1", 0]}}, + } + with pytest.raises(RuntimeError, match="output"): + runner._linked_previews(workflow, "1") + + +def test_run_analysis_captures_both_reports(monkeypatch, runner): + workflow = { + "1": {"class_type": "LoRAModelAnalysis", "inputs": {}}, + "2": {"class_type": "PreviewAny", "inputs": {"source": ["1", 0]}}, + "3": {"class_type": "PreviewAny", "inputs": {"source": ["1", 1]}}, + } + monkeypatch.setattr(runner.api, "_json_request", lambda *args: {"prompt_id": "p"}) + monkeypatch.setattr( + runner.api, + "_wait_for_history", + lambda *args: {"outputs": {"2": {"text": ["standard"]}, "3": {"text": ["cwb"]}}}, + ) + result = runner.run_analysis(workflow, "http://server", 0.1) + assert result == { + "prompt_id": "p", + "comparison_report": "standard", + "cwb_report": "cwb", + }