bvandeusen/FabledScribe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 50 Wiki Activity

fix(chat): always think on qwen3, drop content-based classifier

Browse Source 
Content-based gating (_should_think) was introduced in 87fcaa6 to cut
TTFT on simple prompts, but it has no way to tell that short prompts
like "create a task titled X" are going to trigger a tool call — and
qwen3:14b's tool-call template is unreliable at think=False, producing
intermittent silent generations where output tokens burn but nothing
parses into content or tool_calls.

Reverting to always-on thinking restores the pre-87fcaa6 reliability
of tool emission at the cost of TTFT latency on short conversational
prompts. This also lets us delete the silent-round retry loop (which
can no longer fire) along with its bookkeeping.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Bryan Van Deusen
2026-04-15 21:09:16 -04:00
parent 1261e93ede
commit fddac2aa2f
3 changed files with 123 additions and 343 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/architecture.md
+1 -1
View File
@@ -313,7 +313,7 @@ See [sso-oauth.md](sso-oauth.md) for provider-specific setup instructions.
### Tool Routing
No separate intent router — the main model handles all tool routing directly via Ollama's structured tool-calling output. The model receives the full tool schema list and decides whether to call a tool or respond conversationally. A thinking-mode heuristic (`_should_think()`) detects complex prompts and enables extended reasoning.
No separate intent router — the main model handles all tool routing directly via Ollama's structured tool-calling output. The model receives the full tool schema list and decides whether to call a tool or respond conversationally. Extended reasoning (`think=True`) is always on for qwen3-class models: content-based gating was tried but exposed tool-call template fragility on short tool-intent prompts.
### Tool Loop
src/fabledassistant/services/generation_task.py
+122 -215
View File
@@ -115,72 +115,6 @@ async def _maybe_save_article_discussion_note(
conv_id, exc_info=True,
)
# ---------------------------------------------------------------------------
# Thinking decision
# ---------------------------------------------------------------------------
#
# `_should_think` is the single source of truth for whether a qwen3-class
# model should engage chain-of-thought for a given request. Frontend callers
# should NOT hardcode think=True — leave it False and let the classifier
# decide from message content. An explicit think_requested=True still acts
# as an override for callers (e.g. a future UI toggle or MCP client) that
# want to force extended reasoning regardless of content.
#
# Why gate it: on qwen3:14b, thinking adds 5–20s of latency before the first
# visible content token, and most conversational messages do not benefit.
# Gating by content keeps quick chats fast while preserving reasoning depth
# for prompts that actually need it.
#
# Models that don't support extended reasoning (e.g. llama3, mistral) simply
# ignore the `think` parameter in the Ollama chat request, so the decision
# here is harmless on non-thinking models.
# Keywords that strongly suggest the user wants reasoning / analysis. Matched
# case-insensitively as whole-ish phrases.
_THINK_KEYWORDS: tuple[str, ...] = (
"why", "how does", "how do i", "how would", "how should",
"explain", "analyze", "analyse", "compare", "contrast",
"design", "architect", "architecture", "plan out", "strategize",
"debug", "diagnose", "troubleshoot", "root cause",
"review", "critique", "evaluate", "trade-off", "tradeoff", "trade off",
"pros and cons", "step by step", "walk me through",
"prove", "derive", "figure out", "work through",
"discuss", # covers briefing /discuss-article + /discuss-topic entry points
)
# Messages shorter than this and without any think-keyword are treated as
# simple/conversational and skip the thinking phase.
_SHORT_MESSAGE_CHARS = 80
# Messages longer than this are treated as substantive regardless of keywords.
_LONG_MESSAGE_CHARS = 400
def _should_think(user_content: str, think_requested: bool) -> bool:
"""Return whether extended thinking should be used for this request.
``think_requested`` acts as an explicit override: if True, thinking is
forced on regardless of content. If False (the default), the decision is
made by inspecting the message: long or keyword-bearing messages get
thinking; short conversational messages skip it.
"""
if think_requested:
return True
text = (user_content or "").strip()
if not text:
return False
if len(text) >= _LONG_MESSAGE_CHARS:
return True
lowered = text.lower()
if any(kw in lowered for kw in _THINK_KEYWORDS):
return True
if len(text) < _SHORT_MESSAGE_CHARS:
return False
# Medium-length message with no obvious reasoning cue: default off.
return False
# Human-readable labels for each tool, shown in the status indicator
_TOOL_LABELS: dict[str, str] = {
"create_note": "Creating note/task",
@@ -368,11 +302,12 @@ async def run_generation(
# Emit context event
buf.append_event("context", {"context": context_meta})
# `_should_think` is authoritative — frontend callers pass think=False by
# default and let this classifier decide based on message content. An
# explicit think=True still forces on as an override.
# Always think on qwen3-class models: reasoning mode is the only reliable
# path for the tool-call template. Content-based gating was tried in 87fcaa6
# but exposed silent-generation failures on short tool-intent prompts, since
# the classifier had no way to tell that "create a task" needs a tool call.
think_requested = think
think = _should_think(user_content, think_requested)
think = True
t_start = time.monotonic()
timing: dict = {
@@ -411,153 +346,125 @@ async def run_generation(
t_stream = time.monotonic()
approx_msg_chars = sum(len(str(m.get("content", ""))) for m in messages)
effective_think = think
retried_with_think = False
attempt = 0
while True:
attempt_content_start = len(buf.content_so_far)
attempt_output_tokens_start = timing.get("output_tokens") or 0
attempt_prompt_tokens_start = timing.get("prompt_tokens") or 0
attempt_tool_calls_start = len(round_tool_calls)
logger.info(
"CTX_DIAG attempt_start conv=%d round=%d attempt=%d num_ctx=%d msgs=%d approx_chars=%d think=%s",
conv_id, _round, attempt, num_ctx, len(messages), approx_msg_chars, effective_think,
)
async for chunk in _stream_with_retry(messages, model, tools, effective_think, num_ctx=num_ctx):
if buf.cancel_event.is_set():
cancelled = True
break
if chunk.type == "thinking":
if timing["first_token_ms"] is None:
timing["first_token_ms"] = int((time.monotonic() - t_start) * 1000)
buf.append_event("thinking_chunk", {"chunk": chunk.content})
elif chunk.type == "content":
if timing["ttft_ms"] is None:
now_ms = int((time.monotonic() - t_start) * 1000)
timing["ttft_ms"] = now_ms
if timing["first_token_ms"] is None:
# No thinking phase occurred — first token IS content.
timing["first_token_ms"] = now_ms
else:
# Thinking phase duration = gap between first thinking
# token and first content token.
timing["thinking_ms"] = now_ms - timing["first_token_ms"]
buf.content_so_far += chunk.content
clean = _TOOL_CALL_MARKER.sub("", chunk.content)
if clean:
buf.append_event("chunk", {"chunk": clean})
now = time.monotonic()
if now - last_flush >= DB_FLUSH_INTERVAL:
try:
await _update_message(msg_id, buf.content_so_far, "generating")
except Exception:
logger.warning("Failed periodic flush for message %d", msg_id, exc_info=True)
last_flush = now
elif chunk.type == "done":
if chunk.prompt_tokens is not None:
timing["prompt_tokens"] = (timing["prompt_tokens"] or 0) + chunk.prompt_tokens
if chunk.output_tokens is not None:
timing["output_tokens"] = (timing["output_tokens"] or 0) + chunk.output_tokens
elif chunk.type == "tool_calls" and chunk.tool_calls:
logger.info("Round %d: model returned %d tool call(s)", _round, len(chunk.tool_calls))
for tc in chunk.tool_calls:
fn = tc.get("function", {})
tool_name = fn.get("name", "")
arguments = fn.get("arguments", {})
logger.info("Executing tool: %s(%s)", tool_name, json.dumps(arguments)[:200])
buf.append_event("status", {"status": f"{_TOOL_LABELS.get(tool_name, 'Working')}..."})
t_tool = time.monotonic()
if tool_name == "research_topic":
topic = arguments.get("topic", "")
try:
note = await run_research_pipeline(topic, user_id, model, buf, project_id=workspace_project_id)
result = {
"success": True,
"type": "research_note",
"data": {"id": note.id, "title": note.title},
}
done_text = (
f"\n\n---\n\nResearch complete! I've compiled a note: "
f"**[{note.title}](/notes/{note.id})**."
)
buf.append_event("chunk", {"chunk": done_text})
buf.content_so_far += done_text
except Exception as e:
logger.exception("Research pipeline failed for topic: %s", topic)
err_msg = str(e) or f"{type(e).__name__}: unexpected error"
result = {"success": False, "error": err_msg}
err_text = f"\nResearch failed: {err_msg}"
buf.append_event("chunk", {"chunk": err_text})
buf.content_so_far += err_text
research_completed = True
else:
result = await execute_tool(
user_id, tool_name, arguments,
conv_id=conv_id,
workspace_project_id=workspace_project_id,
)
# Capture RAG scope change for SSE done event
if result.get("type") == "rag_scope_set" and result.get("success"):
new_rag_scope = arguments.get("project_id")
new_rag_scope_label = result.get("scope_label")
timing["tools"].append({"name": tool_name, "ms": int((time.monotonic() - t_tool) * 1000)})
logger.info("Tool %s result: success=%s", tool_name, result.get("success"))
tool_record = {
"function": tool_name,
"arguments": arguments,
"result": result,
"status": "success" if result.get("success") else "error",
}
round_tool_calls.append(tool_record)
all_tool_calls.append(tool_record)
buf.append_event("tool_call", {"tool_call": tool_record})
if cancelled:
round_content_start = len(buf.content_so_far)
round_output_tokens_start = timing.get("output_tokens") or 0
round_prompt_tokens_start = timing.get("prompt_tokens") or 0
logger.info(
"CTX_DIAG round_start conv=%d round=%d num_ctx=%d msgs=%d approx_chars=%d think=%s",
conv_id, _round, num_ctx, len(messages), approx_msg_chars, think,
)
async for chunk in _stream_with_retry(messages, model, tools, think, num_ctx=num_ctx):
if buf.cancel_event.is_set():
cancelled = True
break
attempt_content_added = len(buf.content_so_far) - attempt_content_start
attempt_tokens_added = (timing.get("output_tokens") or 0) - attempt_output_tokens_start
attempt_prompt_tokens = (timing.get("prompt_tokens") or 0) - attempt_prompt_tokens_start
attempt_tool_calls_added = len(round_tool_calls) - attempt_tool_calls_start
headroom = num_ctx - attempt_prompt_tokens if attempt_prompt_tokens else None
is_silent = (
attempt_tool_calls_added == 0
and attempt_content_added == 0
and attempt_tokens_added > 0
)
logger.info(
"CTX_DIAG attempt_end conv=%d round=%d attempt=%d think=%s prompt_tokens=%d output_tokens=%d headroom=%s content_added=%d tool_calls=%d silent=%s",
conv_id, _round, attempt, effective_think, attempt_prompt_tokens, attempt_tokens_added,
headroom, attempt_content_added, attempt_tool_calls_added, is_silent,
)
if (
is_silent
and not effective_think
and not retried_with_think
):
# Silent round: qwen3's tool-call tokens sometimes aren't
# parsed into content or tool_calls. Retry once with
# think=True — reasoning mode produces more reliable output.
logger.warning(
"Silent round %d for conv %d (tokens=%d); retrying with think=True",
_round, conv_id, attempt_tokens_added,
)
buf.append_event("status", {"status": "Retrying with reasoning enabled..."})
effective_think = True
retried_with_think = True
attempt += 1
continue
break
if chunk.type == "thinking":
if timing["first_token_ms"] is None:
timing["first_token_ms"] = int((time.monotonic() - t_start) * 1000)
buf.append_event("thinking_chunk", {"chunk": chunk.content})
elif chunk.type == "content":
if timing["ttft_ms"] is None:
now_ms = int((time.monotonic() - t_start) * 1000)
timing["ttft_ms"] = now_ms
if timing["first_token_ms"] is None:
# No thinking phase occurred — first token IS content.
timing["first_token_ms"] = now_ms
else:
# Thinking phase duration = gap between first thinking
# token and first content token.
timing["thinking_ms"] = now_ms - timing["first_token_ms"]
buf.content_so_far += chunk.content
clean = _TOOL_CALL_MARKER.sub("", chunk.content)
if clean:
buf.append_event("chunk", {"chunk": clean})
now = time.monotonic()
if now - last_flush >= DB_FLUSH_INTERVAL:
try:
await _update_message(msg_id, buf.content_so_far, "generating")
except Exception:
logger.warning("Failed periodic flush for message %d", msg_id, exc_info=True)
last_flush = now
elif chunk.type == "done":
if chunk.prompt_tokens is not None:
timing["prompt_tokens"] = (timing["prompt_tokens"] or 0) + chunk.prompt_tokens
if chunk.output_tokens is not None:
timing["output_tokens"] = (timing["output_tokens"] or 0) + chunk.output_tokens
elif chunk.type == "tool_calls" and chunk.tool_calls:
logger.info("Round %d: model returned %d tool call(s)", _round, len(chunk.tool_calls))
for tc in chunk.tool_calls:
fn = tc.get("function", {})
tool_name = fn.get("name", "")
arguments = fn.get("arguments", {})
logger.info("Executing tool: %s(%s)", tool_name, json.dumps(arguments)[:200])
buf.append_event("status", {"status": f"{_TOOL_LABELS.get(tool_name, 'Working')}..."})
t_tool = time.monotonic()
if tool_name == "research_topic":
topic = arguments.get("topic", "")
try:
note = await run_research_pipeline(topic, user_id, model, buf, project_id=workspace_project_id)
result = {
"success": True,
"type": "research_note",
"data": {"id": note.id, "title": note.title},
}
done_text = (
f"\n\n---\n\nResearch complete! I've compiled a note: "
f"**[{note.title}](/notes/{note.id})**."
)
buf.append_event("chunk", {"chunk": done_text})
buf.content_so_far += done_text
except Exception as e:
logger.exception("Research pipeline failed for topic: %s", topic)
err_msg = str(e) or f"{type(e).__name__}: unexpected error"
result = {"success": False, "error": err_msg}
err_text = f"\nResearch failed: {err_msg}"
buf.append_event("chunk", {"chunk": err_text})
buf.content_so_far += err_text
research_completed = True
else:
result = await execute_tool(
user_id, tool_name, arguments,
conv_id=conv_id,
workspace_project_id=workspace_project_id,
)
# Capture RAG scope change for SSE done event
if result.get("type") == "rag_scope_set" and result.get("success"):
new_rag_scope = arguments.get("project_id")
new_rag_scope_label = result.get("scope_label")
timing["tools"].append({"name": tool_name, "ms": int((time.monotonic() - t_tool) * 1000)})
logger.info("Tool %s result: success=%s", tool_name, result.get("success"))
tool_record = {
"function": tool_name,
"arguments": arguments,
"result": result,
"status": "success" if result.get("success") else "error",
}
round_tool_calls.append(tool_record)
all_tool_calls.append(tool_record)
buf.append_event("tool_call", {"tool_call": tool_record})
round_content_added = len(buf.content_so_far) - round_content_start
round_output_tokens_added = (timing.get("output_tokens") or 0) - round_output_tokens_start
round_prompt_tokens = (timing.get("prompt_tokens") or 0) - round_prompt_tokens_start
headroom = num_ctx - round_prompt_tokens if round_prompt_tokens else None
is_silent = (
not round_tool_calls
and round_content_added == 0
and round_output_tokens_added > 0
)
logger.info(
"CTX_DIAG round_end conv=%d round=%d think=%s prompt_tokens=%d output_tokens=%d headroom=%s content_added=%d tool_calls=%d silent=%s",
conv_id, _round, think, round_prompt_tokens, round_output_tokens_added,
headroom, round_content_added, len(round_tool_calls), is_silent,
)
timing["generation_ms"] = int((time.monotonic() - t_stream) * 1000)
tests/test_should_think.py
-127
View File
@@ -1,127 +0,0 @@
"""Tests for the `_should_think` classifier.
`_should_think` decides whether qwen3-class models should engage chain-of-
thought for a given chat turn. It is the single source of truth: frontend
callers pass `think_requested=False` by default and defer to this function,
while explicit `think_requested=True` acts as an override for curated
analytical entry points.
These tests lock in the content-based behavior so future tweaks don't
silently regress the short / long / keyword boundaries.
"""
import pytest
from fabledassistant.services.generation_task import (
_LONG_MESSAGE_CHARS,
_SHORT_MESSAGE_CHARS,
_should_think,
)
class TestExplicitOverride:
def test_override_forces_on_for_empty(self):
assert _should_think("", think_requested=True) is True
def test_override_forces_on_for_short_greeting(self):
assert _should_think("hi", think_requested=True) is True
def test_override_forces_on_for_medium_no_keyword(self):
text = "just checking in on the status of things for the week"
assert _should_think(text, think_requested=True) is True
class TestEmptyAndWhitespace:
def test_empty_string_off(self):
assert _should_think("", think_requested=False) is False
def test_none_content_off(self):
# _should_think defensively handles None content from upstream callers
assert _should_think(None, think_requested=False) is False # type: ignore[arg-type]
def test_whitespace_only_off(self):
assert _should_think(" \n\t ", think_requested=False) is False
class TestShortMessages:
def test_short_greeting_off(self):
assert _should_think("hi", think_requested=False) is False
def test_short_thanks_off(self):
assert _should_think("thanks!", think_requested=False) is False
def test_short_acknowledgement_off(self):
assert _should_think("ok sounds good", think_requested=False) is False
def test_just_below_short_threshold_off(self):
text = "a" * (_SHORT_MESSAGE_CHARS - 1)
assert _should_think(text, think_requested=False) is False
class TestLongMessages:
def test_at_long_threshold_on(self):
text = "a" * _LONG_MESSAGE_CHARS
assert _should_think(text, think_requested=False) is True
def test_well_above_long_threshold_on(self):
text = "x" * (_LONG_MESSAGE_CHARS * 3)
assert _should_think(text, think_requested=False) is True
class TestMediumMessages:
def test_medium_no_keyword_off(self):
# Between the short and long thresholds with no reasoning cue.
text = "a" * ((_SHORT_MESSAGE_CHARS + _LONG_MESSAGE_CHARS) // 2)
assert _should_think(text, think_requested=False) is False
class TestKeywordTriggers:
@pytest.mark.parametrize(
"text",
[
"why is this failing",
"how does caching work here",
"how do i configure this",
"explain the retry logic",
"analyze the latency breakdown",
"compare gemma3 vs qwen3 for tool use",
"please design the schema for X",
"debug this error",
"troubleshoot the connection issue",
"root cause the outage",
"review this PR",
"critique my approach",
"walk me through the flow",
"step by step instructions please",
"pros and cons of each option",
"help me figure out what's wrong",
"discuss this article", # covers briefing /discuss entry points
],
)
def test_keyword_forces_on(self, text):
assert _should_think(text, think_requested=False) is True
def test_keyword_case_insensitive(self):
assert _should_think("WHY does this break?", think_requested=False) is True
def test_keyword_in_longer_sentence(self):
text = "hey quick one — can you explain what caching does for qwen3"
assert _should_think(text, think_requested=False) is True
class TestNonTriggers:
"""Messages that look chatty and should NOT trigger thinking."""
@pytest.mark.parametrize(
"text",
[
"hey",
"yep",
"no worries",
"got it, thanks",
"good morning",
"remind me later", # no reasoning keyword, short
],
)
def test_chatty_messages_off(self, text):
assert _should_think(text, think_requested=False) is False