main
py 169 lines 5.49 KB
Raw
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 from abc import ABC, abstractmethod
4 from collections.abc import Set
5 from typing import TypeAlias
6
7 import torch
8 import torch.nn as nn
9
10 from vllm.model_executor.layers.pooler import ActivationFn, ClassifierFn, ProjectorFn
11 from vllm.pooling_params import PoolingParams
12 from vllm.tasks import PoolingTask
13 from vllm.v1.pool.metadata import PoolingMetadata
14
15 from .methods import TokenPoolingMethodOutputItem
16
17 TokenPoolerHeadOutputItem: TypeAlias = torch.Tensor | None
18
19
20 class TokenPoolerHead(nn.Module, ABC):
21 @abstractmethod
22 def get_supported_tasks(self) -> Set[PoolingTask]:
23 raise NotImplementedError
24
25 @abstractmethod
26 def forward_chunk(
27 self,
28 pooled_data: TokenPoolingMethodOutputItem,
29 pooling_param: PoolingParams,
30 ) -> TokenPoolerHeadOutputItem:
31 raise NotImplementedError
32
33 def forward(
34 self,
35 pooled_data: list[TokenPoolingMethodOutputItem],
36 pooling_metadata: PoolingMetadata,
37 ) -> list[TokenPoolerHeadOutputItem]:
38 pooling_params = pooling_metadata.pooling_params
39 if len(pooled_data) != len(pooling_params):
40 raise ValueError(
41 f"pooled_data length ({len(pooled_data)}) does not match "
42 f"pooling_params length ({len(pooling_params)})"
43 )
44
45 return [self.forward_chunk(d, p) for d, p in zip(pooled_data, pooling_params)]
46
47
48 class TokenEmbeddingPoolerHead(TokenPoolerHead):
49 def __init__(
50 self,
51 head_dtype: torch.dtype | str | None = None,
52 projector: ProjectorFn | None = None,
53 activation: ActivationFn | None = None,
54 ) -> None:
55 super().__init__()
56
57 self.head_dtype = head_dtype
58 self.projector = projector
59 self.activation = activation
60
61 def extra_repr(self) -> str:
62 attrs = []
63 if self.head_dtype is not None:
64 attrs.append(f"head_dtype={self.head_dtype}")
65 if self.projector is not None:
66 attrs.append("projector=True")
67 if self.activation is not None:
68 attrs.append(f"activation={self.activation.__class__.__name__}")
69 return ", ".join(attrs)
70
71 def get_supported_tasks(self) -> Set[PoolingTask]:
72 return {"token_embed"}
73
74 def forward_chunk(
75 self,
76 pooled_data: TokenPoolingMethodOutputItem,
77 pooling_param: PoolingParams,
78 ) -> TokenPoolerHeadOutputItem:
79 # for unfinished chunked prefill
80 if pooled_data is None:
81 return None
82
83 if self.head_dtype is not None:
84 pooled_data = pooled_data.to(self.head_dtype)
85 # pooled_data shape: [n_tokens, hidden_size]
86
87 # Apply ST projector
88 if self.projector is not None:
89 embeddings = self.projector(pooled_data)
90 else:
91 embeddings = pooled_data
92 # embeddings shape: [n_tokens, embedding_size]
93
94 # for matryoshka representation
95 if pooling_param.dimensions is not None:
96 embeddings = embeddings[..., : pooling_param.dimensions]
97
98 # for normalize
99 if self.activation is not None and pooling_param.use_activation:
100 embeddings = self.activation(embeddings)
101
102 # embeddings shape: [n_tokens, embedding_size]
103 return embeddings
104
105
106 class TokenClassifierPoolerHead(TokenPoolerHead):
107 def __init__(
108 self,
109 classifier: ClassifierFn | None = None,
110 logit_mean: float | None = None,
111 logit_sigma: float | None = None,
112 head_dtype: torch.dtype | str | None = None,
113 activation: ActivationFn | None = None,
114 ) -> None:
115 super().__init__()
116
117 self.classifier = classifier
118 self.logit_mean = logit_mean
119 self.logit_sigma = logit_sigma
120 self.head_dtype = head_dtype
121 self.activation = activation
122
123 def extra_repr(self) -> str:
124 attrs = []
125 if self.head_dtype is not None:
126 attrs.append(f"head_dtype={self.head_dtype}")
127 if self.classifier is not None:
128 attrs.append("classifier=True")
129 if self.logit_mean is not None:
130 attrs.append(f"logit_mean={self.logit_mean}")
131 if self.logit_sigma is not None:
132 attrs.append(f"logit_sigma={self.logit_sigma}")
133 if self.activation is not None:
134 attrs.append(f"activation={self.activation.__class__.__name__}")
135 return ", ".join(attrs)
136
137 def get_supported_tasks(self) -> Set[PoolingTask]:
138 return {"token_classify"}
139
140 def forward_chunk(
141 self,
142 pooled_data: TokenPoolingMethodOutputItem,
143 pooling_param: PoolingParams,
144 ) -> TokenPoolerHeadOutputItem:
145 # for unfinished chunked prefill
146 if pooled_data is None:
147 return None
148
149 if self.head_dtype is not None:
150 pooled_data = pooled_data.to(self.head_dtype)
151 # hidden_states shape: [n_token, hidden_size]
152
153 if self.classifier is not None:
154 logits = self.classifier(pooled_data)
155 else:
156 logits = pooled_data
157 # logits shape: [n_token, num_labels]
158
159 # Affine score calibration: activation((logit - mean) / sigma)
160 if self.logit_mean is not None:
161 logits = logits - self.logit_mean
162 if self.logit_sigma is not None:
163 logits = logits / self.logit_sigma
164
165 if self.activation is not None and pooling_param.use_activation:
166 logits = self.activation(logits)
167
168 # logits shape: [n_token, num_labels]
169 return logits