| from __future__ import annotations |
|
|
| import re |
|
|
| EOS_STOP = "<｜end▁of▁sentence｜>" |
| _NONTEXT_LINE_RE = re.compile(r"(?m)^[ \t]*(?:\[\s*Non(?:[ -]?Text)\s*\]|NonText)[ \t]*(?:\n|$)") |
|
|
|
|
| def _replace_complete_tags(text: str, opening: str, closing: str, *, preserve_content: bool) -> str: |
| """Replace only complete exact tag pairs; malformed input is preserved.""" |
| chunks: list[str] = [] |
| cursor = 0 |
| while True: |
| start = text.find(opening, cursor) |
| if start < 0: |
| chunks.append(text[cursor:]) |
| break |
| end = text.find(closing, start + len(opening)) |
| if end < 0: |
| chunks.append(text[cursor:]) |
| break |
| chunks.append(text[cursor:start]) |
| if preserve_content: |
| chunks.append(text[start + len(opening) : end]) |
| cursor = end + len(closing) |
| return "".join(chunks) |
|
|
|
|
| def clean_model_output(text: str) -> str: |
| """Remove exact model sentinels while preserving recognized content verbatim.""" |
| if text.endswith(EOS_STOP): |
| text = text[: -len(EOS_STOP)] |
| text = _replace_complete_tags(text, "<|ref|>", "<|/ref|>", preserve_content=True) |
| text = _replace_complete_tags(text, "<|det|>", "<|/det|>", preserve_content=False) |
| text = _NONTEXT_LINE_RE.sub("", text) |
| return text.strip() |
|
|
|
|
| def repetition_warning(text: str) -> str | None: |
| """Return a warning for obvious terminal repetition loops.""" |
| lines = [line.strip() for line in text.splitlines() if line.strip()] |
| for index in range(2, len(lines)): |
| if lines[index] == lines[index - 1] == lines[index - 2] and len(lines[index]) >= 8: |
| return f"repeated line detected three times: {lines[index][:80]!r}" |
|
|
| compact = re.sub(r"\s+", " ", text).strip() |
| for width in (64, 48, 32): |
| if len(compact) >= width * 3: |
| tail = compact[-width:] |
| if compact[-width * 3 :] == tail * 3: |
| return f"repeated {width}-character suffix detected" |
| return None |
|
|
