| 210 | self.max_concurrent = max_concurrent |
| 211 | |
| 212 | def embed(self, texts: List[str], max_retries: int = 3) -> List[np.ndarray]: |
| 213 | # Validate all texts before sending |
| 214 | validated_texts = [] |
| 215 | for text in texts: |
| 216 | if not validate_token_length(text, MAX_TOKENS): |
| 217 | # If validation fails, re-clip the text more aggressively |
| 218 | text = clip_to_max_tokens(text, MAX_TOKENS - 10) # Extra buffer |
| 219 | if not validate_token_length(text, MAX_TOKENS): |
| 220 | # Last resort: truncate very aggressively |
| 221 | tokens = get_tokenizer().encode(text, add_special_tokens=False) |
| 222 | safe_tokens = tokens[: MAX_TOKENS - 10] |
| 223 | text = get_tokenizer().decode(safe_tokens, skip_special_tokens=True) |
| 224 | validated_texts.append(text) |
| 225 | |
| 226 | # Retry logic for network resilience under high load |
| 227 | for attempt in range(max_retries): |
| 228 | try: |
| 229 | resp = self.client.embeddings.create( |
| 230 | model="Snowflake/snowflake-arctic-embed-m-v1.5", |
| 231 | input=validated_texts, |
| 232 | encoding_format="float", |
| 233 | ) |
| 234 | # Convert to float32 for downstream usage |
| 235 | return [np.array(e.embedding, dtype=np.float32) for e in resp.data] |
| 236 | except Exception as e: |
| 237 | if attempt == max_retries - 1: |
| 238 | raise e |
| 239 | # Exponential backoff on retry |
| 240 | wait_time = (2**attempt) * 0.5 # 0.5s, 1s, 2s |
| 241 | time.sleep(wait_time) |
| 242 | |
| 243 | def embed_concurrent(self, texts: List[str]) -> List[np.ndarray]: |
| 244 | """ |