clase QwenChat: def __init__(self, modelo, procesador, sistema=Ninguno, herramientas=Ninguno): self.model, self.processor = modelo, procesador self.tokenizer = procesador.tokenizer self.history: lista[dict] = []
if sistema: self.history.append({“role”: “system”, “content”: system}) self.tools = herramientas def usuario(self, contenido): self.history.append({“role”:”user”,”content”:content}); devolver asistente de autodefinición (self, contenido, razonamiento=””): m = {“role”:”assistant”,”content”:content} si razonamiento: m[“reasoning_content”] = razonamiento self.history.append(m); devolver autodefinición resultado_herramienta(self, nombre, resultado): self.history.append({“role”:”herramienta”,”nombre”:nombre, “contenido”: resultado si isinstance(resultado, cadena) else json.dumps(resultado)}) devolver autodefinición _inputs(self, enable_thinking, preserve_thinking): devolver self.processor.apply_chat_template( self.history, tools=self.tools, tokenize=True, add_generación_prompt=Verdadero, return_dict=Verdadero, return_tensors=”pt”, enable_thinking=enable_thinking, preserve_thinking=preserve_thinking, ).to(self.model.device) def generate(self, *, enable_thinking=True, preserve_thinking=False, max_new_tokens=2048, preset=”thinking_general”, stop_criteria=Ninguno, append_to_history=True): inp = self._inputs(enable_thinking, preserve_thinking) cfg = MUESTREO[preset]
gk = dict(**inp, max_new_tokens=max_new_tokens, do_sample=True, temperatura=cfg[“temperature”]top_p=cfg[“top_p”]top_k=cfg[“top_k”]repetition_penalty=1.0, pad_token_id=self.tokenizer.pad_token_id o self.tokenizer.eos_token_id) si stop_criteria no es Ninguno: gk[“stopping_criteria”] = stop_criteria con torch.inference_mode(): out = self.model.generate(**gk) raw = self.tokenizer.decode(out[0, inp[“input_ids”].forma[-1]:], skip_special_tokens=True) think, ans = split_thinking(raw) if append_to_history: self.assistant(ans, razonamiento=think) return think, ans def stream(self, *, enable_thinking=True, preserve_thinking=False, max_new_tokens=2048, preset=”thinking_general”, on_thinking=None, on_answer=None): inp = self._inputs(enable_thinking, preserve_thinking) cfg = MUESTREO[preset]
streamer = TextIteratorStreamer(self.tokenizer, skip_prompt=True, skip_special_tokens=True) gk = dict(**inp, streamer=streamer, max_new_tokens=max_new_tokens, do_sample=True, temperatura=cfg[“temperature”]top_p=cfg[“top_p”]top_k=cfg[“top_k”]pad_token_id=self.tokenizer.pad_token_id o self.tokenizer.eos_token_id) t = threading.Thread(target=self.model.generate, kwargs=gk); t.start() buf, in_think = “”, enable_thinking think_text, respuesta_text = “”, “” for pieza en streamer: buf += pieza if in_think: if THINK_CLOSE en buf: close_at = buf.index(THINK_CLOSE) resid = buf[:close_at]
si on_thinking: on_thinking(resid[len(think_text):]) think_text = resid buf = buf[close_at + len(THINK_CLOSE):]
in_think = Falso si buf y on_answer: on_answer(buf) respuesta_texto = buf; buf = “” else: if on_thinking: on_thinking(pieza) think_text += pieza else: if on_answer: on_answer(pieza) respuesta_texto += pieza t.join() self.assistant(answer_text.strip(), razonamiento=think_text.strip()) return think_text.strip(), respuesta_text.strip() def save(self, ruta): con open(ruta, “w”) como f: json.dump({“history”: self.history, “tools”: self.tools}, f, indent=2) @classmethod def load(cls, modelo, procesador, ruta): con open(ruta) como f: data = json.load(f) c = cls(modelo, procesador, herramientas=data.get(“tools”)) c.history = data[“history”]; devolver clase c ThinkingBudget(StoppingCriteria): def __init__(self, tokenizer, presupuesto: int): self.budget = presupuesto self.open_ids = tokenizer.encode(THINK_OPEN, add_special_tokens=False) self.close_ids = tokenizer.encode(THINK_CLOSE, add_special_tokens=False) self.start = Ninguno def _find(self, seq, aguja): n = len(aguja) para i en el rango(len(seq)-n+1): si seq[i:i+n] == aguja: devolver i devolver Ninguno def __call__(self, input_ids, puntuaciones, **kwargs): seq = input_ids[0].tolist() si self.start es Ninguno: idx = self._find(seq, self.open_ids) si idx no es Ninguno: self.start = idx + len(self.open_ids) devuelve Falso si self._find(seq[self.start:]self.close_ids) no es Ninguno: return False return (len(seq) – self.start) >= self.budget TOOL_CALL_RE = re.compile(r”\s*(\{.*?\})\s*”, re.S) def run_calculate(expr: str) -> str: si existe (c no en “0123456789+-*/().% ” para c en expr): return json.dumps({“error”:”illegal chars”}) intente: return json.dumps({“result”: eval(expr, {“__builtins__”: {}}, {})}) excepto Excepción como e: return json.dumps({“error”: str(e)}) _DOCS = { “qwen3.6”: “Qwen3.6-35B-A3B es un MoE de 35B con 3B parámetros activos y contexto nativo de 262k.”, “deltanet”: “Gated DeltaNet es una variante de atención lineal utilizada en las capas híbridas de Qwen3.6.”, “moe”: “Qwen3.6 utiliza 256 expertos con 8 enrutados + 1 compartido por token.”, } def run_search_docs(q): hits = [v for k,v in _DOCS.items() if k in q.lower()]
return json.dumps({“resultados”: visitas o [“no hits”]}) def run_get_time(): importar fecha y hora como dt return json.dumps({“iso”: dt.datetime.utcnow().isoformat()+”Z”}) TOOL_FNS = { “calcular”: lambda a: run_calculate(a[“expression”]), “search_docs”: lambda a: run_search_docs(a[“query”]), “get_time”: lambda a: run_get_time(), } TOOLS_SCHEMA = [
{“type”:”function”,”function”:{“name”:”calculate”,”description”:”Evaluate arithmetic.”,
“parameters”:{“type”:”object”,”properties”:{“expression”:{“type”:”string”}},”required”:[“expression”]}}}, {“type”:”function”,”function”:{“name”:”search_docs”,”description”:”Buscar documentos internos.”, “parameters”:{“type”:”object”,”properties”:{“query”:{“type”:”string”}},”required”:[“query”]}}}, {“type”:”function”,”function”:{“name”:”get_time”,”description”:”Obtener la hora UTC actual.”, “parameters”:{“type”:”object”,”properties”:{}}}}, ]
if sistema: self.history.append({“role”: “system”, “content”: system}) self.tools = herramientas def usuario(self, contenido): self.history.append({“role”:”user”,”content”:content}); devolver asistente de autodefinición (self, contenido, razonamiento=””): m = {“role”:”assistant”,”content”:content} si razonamiento: m[“reasoning_content”] = razonamiento self.history.append(m); devolver autodefinición resultado_herramienta(self, nombre, resultado): self.history.append({“role”:”herramienta”,”nombre”:nombre, “contenido”: resultado si isinstance(resultado, cadena) else json.dumps(resultado)}) devolver autodefinición _inputs(self, enable_thinking, preserve_thinking): devolver self.processor.apply_chat_template( self.history, tools=self.tools, tokenize=True, add_generación_prompt=Verdadero, return_dict=Verdadero, return_tensors=”pt”, enable_thinking=enable_thinking, preserve_thinking=preserve_thinking, ).to(self.model.device) def generate(self, *, enable_thinking=True, preserve_thinking=False, max_new_tokens=2048, preset=”thinking_general”, stop_criteria=Ninguno, append_to_history=True): inp = self._inputs(enable_thinking, preserve_thinking) cfg = MUESTREO[preset]
gk = dict(**inp, max_new_tokens=max_new_tokens, do_sample=True, temperatura=cfg[“temperature”]top_p=cfg[“top_p”]top_k=cfg[“top_k”]repetition_penalty=1.0, pad_token_id=self.tokenizer.pad_token_id o self.tokenizer.eos_token_id) si stop_criteria no es Ninguno: gk[“stopping_criteria”] = stop_criteria con torch.inference_mode(): out = self.model.generate(**gk) raw = self.tokenizer.decode(out[0, inp[“input_ids”].forma[-1]:], skip_special_tokens=True) think, ans = split_thinking(raw) if append_to_history: self.assistant(ans, razonamiento=think) return think, ans def stream(self, *, enable_thinking=True, preserve_thinking=False, max_new_tokens=2048, preset=”thinking_general”, on_thinking=None, on_answer=None): inp = self._inputs(enable_thinking, preserve_thinking) cfg = MUESTREO[preset]
streamer = TextIteratorStreamer(self.tokenizer, skip_prompt=True, skip_special_tokens=True) gk = dict(**inp, streamer=streamer, max_new_tokens=max_new_tokens, do_sample=True, temperatura=cfg[“temperature”]top_p=cfg[“top_p”]top_k=cfg[“top_k”]pad_token_id=self.tokenizer.pad_token_id o self.tokenizer.eos_token_id) t = threading.Thread(target=self.model.generate, kwargs=gk); t.start() buf, in_think = “”, enable_thinking think_text, respuesta_text = “”, “” for pieza en streamer: buf += pieza if in_think: if THINK_CLOSE en buf: close_at = buf.index(THINK_CLOSE) resid = buf[:close_at]
si on_thinking: on_thinking(resid[len(think_text):]) think_text = resid buf = buf[close_at + len(THINK_CLOSE):]
in_think = Falso si buf y on_answer: on_answer(buf) respuesta_texto = buf; buf = “” else: if on_thinking: on_thinking(pieza) think_text += pieza else: if on_answer: on_answer(pieza) respuesta_texto += pieza t.join() self.assistant(answer_text.strip(), razonamiento=think_text.strip()) return think_text.strip(), respuesta_text.strip() def save(self, ruta): con open(ruta, “w”) como f: json.dump({“history”: self.history, “tools”: self.tools}, f, indent=2) @classmethod def load(cls, modelo, procesador, ruta): con open(ruta) como f: data = json.load(f) c = cls(modelo, procesador, herramientas=data.get(“tools”)) c.history = data[“history”]; devolver clase c ThinkingBudget(StoppingCriteria): def __init__(self, tokenizer, presupuesto: int): self.budget = presupuesto self.open_ids = tokenizer.encode(THINK_OPEN, add_special_tokens=False) self.close_ids = tokenizer.encode(THINK_CLOSE, add_special_tokens=False) self.start = Ninguno def _find(self, seq, aguja): n = len(aguja) para i en el rango(len(seq)-n+1): si seq[i:i+n] == aguja: devolver i devolver Ninguno def __call__(self, input_ids, puntuaciones, **kwargs): seq = input_ids[0].tolist() si self.start es Ninguno: idx = self._find(seq, self.open_ids) si idx no es Ninguno: self.start = idx + len(self.open_ids) devuelve Falso si self._find(seq[self.start:]self.close_ids) no es Ninguno: return False return (len(seq) – self.start) >= self.budget TOOL_CALL_RE = re.compile(r”\s*(\{.*?\})\s*”, re.S) def run_calculate(expr: str) -> str: si existe (c no en “0123456789+-*/().% ” para c en expr): return json.dumps({“error”:”illegal chars”}) intente: return json.dumps({“result”: eval(expr, {“__builtins__”: {}}, {})}) excepto Excepción como e: return json.dumps({“error”: str(e)}) _DOCS = { “qwen3.6”: “Qwen3.6-35B-A3B es un MoE de 35B con 3B parámetros activos y contexto nativo de 262k.”, “deltanet”: “Gated DeltaNet es una variante de atención lineal utilizada en las capas híbridas de Qwen3.6.”, “moe”: “Qwen3.6 utiliza 256 expertos con 8 enrutados + 1 compartido por token.”, } def run_search_docs(q): hits = [v for k,v in _DOCS.items() if k in q.lower()]
return json.dumps({“resultados”: visitas o [“no hits”]}) def run_get_time(): importar fecha y hora como dt return json.dumps({“iso”: dt.datetime.utcnow().isoformat()+”Z”}) TOOL_FNS = { “calcular”: lambda a: run_calculate(a[“expression”]), “search_docs”: lambda a: run_search_docs(a[“query”]), “get_time”: lambda a: run_get_time(), } TOOLS_SCHEMA = [
{“type”:”function”,”function”:{“name”:”calculate”,”description”:”Evaluate arithmetic.”,
“parameters”:{“type”:”object”,”properties”:{“expression”:{“type”:”string”}},”required”:[“expression”]}}}, {“type”:”function”,”function”:{“name”:”search_docs”,”description”:”Buscar documentos internos.”, “parameters”:{“type”:”object”,”properties”:{“query”:{“type”:”string”}},”required”:[“query”]}}}, {“type”:”function”,”function”:{“name”:”get_time”,”description”:”Obtener la hora UTC actual.”, “parameters”:{“type”:”object”,”properties”:{}}}}, ]