auto: sync run_qwen_phase_probe.py

scripts/run_qwen_phase_probe.py

@@ -60,6 +60,8 @@ from src.utils.anchor_geometry import (
     match_anchor_span,
     token_has_leading_whitespace,
 )
+from src.utils.qwen_prompting import format_generation_prompt
+from src.utils.stdio import configure_utf8_stdio
 
 # ─────────────────────────────────────────────────────────────────────────────
 # Constants
@@ -347,8 +349,9 @@ def generate_base_text(
     if tokenizer is None:
         raise ValueError("tokenizer is required")
     device = next(overlay.parameters()).device
+    generation_prompt = format_generation_prompt(tokenizer, prompt)
     encoded = tokenizer(
-        [prompt],
+        [generation_prompt],
         truncation=True,
         max_length=MAX_LENGTH,
         return_tensors="pt",
@@ -627,6 +630,7 @@ def run(
 
 
 def main() -> None:
+    configure_utf8_stdio()
     parser = argparse.ArgumentParser(description="ABPT Phase Probe — Фаза 1 верификации геометрии")
     parser.add_argument("--model", default="Qwen/Qwen3.5-4B",
                         help="HuggingFace model name")

src/model/qwen_anchor_overlay.py

@@ -53,6 +53,7 @@ from src.utils.anchor_geometry import (
     match_anchor_span,
     select_tail_probe_layers,
 )
+from src.utils.qwen_prompting import format_generation_prompt
 from src.data.qwen_anchor_geometry_cases import QwenAnchorGeometryCase, make_qwen_anchor_geometry_cases
 
 
@@ -853,8 +854,9 @@ class QwenAnchorOverlay(nn.Module):
         if self.tokenizer is None:
             raise ValueError("tokenizer is required for generate_with_anchor_bias")
 
+        generation_prompt = format_generation_prompt(self.tokenizer, prompt)
         encoded = self.tokenizer(
-            [prompt],
+            [generation_prompt],
             padding=True,
             truncation=True,
             max_length=max_length,
@@ -1235,8 +1237,9 @@ class QwenAnchorOverlay(nn.Module):
     ) -> dict[str, Any]:
         if self.tokenizer is None:
             raise ValueError("tokenizer is required for _generate_trust_completion")
+        generation_prompt = format_generation_prompt(self.tokenizer, prompt)
         encoded = self.tokenizer(
-            [prompt],
+            [generation_prompt],
             padding=True,
             truncation=True,
             max_length=max_length,

src/utils/qwen_prompting.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+def format_generation_prompt(
+    tokenizer: Any | None,
+    prompt: str,
+    *,
+    disable_thinking: bool = True,
+) -> str:
+    if tokenizer is None:
+        return prompt
+    apply_chat_template = getattr(tokenizer, "apply_chat_template", None)
+    if not callable(apply_chat_template):
+        return prompt
+    messages = [{"role": "user", "content": prompt}]
+    kwargs: dict[str, Any] = {
+        "tokenize": False,
+        "add_generation_prompt": True,
+    }
+    if disable_thinking:
+        kwargs["enable_thinking"] = False
+    try:
+        rendered = apply_chat_template(messages, **kwargs)
+    except TypeError:
+        kwargs.pop("enable_thinking", None)
+        try:
+            rendered = apply_chat_template(messages, **kwargs)
+        except Exception:
+            return prompt
+    except Exception:
+        return prompt
+    return str(rendered)

src/utils/stdio.py

@@ -0,0 +1,11 @@
+from __future__ import annotations
+
+import sys
+
+
+def configure_utf8_stdio() -> None:
+    for stream_name in ("stdout", "stderr"):
+        stream = getattr(sys, stream_name, None)
+        reconfigure = getattr(stream, "reconfigure", None)
+        if callable(reconfigure):
+            reconfigure(encoding="utf-8", errors="replace")