`symfonic.services.budget.truncation`¶

truncation ¶

Deterministic truncation with a fit guarantee.

The rule this module exists to enforce: the returned string, marker included, costs no more than the ceiling it was given. That sounds obvious and the helper this replaces did not hold it — it sliced to max_tokens * chars_per_token characters and then appended "...", so every truncated result was over budget by the width of the marker, and the same module's floor-dividing estimator agreed that it fitted. Two errors in the same direction hide each other.

Holding the guarantee against an arbitrary counter takes a search rather than arithmetic: the caller may have bound a provider tokenizer, and the number of tokens in a prefix is that tokenizer's business. Binary search over the kept character count finds the longest fitting prefix when the counter is monotone in prefix length, and merely a fitting prefix when it is not — but never an over-budget one, because a candidate becomes the answer only after being measured against the ceiling. The guarantee therefore costs nothing to a well-behaved counter and survives a badly-behaved one, which matters because a third party's arithmetic is not something this module gets to assume.

Two consequences worth naming. Truncation is idempotent: the result already fits, so a second pass returns it unchanged — which is what makes it safe to call defensively in a pipeline where two stages both budget. And truncation is monotone in the budget: a larger ceiling never keeps less text, so raising a limit cannot make a prompt smaller.

truncate ¶

truncate(text: str, max_tokens: int, counter: TokenCounter, policy: TruncationPolicy | None = None) -> TruncationResult

Cut text down until it costs at most max_tokens.

Returns the input untouched when it already fits — no marker, no copy, and truncated=False, so a caller can tell "was not cut" from "was cut to exactly the limit".

When the ceiling is too small to hold even the elision marker the result is the empty string rather than a bare marker. A marker on its own is not content: it would consume the entire allowance to communicate that the allowance was consumed.

Source code in src/symfonic/services/budget/truncation.py

def truncate(
    text: str,
    max_tokens: int,
    counter: TokenCounter,
    policy: TruncationPolicy | None = None,
) -> TruncationResult:
    """Cut ``text`` down until it costs at most ``max_tokens``.

    Returns the input untouched when it already fits — no marker, no copy, and
    ``truncated=False``, so a caller can tell "was not cut" from "was cut to
    exactly the limit".

    When the ceiling is too small to hold even the elision marker the result is
    the empty string rather than a bare marker. A marker on its own is not
    content: it would consume the entire allowance to communicate that the
    allowance was consumed.
    """
    if max_tokens < 0:
        raise BudgetPolicyError(f"max_tokens must be >= 0, got {max_tokens}")
    active = policy or _DEFAULT_POLICY

    original = counter.count(text)
    if original <= max_tokens:
        return TruncationResult(text=text, tokens=original, truncated=False)

    low, high, best = 0, len(text), -1
    while low <= high:
        mid = (low + high) // 2
        if counter.count(_compose(text, mid, active)) <= max_tokens:
            best, low = mid, mid + 1
        else:
            high = mid - 1

    # ``best`` is only ever assigned to a candidate that was *measured* under
    # the ceiling, which is what makes the fit guarantee independent of the
    # counter's behaviour: a badly-behaved counter can cost this search the
    # optimal answer, but it cannot buy an over-budget one.
    candidate = _compose(text, best, active) if best >= 0 else ""

    tokens = counter.count(candidate)
    return TruncationResult(
        text=candidate,
        tokens=tokens,
        truncated=True,
        dropped_tokens=original - tokens,
    )

symfonic.services.budget.truncation¶

truncation ¶

truncate ¶

`symfonic.services.budget.truncation`¶