(
self,
attn,
hidden_states,
encoder_hidden_states=None,
attention_mask=None,
temb=None,
scale=1.0,
)
| 2014 | self.to_v_ip = nn.Linear(cross_attention_dim or hidden_size, hidden_size, bias=False) |
| 2015 | |
| 2016 | def __call__( |
| 2017 | self, |
| 2018 | attn, |
| 2019 | hidden_states, |
| 2020 | encoder_hidden_states=None, |
| 2021 | attention_mask=None, |
| 2022 | temb=None, |
| 2023 | scale=1.0, |
| 2024 | ): |
| 2025 | if scale != 1.0: |
| 2026 | logger.warning("`scale` of IPAttnProcessor should be set with `set_ip_adapter_scale`.") |
| 2027 | residual = hidden_states |
| 2028 | |
| 2029 | if attn.spatial_norm is not None: |
| 2030 | hidden_states = attn.spatial_norm(hidden_states, temb) |
| 2031 | |
| 2032 | input_ndim = hidden_states.ndim |
| 2033 | |
| 2034 | if input_ndim == 4: |
| 2035 | batch_size, channel, height, width = hidden_states.shape |
| 2036 | hidden_states = hidden_states.view(batch_size, channel, height * width).transpose(1, 2) |
| 2037 | |
| 2038 | batch_size, sequence_length, _ = ( |
| 2039 | hidden_states.shape if encoder_hidden_states is None else encoder_hidden_states.shape |
| 2040 | ) |
| 2041 | attention_mask = attn.prepare_attention_mask(attention_mask, sequence_length, batch_size) |
| 2042 | |
| 2043 | if attn.group_norm is not None: |
| 2044 | hidden_states = attn.group_norm(hidden_states.transpose(1, 2)).transpose(1, 2) |
| 2045 | |
| 2046 | query = attn.to_q(hidden_states) |
| 2047 | |
| 2048 | if encoder_hidden_states is None: |
| 2049 | encoder_hidden_states = hidden_states |
| 2050 | elif attn.norm_cross: |
| 2051 | encoder_hidden_states = attn.norm_encoder_hidden_states(encoder_hidden_states) |
| 2052 | |
| 2053 | # split hidden states |
| 2054 | end_pos = encoder_hidden_states.shape[1] - self.num_tokens |
| 2055 | encoder_hidden_states, ip_hidden_states = ( |
| 2056 | encoder_hidden_states[:, :end_pos, :], |
| 2057 | encoder_hidden_states[:, end_pos:, :], |
| 2058 | ) |
| 2059 | |
| 2060 | key = attn.to_k(encoder_hidden_states) |
| 2061 | value = attn.to_v(encoder_hidden_states) |
| 2062 | |
| 2063 | query = attn.head_to_batch_dim(query) |
| 2064 | key = attn.head_to_batch_dim(key) |
| 2065 | value = attn.head_to_batch_dim(value) |
| 2066 | |
| 2067 | attention_probs = attn.get_attention_scores(query, key, attention_mask) |
| 2068 | hidden_states = torch.bmm(attention_probs, value) |
| 2069 | hidden_states = attn.batch_to_head_dim(hidden_states) |
| 2070 | |
| 2071 | # for ip-adapter |
| 2072 | ip_key = self.to_k_ip(ip_hidden_states) |
| 2073 | ip_value = self.to_v_ip(ip_hidden_states) |
nothing calls this directly
no test coverage detected