import os import json from dataclasses import dataclass from typing import Any, Dict, List, Optional, Tuple import torch import torch.nn.functional as F from transformers import AutoTokenizer, AutoModel, BitsAndBytesConfig try: from peft import PeftModel _HAS_PEFT = True except Exception: PeftModel = None _HAS_PEFT = False try: from huggingface_hub import snapshot_download _HAS_HUB = True except Exception: snapshot_download = None _HAS_HUB = False def require_flash_attention_2() -> str: if not torch.cuda.is_available(): raise RuntimeError("FlashAttention-2 requires CUDA, but torch.cuda.is_available() is False.") try: import flash_attn # noqa: F401 ver = getattr(flash_attn, "__version__", "0.0.0") major = int(str(ver).split(".")[0]) if major < 2: raise RuntimeError(f"flash-attn version {ver} < 2.0.0") except Exception as e: raise RuntimeError( "FlashAttention-2 is REQUIRED but not available/importable.\n" "Install flash-attn>=2 and ensure it matches your torch/CUDA.\n" f"Import/Version error: {repr(e)}" ) return "flash_attention_2" def build_qwen_query(instruction: str, query: str) -> str: instruction = str(instruction).strip() query = str(query).strip() return f"Instruct: {instruction}\nQuery: {query}" def get_pool_token_id(tok) -> int: eod_id = tok.convert_tokens_to_ids("<|endoftext|>") if eod_id is None or eod_id < 0: eod_id = tok.pad_token_id return eod_id def encode_with_eos_ids(tok, texts: List[str], max_len: int) -> Dict[str, torch.Tensor]: """ Encode texts with guaranteed <|endoftext|> at the end for Qwen3-Embedding pooling. Reserves 1 slot for <|endoftext|>, left-pads to batch max length. Must match the training tokenization exactly. """ pad_id = tok.pad_token_id if tok.pad_token_id is not None else tok.eos_token_id eod_id = get_pool_token_id(tok) # Reserve 1 position for <|endoftext|> enc = tok( [str(t) for t in texts], add_special_tokens=False, truncation=True, max_length=max_len - 1, padding=False, return_attention_mask=False, ) input_ids = [ids + [eod_id] for ids in enc["input_ids"]] attn_mask = [[1] * len(ids) for ids in input_ids] # Left-pad to batch max length T = max(len(ids) for ids in input_ids) if input_ids else 1 input_ids = [[pad_id] * (T - len(ids)) + ids for ids in input_ids] attn_mask = [[0] * (T - len(m)) + m for m in attn_mask] return { "input_ids": torch.tensor(input_ids, dtype=torch.long), "attention_mask": torch.tensor(attn_mask, dtype=torch.long), } def last_token_pool(last_hidden_states: torch.Tensor, attention_mask: torch.Tensor) -> torch.Tensor: """ Left-padding aware last-token pooling (extracts EOS token embedding). """ left_padding = (attention_mask[:, -1].sum() == attention_mask.shape[0]) if left_padding: return last_hidden_states[:, -1] idx = attention_mask.sum(dim=1) - 1 return last_hidden_states[torch.arange(last_hidden_states.size(0), device=last_hidden_states.device), idx] def get_last_hidden_state(model, input_ids, attention_mask): """ Get final hidden state from the model. Provide position_ids for left padding (FlashAttention-2). """ m = model.module if hasattr(model, "module") else model # Compute position_ids for left-padded sequences (required for flash_attention_2) position_ids = attention_mask.long().cumsum(-1) - 1 position_ids.masked_fill_(attention_mask == 0, 0) out = m( input_ids=input_ids, attention_mask=attention_mask, position_ids=position_ids, use_cache=False, return_dict=True, ) if hasattr(out, "last_hidden_state"): return out.last_hidden_state out = m( input_ids=input_ids, attention_mask=attention_mask, position_ids=position_ids, output_hidden_states=True, use_cache=False, return_dict=True, ) return out.hidden_states[-1] def _read_json(path: str) -> Dict[str, Any]: with open(path, "r", encoding="utf-8") as f: return json.load(f) def _resolve_repo_path(repo_id_or_path: str) -> str: # If it's a local directory, use it as-is. if os.path.isdir(repo_id_or_path): return repo_id_or_path # Otherwise treat as HF repo_id and download snapshot. if not _HAS_HUB: raise RuntimeError( "huggingface_hub is required to load by repo_id. " "Install it: pip install huggingface_hub" ) return snapshot_download(repo_id_or_path) @dataclass class EmbedOutput: """ Simplified output: only global embeddings (no section embeddings). """ embedding: torch.Tensor # [N,H], float32 on CPU by default class Chest2Vec: """ Simplified wrapper for global embeddings only: - loads base Qwen3-Embedding - applies LoRA adapter - returns global embeddings via last-token pooling """ def __init__(self, tokenizer, model, device: torch.device): self.tokenizer = tokenizer self.model = model self.device = device self.model.eval() @classmethod def from_pretrained( cls, repo_id_or_path: str, *, device: str = "cuda:0", use_4bit: bool = False, force_flash_attention_2: bool = True, ) -> "Chest2Vec": repo_path = _resolve_repo_path(repo_id_or_path) cfg_path = os.path.join(repo_path, "chest2vec_config.json") if not os.path.isfile(cfg_path): raise FileNotFoundError(f"Missing chest2vec_config.json in {repo_path}") cfg = _read_json(cfg_path) base_model = str(cfg["base_model"]) adapter_subdir = str(cfg.get("adapter_subdir", "contrastive")) if force_flash_attention_2 or bool(cfg.get("require_flash_attention_2", False)): attn_impl = require_flash_attention_2() else: attn_impl = "sdpa" if not _HAS_PEFT: raise RuntimeError("peft is required. Install: pip install peft") device_t = torch.device(device) tokenizer = AutoTokenizer.from_pretrained(base_model, padding_side="left", trust_remote_code=True) if tokenizer.pad_token_id is None: tokenizer.pad_token = tokenizer.eos_token device_map = {"": str(device_t)} # Load base model with FlashAttention-2 if use_4bit: qconf = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_quant_type="nf4", bnb_4bit_use_double_quant=True, bnb_4bit_compute_dtype=torch.bfloat16, ) try: base = AutoModel.from_pretrained( base_model, trust_remote_code=True, attn_implementation=attn_impl, quantization_config=qconf, device_map=device_map, ) except TypeError as e: raise RuntimeError( "Your transformers version does not support attn_implementation=... " "Upgrade transformers to use FlashAttention-2." ) from e else: try: base = AutoModel.from_pretrained( base_model, trust_remote_code=True, attn_implementation=attn_impl, torch_dtype=torch.bfloat16, device_map=device_map, ) except TypeError as e: raise RuntimeError( "Your transformers version does not support attn_implementation=... " "Upgrade transformers to use FlashAttention-2." ) from e # Load adapter from this repo folder adapter_dir = os.path.join(repo_path, adapter_subdir) if not os.path.isfile(os.path.join(adapter_dir, "adapter_config.json")): raise FileNotFoundError(f"adapter_config.json not found under: {adapter_dir}") model = PeftModel.from_pretrained(base, adapter_dir) model.eval() return cls(tokenizer=tokenizer, model=model, device=device_t) @torch.inference_mode() def embed_texts( self, texts: List[str], *, max_len: int = 512, batch_size: int = 16, return_cpu_float32: bool = True, ) -> EmbedOutput: """ Encodes arbitrary texts and returns global embeddings via last-token pooling. Returns: - embedding: [N,H] - global embeddings (L2-normalized) """ # Determine AMP device = self.device if device.type == "cuda": amp_dtype = torch.bfloat16 if torch.cuda.is_bf16_supported() else torch.float16 use_amp = True else: amp_dtype = torch.float32 use_amp = False outs = [] for i in range(0, len(texts), batch_size): chunk = [str(t) for t in texts[i:i + batch_size]] enc = encode_with_eos_ids(self.tokenizer, chunk, max_len) input_ids = enc["input_ids"].to(device, non_blocking=True) attention_mask = enc["attention_mask"].to(device, non_blocking=True) with torch.autocast(device_type=("cuda" if device.type == "cuda" else "cpu"), dtype=amp_dtype, enabled=use_amp): h = get_last_hidden_state(self.model, input_ids, attention_mask) # [B,T,H] # Global embedding: extract EOS token embedding via last-token pooling emb = last_token_pool(h, attention_mask) # [B,H] emb = F.normalize(emb.float(), p=2, dim=-1) outs.append(emb.detach()) embeddings = torch.cat(outs, dim=0) # on device, dtype ~ bf16 # Move to CPU float32 if requested (recommended for retrieval stability) if return_cpu_float32: embeddings_cpu = embeddings.float().cpu() # re-normalize to fix any numerical drift embeddings_cpu = F.normalize(embeddings_cpu, p=2, dim=-1) else: embeddings_cpu = embeddings return EmbedOutput(embedding=embeddings_cpu) @torch.inference_mode() def embed_instruction_query( self, instructions: List[str], queries: List[str], *, max_len: int = 512, batch_size: int = 16, return_cpu_float32: bool = True, ) -> EmbedOutput: """ Embed instruction-query pairs. Returns: - embedding: [N,H] - global embeddings (L2-normalized) """ if len(instructions) != len(queries): raise ValueError("instructions and queries must have the same length.") q_texts = [build_qwen_query(i, q) for i, q in zip(instructions, queries)] return self.embed_texts( q_texts, max_len=max_len, batch_size=batch_size, return_cpu_float32=return_cpu_float32, ) @staticmethod def cosine_topk( query_emb: torch.Tensor, # [Nq,H] CPU float32 recommended cand_emb: torch.Tensor, # [Nd,H] CPU float32 recommended k: int = 10, *, device: str = "cuda", query_batch_size: int = 256, doc_chunk_size: int = 8192, ) -> Tuple[torch.Tensor, torch.Tensor]: """ Chunked cosine top-k, stable in float32. Returns (top_scores [Nq,k], top_indices [Nq,k]) on CPU. """ device_t = torch.device(device) q = F.normalize(query_emb.float(), p=2, dim=-1) d = F.normalize(cand_emb.float(), p=2, dim=-1) Nq, H = q.shape Nd = d.shape[0] k = min(int(k), Nd) top_scores_all = torch.empty((Nq, k), dtype=torch.float32) top_indices_all = torch.empty((Nq, k), dtype=torch.long) for qs in range(0, Nq, query_batch_size): qe = q[qs:qs + query_batch_size].to(device_t, non_blocking=True) bq = qe.size(0) top_scores = torch.full((bq, k), -1e9, device=device_t, dtype=torch.float32) top_indices = torch.full((bq, k), -1, device=device_t, dtype=torch.long) for ds in range(0, Nd, doc_chunk_size): de = d[ds:ds + doc_chunk_size].to(device_t, non_blocking=True) scores = (qe @ de.T).float() chunk = scores.size(1) idx_chunk = torch.arange(ds, ds + chunk, device=device_t, dtype=torch.long).unsqueeze(0).expand(bq, -1) comb_scores = torch.cat([top_scores, scores], dim=1) comb_idx = torch.cat([top_indices, idx_chunk], dim=1) new_scores, new_pos = torch.topk(comb_scores, k, dim=1) new_idx = comb_idx.gather(1, new_pos) top_scores, top_indices = new_scores, new_idx top_scores_all[qs:qs + bq] = top_scores.cpu() top_indices_all[qs:qs + bq] = top_indices.cpu() return top_scores_all, top_indices_all