|
| 1 | +# SPDX-FileCopyrightText: Copyright (c) 2023-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. |
| 2 | +# SPDX-License-Identifier: Apache-2.0 |
| 3 | +# |
| 4 | +# Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | +# you may not use this file except in compliance with the License. |
| 6 | +# You may obtain a copy of the License at |
| 7 | +# |
| 8 | +# http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | +# |
| 10 | +# Unless required by applicable law or agreed to in writing, software |
| 11 | +# distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | +# See the License for the specific language governing permissions and |
| 14 | +# limitations under the License. |
| 15 | + |
| 16 | +"""Extract hidden states from an HF-compatible LLM.""" |
| 17 | + |
| 18 | +import argparse |
| 19 | +import asyncio |
| 20 | +import json |
| 21 | +from pathlib import Path |
| 22 | + |
| 23 | +import torch |
| 24 | +from tqdm import tqdm as tqdm |
| 25 | +from transformers import AutoModel, AutoTokenizer |
| 26 | + |
| 27 | +REMOVE_THINK_CHAT_TEMPLATE = ( |
| 28 | + "{% if '</think>' in content %}{% set content = content.split('</think>')[-1] %}{% endif %}" |
| 29 | +) |
| 30 | + |
| 31 | + |
| 32 | +def parse_args() -> argparse.Namespace: |
| 33 | + parser = argparse.ArgumentParser( |
| 34 | + description="""Collect hidden states from conversations |
| 35 | + by running full conversations through a Hugging Face model.""" |
| 36 | + ) |
| 37 | + |
| 38 | + ## Model & Generation Parameters ## |
| 39 | + parser.add_argument( |
| 40 | + "--model", |
| 41 | + type=str, |
| 42 | + required=True, |
| 43 | + help="Name of the served model.", |
| 44 | + ) |
| 45 | + |
| 46 | + ## Client Parameters ## |
| 47 | + parser.add_argument( |
| 48 | + "--max-seq-len", |
| 49 | + type=int, |
| 50 | + default=3072, |
| 51 | + help="""Maximum number of tokens in a conversation. Longer conversations will be skipped. |
| 52 | + Defaults to 3072 tokens.""", |
| 53 | + ) |
| 54 | + |
| 55 | + ## I/O Parameters ## |
| 56 | + parser.add_argument( |
| 57 | + "--input-file", |
| 58 | + type=Path, |
| 59 | + required=True, |
| 60 | + help="""Path to the input `jsonl` file containing conversations. |
| 61 | + Each entry must have a unique `conversation_id` field and a `conversations` field |
| 62 | + containing a list of messages.""", |
| 63 | + ) |
| 64 | + parser.add_argument( |
| 65 | + "--output-dir", |
| 66 | + type=Path, |
| 67 | + required=True, |
| 68 | + help="""Root directory in which to save the hidden states. |
| 69 | + The data will be saved as a torch (`.pt`) dump file for each conversation.""", |
| 70 | + ) |
| 71 | + parser.add_argument( |
| 72 | + "--debug-max-num-conversations", |
| 73 | + type=int, |
| 74 | + default=None, |
| 75 | + help="""For debugging purposes, limit the number of conversations processed. |
| 76 | + Default is None, meaning no limit.""", |
| 77 | + ) |
| 78 | + |
| 79 | + return parser.parse_args() |
| 80 | + |
| 81 | + |
| 82 | +async def main(args: argparse.Namespace) -> None: |
| 83 | + all_conversations = [] |
| 84 | + with args.input_file.open("r", encoding="utf-8") as f: |
| 85 | + all_conversations.extend([json.loads(line) for line in f if line.strip()]) |
| 86 | + |
| 87 | + print("Loaded", len(all_conversations), "conversations from", args.input_file) |
| 88 | + |
| 89 | + model = AutoModel.from_pretrained(args.model, torch_dtype="auto", device_map="auto") |
| 90 | + num_hidden_layers = getattr(model.config, "num_hidden_layers", None) |
| 91 | + |
| 92 | + tokenizer = AutoTokenizer.from_pretrained(args.model) |
| 93 | + if tokenizer.pad_token is None: |
| 94 | + tokenizer.pad_token = tokenizer.eos_token |
| 95 | + tokenizer.chat_template = tokenizer.chat_template.replace(REMOVE_THINK_CHAT_TEMPLATE, "") |
| 96 | + |
| 97 | + output_dir = args.output_dir |
| 98 | + output_dir.mkdir(parents=True, exist_ok=True) |
| 99 | + num_skipped_too_long = 0 |
| 100 | + num_invalid = 0 |
| 101 | + num_success = 0 |
| 102 | + num_total_conversations = min( |
| 103 | + len(all_conversations), args.debug_max_num_conversations or len(all_conversations) |
| 104 | + ) |
| 105 | + for idx, entry in enumerate( |
| 106 | + tqdm( |
| 107 | + all_conversations[: args.debug_max_num_conversations], |
| 108 | + desc="Processing conversations", |
| 109 | + total=num_total_conversations, |
| 110 | + ) |
| 111 | + ): |
| 112 | + conversation_id = entry.get("conversation_id", "{:08d}".format(idx)) |
| 113 | + conversations = entry["conversations"] |
| 114 | + if not conversations or not isinstance(conversations, list): |
| 115 | + num_invalid += 1 |
| 116 | + continue |
| 117 | + |
| 118 | + # Tokenize and check length |
| 119 | + input_ids = tokenizer.apply_chat_template( |
| 120 | + conversations, return_tensors="pt", add_generation_template=False |
| 121 | + ) |
| 122 | + num_input_tokens = input_ids.shape[1] |
| 123 | + if num_input_tokens <= 10 or num_input_tokens > args.max_seq_len: |
| 124 | + num_skipped_too_long += 1 |
| 125 | + continue |
| 126 | + |
| 127 | + # Get hidden states |
| 128 | + with torch.inference_mode(): |
| 129 | + outputs = model(input_ids=input_ids.to(model.device), output_hidden_states=True) |
| 130 | + if num_hidden_layers is None: |
| 131 | + num_hidden_layers = len(outputs.hidden_states) - 1 |
| 132 | + else: |
| 133 | + assert num_hidden_layers + 1 == len(outputs.hidden_states), ( |
| 134 | + f"Expected {num_hidden_layers}+1 layers of hidden states, but got {len(outputs.hidden_states)}." |
| 135 | + ) |
| 136 | + # Extract hidden states from layers with index (2, N/2, N-3), and the output hidden states |
| 137 | + hidden_states = outputs.hidden_states |
| 138 | + selected_layer_indices = [ |
| 139 | + 2, |
| 140 | + max(0, num_hidden_layers // 2), |
| 141 | + max(1, num_hidden_layers - 3), |
| 142 | + ] |
| 143 | + selected_layer_indices = sorted(set(selected_layer_indices)) |
| 144 | + aux_hidden_states = torch.cat( |
| 145 | + [hidden_states[i].squeeze(0).cpu() for i in selected_layer_indices], dim=-1 |
| 146 | + ) |
| 147 | + output_hidden_states = outputs.last_hidden_state.squeeze(0).cpu() |
| 148 | + output_file = output_dir / f"{conversation_id}.pt" |
| 149 | + num_success += 1 |
| 150 | + with open(output_file, "wb") as f: |
| 151 | + torch.save( |
| 152 | + { |
| 153 | + "input_ids": input_ids.squeeze(0).cpu(), |
| 154 | + "hidden_states": output_hidden_states, |
| 155 | + "aux_hidden_states": aux_hidden_states, |
| 156 | + "conversation_id": conversation_id, |
| 157 | + }, |
| 158 | + f, |
| 159 | + ) |
| 160 | + |
| 161 | + if num_skipped_too_long > 0: |
| 162 | + print(f"Skipped {num_skipped_too_long} conversations due to length constraints.") |
| 163 | + if num_invalid > 0: |
| 164 | + print(f"Skipped {num_invalid} invalid conversations without proper fields.") |
| 165 | + |
| 166 | + if num_success == num_total_conversations: |
| 167 | + print(f"Successfully processed all {num_success} conversations.") |
| 168 | + else: |
| 169 | + print( |
| 170 | + f"Successfully processed {num_success} out of {num_total_conversations} conversations." |
| 171 | + ) |
| 172 | + |
| 173 | + |
| 174 | +if __name__ == "__main__": |
| 175 | + cli_args = parse_args() |
| 176 | + asyncio.run(main(cli_args)) |
0 commit comments