"""Shared thinking-tag parsing for server-based providers. Extracts thinking blocks from raw LLM responses so that thinking content is separated from the answer text. Used by any provider that receives inline thinking tags (MLX Server, vLLM, SGLang, etc.). Supported formats: - Qwen3/DeepSeek: ``...`` - Gemma 4: ``<|channel>thought\\n...`` """ import re # Gemma 4 thinking tag pattern: # <|channel>thought\n ... _GEMMA4_PATTERN = re.compile( r"<\|channel>thought\n(.*?)", re.DOTALL, ) def parse_thinking_tags(text: str) -> tuple[str, int, str | None]: """Extract thinking blocks from response text. Handles multiple formats: **Qwen3 / DeepSeek format:** 1. ``contentanswer`` — standard. 2. ``contentanswer`` — opening tag in prompt template. **Gemma 4 format:** 3. ``<|channel>thought\\ncontentanswer`` — Gemma 4 thinking. **Fallback:** 4. No thinking tags — full text returned as-is. Token counting uses a whitespace-split heuristic (approximate). Providers with access to a tokenizer should override if precision is needed. Args: text: Raw response text potentially containing think tags. Returns: Tuple of (answer_text, thinking_token_count, thinking_text_or_None). """ # --- Qwen3 / DeepSeek: ... --- # Case 1: both tags present matches = re.findall(r"(.*?)", text, re.DOTALL) if matches: all_thinking = "\n".join(matches) thinking_tokens = len(all_thinking.split()) # approximate cleaned = re.sub( r".*?", "", text, flags=re.DOTALL, ).strip() return cleaned, thinking_tokens, all_thinking # Case 2: only present (opening tag was in prompt template) if "" in text: parts = text.split("", 1) thinking_content = parts[0] answer = parts[1].strip() if len(parts) > 1 else "" thinking_tokens = len(thinking_content.split()) # approximate return answer, thinking_tokens, thinking_content # --- Gemma 4: <|channel>thought\n... --- # Case 3: Gemma 4 thinking blocks gemma_matches = _GEMMA4_PATTERN.findall(text) if gemma_matches: all_thinking = "\n".join(gemma_matches) thinking_tokens = len(all_thinking.split()) cleaned = _GEMMA4_PATTERN.sub("", text).strip() return cleaned, thinking_tokens, all_thinking # Case 4: no thinking tags return text, 0, None