(
self,
payload: CreateEmbeddingRequest,
)
| 13134 | self.model.truncate_draft_sequence(seq_id, 0) |
| 13135 | |
| 13136 | def create_embedding( |
| 13137 | self, |
| 13138 | payload: CreateEmbeddingRequest, |
| 13139 | ) -> CreateEmbeddingResponse: |
| 13140 | if not self.is_idle(): |
| 13141 | raise RuntimeError("embedding requests require an idle scheduler") |
| 13142 | self.clear_resident_state() |
| 13143 | try: |
| 13144 | embeddings, total_tokens = self.model.embed(payload.normalized_input()) |
| 13145 | return CreateEmbeddingResponse.from_embeddings( |
| 13146 | model=payload.model, |
| 13147 | embeddings=embeddings, |
| 13148 | total_tokens=total_tokens, |
| 13149 | encoding_format=payload.encoding_format, |
| 13150 | dimensions=payload.dimensions, |
| 13151 | ) |
| 13152 | finally: |
| 13153 | self.clear_resident_state() |
| 13154 | |
| 13155 | @staticmethod |
| 13156 | def request_needs_prompt_logits(request: CompletionRequest) -> bool: |
nothing calls this directly
no test coverage detected