Next: Allow TIs to be either a key or a name in the prompt during our transition to using keys

invokeai/app/invocations/compel.py

@@ -3,9 +3,8 @@
 import torch
 from compel import Compel, ReturnedEmbeddingsType
 from compel.prompt_parser import Blend, Conjunction, CrossAttentionControlSubstitute, FlattenedPrompt, Fragment
-from transformers import CLIPTokenizer
+from transformers import CLIPTextModel, CLIPTokenizer
 
-import invokeai.backend.util.logging as logger
 from invokeai.app.invocations.fields import (
     FieldDescriptions,
     Input,
@@ -14,19 +13,16 @@
     UIComponent,
 )
 from invokeai.app.invocations.primitives import ConditioningOutput
-from invokeai.app.services.model_records import UnknownModelException
 from invokeai.app.services.shared.invocation_context import InvocationContext
-from invokeai.app.util.ti_utils import extract_ti_triggers_from_prompt
+from invokeai.app.util.ti_utils import generate_ti_list
 from invokeai.backend.lora import LoRAModelRaw
-from invokeai.backend.model_manager import ModelType
 from invokeai.backend.model_patcher import ModelPatcher
 from invokeai.backend.stable_diffusion.diffusion.conditioning_data import (
     BasicConditioningInfo,
     ConditioningFieldData,
     ExtraConditioningInfo,
     SDXLConditioningInfo,
 )
