Files
2026-09-15 15:24:03 -04:00

123 lines
5.6 KiB
Python

"""Response size control.
Open WebUI does not truncate tool results anywhere - I checked the whole
path through `process_tool_result`. Whatever a tool returns is serialised
into the conversation verbatim, and with the tool loop allowed up to 256
iterations, one chatty tool can eat a 150k window on its own.
So every response goes through `cap()` on the way out, and truncation is
always *reported* rather than silent: the model is told what it lost and
exactly how to ask for the rest. A model that knows it saw 6000 of 42000
characters behaves very differently from one that thinks it read the page.
"""
from __future__ import annotations
from .settings import settings
# Roughly 3.2 characters per token on Qwen3's tokeniser for dense English
# prose; only used to give the model a sense of cost in continuation hints.
CHARS_PER_TOKEN = 3.2
# The trailing citation line, and the marker `cap()` uses to protect it.
#
# Every other rule in this suite is carried in tool output because the model
# treats tool output as fact; the one rule that never landed - "end a
# tool-assisted answer with its URLs" - lived only in the system prompt, and
# across two graded rounds it fired on 1 of 25 opportunities. The single
# answer that did cite lifted `search_papers`' ready-made `CITE AS:` lines
# verbatim. So the fix is not another instruction: it is to hand over the
# line already formatted the way the answer should carry it.
#
# It goes at the *end*, because that is where the model is composing from
# when it writes its closing lines - the `# Source:` header at the top of a
# fetched page has been available all along and is read as document identity,
# not as something to repeat. Which makes it the one thing in the response
# that must survive trimming, hence the special case in `cap()` below.
SOURCES_SEPARATOR = "\n\n---\n"
SOURCES_PREFIX = "Sources: "
def _detach_sources(text: str) -> tuple[str, str]:
"""Split off a trailing one-line sources block, if there is one."""
marker = SOURCES_SEPARATOR + SOURCES_PREFIX
idx = text.rfind(marker)
if idx == -1:
return text, ""
block = text[idx:]
# One line only. Anything else is a coincidence in the document body, and
# moving a chunk of someone's prose to the end would be worse than trimming.
if "\n" in block[len(SOURCES_SEPARATOR):]:
return text, ""
return text[:idx], block
def cap(text: str, limit: int | None = None, *, note: str | None = None) -> str:
"""Trim `text` to `limit` characters, saying so if anything was cut."""
# A tool result that IS a data: URI is not text and must not be trimmed.
# Open WebUI detects it, replaces the textual result with a one-line
# summary, and passes the image to the model as a vision input - so the
# model sees the picture rather than 300 KB of base64. Cutting it here
# would only produce a corrupt image.
if text.startswith("data:image/"):
return text
limit = min(limit or settings.max_response_chars, settings.max_response_chars)
if len(text) <= limit:
return text
# Trimming from the end would remove the citation line, which is the one
# part of the response that exists to be copied out. Detach it, trim the
# body against a reduced budget, and put it back.
body, sources = _detach_sources(text)
budget = limit - len(sources)
# Cut on a line boundary where one is close by, so the model never sees
# a half-written table row or a truncated URL it might then "repair".
cut = body.rfind("\n", int(budget * 0.85), budget)
if cut == -1:
cut = budget
trimmed = body[:cut].rstrip()
tail = note or "This response was too long and has been trimmed."
return (f"{trimmed}\n\n[TRUNCATED after {cut:,} of {len(body):,} characters. {tail}]"
+ sources)
def paginate(text: str, offset: int, size: int, *, how_to_continue: str) -> str:
"""Return one window of a long document with an explicit continuation hint.
`how_to_continue` is phrased as an instruction the model can act on
directly, because a hint it cannot act on is just wasted tokens.
"""
total = len(text)
offset = max(0, min(offset, total))
end = min(offset + size, total)
window = text[offset:end]
header = f"[characters {offset:,}-{end:,} of {total:,}]"
if end >= total:
footer = "\n\n[END OF DOCUMENT]" if offset else ""
return f"{header}\n{window}{footer}" if offset else window
remaining = total - end
approx_tokens = int(remaining / CHARS_PER_TOKEN)
# This used to end "Only do so if this section did not answer the question."
# That reads as permission to stop, and the model took it: asked for the
# scorers in a football final it read 4,000 characters of a 118,000-character
# article, found the headline result in the lead, stopped - and then supplied
# the scorers, the minutes and the shoot-out misses from memory, wrongly. A
# lead section answers the headline question and almost never carries the
# itemised detail, so the default advice has to point the other way.
return (
f"{header}\n{window}\n\n"
f"[MORE AVAILABLE: {remaining:,} characters (~{approx_tokens:,} tokens) remain. "
f"{how_to_continue.format(offset=end)} "
f"Keep reading if the answer may be further in - what you have not read, you do not "
f"know, and a lead or summary section rarely carries names, dates, figures or tables. "
f"To reach one specific thing, jump to it with search='<term>' instead of paging.]"
)
def bullet_list(items: list[str], *, empty: str) -> str:
return "\n".join(items) if items else empty