From 53e54ea761941cdacc100b4522d4c0445dce2629 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 2 Mar 2026 18:30:21 -0500 Subject: [PATCH] Remove intent router from chat pipeline; raise OLLAMA_NUM_CTX to 16384 The intent classifier (Phase 21) is removed from the main chat generation path. The main model now handles all tool routing natively via Ollama's structured tool-calling API, eliminating misidentification issues caused by the small intent model. Changes: - generation_task.py: remove classify_intent call, intent_task, _WRITE_TOOLS, _TOOL_ACTIONS, _INTENT_TRIGGER_WORDS, _should_skip_intent(), and the entire round-0 intent-first + write-tool confirmation block (~315 lines removed) - research_topic tool calls are now handled inline in the streaming loop: runs run_research_pipeline, streams synthesis to buf, then breaks the round loop (research is still the full response, no model follow-up) - config.py: raise OLLAMA_NUM_CTX default from 8192 to 16384 The quick-capture dedicated classifier (classify_capture_intent) is unchanged. Co-Authored-By: Claude Sonnet 4.6 --- src/fabledassistant/config.py | 3 +- .../services/generation_task.py | 350 ++---------------- 2 files changed, 39 insertions(+), 314 deletions(-) diff --git a/src/fabledassistant/config.py b/src/fabledassistant/config.py index 573cbfd..67d9c68 100644 --- a/src/fabledassistant/config.py +++ b/src/fabledassistant/config.py @@ -28,8 +28,7 @@ class Config: # Falls back to OLLAMA_MODEL if not set. Can also be overridden per-user via settings. OLLAMA_INTENT_MODEL: str = os.environ.get("OLLAMA_INTENT_MODEL", "qwen2.5:7b") # KV cache context window for generation. Lower = less VRAM, less throughput impact. - # 8192 is sufficient for most conversations; raise if you paste large documents. - OLLAMA_NUM_CTX: int = int(os.environ.get("OLLAMA_NUM_CTX", "8192")) + OLLAMA_NUM_CTX: int = int(os.environ.get("OLLAMA_NUM_CTX", "16384")) SECRET_KEY: str = _read_secret("SECRET_KEY", "SECRET_KEY_FILE", "dev-secret-change-me") SECURE_COOKIES: bool = os.environ.get("SECURE_COOKIES", "").lower() in ("1", "true", "yes") LOG_LEVEL: str = os.environ.get("LOG_LEVEL", "INFO") diff --git a/src/fabledassistant/services/generation_task.py b/src/fabledassistant/services/generation_task.py index e30a136..16c730c 100644 --- a/src/fabledassistant/services/generation_task.py +++ b/src/fabledassistant/services/generation_task.py @@ -24,7 +24,6 @@ from fabledassistant.services.generation_buffer import ( ) from fabledassistant.services.llm import ChatChunk, build_context, generate_completion, stream_chat, stream_chat_with_tools, summarize_history_for_context, wait_for_model_loaded from fabledassistant.services.chat import update_conversation_title -from fabledassistant.services.intent import IntentResult, classify_intent from fabledassistant.services.logging import log_generation from fabledassistant.services.settings import get_setting from fabledassistant.services.tools import get_tools_for_user, execute_tool @@ -58,67 +57,6 @@ _TOOL_LABELS: dict[str, str] = { "research_topic": "Researching topic", } -# Tools that write data and require explicit user confirmation before executing. -_WRITE_TOOLS: frozenset[str] = frozenset({ - "create_task", "create_note", "update_note", "delete_note", "delete_task", - "create_event", "update_event", "delete_event", -}) - -# Action phrases used in the acknowledgment prompt — "You are about to: {action}" -_TOOL_ACTIONS: dict[str, str] = { - "create_task": "create a task", - "create_note": "create a new note", - "update_note": "update an existing note", - "delete_note": "permanently delete a note", - "delete_task": "permanently delete a task", - "get_note": "read a note", - "list_notes": "list notes", - "list_tasks": "look up tasks", - "search_notes": "search through notes", - "create_event": "schedule a calendar event", - "list_events": "check the calendar", - "search_events": "search calendar events", - "update_event": "update a calendar event", - "delete_event": "remove a calendar event", - "list_calendars": "list available calendars", -} - - -# Words that strongly suggest a tool call is needed. -# If none of these appear in a short message, skip intent classification. -_INTENT_TRIGGER_WORDS: frozenset[str] = frozenset({ - # Creation - "create", "add", "make", "new", "write", "set", - # Objects / tools - "note", "notes", "task", "tasks", "event", "calendar", "reminder", "todo", - "meeting", "appointment", "schedule", "due", "deadline", - # Read / search - "find", "search", "look", "show", "list", "get", "read", "open", "fetch", - # Research / web - "research", "investigate", "compile", "report", "google", "web", - # Mutation - "update", "edit", "change", "rename", "move", "reschedule", "delete", - "remove", "cancel", "complete", "finish", "mark", "tag", "untag", "append", - # Dates / times (might trigger calendar tools) - "today", "tomorrow", "yesterday", "next", "last", "week", "month", - "monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday", - # Misc triggers - "overdue", "priority", "high", "urgent", "remind", "alert", -}) - - -def _should_skip_intent(message: str) -> bool: - """Return True if the message is clearly conversational and needs no tool. - - Skips intent classification for short messages (≤ 10 words) that contain - none of the trigger words. This saves a model call (~400-800ms) for simple - exchanges like "thanks", "okay", "can you explain that more?", etc. - """ - words = message.lower().split() - if len(words) > 10: - return False - return not any(w in _INTENT_TRIGGER_WORDS for w in words) - async def _generate_title(messages: list[dict], model: str) -> str: """Ask the LLM for a concise conversation title.""" @@ -225,19 +163,9 @@ async def run_generation( ) intent_model = intent_model_setting or Config.OLLAMA_INTENT_MODEL or model - # research_topic is handled exclusively by the intent-first pipeline (round 0). - # Remove it from the tool list given to the main model so it can never be called - # directly via the streaming tool loop (which would just return a placeholder result - # and cause the model to retry in a loop). - _INTENT_ONLY_TOOLS = frozenset({"research_topic"}) - stream_tools = [ - t for t in tools - if t.get("function", {}).get("name") not in _INTENT_ONLY_TOOLS - ] - logger.info( - "Starting generation for conv %d: model=%s, intent_model=%s, tools=%d (stream_tools=%d)", - conv_id, model, intent_model, len(tools), len(stream_tools), + "Starting generation for conv %d: model=%s, intent_model=%s, tools=%d", + conv_id, model, intent_model, len(tools), ) # Phase 2: Summarize long conversation history if needed. @@ -247,10 +175,7 @@ async def run_generation( buf.append_event("status", {"status": "Summarizing conversation history..."}) history_to_use, history_summary = await summarize_history_for_context(history, intent_model) - # Phase 3: Build context, classify intent, and wait for model — all in parallel. - # build_context is fast DB/search ops that don't need the main model. - # classify_intent uses the small intent model, not the main model. - # wait_for_model_loaded polls /api/ps so the main stream starts without 500 errors. + # Phase 3: Build context and wait for model in parallel. model_load_task = asyncio.create_task(wait_for_model_loaded(model, timeout=90.0)) context_task = asyncio.create_task(build_context( @@ -259,19 +184,6 @@ async def run_generation( include_note_ids=include_note_ids, )) - intent_task: asyncio.Task[IntentResult] | None = None - t_intent = time.monotonic() - if tools and not _should_skip_intent(user_content): - intent_history = [ - m for m in history_to_use - if m.get("role") in ("user", "assistant") and m.get("content") - ][-6:] - intent_task = asyncio.create_task( - classify_intent(user_content, tools, intent_model, history=intent_history) - ) - elif tools: - logger.debug("Skipping intent classification for short/conversational message") - messages, context_meta = await context_task # Emit context event @@ -287,7 +199,6 @@ async def run_generation( t_start = time.monotonic() timing: dict = { - "intent_ms": None, "tools": [], "ttft_ms": None, "generation_ms": None, @@ -299,221 +210,19 @@ async def run_generation( try: cancelled = False + research_completed = False for _round in range(MAX_TOOL_ROUNDS + 1): round_tool_calls: list[dict] = [] logger.info("Generation round %d started for conv %d (model=%s)", _round, conv_id, model) - # --- Round 0 with tools: intent-first pipeline --- - # Wait for intent result, then act immediately. The ack sentence - # (embedded in the intent JSON) is streamed at TTFT (~400ms), then - # the tool runs while the user is reading it. - if _round == 0 and tools and intent_task is not None: - intent = await intent_task - timing["intent_ms"] = int((time.monotonic() - t_intent) * 1000) - - if intent.should_execute: - tool_name = intent.tool_name - - # Stream ack immediately — this becomes TTFT - ack_text = (intent.ack or "").strip() - if ack_text: - ack_with_newline = ack_text + "\n\n" - buf.append_event("chunk", {"chunk": ack_with_newline}) - buf.content_so_far += ack_with_newline - if timing["ttft_ms"] is None: - timing["ttft_ms"] = int((time.monotonic() - t_start) * 1000) - - if tool_name == "research_topic": - topic = intent.arguments.get("topic", "") - if not ack_text: - fallback_ack = f"I'll research '{topic}' and compile a note.\n\n" - buf.append_event("chunk", {"chunk": fallback_ack}) - buf.content_so_far += fallback_ack - if timing["ttft_ms"] is None: - timing["ttft_ms"] = int((time.monotonic() - t_start) * 1000) - try: - note = await run_research_pipeline( - topic, user_id, model, intent_model, buf - ) - 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 - tool_record = { - "function": "research_topic", - "arguments": {"topic": topic}, - "result": { - "success": True, - "type": "research_note", - "data": {"id": note.id, "title": note.title}, - }, - "status": "success", - } - all_tool_calls.append(tool_record) - buf.append_event("tool_call", {"tool_call": tool_record}) - except Exception as e: - logger.exception("Research pipeline failed for topic: %s", topic) - err_text = f"\nResearch failed: {e}" - buf.append_event("chunk", {"chunk": err_text}) - buf.content_so_far += err_text - break # research IS the full response - - confirmed = True - if tool_name in _WRITE_TOOLS: - loop = asyncio.get_running_loop() - confirm_future: asyncio.Future = loop.create_future() - buf.confirmation_future = confirm_future - buf.pending_tool = { - "function": tool_name, - "arguments": intent.arguments, - "label": _TOOL_LABELS.get(tool_name, "Action"), - } - buf.append_event("status", {"status": "Waiting for confirmation..."}) - buf.append_event("tool_pending", {"tool_pending": buf.pending_tool}) - - cancel_task = asyncio.create_task(buf.cancel_event.wait()) - confirmed = False - try: - done_set, _ = await asyncio.wait( - {confirm_future, cancel_task}, - timeout=120.0, - return_when=asyncio.FIRST_COMPLETED, - ) - if cancel_task in done_set: - cancelled = True - elif confirm_future in done_set: - try: - confirmed = bool(confirm_future.result()) - except Exception: - confirmed = False - except Exception: - confirmed = False - finally: - cancel_task.cancel() - buf.confirmation_future = None - buf.pending_tool = None - - if not confirmed: - if not cancelled: - declined_record = { - "function": tool_name, - "arguments": intent.arguments, - "result": {"success": False, "error": "Declined"}, - "status": "declined", - } - all_tool_calls.append(declined_record) - buf.append_event("tool_call", {"tool_call": declined_record}) - - if confirmed: - buf.append_event("status", {"status": f"{_TOOL_LABELS.get(tool_name, 'Working')}..."}) - t_tool = time.monotonic() - result = await execute_tool(user_id, tool_name, intent.arguments) - timing["tools"].append({"name": tool_name, "ms": int((time.monotonic() - t_tool) * 1000)}) - logger.info("Intent-routed tool %s result: success=%s", tool_name, result.get("success")) - - tool_record = { - "function": tool_name, - "arguments": intent.arguments, - "result": result, - "status": "success" if result.get("success") else "error", - } - all_tool_calls.append(tool_record) - buf.append_event("tool_call", {"tool_call": tool_record}) - - messages.append({ - "role": "assistant", - "content": buf.content_so_far, - "tool_calls": [ - {"function": {"name": tool_name, "arguments": intent.arguments}} - ], - }) - messages.append({"role": "tool", "content": json.dumps(result)}) - continue # Round 1: stream response with tool result - - # Declined write tool — fall through to fresh stream. - if cancelled: - break - - # No tool (or declined write tool): stream directly, no queue. - buf.append_event("status", {"status": "Generating response..."}) - t_stream = time.monotonic() - - async for chunk in _stream_with_retry(messages, model, stream_tools, think): - if buf.cancel_event.is_set(): - cancelled = True - break - - if chunk.type == "content": - if timing["ttft_ms"] is None: - timing["ttft_ms"] = int((time.monotonic() - t_start) * 1000) - 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 == "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() - result = await execute_tool(user_id, tool_name, arguments) - 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}) - - timing["generation_ms"] = int((time.monotonic() - t_stream) * 1000) - - if cancelled: - break - if not round_tool_calls: - break - - buf.content_so_far = _TOOL_CALL_MARKER.sub("", buf.content_so_far) - messages.append({ - "role": "assistant", - "content": buf.content_so_far, - "tool_calls": [ - {"function": {"name": tc["function"], "arguments": tc["arguments"]}} - for tc in round_tool_calls - ], - }) - for tc in round_tool_calls: - messages.append({"role": "tool", "content": json.dumps(tc["result"])}) - buf.content_so_far = "" - continue - - # --- Rounds 1+ (and round 0 with no tools) --- if cancelled: break 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, stream_tools, think): + + async for chunk in _stream_with_retry(messages, model, tools, think): if buf.cancel_event.is_set(): cancelled = True break @@ -525,12 +234,10 @@ async def run_generation( if timing["ttft_ms"] is None: timing["ttft_ms"] = int((time.monotonic() - t_start) * 1000) buf.content_so_far += chunk.content - # Filter out "[TOOL_CALLS]" marker from streaming output clean = _TOOL_CALL_MARKER.sub("", chunk.content) if clean: buf.append_event("chunk", {"chunk": clean}) - # Periodic DB flush now = time.monotonic() if now - last_flush >= DB_FLUSH_INTERVAL: try: @@ -549,7 +256,31 @@ async def run_generation( buf.append_event("status", {"status": f"{_TOOL_LABELS.get(tool_name, 'Working')}..."}) t_tool = time.monotonic() - result = await execute_tool(user_id, tool_name, arguments) + if tool_name == "research_topic": + topic = arguments.get("topic", "") + try: + note = await run_research_pipeline(topic, user_id, model, intent_model, buf) + 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) + result = {"success": False, "error": str(e)} + err_text = f"\nResearch failed: {e}" + 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) + 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")) @@ -569,18 +300,18 @@ async def run_generation( logger.info("Generation cancelled for conv %d", conv_id) break - # If no tool calls this round, the LLM gave its final text response + if research_completed: + logger.info("Research complete for conv %d, ending generation", conv_id) + break + if not round_tool_calls: logger.info("Round %d: no tool calls, final content length=%d", _round, len(buf.content_so_far)) break logger.info("Round %d: %d tool call(s) executed, starting next round", _round, len(round_tool_calls)) - # Strip model artifacts like "[TOOL_CALLS]" from content buf.content_so_far = _TOOL_CALL_MARKER.sub("", buf.content_so_far) - # Append assistant tool_call message and tool results to conversation - # for the next round messages.append({ "role": "assistant", "content": buf.content_so_far, @@ -589,14 +320,9 @@ async def run_generation( for tc in round_tool_calls ], }) - for tc in round_tool_calls: - messages.append({ - "role": "tool", - "content": json.dumps(tc["result"]), - }) + messages.append({"role": "tool", "content": json.dumps(tc["result"])}) - # Reset content for the next round (LLM will produce a new response) buf.content_so_far = "" # Strip model artifacts from final content @@ -614,8 +340,8 @@ async def run_generation( timing["total_ms"] = int((time.monotonic() - t_start) * 1000) logger.info( - "Generation timing for conv %d: total=%dms ttft=%s intent=%s tools=%s generation=%s", - conv_id, timing["total_ms"], timing["ttft_ms"], timing["intent_ms"], + "Generation timing for conv %d: total=%dms ttft=%s tools=%s generation=%s", + conv_id, timing["total_ms"], timing["ttft_ms"], [(t["name"], t["ms"]) for t in timing["tools"]], timing["generation_ms"], ) try: