From df4c52412d8089e2c6590e414dad2dd425ac53a0 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 27 Feb 2026 18:24:15 -0500 Subject: [PATCH] Phase 22b: Parallel research fetching, streaming synthesis, intent optimizations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit research.py: - Parallelize all 5 SearXNG queries concurrently (200ms stagger via asyncio.gather) - Parallelize all URL fetches in parallel (asyncio.gather) — up to 15 URLs at once instead of sequential fetches; biggest performance win (was O(n) × 15s, now ~15s flat) - _synthesize_note accepts buf: when provided uses stream_chat (num_ctx=16384, num_predict=8192) to emit tokens into the chat buffer in real time so users see the note being written; falls back to generate_completion when buf=None - Added \n\n---\n\n separator before "Research complete!" to cleanly mark boundary after streamed synthesis content intent.py: - classify_intent passes num_ctx=4096 to generate_completion — reduces VRAM pressure and prefill time for the intent model call on every single request generation_task.py: - _INTENT_TRIGGER_WORDS frozenset (~50 action/object/date words) + _should_skip_intent() skips intent classification for short messages (≤10 words) with no trigger words; saves 400-800ms model call for conversational replies ("thanks", "okay", etc.) - Added \n\n---\n\n separator before research "done" text in research_topic branch Co-Authored-By: Claude Sonnet 4.6 --- .../services/generation_task.py | 42 ++++++++- src/fabledassistant/services/intent.py | 2 +- src/fabledassistant/services/research.py | 86 ++++++++++++------- summary.md | 43 ++++++++-- 4 files changed, 133 insertions(+), 40 deletions(-) diff --git a/src/fabledassistant/services/generation_task.py b/src/fabledassistant/services/generation_task.py index b688541..7a6fca0 100644 --- a/src/fabledassistant/services/generation_task.py +++ b/src/fabledassistant/services/generation_task.py @@ -97,6 +97,42 @@ _TOOL_ACTIONS: dict[str, str] = { } +# 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.""" # Build conversation text like summarize_conversation_as_note @@ -227,7 +263,7 @@ async def run_generation( intent_task: asyncio.Task[IntentResult] | None = None t_intent = time.monotonic() - if tools: + 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") @@ -235,6 +271,8 @@ async def run_generation( 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 @@ -301,7 +339,7 @@ async def run_generation( topic, user_id, model, intent_model, buf ) done_text = ( - f"Research complete! I've compiled a note: " + f"\n\n---\n\nResearch complete! I've compiled a note: " f"**[{note.title}](/notes/{note.id})**." ) buf.append_event("chunk", {"chunk": done_text}) diff --git a/src/fabledassistant/services/intent.py b/src/fabledassistant/services/intent.py index f7f2484..0473bf6 100644 --- a/src/fabledassistant/services/intent.py +++ b/src/fabledassistant/services/intent.py @@ -150,7 +150,7 @@ async def classify_intent( messages.append({"role": "user", "content": user_message}) try: - raw = await generate_completion(messages, model, max_tokens=350) + raw = await generate_completion(messages, model, max_tokens=350, num_ctx=4096) except Exception: logger.warning("Intent classification LLM call failed", exc_info=True) return IntentResult() diff --git a/src/fabledassistant/services/research.py b/src/fabledassistant/services/research.py index 16e8e83..dfbfa3b 100644 --- a/src/fabledassistant/services/research.py +++ b/src/fabledassistant/services/research.py @@ -8,7 +8,7 @@ import re import httpx from fabledassistant.config import Config -from fabledassistant.services.llm import fetch_url_content, generate_completion +from fabledassistant.services.llm import fetch_url_content, generate_completion, stream_chat from fabledassistant.services.notes import create_note from fabledassistant.models.note import Note @@ -38,32 +38,45 @@ async def run_research_pipeline( queries = await _generate_sub_queries(topic, intent_model) logger.info("Research: generated %d sub-queries for topic '%s'", len(queries), topic) - # Step 2: Search and fetch - all_sources: list[dict] = [] - seen_urls: set[str] = set() - - for i, query in enumerate(queries): + # Step 2: Search all queries in parallel (200 ms stagger to avoid hammering SearXNG) + async def _search_with_stagger(i: int, query: str) -> tuple[str, list[dict]]: if i > 0: - await asyncio.sleep(1.0) # avoid hammering SearXNG + await asyncio.sleep(0.2 * i) buf.append_event("status", {"status": f"Searching: {query}..."}) results = await _search_searxng(query) logger.info("Research: query '%s' → %d results", query, len(results)) + return query, results + search_results = await asyncio.gather( + *[_search_with_stagger(i, q) for i, q in enumerate(queries)] + ) + + # Deduplicate URLs across all queries + seen_urls: set[str] = set() + url_tasks: list[tuple[str, dict, str]] = [] # (url, result_dict, query) + for query, results in search_results: for result in results[:PAGES_PER_QUERY]: url = result.get("url", "") - if not url or url in seen_urls: - continue - seen_urls.add(url) - title = result.get("title", url) - buf.append_event("status", {"status": f"Reading: {title[:60]}..."}) - content = await fetch_url_content(url) - all_sources.append({ - "url": url, - "title": title, - "query": query, - "snippet": result.get("snippet", ""), - "content": content, - }) + if url and url not in seen_urls: + seen_urls.add(url) + url_tasks.append((url, result, query)) + + # Fetch all unique URLs in parallel + async def _fetch_source(url: str, result: dict, query: str) -> dict: + title = result.get("title", url) + buf.append_event("status", {"status": f"Reading: {title[:60]}..."}) + content = await fetch_url_content(url) + return { + "url": url, + "title": title, + "query": query, + "snippet": result.get("snippet", ""), + "content": content, + } + + all_sources: list[dict] = list(await asyncio.gather( + *[_fetch_source(url, result, query) for url, result, query in url_tasks] + )) if not all_sources: raise ValueError(f"No results found for '{topic}'") @@ -84,9 +97,9 @@ async def run_research_pipeline( len(good_sources), len(all_sources), len(synthesis_sources), ) - # Step 4: Synthesize + # Step 4: Synthesize (streams tokens into chat as the note is being written) buf.append_event("status", {"status": f"Synthesizing report from {len(synthesis_sources)} sources..."}) - title, body = await _synthesize_note(topic, synthesis_sources, model) + title, body = await _synthesize_note(topic, synthesis_sources, model, buf) # Step 5: Create note buf.append_event("status", {"status": "Saving note..."}) @@ -175,11 +188,13 @@ async def _synthesize_note( topic: str, sources: list[dict], model: str, + buf=None, ) -> tuple[str, str]: """Synthesize a comprehensive markdown research document from fetched sources. Returns (title, body_markdown). - Uses an extended context window so the output can be several thousand words. + When buf is provided, tokens are streamed into the chat buffer in real time + so the user can see the note being written. Uses an extended context window. """ sources_text_parts = [] for i, s in enumerate(sources, 1): @@ -222,13 +237,24 @@ async def _synthesize_note( }, ] - raw = await generate_completion( - messages, - model, - max_tokens=8192, - num_ctx=16384, - ) - raw = raw.strip() + if buf is not None: + # Stream tokens into the chat buffer so the user sees the note being written + raw_parts: list[str] = [] + async for token in stream_chat( + messages, model, options={"num_ctx": 16384, "num_predict": 8192} + ): + raw_parts.append(token) + buf.append_event("chunk", {"chunk": token}) + buf.content_so_far += token + raw = "".join(raw_parts).strip() + else: + raw = await generate_completion( + messages, + model, + max_tokens=8192, + num_ctx=16384, + ) + raw = raw.strip() # Extract title from first # heading lines = raw.splitlines() diff --git a/summary.md b/summary.md index 4d4620d..54c074c 100644 --- a/summary.md +++ b/summary.md @@ -12,7 +12,7 @@ > Include file-level details in the commit body when the change is non-trivial. ## Last Updated -2026-02-26 — Phase 21: Intent-first pipeline, visible acknowledgment, KV-stable system prompt +2026-02-27 — Phase 22: SearXNG web research pipeline + performance improvements (parallel fetching, streaming synthesis, intent skip heuristic, intent num_ctx) ## Project Overview Fabled Assistant is a self-hosted note-taking and task-tracking application with @@ -260,7 +260,7 @@ fabledassistant/ │ ├── __init__.py │ ├── app.py # Quart app factory: SPA via 404 handler, JSON 404/500 for API, request logging, security headers (after_request) │ ├── auth.py # Auth decorators: login_required, admin_required, get_current_user_id — shared _check_auth() helper -│ ├── config.py # Config from env vars + Docker secrets file support (_read_secret) + SECURE_COOKIES + TRUST_PROXY_HEADERS + OLLAMA_NUM_CTX (KV cache window, default 8192) + OIDC_ISSUER/CLIENT_ID/CLIENT_SECRET/SCOPES + LOCAL_AUTH_ENABLED + oidc_enabled() classmethod +│ ├── config.py # Config from env vars + Docker secrets file support (_read_secret) + SECURE_COOKIES + TRUST_PROXY_HEADERS + OLLAMA_NUM_CTX (KV cache window, default 8192) + OIDC_ISSUER/CLIENT_ID/CLIENT_SECRET/SCOPES + LOCAL_AUTH_ENABLED + oidc_enabled() classmethod + SEARXNG_URL + searxng_enabled() classmethod │ ├── rate_limit.py # In-memory sliding-window rate limiter (asyncio.Lock + defaultdict); is_rate_limited(key, max, window) │ ├── models/ │ │ ├── __init__.py # async_session factory, Base, imports all models @@ -284,12 +284,13 @@ fabledassistant/ │ │ ├── oauth.py # OIDC/OAuth2 service: get_oidc_config (discovery, cached), build_auth_url (PKCE), exchange_code, get_userinfo, find_or_create_oauth_user (sub lookup → email auto-link → create) │ │ ├── backup.py # Backup/restore: export_full_backup, export_user_backup, restore_full_backup │ │ ├── notes.py # CRUD with user_id isolation, is_task filter, convert, backlinks, search_notes_for_context -│ │ ├── llm.py # Ollama interaction: build_context with user_id, streaming (stream_chat + stream_chat_with_tools), ChatChunk dataclass, URL fetching; uses Config.OLLAMA_NUM_CTX for KV cache window +│ │ ├── llm.py # Ollama interaction: build_context with user_id, streaming (stream_chat + stream_chat_with_tools), ChatChunk dataclass, URL fetching; generate_completion accepts num_ctx kwarg; uses Config.OLLAMA_NUM_CTX for KV cache window │ │ ├── chat.py # Conversation CRUD with user_id isolation, add_message, save/summarize as note (LLM-titled, chat-tagged) │ │ ├── generation_buffer.py # In-memory SSE event buffer with cancel_event, reconnect support, auto-cleanup; supports chat (int keys) and assist (string keys) -│ │ ├── generation_task.py # Background asyncio tasks: run_generation (chat, DB flush, titles, intent-first pipeline + tool loop) + run_assist_generation (lightweight, no DB) -│ │ ├── intent.py # Intent routing: classify_intent() makes fast non-streaming LLM call; IntentResult has ack field (one-sentence acknowledgment streamed as TTFT) -│ │ ├── tools.py # LLM tool definitions (create/delete note+task, update_note w/tag management, get_note, list_notes, search_notes w/type filter, list_tasks, full CalDAV suite incl. search_todos) + execute_tool dispatcher +│ │ ├── generation_task.py # Background asyncio tasks: run_generation (chat, DB flush, titles, intent-first pipeline + tool loop) + run_assist_generation (lightweight, no DB); _INTENT_TRIGGER_WORDS + _should_skip_intent() for skipping intent on conversational messages +│ │ ├── intent.py # Intent routing: classify_intent() makes fast non-streaming LLM call (num_ctx=4096); IntentResult has ack field (one-sentence acknowledgment streamed as TTFT) +│ │ ├── tools.py # LLM tool definitions (create/delete note+task, update_note w/tag management, get_note, list_notes, search_notes w/type filter, list_tasks, full CalDAV suite incl. search_todos, search_web, research_topic when SearXNG enabled) + execute_tool dispatcher +│ │ ├── research.py # SearXNG research pipeline: parallel query/fetch, streaming synthesis; run_research_pipeline() → Note; constants SEARXNG_QUERIES=5, PAGES_PER_QUERY=3, MAX_SYNTHESIS_SOURCES=12, CHARS_PER_SOURCE=2000 │ │ ├── tag_suggestions.py # LLM-powered tag suggestions: suggest_tags() builds prompt with existing tags, calls generate_completion, parses JSON response │ │ ├── caldav.py # CalDAV integration: full event lifecycle (create/list/search/update/delete), todos (create/list/search/update/complete/delete), list_calendars, timezone (ZoneInfo), reminders (VALARM), attendees, multi-calendar search │ │ ├── settings.py # Settings CRUD with user_id isolation: get_setting, set_setting, set_settings_batch, get_all_settings @@ -651,6 +652,35 @@ When adding a new migration, follow these conventions: "browse/list notes", tag management via update_note (tag_mode add/remove), search_todos. `generate_completion` (used by intent classifier) retries on HTTP 500 (3 attempts, 3s/6s delays) to handle cold model loading without failing intent classification. +- **Intent performance optimizations:** + - **Intent skip heuristic:** `_should_skip_intent(msg)` in `generation_task.py` skips the intent + model call entirely for short messages (≤10 words) that contain none of the `_INTENT_TRIGGER_WORDS` + frozenset (~50 action/object/date words). Saves 400–800ms for conversational replies like "thanks", + "can you explain that?", "okay" without risking missed tool calls on longer or action-verb messages. + - **Intent `num_ctx=4096`:** Intent classification calls use a 4k context window (override) instead + of the default, reducing VRAM pressure and prefill time on every request. +- **Web research pipeline (Phase 22):** `research.py` implements a full autonomous research pipeline + triggered by "research X and make a note" (intent routes to `research_topic` tool) or via the 🔍 + Research button in ChatView (sends "Research: {topic}" message). + Pipeline stages: + 1. Intent model generates 5 focused sub-queries as a JSON array (falls back to `[topic]` on parse failure) + 2. All 5 SearXNG queries execute in parallel with 200ms stagger (avoids hammering rate limiter) + 3. All unique URLs (up to 15) fetched in parallel via `asyncio.gather`; duplicates deduplicated by URL + 4. Failed fetches filtered; up to 12 sources passed to synthesis LLM + 5. Synthesis uses `stream_chat` with `num_ctx=16384`, `num_predict=8192` — tokens stream into + the chat buffer in real time (user sees the note being written); minimum 2500 words, 6+ topic-appropriate + sections (topic determines section structure, not hardcoded), detailed prose, `## Sources` section + 6. Note created with `tags=["research"]`; "Research complete!" separator appended to chat + Also includes `search_web` tool for lightweight single-query searches (no note created; results + returned to LLM for conversational answer). Both tools only added to tool list when `SEARXNG_URL` is set. + SearXNG `429` handling: 3-attempt retry with exponential backoff per query. + Sub-query JSON parsed with `json.JSONDecoder().raw_decode()` to handle trailing model text. + Frontend: `ToolCallCard.vue` handles `web_search` (external links), `research_pending`, `research_note` types. + Settings: `GET /api/settings/search?q=` proxy + Search Test section in SettingsView. + Settings UI: 2-column grid layout (paired simple cards, `full-width` for complex sections). + Docker: `OLLAMA_NUM_PARALLEL=2` to prevent research and live chat queuing each other. + SearXNG setup: add app server IP to `botdetection.ip_lists.pass_ip` in SearXNG `settings.yml` + to bypass rate limiter for trusted backend requests while keeping it active for public users. - **CalDAV calendar integration:** Per-user CalDAV settings (URL, username, password, calendar name, timezone). LLM tools: `create_event` (all_day, recurrence, timezone, reminder_minutes, attendees, calendar_name), `list_events`, `search_events`, `update_event`, `delete_event`, `list_calendars`, @@ -744,7 +774,6 @@ When adding a new migration, follow these conventions: - Calendar/timeline view for tasks - Import/export (Markdown files, JSON) - Email integration (read/send emails from chat via IMAP/SMTP tools) -- Web search support (LLM tool to search the web and summarize results) - Session invalidation on user deletion - **Note relation map (Obsidian-style graph):** Interactive force-directed graph visualizing connections between notes via wikilinks (`[[Title]]`) and shared tags. Nodes = notes/tasks; edges = wikilink references or