Views
No views yet
ollama run huihui_ai/qwen3-abliterated:30btransformers library:
You can try using /no_think to toggle think mode, but it’s not guaranteed to work every time.1from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig, TextStreamer
2import torch
3import os
4import signal
5
6cpu_count = os.cpu_count()
7print(f"Number of CPU cores in the system: {cpu_count}")
8half_cpu_count = cpu_count // 2
9os.environ["MKL_NUM_THREADS"] = str(half_cpu_count)
10os.environ["OMP_NUM_THREADS"] = str(half_cpu_count)
11torch.set_num_threads(half_cpu_count)
12
13print(f"PyTorch threads: {torch.get_num_threads()}")
14print(f"MKL threads: {os.getenv('MKL_NUM_THREADS')}")
15print(f"OMP threads: {os.getenv('OMP_NUM_THREADS')}")
16
17# Load the model and tokenizer
18NEW_MODEL_ID = "huihui-ai/Qwen3-30B-A3B-abliterated"
19print(f"Load Model {NEW_MODEL_ID} ... ")
20quant_config_4 = BitsAndBytesConfig(
21 load_in_4bit=True,
22 bnb_4bit_compute_dtype=torch.bfloat16,
23 bnb_4bit_use_double_quant=True,
24 llm_int14_enable_fp32_cpu_offload=True,
25)
26
27model = AutoModelForCausalLM.from_pretrained(
28 NEW_MODEL_ID,
29 device_map="auto",
30 trust_remote_code=True,
31 #quantization_config=quant_config_4,
32 torch_dtype=torch.bfloat16
33)
34tokenizer = AutoTokenizer.from_pretrained(NEW_MODEL_ID, trust_remote_code=True)
35if tokenizer.pad_token is None:
36 tokenizer.pad_token = tokenizer.eos_token
37tokenizer.pad_token_id = tokenizer.eos_token_id
38
39messages = []
40enable_thinking = True
41skip_prompt=True
42skip_special_tokens=True
43
44def apply_chat_template(tokenizer, messages, enable_thinking, add_generation_prompt=True):
45 input_ids = tokenizer.apply_chat_template(
46 messages,
47 tokenize=False,
48 add_generation_prompt=add_generation_prompt,
49 )
50 if not enable_thinking:
51 input_ids += "\n<think>\n\n</think>\n"
52 return input_ids
53
54class CustomTextStreamer(TextStreamer):
55 def __init__(self, tokenizer, skip_prompt=True, skip_special_tokens=True):
56 super().__init__(tokenizer, skip_prompt=skip_prompt, skip_special_tokens=skip_special_tokens)
57 self.generated_text = ""
58 self.stop_flag = False
59
60 def on_finalized_text(self, text: str, stream_end: bool = False):
61 self.generated_text += text
62 print(text, end="", flush=True)
63 if self.stop_flag:
64 raise StopIteration
65
66 def stop_generation(self):
67 self.stop_flag = True
68
69def generate_stream(model, tokenizer, messages, enable_thinking, skip_prompt, skip_special_tokens, max_new_tokens):
70 formatted_prompt = apply_chat_template(tokenizer, messages, enable_thinking)
71 input_ids = tokenizer(
72 formatted_prompt,
73 return_tensors="pt",
74 return_attention_mask=True,
75 padding=False
76 )
77
78 tokens = input_ids['input_ids'].to(model.device)
79 attention_mask = input_ids['attention_mask'].to(model.device)
80
81 streamer = CustomTextStreamer(tokenizer, skip_prompt=skip_prompt, skip_special_tokens=skip_special_tokens)
82
83 def signal_handler(sig, frame):
84 streamer.stop_generation()
85 print("\n[Generation stopped by user with Ctrl+C]")
86
87 signal.signal(signal.SIGINT, signal_handler)
88
89 print("Response: ", end="", flush=True)
90 try:
91 generated_ids = model.generate(
92 tokens,
93 attention_mask=attention_mask,
94 use_cache=False,
95 max_new_tokens=max_new_tokens,
96 do_sample=True,
97 pad_token_id=tokenizer.pad_token_id,
98 streamer=streamer
99 )
100 del generated_ids
101 except StopIteration:
102 print("\n[Stopped by user]")
103
104 del input_ids, attention_mask
105 torch.cuda.empty_cache()
106 signal.signal(signal.SIGINT, signal.SIG_DFL)
107
108 return streamer.generated_text, streamer.stop_flag
109
110while True:
111 user_input = input("User: ").strip()
112 if user_input.lower() == "/exit":
113 print("Exiting chat.")
114 break
115 if user_input.lower() == "/clear":
116 messages = []
117 print("Chat history cleared. Starting a new conversation.")
118 continue
119 if user_input.lower() == "/no_think":
120 if enable_thinking:
121 enable_thinking = False
122 print("Thinking = False.")
123 else:
124 enable_thinking = True
125 print("Thinking = True.")
126 continue
127 if user_input.lower() == "/skip_prompt":
128 if skip_prompt:
129 skip_prompt = False
130 print("skip_prompt = False.")
131 else:
132 skip_prompt = True
133 print("skip_prompt = True.")
134 continue
135 if user_input.lower() == "/skip_special_tokens":
136 if skip_special_tokens:
137 skip_special_tokens = False
138 print("skip_special_tokens = False.")
139 else:
140 skip_special_tokens = True
141 print("skip_special_tokens = True.")
142 continue
143 if not user_input:
144 print("Input cannot be empty. Please enter something.")
145 continue
146 messages.append({"role": "user", "content": user_input})
147 response, stop_flag = generate_stream(model, tokenizer, messages, enable_thinking, skip_prompt, skip_special_tokens, 14192)
148 print("", flush=True)
149 if stop_flag:
150 continue
151 messages.append({"role": "assistant", "content": response}) bc1qqnkhuchxw0zqjh2ku3lu14hq145hc6gy1414uk70ge