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 <noreply@anthropic.com>
This commit is contained in:
@@ -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")
|
||||
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user