-from invokeai.backend.textual_inversion import TextualInversionModelRaw
 from invokeai.backend.util.devices import torch_dtype
 
 from .baseinvocation import (
@@ -70,7 +66,11 @@ class CompelInvocation(BaseInvocation):
     @torch.no_grad()
     def invoke(self, context: InvocationContext) -> ConditioningOutput:
         tokenizer_info = context.models.load(**self.clip.tokenizer.model_dump())
+        tokenizer_model = tokenizer_info.model
+        assert isinstance(tokenizer_model, CLIPTokenizer)
         text_encoder_info = context.models.load(**self.clip.text_encoder.model_dump())
+        text_encoder_model = text_encoder_info.model
+        assert isinstance(text_encoder_model, CLIPTextModel)
 
         def _lora_loader() -> Iterator[Tuple[LoRAModelRaw, float]]:
             for lora in self.clip.loras:
@@ -82,30 +82,20 @@ def _lora_loader() -> Iterator[Tuple[LoRAModelRaw, float]]:
 
         # loras = [(context.models.get(**lora.dict(exclude={"weight"})).context.model, lora.weight) for lora in self.clip.loras]
 
-        ti_list = []
-        for trigger in extract_ti_triggers_from_prompt(self.prompt):
-            name = trigger[1:-1]
-            try:
-                loaded_model = context.models.load(key=name).model
-                assert isinstance(loaded_model, TextualInversionModelRaw)
-                ti_list.append((name, loaded_model))
-            except UnknownModelException:
-                # print(e)
-                # import traceback
-                # print(traceback.format_exc())
-                print(f'Warn: trigger: "{trigger}" not found')
+        ti_list = generate_ti_list(self.prompt, text_encoder_info.config.base, context)
 
         with (
-            ModelPatcher.apply_ti(tokenizer_info.model, text_encoder_info.model, ti_list) as (
+            ModelPatcher.apply_ti(tokenizer_model, text_encoder_model, ti_list) as (
                 tokenizer,
                 ti_manager,
             ),
             text_encoder_info as text_encoder,
             # Apply the LoRA after text_encoder has been moved to its target device for faster patching.
             ModelPatcher.apply_lora_text_encoder(text_encoder, _lora_loader()),
             # Apply CLIP Skip after LoRA to prevent LoRA application from failing on skipped layers.
-            ModelPatcher.apply_clip_skip(text_encoder_info.model, self.clip.skipped_layers),
+            ModelPatcher.apply_clip_skip(text_encoder_model, self.clip.skipped_layers),
         ):
+            assert isinstance(text_encoder, CLIPTextModel)
             compel = Compel(
                 tokenizer=tokenizer,
                 text_encoder=text_encoder,
@@ -155,7 +145,11 @@ def run_clip_compel(
         zero_on_empty: bool,
     ) -> Tuple[torch.Tensor, Optional[torch.Tensor], Optional[ExtraConditioningInfo]]:
         tokenizer_info = context.models.load(**clip_field.tokenizer.model_dump())
+        tokenizer_model = tokenizer_info.model
+        assert isinstance(tokenizer_model, CLIPTokenizer)
         text_encoder_info = context.models.load(**clip_field.text_encoder.model_dump())
+        text_encoder_model = text_encoder_info.model
+        assert isinstance(text_encoder_model, CLIPTextModel)
 
         # return zero on empty
         if prompt == "" and zero_on_empty:
@@ -189,34 +183,20 @@ def _lora_loader() -> Iterator[Tuple[LoRAModelRaw, float]]:
 
         # loras = [(context.models.get(**lora.dict(exclude={"weight"})).context.model, lora.weight) for lora in self.clip.loras]
 
-        ti_list = []
-        for trigger in extract_ti_triggers_from_prompt(prompt):
-            name = trigger[1:-1]
-            try:
-                ti_model = context.models.load_by_attrs(
-                    model_name=name, base_model=text_encoder_info.config.base, model_type=ModelType.TextualInversion
-                ).model
-                assert isinstance(ti_model, TextualInversionModelRaw)
-                ti_list.append((name, ti_model))
-            except UnknownModelException:
-                # print(e)
-                # import traceback
-                # print(traceback.format_exc())
-                logger.warning(f'trigger: "{trigger}" not found')
-            except ValueError:
-                logger.warning(f'trigger: "{trigger}" more than one similarly-named textual inversion models')
+        ti_list = generate_ti_list(prompt, text_encoder_info.config.base, context)
 
         with (
-            ModelPatcher.apply_ti(tokenizer_info.model, text_encoder_info.model, ti_list) as (
+            ModelPatcher.apply_ti(tokenizer_model, text_encoder_model, ti_list) as (
                 tokenizer,
                 ti_manager,
             ),
             text_encoder_info as text_encoder,
             # Apply the LoRA after text_encoder has been moved to its target device for faster patching.
             ModelPatcher.apply_lora(text_encoder, _lora_loader(), lora_prefix),
             # Apply CLIP Skip after LoRA to prevent LoRA application from failing on skipped layers.
-            ModelPatcher.apply_clip_skip(text_encoder_info.model, clip_field.skipped_layers),
+            ModelPatcher.apply_clip_skip(text_encoder_model, clip_field.skipped_layers),
         ):
+            assert isinstance(text_encoder, CLIPTextModel)
             compel = Compel(
                 tokenizer=tokenizer,
                 text_encoder=text_encoder,

invokeai/app/util/ti_utils.py

@@ -1,8 +1,47 @@
 import re
+from typing import List, Tuple
 
+import invokeai.backend.util.logging as logger
+from invokeai.app.services.model_records import UnknownModelException
+from invokeai.app.services.shared.invocation_context import InvocationContext
+from invokeai.backend.model_manager.config import BaseModelType, ModelType
+from invokeai.backend.textual_inversion import TextualInversionModelRaw
 
-def extract_ti_triggers_from_prompt(prompt: str) -> list[str]:
-    ti_triggers = []
+
+def extract_ti_triggers_from_prompt(prompt: str) -> List[str]:
+    ti_triggers: List[str] = []
     for trigger in re.findall(r"<[a-zA-Z0-9., _-]+>", prompt):
-        ti_triggers.append(trigger)
+        ti_triggers.append(str(trigger))
     return ti_triggers
+
+
+def generate_ti_list(
+    prompt: str, base: BaseModelType, context: InvocationContext
+) -> List[Tuple[str, TextualInversionModelRaw]]:
+    ti_list: List[Tuple[str, TextualInversionModelRaw]] = []
+    for trigger in extract_ti_triggers_from_prompt(prompt):
+        name_or_key = trigger[1:-1]
+        try:
+            loaded_model = context.models.load(key=name_or_key)
+            model = loaded_model.model
+            assert isinstance(model, TextualInversionModelRaw)
+            assert loaded_model.config.base == base
+            ti_list.append((name_or_key, model))
+        except UnknownModelException:
+            try:
+                loaded_model = context.models.load_by_attrs(
+                    model_name=name_or_key, base_model=base, model_type=ModelType.TextualInversion
+                )
+                model = loaded_model.model
+                assert isinstance(model, TextualInversionModelRaw)
+                assert loaded_model.config.base == base
+                ti_list.append((name_or_key, model))
+            except UnknownModelException:
+                pass
+        except ValueError:
+            logger.warning(f'trigger: "{trigger}" more than one similarly-named textual inversion models')
+        except AssertionError:
+            logger.warning(f'trigger: "{trigger}" not a valid textual inversion model for this graph')
+        except Exception:
+            logger.warning(f'Failed to load TI model for trigger: "{trigger}"')
+    return ti_list