Spaces:
Sleeping
Sleeping
File size: 2,771 Bytes
0bb4dfa | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | """Centralized sanitizer for LLM responses.
CCAI conversations don't work well when thinking traces leak into chat or
into orchestrator/summarizer/Credential-Summary inputs, so every LLM
response funnels through `strip_thinking` before being stored, displayed,
or forwarded to another LLM.
"""
from __future__ import annotations
import re
# Top-level reasoning blocks emitted as XML-ish tags. DOTALL so we catch
# multi-line reasoning blocks; non-greedy so adjacent blocks don't merge.
_THINK_TAG_RE = re.compile(r"<think>.*?</think>", re.DOTALL | re.IGNORECASE)
_REASONING_BLOCK_RE = re.compile(
r"<(reasoning|reflection|inner_thoughts|scratchpad|analysis|plan)>.*?</\1>",
re.DOTALL | re.IGNORECASE,
)
# Bare "thought:" / "reasoning:" prologues some models emit before content
# (only at the very start of the response, otherwise we'd nuke the body).
_PROLOGUE_RE = re.compile(
r"^\s*(thought|thinking|reasoning|analysis|scratchpad)\s*:\s*"
r".*?(?=\n\n|\Z)",
re.DOTALL | re.IGNORECASE,
)
# Some providers wrap thinking in special framing tokens. We try to strip
# the *paired* form (open ... close) first so the body in between is
# removed, and fall back to stripping any leftover bare markers.
_PAIRED_FRAMING_RES = [
re.compile(r"<\|reasoning\|>.*?<\|/reasoning\|>", re.DOTALL | re.IGNORECASE),
re.compile(r"<\|think\|>.*?<\|/think\|>", re.DOTALL | re.IGNORECASE),
]
_FRAMING_TOKENS = ["<|reasoning|>", "<|/reasoning|>", "<|think|>", "<|/think|>"]
def strip_thinking(text: str | None) -> str:
"""Return `text` with all reasoning artifacts removed.
Safe to call on empty, whitespace-only, or None inputs (returns empty
string in those cases). Idempotent: calling twice yields the same result.
"""
if not text:
return ""
out = _THINK_TAG_RE.sub("", text)
out = _REASONING_BLOCK_RE.sub("", out)
for paired in _PAIRED_FRAMING_RES:
out = paired.sub("", out)
for tok in _FRAMING_TOKENS:
out = out.replace(tok, "")
out = _PROLOGUE_RE.sub("", out)
return out.strip()
def response_has_thinking(text: str | None, msg: dict | None = None) -> bool:
"""Return True if the raw response had any thinking artifact.
Checks both the textual content and any `reasoning_content` /
`reasoning` fields the OpenAI-compat client may have surfaced.
"""
if msg is not None:
if msg.get("reasoning_content") or msg.get("reasoning"):
return True
if not text:
return False
if _THINK_TAG_RE.search(text):
return True
if _REASONING_BLOCK_RE.search(text):
return True
if any(tok in text for tok in _FRAMING_TOKENS):
return True
if _PROLOGUE_RE.match(text):
return True
return False
|