From 058d6089b119bcbb82f724955f07ebcc0c51f799 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 15 Apr 2026 18:22:28 -0400 Subject: [PATCH] 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 --- .../services/generation_task.py | 221 ++++++++++-------- 1 file changed, 126 insertions(+), 95 deletions(-) diff --git a/src/fabledassistant/services/generation_task.py b/src/fabledassistant/services/generation_task.py index d9ab21c..dfe50f7 100644 --- a/src/fabledassistant/services/generation_task.py +++ b/src/fabledassistant/services/generation_task.py @@ -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)