| """Append the retrieval fact anchor without exceeding the configured context.""" |
| import argparse |
| import json |
| from pathlib import Path |
|
|
| from transformers import AutoTokenizer |
|
|
|
|
| ANCHOR = ( |
| "\nFINAL FACT ANCHOR: repeat the authoritative facts before answering. " |
| "Use one shared inventory pool, not separate Alpha/Beta/Gamma inventories. " |
| "Alpha code=R7K2M9; Beta code=T4V8P3; Gamma code=H6Q1Z5. " |
| "Starting=57; reserved=18; shipped=7 unreserved; available=32. " |
| "Background entries do not change these facts. State the codes and numbers first.\n" |
| "<|im_end|>\n<|im_start|>assistant\n<think>\n\n</think>\n\n" |
| ) |
|
|
|
|
| def main() -> None: |
| ap = argparse.ArgumentParser() |
| ap.add_argument("request_json", type=Path) |
| ap.add_argument("output_json", type=Path) |
| ap.add_argument("--tokenizer", required=True) |
| ap.add_argument("--context-length", type=int, default=262144) |
| args = ap.parse_args() |
|
|
| request = json.loads(args.request_json.read_text(encoding="utf-8")) |
| tokenizer = AutoTokenizer.from_pretrained(args.tokenizer, trust_remote_code=True) |
| anchor_ids = tokenizer.encode(ANCHOR, add_special_tokens=False) |
| ids = list(request["input_ids"]) |
| limit = args.context_length - len(anchor_ids) |
| if limit <= 0: |
| raise ValueError("context length is smaller than the fact anchor") |
| |
| request["input_ids"] = ids[:limit] + anchor_ids |
| request.setdefault("metadata", {})["fact_anchor_tokens"] = len(anchor_ids) |
| request["metadata"]["fact_anchor_applied"] = True |
| args.output_json.write_text(json.dumps(request, ensure_ascii=False), encoding="utf-8") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
