fix(chat): retry silent rounds with think=True before falling back
Qwen3's tool-call tokens sometimes fail to parse into either content or tool_calls, burning output tokens and producing empty bubbles. Detect the signature within a round (empty content, no tool calls, eval_count > 0) and re-run the same round once with reasoning mode enabled, which emits more reliable output. The post-loop fallback remains as the final catch. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -410,103 +410,134 @@ async def run_generation(
|
||||
buf.append_event("status", {"status": "Generating response..." if _round == 0 else "Composing response..."})
|
||||
t_stream = time.monotonic()
|
||||
|
||||
async for chunk in _stream_with_retry(messages, model, tools, think, num_ctx=num_ctx):
|
||||
if buf.cancel_event.is_set():
|
||||
cancelled = True
|
||||
round_content_start = len(buf.content_so_far)
|
||||
round_tokens_start = timing.get("output_tokens") or 0
|
||||
effective_think = think
|
||||
retried_with_think = False
|
||||
|
||||
while True:
|
||||
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:
|
||||
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_tokens_added = (timing.get("output_tokens") or 0) - round_tokens_start
|
||||
if (
|
||||
not round_tool_calls
|
||||
and round_content_added == 0
|
||||
and round_tokens_added > 0
|
||||
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, round_tokens_added,
|
||||
)
|
||||
buf.append_event("status", {"status": "Retrying with reasoning enabled..."})
|
||||
effective_think = True
|
||||
retried_with_think = True
|
||||
continue
|
||||
break
|
||||
|
||||
timing["generation_ms"] = int((time.monotonic() - t_stream) * 1000)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user