bvandeusen/FabledScribe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 50 Wiki Activity

Phase 21: Intent-first pipeline, visible ack, KV-stable system prompt

Browse Source 
Pipeline changes (generation_task.py, intent.py):
- Remove optimistic streaming queue/race (_drain_queue deleted)
- Remove _generate_acknowledgment — ack now embedded in intent JSON
- Round 0: await intent (~400ms), stream ack immediately as TTFT,
  then execute tool sequentially; chat-only streams directly
- IntentResult.ack: one-sentence acknowledgment, intent max_tokens 200→350
- _parse_intent extracts and trims ack field

KV cache stability (llm.py, generation_buffer.py, generation_task.py):
- build_context: replace cached_note_ids with include_note_ids
- Auto-found notes populate context_meta["auto_notes"] for sidebar but
  are NOT injected into system prompt (--- Related Notes --- removed)
- Explicitly included notes injected as --- Included Notes ---
- _conv_note_cache dict + get/set/clear functions removed from generation_buffer.py
- All clear_conv_note_cache() calls removed

Cold model retry (llm.py):
- generate_completion (used by classify_intent) retries on HTTP 500:
  3 attempts with 3s/6s delays — prevents intent failure during cold load

API + frontend (routes/chat.py, stores/chat.ts, views/ChatView.vue, components/ChatPanel.vue):
- exclude_note_ids → include_note_ids throughout
- ChatView sidebar: Suggested (auto-found, + to include) + In Context (× to remove)
- ChatPanel: remove exclude button from context pills; no IDs passed to sendMessage

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Bryan Van Deusen
2026-02-26 22:34:54 -05:00
parent 316a85e13b
commit e119331645
9 changed files with 237 additions and 353 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/fabledassistant/services/generation_task.py
+29 -187
View File
@@ -21,9 +21,6 @@ from fabledassistant.models.conversation import Message
from fabledassistant.services.generation_buffer import (
GenerationBuffer,
GenerationState,
clear_conv_note_cache,
get_conv_note_cache,
set_conv_note_cache,
)
from fabledassistant.services.llm import ChatChunk, build_context, generate_completion, stream_chat, stream_chat_with_tools, summarize_history_for_context
from fabledassistant.services.chat import update_conversation_title
@@ -97,39 +94,6 @@ _TOOL_ACTIONS: dict[str, str] = {
}
async def _generate_acknowledgment(user_content: str, tool_name: str, model: str) -> str:
"""Generate a brief conversational acknowledgment that runs in parallel with tool execution.
Uses the intent model (small, fast, already in VRAM) so the sentence is ready
within ~200-400ms. Returned string includes a trailing double-newline so the
main LLM response starts on a new paragraph.
"""
action = _TOOL_ACTIONS.get(tool_name, "work on that")
messages = [
{
"role": "system",
"content": (
"You are a helpful assistant. Write ONE short, natural sentence acknowledging "
"what you are about to do. Vary your phrasing — do not always start with "
"'Let me'. Be warm and conversational. Do not answer the question yet. "
"Output only the sentence, nothing else."
),
},
{
"role": "user",
"content": f"User said: {user_content}\nYou are about to: {action}",
},
]
try:
ack = await generate_completion(messages, model, max_tokens=40)
ack = ack.strip()
if ack:
return ack + "\n\n"
except Exception:
logger.warning("Failed to generate acknowledgment", exc_info=True)
return ""
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
@@ -175,26 +139,6 @@ async def _update_message(
await session.commit()
async def _drain_queue(
prefetched: list[ChatChunk],
queue: asyncio.Queue,
) -> AsyncGenerator[ChatChunk, None]:
"""Yield pre-fetched chunks then drain remaining chunks from the queue.
A None sentinel in the queue signals the stream is finished.
A BaseException in the queue is re-raised so callers see the error.
"""
for chunk in prefetched:
yield chunk
while True:
item = await queue.get()
if item is None:
break
if isinstance(item, BaseException):
raise item
yield item
async def _stream_with_retry(
messages: list[dict],
model: str,
@@ -239,7 +183,7 @@ async def run_generation(
conv_title: str,
user_content: str,
context_note_id: int | None = None,
exclude_note_ids: list[int] | None = None,
include_note_ids: list[int] | None = None,
think: bool = False,
) -> None:
"""Stream LLM response into buffer with periodic DB flushes."""
@@ -266,17 +210,13 @@ 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. Start intent classification concurrently when tools
# are available so it runs in parallel with the embedding/DB work in build_context.
# We only block on context (need messages to stream) — intent result is consumed
# later via a race with the first streaming token.
cached_note_ids = get_conv_note_cache(conv_id) or None
# Phase 3: Build context and start intent classification in parallel.
# We block on context (need messages to stream) — intent is consumed
# after context is ready, at the start of round 0.
context_task = asyncio.create_task(build_context(
user_id, history_to_use, context_note_id, user_content,
exclude_note_ids=exclude_note_ids,
history_summary=history_summary,
cached_note_ids=cached_note_ids,
include_note_ids=include_note_ids,
))
intent_task: asyncio.Task[IntentResult] | None = None
@@ -292,11 +232,6 @@ async def run_generation(
messages, context_meta = await context_task
# Update the note cache with whatever notes ended up in context.
new_note_ids = context_meta.get("auto_note_ids") or []
if new_note_ids:
set_conv_note_cache(conv_id, new_note_ids)
# Emit context event
buf.append_event("context", {"context": context_meta})
@@ -319,89 +254,27 @@ async def run_generation(
round_tool_calls: list[dict] = []
logger.info("Generation round %d started for conv %d (model=%s)", _round, conv_id, model)
# --- Round 0 with tools: optimistic streaming ---
# Start the main stream immediately (into a queue) while the intent
# classifier finishes in the background. Race the two:
# • Intent wins before first token → check for tool call, cancel stream if needed
# • First token wins → discard intent, stream has already started
# --- 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:
stream_queue: asyncio.Queue[ChatChunk | BaseException | None] = asyncio.Queue(maxsize=256)
intent = await intent_task
timing["intent_ms"] = int((time.monotonic() - t_intent) * 1000)
async def _fill_queue() -> None:
# Retry on Ollama 500 (model cold-loading race) up to 2 times.
last_exc: BaseException | None = None
for attempt in range(3):
if attempt > 0:
wait = 3.0 * attempt
logger.warning(
"Ollama stream 500 (attempt %d/3), retrying in %.0fs", attempt, wait
)
await asyncio.sleep(wait)
try:
async for c in stream_chat_with_tools(messages, model, tools=tools, think=think):
await stream_queue.put(c)
last_exc = None
break
except httpx.HTTPStatusError as exc:
last_exc = exc
if exc.response.status_code != 500:
break # non-500 errors are not retryable
except BaseException as exc:
last_exc = exc
break
if last_exc is not None:
await stream_queue.put(last_exc)
await stream_queue.put(None)
stream_fill_task = asyncio.create_task(_fill_queue())
buf.append_event("status", {"status": "Generating response..."})
queue_peek = asyncio.create_task(stream_queue.get())
race_done, _ = await asyncio.wait(
{intent_task, queue_peek},
return_when=asyncio.FIRST_COMPLETED,
)
intent = IntentResult()
prefetched: list[ChatChunk] = []
use_tool_path = False
if intent_task in race_done and queue_peek not in race_done:
# Intent finished before the first streaming token arrived.
timing["intent_ms"] = int((time.monotonic() - t_intent) * 1000)
intent = intent_task.result()
if intent.should_execute:
# Cancel the optimistic stream — we'll execute a tool instead.
stream_fill_task.cancel()
queue_peek.cancel()
use_tool_path = True
else:
# No tool needed — collect the first chunk and keep streaming.
first = await queue_peek
if isinstance(first, BaseException):
raise first
if first is not None:
prefetched.append(first)
else:
# Stream produced a token before intent finished (or simultaneously).
# Discard intent — the response is already on its way.
if not intent_task.done():
intent_task.cancel()
else:
timing["intent_ms"] = int((time.monotonic() - t_intent) * 1000)
first = queue_peek.result() if queue_peek in race_done else await queue_peek
if isinstance(first, BaseException):
raise first
if first is not None:
prefetched.append(first)
if use_tool_path:
# === Tool path (intent won the race) ===
if intent.should_execute:
tool_name = intent.tool_name
confirmed = True
# 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)
confirmed = True
if tool_name in _WRITE_TOOLS:
loop = asyncio.get_running_loop()
confirm_future: asyncio.Future = loop.create_future()
@@ -449,24 +322,11 @@ async def run_generation(
if confirmed:
buf.append_event("status", {"status": f"{_TOOL_LABELS.get(tool_name, 'Working')}..."})
t_tool = time.monotonic()
result, ack_text = await asyncio.gather(
execute_tool(user_id, tool_name, intent.arguments),
_generate_acknowledgment(user_content, tool_name, intent_model),
)
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"))
if ack_text:
buf.append_event("chunk", {"chunk": ack_text})
buf.content_so_far += ack_text
if timing["ttft_ms"] is None:
timing["ttft_ms"] = int((time.monotonic() - t_start) * 1000)
if result.get("success") and tool_name in {"create_task", "create_note", "update_note", "delete_note", "delete_task"}:
clear_conv_note_cache(conv_id)
tool_record = {
"function": tool_name,
"arguments": intent.arguments,
@@ -478,35 +338,23 @@ async def run_generation(
messages.append({
"role": "assistant",
"content": ack_text,
"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 the response incorporating tool result
messages.append({"role": "tool", "content": json.dumps(result)})
continue # Round 1: stream response with tool result
# Declined write tool — fall through to stream a fresh response.
# Declined write tool — fall through to fresh stream.
if cancelled:
break
# === Stream path for round 0 ===
# Either intent said no-tool, or a write tool was declined.
# For no-tool: drain the already-started queue (prefetched + remaining).
# For declined: start a fresh stream (queue was cancelled).
# No tool (or declined write tool): stream directly, no queue.
buf.append_event("status", {"status": "Generating response..."})
t_stream = time.monotonic()
if use_tool_path:
# Declined write tool — the optimistic stream was cancelled, start fresh.
stream_source: AsyncGenerator = _stream_with_retry(messages, model, tools, think)
else:
stream_source = _drain_queue(prefetched, stream_queue)
async for chunk in stream_source:
async for chunk in _stream_with_retry(messages, model, tools, think):
if buf.cancel_event.is_set():
cancelled = True
break
@@ -541,9 +389,6 @@ async def run_generation(
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"))
if result.get("success") and tool_name in {"create_task", "create_note", "update_note", "delete_note", "delete_task"}:
clear_conv_note_cache(conv_id)
tool_record = {
"function": tool_name,
"arguments": arguments,
@@ -618,9 +463,6 @@ async def run_generation(
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"))
if result.get("success") and tool_name in {"create_task", "create_note", "update_note", "delete_note", "delete_task"}:
clear_conv_note_cache(conv_id)
tool_record = {
"function": tool_name,
"arguments": arguments,