edgevoid/freetoken
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[Feature] unify reasoning effort on one shared scale (#5)

Browse Source 
* feat(server): accept every reasoning-effort dialect

* refactor(server): derive thinking gears from the checkpoint, not a registry
This commit is contained in:
Xiaoze Fan
2026-08-15 23:40:52 -07:00
committed by GitHub
parent a2572083e9
commit d7cd6ed88b
19 changed files with 1227 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/freetoken/launch.py
+8 -5
View File
@@ -724,9 +724,12 @@ def prepare_dsh(ctx: LaunchContext) -> CommandSpec:
``freetoken-launch.settings.yaml`` for this invocation only, so the user's
``settings.yaml`` is never read or written. The llm-deepseek adapter (route
``deepseek-official``) is used rather than a custom llm-pi-ai provider: it
replays tool-call arguments and assistant content verbatim, which prefix
caching depends on. The dummy DEEPSEEK_API_KEY satisfies dsh's non-empty
key requirement; FreeToken itself is unauthenticated."""
replays tool-call arguments byte-verbatim and reasoning as
``reasoning_content``, while llm-pi-ai's JSON round-trip can change argument
values and key order — drift that survives render canonicalization, cuts
the prefix cache, and shows the model a rewrite of its own output. The
dummy DEEPSEEK_API_KEY satisfies dsh's non-empty key requirement; FreeToken
itself is unauthenticated."""
settings_path = _dsh_home() / DSH_LAUNCH_SETTINGS_NAME
patch_path = _dsh_home() / DSH_LAUNCH_PATCH_NAME
if not ctx.dry_run:
@@ -766,8 +769,8 @@ def prepare_dsh(ctx: LaunchContext) -> CommandSpec:
patch = [{"id": "settings", "config": {"path": str(settings_path)}}]
_write_text_with_backup(patch_path, yaml.safe_dump(patch, sort_keys=False))
# The `dsh web` alias rejects launcher flags like --patch; use the explicit
# --profile form and normalize a leading alias or --profile from extra args.
# The explicit --profile form (equivalent to the `web` alias) lets the
# normalization below swap the profile from extra args.
profile, app_args = "web", list(ctx.extra_args)
if app_args and app_args[0] == "web":
app_args = app_args[1:]
python/freetoken/server/anthropic_api.py
+6 -6
View File
@@ -284,17 +284,17 @@ def convert_anthropic_prompt(
selected = req.tool_choice.name if (req.tool_choice and req.tool_choice.type == "tool") else None
template_tools, parser_tools = split_tool_lists(raw_tools, selected)
# Native extended-thinking toggle -> template kwargs, through the per-family
# mapping in model_meta (a bare enable_thinking bool is inert for templates
# that read a different knob, e.g. M3's thinking_mode).
from .model_meta import think_toggle_kwargs
# Native extended-thinking toggle -> template kwargs, broadcast in every
# spelling the ecosystem's templates read (a bare enable_thinking bool is
# inert for templates that read a different knob, e.g. M3's thinking_mode).
from .model_meta import thinking_toggle_kwargs
ctk: dict[str, Any] = {}
if req.thinking:
if req.thinking.get("type") == "enabled":
ctk = think_toggle_kwargs(reasoning_parser, True)
ctk = thinking_toggle_kwargs(True)
elif req.thinking.get("type") == "disabled":
ctk = think_toggle_kwargs(reasoning_parser, False)
ctk = thinking_toggle_kwargs(False)
return render_messages(messages), template_tools, parser_tools, ctk
python/freetoken/server/api_models.py
+9
View File
@@ -78,6 +78,11 @@ class ChatCompletionRequest(BaseModel):
frequency_penalty: float = 0.0
chat_template_kwargs: dict[str, Any] = Field(default_factory=dict)
reasoning_effort: str | None = None
# DeepSeek-wire thinking toggle ({"type": "enabled"|"disabled"}). Any so a
# foreign shape stays ignored (extra="allow" swallowed it before this field
# existed) instead of becoming a bare 422 at the route boundary; the handler
# reads the dict form and 400s only on an unknown "type" value.
thinking: Any | None = None
ignore_eos: bool = False
tools: list[Tool] | None = None
tool_choice: Literal["none", "auto", "required"] | ToolChoiceObject | None = None
@@ -133,6 +138,10 @@ class ModelCard(BaseModel):
# `max_model_len` is vLLM/SGLang's, `context_length` what most other clients look for.
max_model_len: int | None = None
context_length: int | None = None
# The checkpoint's probed effort vocabulary (freetoken.tokenizer.effort); None
# (not []) when the model has no effort knob or the probe could not run.
supported_reasoning_efforts: list[str] | None = None
default_reasoning_effort: str | None = None
class ModelList(BaseModel):
python/freetoken/server/api_server.py
+45 -17
View File
@@ -178,6 +178,8 @@ class FrontendManager:
# the generation path. The lock dedupes concurrent first builds.
_frontend_tokenizer: Any = None
_frontend_tokenizer_lock: Any = field(default_factory=threading.Lock)
# One-shot guard for warm_frontend_tokenizer(); benign if two polls race it.
_frontend_warm_started: bool = False
def __post_init__(self) -> None:
if self.stats is None:
@@ -194,6 +196,22 @@ class FrontendManager:
self._frontend_tokenizer = TokenizeManager(load_tokenizer(self.config.model_path))
return self._frontend_tokenizer
def warm_frontend_tokenizer(self) -> None:
"""Build the frontend tokenizer and probe its thinking profile off-thread,
once — /v1/cache/status polls call this so the gear picker self-populates
without ever blocking the event loop on a tokenizer load."""
if self._frontend_warm_started:
return
self._frontend_warm_started = True
def _warm() -> None:
try:
self.frontend_tokenizer().thinking_profile()
except Exception: # noqa: BLE001 -- warmup only; real faults surface on use
pass
threading.Thread(target=_warm, daemon=True, name="frontend-tokenizer-warm").start()
def new_user(self) -> int:
if self.maintenance_state != "serving":
raise AdmissionClosedError(
@@ -651,6 +669,27 @@ def _cache_limits(geo: dict, unit_bytes: dict, pool_budget: int, floors: dict) -
}
def _reasoning_geometry(state: Any) -> dict | None:
"""The ``geometry.reasoning`` block, from the frontend tokenizer's probed
thinking profile. Peeks rather than builds: a cold tokenizer only kicks the
warmup thread (the profile itself is a handful of microsecond renders once
the tokenizer exists, safe to run inline)."""
manager = getattr(state, "_frontend_tokenizer", None)
if manager is None:
state.warm_frontend_tokenizer()
return None
from .model_meta import derive_think_gears
derived = derive_think_gears(
manager.thinking_profile(),
parser_configured=bool(getattr(state.config, "reasoning_parser", None)),
)
if derived is None:
return None
gears, default, kwargs = derived
return {"gears": list(gears), "default": default, "kwargs": kwargs}
def cache_geometry(state: Any) -> dict:
"""Current cache geometry for the desktop cache panel. Each pool size resolves
most-recent-truth first: the last rebuild's result, else the running UserReply snapshot
@@ -705,24 +744,13 @@ def cache_geometry(state: Any) -> dict:
}
except Exception:
unit_bytes = {"kv_per_token": 0, "moe_per_expert": 0, "mamba_per_slot": 0, "swa_per_token": 0}
# Model-family thinking control: the gears a client can offer and the chat_template_kwargs
# each selects, so the desktop can render a think/effort control (and the shell its /think)
# and send the right kwargs on /v1/chat/completions. None when the model has no controllable
# thinking (or a dummy/absent config).
from .model_meta import think_chat_template_kwargs, think_spec
# Thinking control: the gears a client can offer and the chat_template_kwargs
# each selects, derived from the checkpoint's own probed template (no
# per-family registry). None until the frontend tokenizer is warm — the
# first status poll kicks the warmup thread and later polls see the gears,
# so the picker self-populates without ever blocking this route.
try:
r_parser = getattr(config, "reasoning_parser", None)
r_gears, r_default = think_spec(r_parser)
reasoning = (
{
"gears": list(r_gears),
"default": r_default,
"kwargs": {g: think_chat_template_kwargs(r_parser, g) for g in r_gears},
}
if r_gears
else None
)
reasoning = _reasoning_geometry(state)
except Exception:
reasoning = None
geo = {
python/freetoken/server/generation.py
+30
View File
@@ -307,6 +307,36 @@ async def count_prompt_tokens(
return int(input_ids.numel())
async def prerender_error(spec: GenSpec, state: Any) -> GenerationError | None:
"""Render ``spec``'s prompt frontend-side, returning the failure a streaming
adapter should surface as an HTTP 400 *before* committing an SSE stream —
once headers go out, a template rejection can only ride in-stream, where
some agents show nothing but "empty response". Render only; the worker
still renders and encodes authoritatively. Best-effort: a state without a
frontend tokenizer, or one that fails to *initialize*, skips validation
rather than blocking the generation path.
"""
build = getattr(state, "frontend_tokenizer", None)
if build is None:
return None
msg = TokenizeMsg(
uid=0,
text=spec.messages,
sampling_params=SamplingParams(),
chat_template_kwargs=spec.chat_template_kwargs,
tools=spec.template_tools,
)
try:
manager = await asyncio.to_thread(build)
except Exception: # noqa: BLE001 -- server fault, not this request's problem
return None
try:
await asyncio.to_thread(manager.render_prompt, msg)
except Exception as exc: # noqa: BLE001 -- mirror the worker's classification
return GenerationError(f"could not encode request: {exc}")
return None
def _make_reasoning_parser(spec: GenSpec, state: Any) -> ReasoningParser | None:
"""Build a reasoning parser for this generation, or None if the server has no
reasoning parser configured. ``force_reasoning`` matches the encode-side
python/freetoken/server/model_meta.py
+81 -54
View File
@@ -11,60 +11,80 @@ from __future__ import annotations
import math
from typing import Any, Tuple
def think_spec(reasoning_parser: str | None) -> Tuple[Tuple[str, ...], str | None]:
"""Return ``(gears, default_gear)`` a client can offer for a model family, keyed by its
configured reasoning parser. ``((), None)`` when the model has no controllable thinking.
Verified per family against each model's chat template / encoder."""
if reasoning_parser == "gpt_oss":
return ("low", "medium", "high"), "medium" # always-on, 3-level effort
if reasoning_parser == "deepseekv32":
return ("off", "on", "max"), "off" # thinking on/off + a max-effort gear
if reasoning_parser == "minimax":
return ("on",), "on" # template always opens a think block; no off path
if reasoning_parser == "minimax_m3":
# M3's template takes thinking_mode disabled/adaptive/enabled; adaptive
# (the template's own default) lets the model decide per turn.
return ("off", "adaptive", "on"), "adaptive"
if reasoning_parser == "gemma4":
return ("off", "on"), "off" # gemma's template defaults thinking off
if reasoning_parser in ("qwen3", "glm"):
return ("off", "on"), "on"
return (), None
from freetoken.tokenizer.effort import (
EFFORT_SCALE,
OPENAI_EFFORT_TRIPLE,
THINKING_ADAPTIVE_KWARGS,
THINKING_OFF_KWARGS,
THINKING_ON_KWARGS,
ThinkingProfile,
)
def think_chat_template_kwargs(reasoning_parser: str | None, gear: str | None) -> dict:
"""The ``chat_template_kwargs`` that select ``gear`` for the model family."""
if gear is None:
return {}
if reasoning_parser == "gpt_oss":
return {"reasoning_effort": gear}
if reasoning_parser == "deepseekv32":
if gear == "max":
return {"enable_thinking": True, "reasoning_effort": "max"}
return {"enable_thinking": gear == "on"}
if reasoning_parser == "minimax":
return {} # always thinks; its template reads no knob
if reasoning_parser == "minimax_m3":
mode = {"off": "disabled", "adaptive": "adaptive", "on": "enabled"}[gear]
return {"thinking_mode": mode}
return {"enable_thinking": gear == "on"} # qwen3, glm, gemma4
def think_toggle_kwargs(reasoning_parser: str | None, enabled: bool) -> dict:
def thinking_toggle_kwargs(enabled: bool) -> dict:
"""``chat_template_kwargs`` for a protocol-level thinking on/off toggle
(Anthropic ``thinking.type``, Responses ``reasoning.effort``), routed through
the same per-family mapping as the chat-completions gears -- a hardcoded
``enable_thinking`` is inert for templates that read a different knob (M3's
``thinking_mode``). A family without the requested direction returns ``{}``;
with no configured parser the protocol-generic key is kept."""
gears, _default = think_spec(reasoning_parser)
(Anthropic ``thinking.type``, DeepSeek ``thinking``, Responses
``reasoning.effort``): every spelling the ecosystem's templates read,
broadcast at once. A template picks the knob it knows and ignores the rest
(Jinja never sees undeclared variables), so no per-family routing exists."""
return dict(THINKING_ON_KWARGS if enabled else THINKING_OFF_KWARGS)
def derive_think_gears(
profile: ThinkingProfile, parser_configured: bool
) -> Tuple[Tuple[str, ...], str | None, dict] | None:
"""``(gears, default_gear, kwargs_per_gear)`` for the /v1/cache/status
``geometry.reasoning`` block, derived from the checkpoint's probed thinking
controls -- the checkpoint owns this knowledge; nothing here is keyed by
model family. ``None`` when there is nothing controllable to offer.
A template that grades effort without validating it gets the OpenAI triple
(the only vocabulary such a template is known to understand); an always-on
model with a reasoning parser but no observable knob shows a single "on"
gear so clients can still label the state."""
efforts = profile.efforts
gears: list[str] = []
kwargs: dict[str, dict] = {}
if profile.toggleable:
gears.append("off")
kwargs["off"] = dict(THINKING_OFF_KWARGS)
if profile.has_adaptive:
gears.append("adaptive")
kwargs["adaptive"] = dict(THINKING_ADAPTIVE_KWARGS)
if efforts.consumes_effort:
names = (
[n for n in efforts.supported if n in EFFORT_SCALE]
if efforts.validates
else list(OPENAI_EFFORT_TRIPLE)
)
for name in sorted(names, key=lambda n: EFFORT_SCALE[n]):
gears.append(name)
gear_kwargs = dict(THINKING_ON_KWARGS) if profile.toggleable else {}
gear_kwargs["reasoning_effort"] = name
kwargs[name] = gear_kwargs
elif profile.toggleable:
gears.append("on")
kwargs["on"] = dict(THINKING_ON_KWARGS)
elif parser_configured:
# Always-thinking family (minimax): no knob, but the state is real.
gears.append("on")
kwargs["on"] = {}
if not gears:
return {"enable_thinking": enabled}
gear = "on" if enabled else "off"
if gear not in gears:
return {}
return think_chat_template_kwargs(reasoning_parser, gear)
return None
if profile.default_state == "off" and "off" in gears:
default = "off"
elif profile.default_state == "adaptive" and "adaptive" in gears:
default = "adaptive"
elif efforts.consumes_effort:
default = efforts.default if efforts.default in gears else (
"medium" if "medium" in gears else gears[-1]
)
else:
default = "on" if "on" in gears else gears[-1]
return tuple(gears), default, kwargs
_THINKING_KWARG_KEYS = ("enable_thinking", "thinking", "thinking_mode", "reasoning_effort")
@@ -72,22 +92,29 @@ _DISABLE_EFFORTS = ("none", "off")
def effort_toggle_kwargs(
reasoning_parser: str | None,
effort: str | None,
chat_template_kwargs: dict | None,
thinking_type: str | None = None,
) -> dict:
"""Fold a protocol-level reasoning-effort request into the template kwargs.
An explicit thinking-related key wins wholesale; unrelated extras ride along.
Effort "none"/"off" (case-insensitive) disables thinking; any other or absent
effort enables it, forwarded for templates that grade it (gpt-oss)."""
effort enables it, forwarded for templates that grade it (quantized against
the checkpoint's probed vocabulary at render time). ``thinking_type`` is the
DeepSeek-wire ``thinking: {"type": ...}`` toggle; when present it decides
the on/off direction outright, "disabled" winning over any effort."""
ctk = dict(chat_template_kwargs or {})
if any(key in ctk for key in _THINKING_KWARG_KEYS):
return ctk
if isinstance(effort, str):
effort = effort.strip().lower()
disabled = effort in _DISABLE_EFFORTS
mapped = dict(think_toggle_kwargs(reasoning_parser, not disabled))
if effort and not disabled:
if thinking_type == "disabled":
disabled = True
elif thinking_type == "enabled":
disabled = False
mapped = thinking_toggle_kwargs(not disabled)
if effort and not disabled and effort not in _DISABLE_EFFORTS:
mapped.setdefault("reasoning_effort", effort)
mapped.update(ctk)
return mapped
python/freetoken/server/openai_api.py
+68 -9
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import asyncio
import json
import time
import uuid
@@ -10,6 +11,7 @@ from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, StreamingResponse
from freetoken.core import SamplingParams
from freetoken.message import TokenizeMsg
from freetoken.tokenizer.effort import EFFORT_SCALE, KNOWN_REASONING_EFFORTS
from .api_models import (
ChatCompletionRequest,
@@ -31,23 +33,39 @@ from .generation import (
ToolCallStart,
generate_events,
generate_full,
prerender_error,
render_messages,
resolve_sampling,
submit_generation,
)
#: The wire superset plus "off", DeepSeek's disable synonym that
#: effort_toggle_kwargs has always honored.
_ACCEPTED_EFFORTS = (*KNOWN_REASONING_EFFORTS, "off")
def _thinking_type(req: Any) -> str | None:
"""The DeepSeek-wire thinking toggle, or None for absent/foreign shapes
(which stay ignored, as extra="allow" ignored them before the field existed)."""
if isinstance(req.thinking, dict):
value = req.thinking.get("type")
if value in ("enabled", "disabled"):
return value
return None
def chat_request_to_genspec(
req: ChatCompletionRequest,
model_sampling: dict[str, Any],
reasoning_parser: str | None = None,
) -> GenSpec:
"""OpenAI ChatCompletionRequest -> GenSpec (the OpenAI 'to_sampling_params')."""
from .model_meta import effort_toggle_kwargs
ctk = req.chat_template_kwargs
if req.reasoning_effort:
ctk = effort_toggle_kwargs(reasoning_parser, req.reasoning_effort, ctk)
thinking_type = _thinking_type(req)
if req.reasoning_effort or thinking_type:
ctk = effort_toggle_kwargs(req.reasoning_effort, ctk, thinking_type=thinking_type)
return GenSpec(
messages=render_messages([m.model_dump(exclude_none=True) for m in req.messages]),
sampling_params=resolve_sampling(
@@ -115,11 +133,14 @@ def register_openai_routes(
state = get_state()
model_id = _served_model_name(state)
ctx = _model_context_length(state)
efforts, default_effort = await _effort_fields(state)
return ModelList(data=[ModelCard(
id=model_id,
root=state.config.model_path,
max_model_len=ctx,
context_length=ctx,
supported_reasoning_efforts=efforts,
default_reasoning_effort=default_effort,
)])
@@ -140,13 +161,35 @@ async def handle_chat_completion(
)
if req.n != 1:
return create_error_response("Only n=1 is supported", param="n")
# Case/whitespace and the "off" disable synonym stay accepted here because
# effort_toggle_kwargs normalizes and honors them downstream.
effort = req.reasoning_effort.strip().lower() if isinstance(req.reasoning_effort, str) else None
if effort and effort not in _ACCEPTED_EFFORTS:
return create_error_response(
f"reasoning_effort must be one of {', '.join(_ACCEPTED_EFFORTS)}; "
f"got {req.reasoning_effort!r}",
param="reasoning_effort",
)
if isinstance(req.thinking, dict):
thinking_type = req.thinking.get("type")
if thinking_type is not None and thinking_type not in ("enabled", "disabled"):
return create_error_response(
f"thinking.type must be 'enabled' or 'disabled'; got {thinking_type!r}",
param="thinking",
)
try:
spec = chat_request_to_genspec(
req, model_sampling, reasoning_parser=getattr(state.config, "reasoning_parser", None)
)
spec = chat_request_to_genspec(req, model_sampling)
except ValueError as exc:
return create_error_response(str(exc))
if req.stream:
# Non-stream requests already surface render failures as a clean 400
# through GenerationError; only the stream path needs the pre-check.
err = await prerender_error(spec, state)
if err is not None:
return create_error_response(str(err), code=err.code)
uid = await submit_generation(spec, state)
if req.stream:
@@ -193,9 +236,7 @@ async def stream_chat_completion_chunks(
) -> AsyncIterator[bytes]:
"""Format generate_events() into the OpenAI chat.completion.chunk SSE stream."""
if spec is None:
spec = chat_request_to_genspec(
req, {}, reasoning_parser=getattr(state.config, "reasoning_parser", None)
)
spec = chat_request_to_genspec(req, {})
yield _sse(
_chat_chunk(
req,
@@ -610,6 +651,24 @@ def _is_token_prompt(prompt: Any) -> bool:
)
async def _effort_fields(state: Any) -> tuple[list[str] | None, str | None]:
"""The checkpoint's probed effort vocabulary for /v1/models, or (None, None)
when there is no frontend tokenizer, it fails to build, or the model has no
effort knob — a metadata route must never 500 over this."""
build = getattr(state, "frontend_tokenizer", None)
if build is None:
return None, None
try:
manager = await asyncio.to_thread(build)
profile = await asyncio.to_thread(manager.effort_profile)
except Exception: # noqa: BLE001 -- metadata only; the generation path reports real faults
return None, None
if not profile.consumes_effort:
return None, None
ordered = sorted(profile.supported, key=lambda name: -EFFORT_SCALE.get(name, 0.0))
return ordered, profile.default
def _served_model_name(state: Any) -> str:
return getattr(state.config, "served_model_name", None) or state.config.model_path
python/freetoken/server/responses_api.py
+1 -1
View File
@@ -234,7 +234,7 @@ def convert_responses_to_genspec(
ctk = dict(getattr(req, "chat_template_kwargs", None) or {})
if req.reasoning:
ctk = effort_toggle_kwargs(reasoning_parser, req.reasoning.get("effort"), ctk)
ctk = effort_toggle_kwargs(req.reasoning.get("effort"), ctk)
return GenSpec(
messages=render_messages(messages),
python/freetoken/tokenizer/effort.py
+228
View File
@@ -0,0 +1,228 @@
"""Reasoning-effort dialect handling.
Each checkpoint's template/encoder accepts only its own effort vocabulary and
hard-fails on the rest, while clients speak whatever dialect their provider
taught them. Named levels project onto the numeric scale vLLM and SGLang share,
and out-of-vocabulary values quantize to the nearest supported gear instead of
failing the request. The vocabulary is probed from the checkpoint's own
template, never from a static table: a parser-family registry cannot be keyed
correctly (Qwen3 and Qwen3.8 resolve to the same parser but only the latter
grades effort).
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Callable
#: Values must match vLLM's and SGLang's inkling table so the ecosystems agree
#: on what "medium" means relative to "xhigh".
EFFORT_SCALE: dict[str, float] = {
"none": 0.0,
"minimal": 0.1,
"low": 0.2,
"medium": 0.7,
"high": 0.9,
"xhigh": 0.99,
"max": 0.99,
}
KNOWN_REASONING_EFFORTS = tuple(EFFORT_SCALE)
#: OpenAI's effort triple -- the common-denominator vocabulary offered for a
#: template that grades effort without validating it (nothing observable
#: narrows its gears, so offer what every dialect understands).
OPENAI_EFFORT_TRIPLE = ("low", "medium", "high")
#: Protocol-level thinking toggles, broadcast in every spelling the ecosystem's
#: templates read (``enable_thinking`` bool: qwen/glm/gemma/dsv4;
#: ``thinking_mode`` string: minimax-m3). Jinja ignores undeclared variables,
#: so a template simply picks the knob it knows -- no per-family routing needed.
THINKING_OFF_KWARGS = {"enable_thinking": False, "thinking_mode": "disabled"}
THINKING_ON_KWARGS = {"enable_thinking": True, "thinking_mode": "enabled"}
THINKING_ADAPTIVE_KWARGS = {"thinking_mode": "adaptive"}
@dataclass(frozen=True)
class EffortProfile:
"""What one checkpoint's template/encoder accepts, learned by probing it.
``default`` is the supported name whose rendering is byte-identical to
passing no effort at all. ``consumes_effort`` False means no probe round
ever changed its output or raised -- the template ignores the knob, so
requests should not carry it. ``validates`` True means the probe observed a
rejection: only then is ``supported`` a real vocabulary rather than "this
template interpolates anything".
"""
supported: frozenset[str]
default: str | None
consumes_effort: bool
validates: bool = False
@dataclass(frozen=True)
class ThinkingProfile:
"""A checkpoint's thinking controls, learned by probing its template.
``toggleable``: the off/on broadcasts render differently. ``has_adaptive``:
thinking_mode "adaptive" is a third distinct state (minimax-m3).
``default_state``: which state the bare render matches ("on" when not
toggleable or unmatched).
"""
efforts: EffortProfile
toggleable: bool
has_adaptive: bool
default_state: str # "on" | "off" | "adaptive"
#: A gear farther than this on the scale misrepresents the request; drop the
#: value and let the template default apply. Keeps OpenAI's "medium" from
#: escalating to the DSV4 encoder's absolute-maximum "high" gear (0.2 away),
#: matching vLLM's DSV4 mapping, while "high" still reaches Qwen's "xhigh"
#: (0.09 away).
_MAX_QUANTIZE_DISTANCE = 0.15
def quantize_effort(value: Any, profile: EffortProfile) -> str | None:
"""Map a client's effort onto ``profile``; ``None`` means "send nothing".
In-vocabulary values pass through untouched. Other named levels land on the
nearest supported gear within ``_MAX_QUANTIZE_DISTANCE`` -- except "max",
which is reachable only by its own name (vLLM's DSV4 rule: an extreme
opt-in gear must never be entered by rounding). Everything else drops to
the template default. With "max" excluded the remaining scale values are
unique, so quantization is deterministic across processes.
"""
if not profile.consumes_effort:
return None
if isinstance(value, str) and value in profile.supported:
return value
position = EFFORT_SCALE.get(value) if isinstance(value, str) else None
if position is None:
return None
ranked = sorted(
(name for name in profile.supported if name != "max"),
key=lambda name: (abs(EFFORT_SCALE[name] - position), -EFFORT_SCALE[name]),
)
if ranked and abs(EFFORT_SCALE[ranked[0]] - position) <= _MAX_QUANTIZE_DISTANCE:
return ranked[0]
return None
#: One tool flips tool-conditional paths (the DSV4 encoder grades effort only
#: in thinking mode, which tools force).
_PROBE_TOOLS = [
{
"type": "function",
"function": {
"name": "noop",
"description": "No-op probe tool.",
"parameters": {"type": "object", "properties": {}},
},
}
]
#: (extra chat_template_kwargs, tools) per probe round: templates read effort
#: unconditionally, under tool-forced thinking, or only under an explicit
#: thinking opt-in -- one round per shape.
_PROBE_ROUNDS: tuple[tuple[dict[str, Any], list[dict[str, Any]] | None], ...] = (
({}, None),
({}, _PROBE_TOOLS),
({"enable_thinking": True}, None),
)
def probe_effort_profile(
render: Callable[[dict[str, Any], list[dict[str, Any]] | None], Any],
) -> EffortProfile:
"""Learn a checkpoint's effort vocabulary by rendering probes through it.
``render(chat_template_kwargs, tools)`` returns a comparable rendering and
raises on rejection. A round whose no-effort baseline raises is skipped:
the template rejected the probe conversation shape, not the effort.
"""
rejected: set[str] = set()
diverged: set[str] = set()
matches_baseline: dict[str, bool] = {name: True for name in KNOWN_REASONING_EFFORTS}
ran_rounds = 0
for base_kwargs, tools in _PROBE_ROUNDS:
try:
baseline = render(dict(base_kwargs), tools)
except Exception: # noqa: BLE001 -- template rejects the probe shape, not the effort
continue
ran_rounds += 1
for name in KNOWN_REASONING_EFFORTS:
try:
rendering = render({**base_kwargs, "reasoning_effort": name}, tools)
except Exception: # noqa: BLE001 -- any raise means "not accepted"
rejected.add(name)
matches_baseline[name] = False
continue
if _renderings_differ(rendering, baseline):
diverged.add(name)
matches_baseline[name] = False
if ran_rounds == 0:
# Nothing learnable: sending no effort is the only safe rendering.
return EffortProfile(supported=frozenset(), default=None, consumes_effort=False)
supported = frozenset(name for name in KNOWN_REASONING_EFFORTS if name not in rejected)
consumes = bool(rejected or diverged)
default = None
if consumes:
defaults = [name for name in supported if matches_baseline[name]]
if defaults:
default = max(defaults, key=lambda name: EFFORT_SCALE[name])
return EffortProfile(
supported=supported,
default=default,
consumes_effort=consumes,
validates=bool(rejected),
)
def probe_thinking_profile(
render: Callable[[dict[str, Any], list[dict[str, Any]] | None], Any],
efforts: EffortProfile,
) -> ThinkingProfile:
"""Learn a checkpoint's thinking-toggle behavior by rendering the broadcast
kwargs through its template. Same contract as ``probe_effort_profile``; a
template that rejects any toggle probe is treated as not toggleable."""
try:
baseline = render({}, None)
off = render(dict(THINKING_OFF_KWARGS), None)
on = render(dict(THINKING_ON_KWARGS), None)
except Exception: # noqa: BLE001 -- can't observe the toggle; assume none
return ThinkingProfile(efforts=efforts, toggleable=False, has_adaptive=False, default_state="on")
toggleable = _renderings_differ(off, on)
has_adaptive = False
adaptive = None
if toggleable:
try:
adaptive = render(dict(THINKING_ADAPTIVE_KWARGS), None)
has_adaptive = _renderings_differ(adaptive, off) and _renderings_differ(adaptive, on)
except Exception: # noqa: BLE001 -- adaptive is not a state this template knows
has_adaptive = False
default_state = "on"
if toggleable:
if not _renderings_differ(baseline, off):
default_state = "off"
elif has_adaptive and not _renderings_differ(baseline, adaptive):
default_state = "adaptive"
return ThinkingProfile(
efforts=efforts,
toggleable=toggleable,
has_adaptive=has_adaptive,
default_state=default_state,
)
def _renderings_differ(a: Any, b: Any) -> bool:
try:
return bool(a != b)
except Exception: # noqa: BLE001 -- exotic tensor comparison; treat as divergence
return True
python/freetoken/tokenizer/tokenize.py
+105 -29
View File
@@ -3,17 +3,24 @@ from __future__ import annotations
import importlib.util
import json
import os
import threading
from types import ModuleType
from typing import Any, List
import torch
from freetoken.message import TokenizeMsg
from freetoken.utils import init_logger
from transformers import PreTrainedTokenizerBase
#: Reasoning-effort values accepted by the DeepSeek-V4 encoder. Anything else
#: (notably OpenAI's default ``"medium"``) makes ``encoding_dsv4.render_message``
#: raise an assertion, so unsupported values are normalized to ``None``.
VALID_REASONING_EFFORTS = ("max", "high")
from .effort import (
EffortProfile,
ThinkingProfile,
probe_effort_profile,
probe_thinking_profile,
quantize_effort,
)
logger = init_logger(__name__)
def resolve_thinking_mode(chat_template_kwargs: dict[str, Any] | None, tools: Any | None) -> str:
@@ -36,47 +43,114 @@ def resolve_thinking_mode(chat_template_kwargs: dict[str, Any] | None, tools: An
return mode
def normalize_reasoning_effort(value: Any | None) -> str | None:
"""Drop reasoning-effort values the dsv4 encoder cannot accept (-> ``None``)."""
return value if value in VALID_REASONING_EFFORTS else None
_EFFORT_PROBE_MESSAGES = [{"role": "user", "content": "ping"}]
class TokenizeManager:
def __init__(self, tokenizer: PreTrainedTokenizerBase) -> None:
self.tokenizer = tokenizer
self._dsv4_encoder = _load_dsv4_encoder_if_needed(tokenizer)
self._effort_profile: EffortProfile | None = None
self._thinking_profile: ThinkingProfile | None = None
self._effort_lock = threading.Lock()
self._logged_effort_maps: set[tuple[Any, str | None]] = set()
def tokenize(self, msgs: List[TokenizeMsg]) -> List[torch.Tensor]:
results: List[torch.Tensor] = []
# TODO: batch tokenization
for msg in msgs:
if isinstance(msg.text, list):
chat_template_kwargs = msg.chat_template_kwargs or {}
if self._dsv4_encoder is not None:
prompt = _apply_dsv4_chat_encoder(
self._dsv4_encoder,
msg.text,
msg.tools,
chat_template_kwargs,
)
else:
if msg.tools is not None:
chat_template_kwargs = {**chat_template_kwargs, "tools": msg.tools}
prompt = self.tokenizer.apply_chat_template(
msg.text,
tokenize=False,
add_generation_prompt=True,
**chat_template_kwargs,
)
assert isinstance(prompt, str)
else:
prompt = msg.text
prompt = self.render_prompt(msg)
input_ids: torch.Tensor = ( # type: ignore
self.tokenizer.encode(prompt, return_tensors="pt")
)
results.append(input_ids.view(-1).to(torch.int32))
return results
def render_prompt(self, msg: TokenizeMsg) -> str:
"""The template/encoder half of ``tokenize``, exposed so the frontend can
validate a request before committing an SSE stream. Sanitizes
``reasoning_effort`` first: every render path (worker, frontend
validation, count_tokens) must quantize identically."""
if not isinstance(msg.text, list):
return msg.text
return self._render(
msg.text, msg.tools, self._sanitize_effort(msg.chat_template_kwargs or {})
)
def _render(
self,
messages: list[dict[str, Any]],
tools: list[dict[str, Any]] | None,
chat_template_kwargs: dict[str, Any],
) -> str:
"""Raw render, no effort sanitation — the probe needs unsupported values
to actually reach the template so rejection is observable."""
if self._dsv4_encoder is not None:
return _apply_dsv4_chat_encoder(
self._dsv4_encoder, messages, tools, chat_template_kwargs
)
if tools is not None:
chat_template_kwargs = {**chat_template_kwargs, "tools": tools}
prompt = self.tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True,
**chat_template_kwargs,
)
assert isinstance(prompt, str)
return prompt
def effort_profile(self) -> EffortProfile:
"""The checkpoint's effort vocabulary, probed on first use and cached
for the process lifetime."""
with self._effort_lock:
if self._effort_profile is None:
self._effort_profile = probe_effort_profile(self._probe_render)
logger.info(
"reasoning-effort profile: supported=%s default=%s",
sorted(self._effort_profile.supported) or "(none)",
self._effort_profile.default,
)
return self._effort_profile
def thinking_profile(self) -> ThinkingProfile:
"""The checkpoint's thinking controls (toggle behavior + effort
vocabulary), probed on first use and cached for the process lifetime.
Feeds the /v1/cache/status gear derivation."""
efforts = self.effort_profile()
with self._effort_lock:
if self._thinking_profile is None:
self._thinking_profile = probe_thinking_profile(self._probe_render, efforts)
return self._thinking_profile
def _probe_render(
self, kwargs: dict[str, Any], tools: list[dict[str, Any]] | None
) -> str:
return self._render(_EFFORT_PROBE_MESSAGES, tools, kwargs)
def _sanitize_effort(self, chat_template_kwargs: dict[str, Any]) -> dict[str, Any]:
if "reasoning_effort" not in chat_template_kwargs:
return chat_template_kwargs
raw = chat_template_kwargs.get("reasoning_effort")
mapped = quantize_effort(raw, self.effort_profile())
if mapped == raw:
return chat_template_kwargs
# raw is client-controlled and may be unhashable (a JSON list/dict).
key = (raw if isinstance(raw, str) else repr(raw), mapped)
if key not in self._logged_effort_maps:
self._logged_effort_maps.add(key)
logger.info(
"reasoning_effort %r is not supported by this checkpoint; using %s",
raw,
mapped if mapped is not None else "the template default",
)
sanitized = dict(chat_template_kwargs)
if mapped is None:
del sanitized["reasoning_effort"]
else:
sanitized["reasoning_effort"] = mapped
return sanitized
def _load_dsv4_encoder_if_needed(tokenizer: PreTrainedTokenizerBase) -> ModuleType | None:
if getattr(tokenizer, "chat_template", None):
@@ -110,10 +184,12 @@ def _apply_dsv4_chat_encoder(
if tools:
_attach_tools_to_dsv4_messages(rendered_messages, tools)
# No effort filtering here: the caller sanitized already, and the probe
# needs raw values to reach the encoder's own validation.
return encoder.encode_messages(
rendered_messages,
thinking_mode=resolve_thinking_mode(chat_template_kwargs, tools),
reasoning_effort=normalize_reasoning_effort(chat_template_kwargs.get("reasoning_effort")),
reasoning_effort=chat_template_kwargs.get("reasoning_effort"),
)
tests/server/test_anthropic_api.py
+13 -18
View File
@@ -158,10 +158,10 @@ def test_convert_system_role_message_and_unknown_block():
assert spec.messages[2]["content"] == "hi"
def test_convert_thinking_toggle_routes_through_family_mapping():
"""A hardcoded {"enable_thinking": bool} is inert for M3, whose template
reads thinking_mode -- the toggle must go through the same model_meta
mapping as the chat-completions gears."""
def test_convert_thinking_toggle_broadcasts_every_spelling():
"""The toggle is broadcast in every spelling templates read (enable_thinking
bool + M3's thinking_mode); each template picks the knob it knows and Jinja
ignores the rest, so the kwargs are family-independent."""
def _req(ttype):
return AnthropicMessagesRequest.model_validate(
{
@@ -171,18 +171,13 @@ def test_convert_thinking_toggle_routes_through_family_mapping():
}
)
spec = A.convert_anthropic_to_genspec(_req("enabled"), {}, reasoning_parser="minimax_m3")
assert spec.chat_template_kwargs == {"thinking_mode": "enabled"}
spec = A.convert_anthropic_to_genspec(_req("disabled"), {}, reasoning_parser="minimax_m3")
assert spec.chat_template_kwargs == {"thinking_mode": "disabled"}
# enable_thinking families keep their key; no parser keeps the generic key
spec = A.convert_anthropic_to_genspec(_req("enabled"), {}, reasoning_parser="qwen3")
assert spec.chat_template_kwargs == {"enable_thinking": True}
spec = A.convert_anthropic_to_genspec(_req("disabled"), {})
assert spec.chat_template_kwargs == {"enable_thinking": False}
# a family without the requested direction sets nothing (minimax has no off)
spec = A.convert_anthropic_to_genspec(_req("disabled"), {}, reasoning_parser="minimax")
assert spec.chat_template_kwargs == {}
on = {"enable_thinking": True, "thinking_mode": "enabled"}
off = {"enable_thinking": False, "thinking_mode": "disabled"}
for parser in ("minimax_m3", "qwen3", "minimax", None):
spec = A.convert_anthropic_to_genspec(_req("enabled"), {}, reasoning_parser=parser)
assert spec.chat_template_kwargs == on, parser
spec = A.convert_anthropic_to_genspec(_req("disabled"), {}, reasoning_parser=parser)
assert spec.chat_template_kwargs == off, parser
def test_convert_thinking_only_assistant_message_keeps_empty_content():
@@ -598,11 +593,11 @@ def test_convert_native_thinking_toggle():
on = A.convert_anthropic_to_genspec(
AnthropicMessagesRequest.model_validate({**base, "thinking": {"type": "enabled", "budget_tokens": 1024}}), {}
)
assert on.chat_template_kwargs == {"enable_thinking": True}
assert on.chat_template_kwargs == {"enable_thinking": True, "thinking_mode": "enabled"}
off = A.convert_anthropic_to_genspec(
AnthropicMessagesRequest.model_validate({**base, "thinking": {"type": "disabled"}}), {}
)
assert off.chat_template_kwargs == {"enable_thinking": False}
assert off.chat_template_kwargs == {"enable_thinking": False, "thinking_mode": "disabled"}
absent = A.convert_anthropic_to_genspec(AnthropicMessagesRequest.model_validate(base), {})
assert absent.chat_template_kwargs == {}
tests/server/test_effort_dialect.py
+243
View File
@@ -0,0 +1,243 @@
"""Effort/thinking dialect handling at the OpenAI API layer.
Covers the wire-level half of the reasoning-effort pipeline: the superset
validation and DeepSeek ``thinking`` toggle in ``handle_chat_completion``, the
pre-stream render validation, and the probed vocabulary on ``/v1/models``.
The quantization itself is covered in tests/tokenizer/test_effort.py.
"""
from __future__ import annotations
import asyncio
import json
from types import SimpleNamespace
from fastapi import FastAPI
from fastapi.responses import JSONResponse, StreamingResponse
from fastapi.testclient import TestClient
from freetoken.message import TokenizeMsg, UserReply
from freetoken.server.model_meta import effort_toggle_kwargs
from freetoken.server.openai_api import (
ChatCompletionRequest,
handle_chat_completion,
register_openai_routes,
)
from freetoken.tokenizer.effort import EffortProfile
def run(coro):
return asyncio.run(coro)
class FakeState:
def __init__(self, reasoning_parser: str | None = None) -> None:
self.config = SimpleNamespace(
model_path="/models/unit-model",
served_model_name="unit-model",
tool_call_parser="llama3",
reasoning_parser=reasoning_parser,
)
self.sent: TokenizeMsg | None = None
def new_user(self) -> int:
return 42
async def send_one(self, msg):
self.sent = msg
async def wait_for_ack(self, uid: int):
yield UserReply(uid=uid, incremental_output="ok", finished=True, finish_reason="stop")
class FakeManager:
def __init__(self, profile: EffortProfile | None = None, render_error: Exception | None = None):
self._profile = profile
self._render_error = render_error
def effort_profile(self) -> EffortProfile:
assert self._profile is not None
return self._profile
def render_prompt(self, msg) -> str:
if self._render_error is not None:
raise self._render_error
return "rendered"
def chat_request(**overrides) -> ChatCompletionRequest:
payload = {
"model": "unit-model",
"messages": [{"role": "user", "content": "hi"}],
**overrides,
}
return ChatCompletionRequest(**payload)
# --------------------------------------------------------------------------- #
# effort_toggle_kwargs: the DeepSeek thinking toggle folds into template kwargs.
# --------------------------------------------------------------------------- #
OFF = {"enable_thinking": False, "thinking_mode": "disabled"}
ON = {"enable_thinking": True, "thinking_mode": "enabled"}
def test_thinking_disabled_wins_over_an_effort():
ctk = effort_toggle_kwargs("high", {}, thinking_type="disabled")
assert ctk == OFF
def test_thinking_enabled_forwards_the_effort():
ctk = effort_toggle_kwargs("high", {}, thinking_type="enabled")
assert ctk == {**ON, "reasoning_effort": "high"}
def test_thinking_enabled_alone_turns_thinking_on():
ctk = effort_toggle_kwargs(None, {}, thinking_type="enabled")
assert ctk == ON
def test_explicit_template_kwargs_still_win_wholesale():
ctk = effort_toggle_kwargs("high", {"enable_thinking": False}, thinking_type="enabled")
assert ctk == {"enable_thinking": False}
# --------------------------------------------------------------------------- #
# handle_chat_completion: superset validation and the pre-stream render check.
# --------------------------------------------------------------------------- #
def test_unknown_reasoning_effort_is_a_400():
response = run(
handle_chat_completion(chat_request(reasoning_effort="banana"), None, FakeState(), {})
)
assert isinstance(response, JSONResponse)
assert response.status_code == 400
assert "reasoning_effort" in json.loads(response.body)["error"]["message"]
def test_unknown_thinking_type_is_a_400():
response = run(
handle_chat_completion(
chat_request(thinking={"type": "sideways"}), None, FakeState(), {}
)
)
assert isinstance(response, JSONResponse)
assert response.status_code == 400
def test_thinking_disabled_reaches_the_tokenizer_as_enable_thinking_false():
state = FakeState(reasoning_parser="qwen3")
response = run(
handle_chat_completion(
chat_request(thinking={"type": "disabled"}), None, state, {}
)
)
assert not isinstance(response, JSONResponse) # plain successful completion
assert state.sent is not None
assert state.sent.chat_template_kwargs == OFF
def test_off_and_mixed_case_efforts_stay_accepted():
# effort_toggle_kwargs has always normalized case/whitespace and honored
# "off" as a disable synonym; the superset gate must not reject them.
for effort, expected in (
("off", OFF),
("High", {**ON, "reasoning_effort": "high"}),
(" high ", {**ON, "reasoning_effort": "high"}),
):
state = FakeState(reasoning_parser="qwen3")
response = run(
handle_chat_completion(chat_request(reasoning_effort=effort), None, state, {})
)
assert not isinstance(response, JSONResponse), effort
assert state.sent.chat_template_kwargs == expected, effort
def test_empty_effort_is_treated_as_absent():
state = FakeState(reasoning_parser="qwen3")
response = run(
handle_chat_completion(chat_request(reasoning_effort=""), None, state, {})
)
assert not isinstance(response, JSONResponse)
assert state.sent.chat_template_kwargs == {}
def test_foreign_thinking_shapes_stay_ignored():
# extra="allow" swallowed any thinking shape before the field existed;
# a bare string, a bool, or a typeless dict must keep working unchanged.
for shape in ("enabled", True, {}, {"budget_tokens": 1024}):
state = FakeState(reasoning_parser="qwen3")
response = run(
handle_chat_completion(chat_request(thinking=shape), None, state, {})
)
assert not isinstance(response, JSONResponse), shape
assert state.sent.chat_template_kwargs == {}, shape
def test_anthropic_style_thinking_dict_works():
state = FakeState(reasoning_parser="qwen3")
response = run(
handle_chat_completion(
chat_request(thinking={"type": "enabled", "budget_tokens": 1024}), None, state, {}
)
)
assert not isinstance(response, JSONResponse)
assert state.sent.chat_template_kwargs == ON
def test_stream_returns_400_when_the_template_rejects_the_render():
state = FakeState()
state.frontend_tokenizer = lambda: FakeManager(
render_error=ValueError("Unexpected reasoning effort high.")
)
response = run(handle_chat_completion(chat_request(stream=True), None, state, {}))
assert isinstance(response, JSONResponse)
assert response.status_code == 400
message = json.loads(response.body)["error"]["message"]
assert message.startswith("could not encode request")
assert state.sent is None # rejected before submission
def test_stream_proceeds_without_a_frontend_tokenizer():
# Minimal embeddings (and the unit FakeState) have no frontend tokenizer;
# validation degrades to the old worker-side path instead of blocking.
response = run(handle_chat_completion(chat_request(stream=True), None, FakeState(), {}))
assert isinstance(response, StreamingResponse)
# --------------------------------------------------------------------------- #
# /v1/models: the probed vocabulary is published; absence stays None.
# --------------------------------------------------------------------------- #
def _models_payload(state) -> dict:
app = FastAPI()
register_openai_routes(app, lambda: state, dict)
with TestClient(app) as client:
response = client.get("/v1/models")
assert response.status_code == 200
return response.json()["data"][0]
def test_v1_models_publishes_the_probed_efforts():
state = FakeState()
state.frontend_tokenizer = lambda: FakeManager(
profile=EffortProfile(
supported=frozenset({"xhigh", "medium", "low"}),
default="xhigh",
consumes_effort=True,
)
)
card = _models_payload(state)
assert card["supported_reasoning_efforts"] == ["xhigh", "medium", "low"]
assert card["default_reasoning_effort"] == "xhigh"
def test_v1_models_omits_efforts_without_a_frontend_tokenizer():
card = _models_payload(FakeState())
assert card["supported_reasoning_efforts"] is None
assert card["default_reasoning_effort"] is None
def test_v1_models_omits_efforts_for_models_without_the_knob():
state = FakeState()
state.frontend_tokenizer = lambda: FakeManager(
profile=EffortProfile(supported=frozenset(), default=None, consumes_effort=False)
)
card = _models_payload(state)
assert card["supported_reasoning_efforts"] is None
assert card["default_reasoning_effort"] is None
tests/server/test_minimax_m3_parsers.py
+12 -3
View File
@@ -590,8 +590,17 @@ def test_detect_and_parse_multiple_wrappers_and_inter_block_text():
"gear,mode", [("off", "disabled"), ("adaptive", "adaptive"), ("on", "enabled")]
)
def test_think_gears(gear, mode):
from freetoken.server.model_meta import think_chat_template_kwargs, think_spec
"""M3's three thinking states are discovered from its template behavior
(thinking_mode disabled/adaptive/enabled, template default adaptive)."""
from freetoken.server.model_meta import derive_think_gears
from freetoken.tokenizer.effort import EffortProfile, probe_thinking_profile
gears, default = think_spec("minimax_m3")
def m3_render(kwargs, tools):
# The template reads thinking_mode only; adaptive is its own default.
return f"m3|{kwargs.get('thinking_mode', 'adaptive')}"
no_efforts = EffortProfile(supported=frozenset(), default=None, consumes_effort=False)
profile = probe_thinking_profile(m3_render, no_efforts)
gears, default, kwargs = derive_think_gears(profile, parser_configured=True)
assert gears == ("off", "adaptive", "on") and default == "adaptive"
assert think_chat_template_kwargs("minimax_m3", gear) == {"thinking_mode": mode}
assert kwargs[gear]["thinking_mode"] == mode
tests/server/test_openai_api.py
+19 -17
View File
@@ -163,7 +163,9 @@ def test_chat_request_reasoning_replay_field_aliases():
def test_chat_reasoning_effort_enables_thinking():
spec = chat_request_to_genspec(chat_request(reasoning_effort="high"), {})
assert spec.chat_template_kwargs == {"enable_thinking": True, "reasoning_effort": "high"}
assert spec.chat_template_kwargs == {
"enable_thinking": True, "thinking_mode": "enabled", "reasoning_effort": "high"
}
# an explicit thinking-related chat_template_kwargs key wins over the mapping
spec = chat_request_to_genspec(
@@ -175,7 +177,9 @@ def test_chat_reasoning_effort_enables_thinking():
spec = chat_request_to_genspec(
chat_request(reasoning_effort="none", chat_template_kwargs={"custom_var": 1}), {}
)
assert spec.chat_template_kwargs == {"enable_thinking": False, "custom_var": 1}
assert spec.chat_template_kwargs == {
"enable_thinking": False, "thinking_mode": "disabled", "custom_var": 1
}
# absent effort -> kwargs pass through untouched
assert chat_request_to_genspec(chat_request(), {}).chat_template_kwargs == {}
@@ -184,25 +188,23 @@ def test_chat_reasoning_effort_enables_thinking():
def test_chat_reasoning_effort_none_disables_thinking():
# vLLM-compatible semantics: an explicit effort "none" DISABLES thinking.
spec = chat_request_to_genspec(chat_request(reasoning_effort="none"), {})
assert spec.chat_template_kwargs == {"enable_thinking": False}
assert spec.chat_template_kwargs == {"enable_thinking": False, "thinking_mode": "disabled"}
def test_chat_reasoning_effort_routes_through_family_mapping():
"""The toggle goes through model_meta's per-family mapping -- for M3 that is
thinking_mode, not the (inert) enable_thinking key."""
def test_chat_reasoning_effort_broadcasts_every_toggle_spelling():
"""The toggle is broadcast in every spelling templates read (enable_thinking
bool + M3's thinking_mode); each template picks the knob it knows and Jinja
ignores the rest, so no per-family routing exists."""
on = chat_request(reasoning_effort="high")
spec = chat_request_to_genspec(on, {}, reasoning_parser="minimax_m3")
assert spec.chat_template_kwargs == {"thinking_mode": "enabled", "reasoning_effort": "high"}
spec = chat_request_to_genspec(on, {})
assert spec.chat_template_kwargs == {
"enable_thinking": True, "thinking_mode": "enabled", "reasoning_effort": "high"
}
off = chat_request(reasoning_effort="none")
spec = chat_request_to_genspec(off, {}, reasoning_parser="minimax_m3")
assert spec.chat_template_kwargs == {"thinking_mode": "disabled"}
spec = chat_request_to_genspec(off, {})
assert spec.chat_template_kwargs == {"enable_thinking": False, "thinking_mode": "disabled"}
# gpt-oss: the template grades effort and has no off gear
spec = chat_request_to_genspec(on, {}, reasoning_parser="gpt_oss")
assert spec.chat_template_kwargs == {"reasoning_effort": "high"}
spec = chat_request_to_genspec(off, {}, reasoning_parser="gpt_oss")
assert spec.chat_template_kwargs == {}
def test_glm_reasoning_parser_honors_disabled_thinking_with_tools():
@@ -211,11 +213,11 @@ def test_glm_reasoning_parser_honors_disabled_thinking_with_tools():
from freetoken.server.generation import _make_reasoning_parser
state = FakeState([], reasoning_parser="glm")
off = chat_request_to_genspec(chat_request(reasoning_effort="none"), {}, reasoning_parser="glm")
off = chat_request_to_genspec(chat_request(reasoning_effort="none"), {})
parser = _make_reasoning_parser(off, state)
assert parser is not None and parser.detector.force_reasoning is False
on = chat_request_to_genspec(chat_request(), {}, reasoning_parser="glm")
on = chat_request_to_genspec(chat_request(), {})
parser = _make_reasoning_parser(on, state)
assert parser is not None and parser.detector.force_reasoning is True
tests/server/test_responses_api.py
+17 -15
View File
@@ -865,7 +865,9 @@ def test_convert_reasoning_field_enables_thinking():
{"model": "m", "input": "hi", "reasoning": {"effort": "high"}}
)
spec = RP.convert_responses_to_genspec(req, {})
assert spec.chat_template_kwargs == {"enable_thinking": True, "reasoning_effort": "high"}
assert spec.chat_template_kwargs == {
"enable_thinking": True, "thinking_mode": "enabled", "reasoning_effort": "high"
}
# an explicit thinking-related chat_template_kwargs key wins over the mapping
req2 = ResponsesRequest.model_validate(
@@ -881,7 +883,7 @@ def test_convert_reasoning_field_enables_thinking():
"chat_template_kwargs": {"custom_var": 1}}
)
assert RP.convert_responses_to_genspec(req3, {}).chat_template_kwargs == {
"enable_thinking": False, "custom_var": 1,
"enable_thinking": False, "thinking_mode": "disabled", "custom_var": 1,
}
# absent reasoning -> no kwargs
@@ -895,25 +897,25 @@ def test_convert_reasoning_effort_none_disables_thinking():
{"model": "m", "input": "hi", "reasoning": {"effort": "none"}}
)
assert RP.convert_responses_to_genspec(req, {}).chat_template_kwargs == {
"enable_thinking": False
"enable_thinking": False, "thinking_mode": "disabled"
}
def test_convert_reasoning_toggle_routes_through_family_mapping():
"""The toggle goes through model_meta's per-family mapping -- for M3 that is
thinking_mode, not the (inert) enable_thinking key."""
def test_convert_reasoning_toggle_broadcasts_every_spelling():
"""The toggle is broadcast in every spelling templates read; a template
picks the knob it knows (M3: thinking_mode) and ignores the rest, so the
kwargs are family-independent."""
req = ResponsesRequest.model_validate(
{"model": "m", "input": "hi", "reasoning": {"effort": "high"}}
)
spec = RP.convert_responses_to_genspec(req, {}, reasoning_parser="minimax_m3")
assert spec.chat_template_kwargs == {
"thinking_mode": "enabled", "reasoning_effort": "high",
}
on = {"enable_thinking": True, "thinking_mode": "enabled", "reasoning_effort": "high"}
off = ResponsesRequest.model_validate(
{"model": "m", "input": "hi", "reasoning": {"effort": "none"}}
)
spec = RP.convert_responses_to_genspec(off, {}, reasoning_parser="minimax_m3")
assert spec.chat_template_kwargs == {"thinking_mode": "disabled"}
# gpt-oss: the template grades effort and has no off gear
spec = RP.convert_responses_to_genspec(req, {}, reasoning_parser="gpt_oss")
assert spec.chat_template_kwargs == {"reasoning_effort": "high"}
for parser in ("minimax_m3", "gpt_oss", None):
spec = RP.convert_responses_to_genspec(req, {}, reasoning_parser=parser)
assert spec.chat_template_kwargs == on, parser
spec = RP.convert_responses_to_genspec(off, {}, reasoning_parser=parser)
assert spec.chat_template_kwargs == {
"enable_thinking": False, "thinking_mode": "disabled"
}, parser
tests/server/test_think_gears.py
+107
View File
@@ -0,0 +1,107 @@
"""derive_think_gears: the probed replacement for the per-family gear registry.
Each case fakes one model family's template behavior and asserts the derived
gears match (or improve on) what the deleted ``think_spec`` registry hardcoded.
"""
from __future__ import annotations
from freetoken.server.model_meta import derive_think_gears
from freetoken.tokenizer.effort import (
EffortProfile,
probe_effort_profile,
probe_thinking_profile,
)
def profile_for(render):
return probe_thinking_profile(render, probe_effort_profile(render))
def test_qwen3_style_on_off_toggle():
# Old registry row: ("off", "on"), default "on".
def render(kwargs, tools):
return f"qwen3|think={kwargs.get('enable_thinking', True)}"
gears, default, kwargs = derive_think_gears(profile_for(render), parser_configured=True)
assert gears == ("off", "on") and default == "on"
assert kwargs["off"]["enable_thinking"] is False
assert kwargs["on"]["enable_thinking"] is True
def test_qwen38_style_graded_efforts():
# The registry had no row for Qwen3.8's gears -- it showed off/on. Derived:
# the template's real vocabulary, ascending, with the off toggle.
def render(kwargs, tools):
if kwargs.get("enable_thinking") is False:
return "qwen38|off"
effort = kwargs.get("reasoning_effort", "xhigh")
if effort not in ("xhigh", "medium", "low"):
raise ValueError(f"Unexpected reasoning effort {effort}")
return f"qwen38|{effort}"
gears, default, kwargs = derive_think_gears(profile_for(render), parser_configured=True)
assert gears == ("off", "low", "medium", "xhigh") and default == "xhigh"
assert kwargs["medium"]["reasoning_effort"] == "medium"
assert kwargs["medium"]["enable_thinking"] is True
assert kwargs["off"]["enable_thinking"] is False
def test_gemma4_style_default_off():
# Old registry row: ("off", "on"), default "off".
def render(kwargs, tools):
return f"gemma|think={bool(kwargs.get('enable_thinking'))}"
gears, default, _ = derive_think_gears(profile_for(render), parser_configured=True)
assert gears == ("off", "on") and default == "off"
def test_gpt_oss_style_always_on_graded():
# Old registry row: ("low", "medium", "high"), default "medium". The
# template grades effort but never validates it, so the derived vocabulary
# falls back to the OpenAI triple rather than every known name.
def render(kwargs, tools):
return f"harmony|{kwargs.get('reasoning_effort', 'medium')}"
gears, default, kwargs = derive_think_gears(profile_for(render), parser_configured=True)
assert gears == ("low", "medium", "high") and default == "medium"
assert kwargs["high"] == {"reasoning_effort": "high"} # no toggle: always on
def test_minimax_style_always_on_no_knob():
# Old registry row: ("on",), default "on", kwargs {}.
def render(kwargs, tools):
return "minimax prompt"
gears, default, kwargs = derive_think_gears(profile_for(render), parser_configured=True)
assert gears == ("on",) and default == "on"
assert kwargs["on"] == {}
def test_no_reasoning_parser_offers_nothing():
# Old registry: unknown parser -> ((), None) -> reasoning block absent.
def render(kwargs, tools):
return "plain prompt"
assert derive_think_gears(profile_for(render), parser_configured=False) is None
def test_dsv4_style_toggle_plus_efforts():
# Old registry row: ("off", "on", "max"), default "off". Derived: the
# encoder's full vocabulary replaces the curated "on" (its low gear renders
# exactly what "on" did), keeping default off.
def render(kwargs, tools):
thinking = (
bool(tools)
or bool(kwargs.get("enable_thinking"))
or kwargs.get("thinking_mode") == "enabled"
)
if not thinking:
return "dsv4|chat"
effort = kwargs.get("reasoning_effort") or "low"
assert effort in ("low", "high", "max"), f"Invalid reasoning effort: {effort}"
return f"dsv4|think|{effort}"
gears, default, kwargs = derive_think_gears(profile_for(render), parser_configured=True)
assert gears == ("off", "low", "high", "max") and default == "off"
assert kwargs["max"]["reasoning_effort"] == "max"
assert kwargs["max"]["enable_thinking"] is True
tests/tokenizer/test_effort.py
+133
View File
@@ -0,0 +1,133 @@
"""Unit tests for the reasoning-effort dialect layer (tokenizer/effort.py)."""
from __future__ import annotations
from freetoken.tokenizer.effort import (
EFFORT_SCALE,
EffortProfile,
KNOWN_REASONING_EFFORTS,
probe_effort_profile,
quantize_effort,
)
QWEN38 = EffortProfile(
supported=frozenset({"xhigh", "medium", "low"}), default="xhigh", consumes_effort=True
)
DSV4_OFFICIAL = EffortProfile(
supported=frozenset({"low", "high", "max"}), default="low", consumes_effort=True
)
IGNORES = EffortProfile(supported=frozenset(KNOWN_REASONING_EFFORTS), default=None, consumes_effort=False)
def test_in_vocabulary_values_pass_through():
for name in ("xhigh", "medium", "low"):
assert quantize_effort(name, QWEN38) == name
for name in ("low", "high", "max"):
assert quantize_effort(name, DSV4_OFFICIAL) == name
def test_deepseek_dialect_high_lands_on_qwen_xhigh():
# high (0.9) is nearer xhigh (0.99) than medium (0.7)
assert quantize_effort("high", QWEN38) == "xhigh"
def test_openai_dialect_medium_drops_to_the_dsv4_default():
# medium (0.7) is 0.2 from the nearest gear -- beyond the quantize
# threshold, so nothing is sent and the encoder default (low) applies;
# anything else would silently escalate OpenAI-default traffic to the
# encoder's absolute-maximum "high" prompt. Matches vLLM's DSV4 mapping.
assert quantize_effort("medium", DSV4_OFFICIAL) is None
def test_xhigh_lands_on_dsv4_high_never_max():
# "max" is an extreme opt-in gear, reachable only by its own name
# (vLLM's DSV4 rule) -- rounding must not enter it.
assert quantize_effort("xhigh", DSV4_OFFICIAL) == "high"
def test_max_lands_on_qwen_xhigh():
assert quantize_effort("max", QWEN38) == "xhigh"
def test_far_values_drop_instead_of_rounding():
profile = EffortProfile(
supported=frozenset({"low", "xhigh"}), default=None, consumes_effort=True
)
assert quantize_effort("minimal", profile) == "low" # 0.1 away
assert quantize_effort("medium", profile) is None # 0.29 to the nearest gear
def test_unknown_and_non_string_values_drop_to_template_default():
assert quantize_effort("banana", QWEN38) is None
assert quantize_effort(3, QWEN38) is None
assert quantize_effort(None, QWEN38) is None
def test_effort_ignoring_template_sends_nothing():
assert quantize_effort("high", IGNORES) is None
assert quantize_effort("xhigh", IGNORES) is None
# --------------------------------------------------------------------------- #
# Probe tests: fake render callables standing in for real templates/encoders.
# --------------------------------------------------------------------------- #
def _qwen38_render(kwargs, tools):
# Validates unconditionally (enable_thinking undefined counts as on) and
# renders a distinct effort preamble per gear, default xhigh.
effort = kwargs.get("reasoning_effort", "xhigh")
if effort not in ("xhigh", "medium", "low"):
raise ValueError(f"Unexpected reasoning effort {effort}")
preamble = {"xhigh": "think hard", "medium": "", "low": "think briefly"}[effort]
return f"{preamble}|tools={bool(tools)}"
def test_probe_learns_the_qwen38_vocabulary():
profile = probe_effort_profile(_qwen38_render)
assert profile.supported == frozenset({"xhigh", "medium", "low"})
assert profile.default == "xhigh"
assert profile.consumes_effort
assert profile.validates # rejections observed -> the vocabulary is real
def _dsv4_render(kwargs, tools):
# Grades effort only in thinking mode (tools force it); asserts on unknown
# values there; "low" renders the empty preamble, matching the default.
effort = kwargs.get("reasoning_effort") or "low"
if not tools:
return "chat prompt"
assert effort in ("low", "high", "max"), f"Invalid reasoning effort: {effort}"
preamble = {"low": "", "high": "absolute maximum", "max": "beyond maximum"}[effort]
return f"{preamble}|thinking"
def test_probe_learns_the_dsv4_vocabulary_through_the_tools_round():
profile = probe_effort_profile(_dsv4_render)
assert profile.supported == frozenset({"low", "high", "max"})
assert profile.default == "low"
assert profile.consumes_effort
assert profile.validates
def test_probe_marks_an_ignoring_template_as_not_consuming():
profile = probe_effort_profile(lambda kwargs, tools: f"same|tools={bool(tools)}")
assert not profile.consumes_effort
assert not profile.validates
assert profile.default is None
def test_probe_marks_an_interpolating_template_as_not_validating():
# Grades effort (renders differ) but rejects nothing: consumes without a
# trustworthy vocabulary.
profile = probe_effort_profile(lambda kwargs, tools: f"p|{kwargs.get('reasoning_effort')}")
assert profile.consumes_effort
assert not profile.validates
def test_probe_skips_rounds_whose_baseline_fails():
def render(kwargs, tools):
if tools: # template rejects tools outright -- round is uninformative
raise RuntimeError("no tools supported")
return _qwen38_render(kwargs, tools)
profile = probe_effort_profile(render)
assert profile.supported == frozenset({"xhigh", "medium", "low"})
assert profile.consumes_effort
tests/tokenizer/test_resolve_thinking_mode.py
+1 -10
View File
@@ -1,7 +1,7 @@
"""Unit tests for the shared thinking-mode resolver (tokenizer/tokenize.py)."""
from __future__ import annotations
from freetoken.tokenizer.tokenize import normalize_reasoning_effort, resolve_thinking_mode
from freetoken.tokenizer.tokenize import resolve_thinking_mode
def test_default_is_chat():
@@ -26,12 +26,3 @@ def test_explicit_chat_mode():
def test_invalid_mode_falls_back_to_chat():
assert resolve_thinking_mode({"thinking_mode": "bogus"}, None) == "chat"
def test_normalize_reasoning_effort():
assert normalize_reasoning_effort("max") == "max"
assert normalize_reasoning_effort("high") == "high"
# OpenAI's default "medium" (and anything else) is rejected by the dsv4 encoder.
assert normalize_reasoning_effort("medium") is None
assert normalize_reasoning_effort(None) is None
assert normalize_reasoning_effort("low") is None
tests/tokenizer/test_tokenize.py
+101
View File
@@ -179,3 +179,104 @@ def test_dsv4_arguments_str_normalization():
for bad in ("[1,2]", "5", "true", '"x"', "not json", [1, 2], 5):
with pytest.raises(ValueError):
_dsv4_arguments_str(bad)
class Qwen38LikeTokenizer:
"""Fake whose template grades effort like Qwen3.8: validates the vocabulary
whenever thinking is not explicitly off, distinct preamble per gear."""
def __init__(self) -> None:
self.chat_template_kwargs = None
def apply_chat_template(self, messages, **kwargs):
self.chat_template_kwargs = kwargs
if kwargs.get("enable_thinking") is not False:
effort = kwargs.get("reasoning_effort", "xhigh")
if effort not in ("xhigh", "medium", "low"):
raise ValueError(f"Unexpected reasoning effort {effort}")
return f"prompt effort={effort}"
return "prompt effort=off"
def encode(self, prompt, return_tensors=None):
return torch.tensor([[7, 8]], dtype=torch.long)
def test_tokenize_quantizes_foreign_effort_onto_the_template_vocabulary():
"""DeepSeek-dialect "high" must reach a Qwen3.8-style template as its
nearest supported gear, not raw (raw would raise_exception)."""
tokenizer = Qwen38LikeTokenizer()
manager = TokenizeManager(tokenizer)
msg = TokenizeMsg(
uid=1,
text=[{"role": "user", "content": "hello"}],
sampling_params=SamplingParams(),
chat_template_kwargs={"enable_thinking": True, "reasoning_effort": "high"},
)
manager.tokenize([msg])
assert tokenizer.chat_template_kwargs["reasoning_effort"] == "xhigh"
assert tokenizer.chat_template_kwargs["enable_thinking"] is True
# the caller's kwargs stay untouched
assert msg.chat_template_kwargs["reasoning_effort"] == "high"
def test_tokenize_drops_effort_for_templates_that_ignore_it():
tokenizer = FakeTokenizer() # renders the same prompt regardless of kwargs
manager = TokenizeManager(tokenizer)
msg = TokenizeMsg(
uid=1,
text=[{"role": "user", "content": "hello"}],
sampling_params=SamplingParams(),
chat_template_kwargs={"reasoning_effort": "high"},
)
manager.tokenize([msg])
assert "reasoning_effort" not in tokenizer.chat_template_kwargs
def test_tokenize_drops_far_effort_for_the_dsv4_encoder(tmp_path):
"""An OpenAI-dialect "medium" has no nearby dsv4 gear, so nothing is sent
and the encoder default ("low") applies -- never a silent escalation to
the absolute-maximum "high" prompt."""
encoding_dir = tmp_path / "encoding"
encoding_dir.mkdir()
(encoding_dir / "encoding_dsv4.py").write_text(
"""
SEEN = []
def encode_messages(messages, thinking_mode, reasoning_effort=None):
effort = reasoning_effort or "low"
assert effort in ("low", "high", "max"), f"Invalid reasoning effort: {effort}"
SEEN.append(reasoning_effort)
return f"dsv4 prompt effort={effort}"
""".lstrip()
)
tokenizer = FakeDsv4Tokenizer(tmp_path)
manager = TokenizeManager(tokenizer)
msg = TokenizeMsg(
uid=1,
text=[{"role": "user", "content": "hello"}],
sampling_params=SamplingParams(),
chat_template_kwargs={"reasoning_effort": "medium"},
)
manager.tokenize([msg])
assert tokenizer.prompt == "dsv4 prompt effort=low"
def test_tokenize_survives_an_unhashable_effort():
tokenizer = Qwen38LikeTokenizer()
manager = TokenizeManager(tokenizer)
msg = TokenizeMsg(
uid=1,
text=[{"role": "user", "content": "hello"}],
sampling_params=SamplingParams(),
chat_template_kwargs={"reasoning_effort": ["high"]}, # legal JSON on the wire
)
manager.tokenize([msg])
assert "reasoning_effort" not in tokenizer.chat_template_kwargs