Phase 22b: Parallel research fetching, streaming synthesis, intent optimizations
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 <noreply@anthropic.com>
This commit is contained in:
@@ -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})
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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()
|
||||
|
||||
+36
-7
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user