bvandeusenandClaude Opus 4.6
734ccc337f
Adds 38 parametrized tests for the _should_think classifier covering the explicit-override path, empty/whitespace content, short/medium/long length boundaries, case-insensitive keyword matching, and a chatty-message negative set. These pin the content-based semantics so future tweaks to the keyword list or length thresholds surface regressions immediately instead of going unnoticed behind subtle latency changes. Also drops the `think=True` overrides from the briefing /discuss-article and /discuss-topic entry points. With `"discuss"` added to _THINK_KEYWORDS, those canned prompts trip the classifier naturally, so the overrides were redundant — keeping a uniform "classifier is authoritative" rule makes the code easier to reason about. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
615 lines
25 KiB
Python
615 lines
25 KiB
Python
"""Background asyncio task for LLM generation.
|
||
|
||
Streams from Ollama into a GenerationBuffer, periodically flushing to DB.
|
||
Runs independently of any HTTP connection.
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import re
|
||
import time
|
||
from collections.abc import AsyncGenerator
|
||
|
||
import httpx
|
||
|
||
from sqlalchemy import update
|
||
|
||
from fabledassistant.config import Config
|
||
from fabledassistant.models import async_session
|
||
from fabledassistant.models.conversation import Message
|
||
from fabledassistant.services.generation_buffer import (
|
||
GenerationBuffer,
|
||
GenerationState,
|
||
)
|
||
from fabledassistant.services.llm import ChatChunk, build_context, generate_completion, pick_num_ctx, stream_chat, stream_chat_with_tools, summarize_history_for_context
|
||
from fabledassistant.services.chat import update_conversation_title
|
||
from fabledassistant.services.settings import get_setting
|
||
from fabledassistant.services.logging import log_generation
|
||
from fabledassistant.services.tools import get_tools_for_user, execute_tool
|
||
from fabledassistant.services.research import run_research_pipeline
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Mistral prefixes tool-call responses with "[TOOL_CALLS]" as visible text
|
||
_TOOL_CALL_MARKER = re.compile(r"^\s*\[TOOL_CALLS\]\s*", re.IGNORECASE)
|
||
|
||
DB_FLUSH_INTERVAL = 5.0 # seconds between partial DB flushes
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Thinking decision
|
||
# ---------------------------------------------------------------------------
|
||
#
|
||
# `_should_think` is the single source of truth for whether a qwen3-class
|
||
# model should engage chain-of-thought for a given request. Frontend callers
|
||
# should NOT hardcode think=True — leave it False and let the classifier
|
||
# decide from message content. An explicit think_requested=True still acts
|
||
# as an override for callers (e.g. a future UI toggle or MCP client) that
|
||
# want to force extended reasoning regardless of content.
|
||
#
|
||
# Why gate it: on qwen3:14b, thinking adds 5–20s of latency before the first
|
||
# visible content token, and most conversational messages do not benefit.
|
||
# Gating by content keeps quick chats fast while preserving reasoning depth
|
||
# for prompts that actually need it.
|
||
#
|
||
# Models that don't support extended reasoning (e.g. llama3, mistral) simply
|
||
# ignore the `think` parameter in the Ollama chat request, so the decision
|
||
# here is harmless on non-thinking models.
|
||
|
||
|
||
# Keywords that strongly suggest the user wants reasoning / analysis. Matched
|
||
# case-insensitively as whole-ish phrases.
|
||
_THINK_KEYWORDS: tuple[str, ...] = (
|
||
"why", "how does", "how do i", "how would", "how should",
|
||
"explain", "analyze", "analyse", "compare", "contrast",
|
||
"design", "architect", "architecture", "plan out", "strategize",
|
||
"debug", "diagnose", "troubleshoot", "root cause",
|
||
"review", "critique", "evaluate", "trade-off", "tradeoff", "trade off",
|
||
"pros and cons", "step by step", "walk me through",
|
||
"prove", "derive", "figure out", "work through",
|
||
"discuss", # covers briefing /discuss-article + /discuss-topic entry points
|
||
)
|
||
|
||
# Messages shorter than this and without any think-keyword are treated as
|
||
# simple/conversational and skip the thinking phase.
|
||
_SHORT_MESSAGE_CHARS = 80
|
||
|
||
# Messages longer than this are treated as substantive regardless of keywords.
|
||
_LONG_MESSAGE_CHARS = 400
|
||
|
||
|
||
def _should_think(user_content: str, think_requested: bool) -> bool:
|
||
"""Return whether extended thinking should be used for this request.
|
||
|
||
``think_requested`` acts as an explicit override: if True, thinking is
|
||
forced on regardless of content. If False (the default), the decision is
|
||
made by inspecting the message: long or keyword-bearing messages get
|
||
thinking; short conversational messages skip it.
|
||
"""
|
||
if think_requested:
|
||
return True
|
||
text = (user_content or "").strip()
|
||
if not text:
|
||
return False
|
||
if len(text) >= _LONG_MESSAGE_CHARS:
|
||
return True
|
||
lowered = text.lower()
|
||
if any(kw in lowered for kw in _THINK_KEYWORDS):
|
||
return True
|
||
if len(text) < _SHORT_MESSAGE_CHARS:
|
||
return False
|
||
# Medium-length message with no obvious reasoning cue: default off.
|
||
return False
|
||
|
||
|
||
# Human-readable labels for each tool, shown in the status indicator
|
||
_TOOL_LABELS: dict[str, str] = {
|
||
"create_note": "Creating note/task",
|
||
"update_note": "Updating note/task",
|
||
"delete_note": "Deleting note/task",
|
||
"read_note": "Reading note",
|
||
"list_notes": "Listing notes",
|
||
"list_tasks": "Searching tasks",
|
||
"search_notes": "Searching notes (semantic)",
|
||
"create_event": "Creating calendar event",
|
||
"list_events": "Searching calendar",
|
||
"search_events": "Searching calendar",
|
||
"update_event": "Updating calendar event",
|
||
"delete_event": "Removing calendar event",
|
||
"list_calendars": "Listing calendars",
|
||
"search_web": "Searching the web",
|
||
"research_topic": "Researching topic",
|
||
}
|
||
|
||
|
||
async def _generate_title(messages: list[dict], user_id: int) -> str:
|
||
"""Ask the LLM for a concise conversation title.
|
||
|
||
Only uses user messages to avoid feeding tool-call JSON, system prompt
|
||
fragments, or other noise into the title generator. Caps input length
|
||
to keep the task fast and focused.
|
||
"""
|
||
user_texts = []
|
||
for m in messages:
|
||
if m["role"] == "user":
|
||
content = (m.get("content") or "").strip()
|
||
if content:
|
||
user_texts.append(content[:300])
|
||
if not user_texts:
|
||
return ""
|
||
# First + last user messages capture intent best
|
||
if len(user_texts) > 2:
|
||
user_texts = [user_texts[0], user_texts[-1]]
|
||
|
||
prompt_messages = [
|
||
{
|
||
"role": "user",
|
||
"content": (
|
||
"Generate a concise 3-8 word title for a conversation that started with:\n\n"
|
||
+ "\n\n".join(user_texts)
|
||
+ "\n\nReply with ONLY the title. No quotes, no punctuation, no explanation."
|
||
),
|
||
},
|
||
]
|
||
bg_model = await get_setting(user_id, "background_model", Config.OLLAMA_BACKGROUND_MODEL)
|
||
title = await generate_completion(prompt_messages, bg_model, max_tokens=30, num_ctx=1024)
|
||
# Strip common LLM noise: quotes, thinking tags, role labels
|
||
title = title.strip().strip('"\'').strip()
|
||
for prefix in ("Title:", "title:", "Assistant:", "User:"):
|
||
if title.startswith(prefix):
|
||
title = title[len(prefix):].strip()
|
||
# Drop anything after a newline (model sometimes adds explanation)
|
||
if "\n" in title:
|
||
title = title.split("\n")[0].strip()
|
||
return title[:80] if title else ""
|
||
|
||
|
||
async def _update_message(
|
||
message_id: int,
|
||
content: str,
|
||
status: str,
|
||
tool_calls: list[dict] | None = None,
|
||
) -> None:
|
||
values: dict = {"content": content, "status": status}
|
||
if tool_calls is not None:
|
||
values["tool_calls"] = tool_calls
|
||
async with async_session() as session:
|
||
await session.execute(
|
||
update(Message)
|
||
.where(Message.id == message_id)
|
||
.values(**values)
|
||
)
|
||
await session.commit()
|
||
|
||
|
||
async def _stream_with_retry(
|
||
messages: list[dict],
|
||
model: str,
|
||
tools: list[dict],
|
||
think: bool,
|
||
num_ctx: int | None = None,
|
||
) -> AsyncGenerator[ChatChunk, None]:
|
||
"""stream_chat_with_tools with automatic retry on Ollama 500 errors.
|
||
|
||
500s occur when Ollama is still loading a model or handling a concurrent
|
||
request (e.g. tag suggestions racing with round 1). Retries up to 2 times
|
||
with a short delay — by which point the model is warm and other calls done.
|
||
"""
|
||
last_exc: BaseException | None = None
|
||
for attempt in range(3):
|
||
if attempt > 0:
|
||
delay = 3.0 * attempt
|
||
logger.warning(
|
||
"Ollama stream 500 (attempt %d/3), retrying in %.0fs", attempt, delay
|
||
)
|
||
await asyncio.sleep(delay)
|
||
try:
|
||
async for chunk in stream_chat_with_tools(messages, model, tools=tools, think=think, num_ctx=num_ctx):
|
||
yield chunk
|
||
return
|
||
except httpx.HTTPStatusError as exc:
|
||
last_exc = exc
|
||
if exc.response.status_code != 500:
|
||
break # non-500 is not retryable
|
||
except BaseException as exc:
|
||
last_exc = exc
|
||
break
|
||
if last_exc is not None:
|
||
raise last_exc
|
||
|
||
|
||
async def run_generation(
|
||
buf: GenerationBuffer,
|
||
history: list[dict],
|
||
model: str,
|
||
user_id: int,
|
||
conv_id: int,
|
||
conv_title: str,
|
||
user_content: str,
|
||
context_note_id: int | None = None,
|
||
include_note_ids: list[int] | None = None,
|
||
excluded_note_ids: list[int] | None = None,
|
||
think: bool = False,
|
||
rag_project_id: int | None = None,
|
||
workspace_project_id: int | None = None,
|
||
user_timezone: str | None = None,
|
||
voice_mode: bool = False,
|
||
) -> None:
|
||
"""Stream LLM response into buffer with periodic DB flushes."""
|
||
MAX_TOOL_ROUNDS = 6
|
||
msg_id = buf.assistant_message_id
|
||
|
||
buf.append_event("status", {"status": "Building context..."})
|
||
|
||
# Phase 1: Resolve the tools list for this user.
|
||
tools = await get_tools_for_user(user_id)
|
||
|
||
logger.info(
|
||
"Starting generation for conv %d: model=%s, tools=%d",
|
||
conv_id, model, len(tools),
|
||
)
|
||
|
||
# Phase 2: Summarize long conversation history if needed.
|
||
history_to_use = history
|
||
history_summary: str | None = None
|
||
if len(history) > 30: # matches _HISTORY_SUMMARY_THRESHOLD in llm.py
|
||
buf.append_event("status", {"status": "Summarizing conversation history..."})
|
||
history_to_use, history_summary = await summarize_history_for_context(history, model)
|
||
|
||
# Phase 3: Build context.
|
||
# Note: Ollama lazy-loads models on the first /api/chat request, so polling
|
||
# /api/ps for model readiness only causes delay. We proceed immediately and
|
||
# let Ollama handle loading on demand.
|
||
|
||
# Fetch voice_speech_style from user settings when voice_mode is active.
|
||
voice_speech_style = "conversational"
|
||
if voice_mode:
|
||
from fabledassistant.services.settings import get_setting
|
||
voice_speech_style = await get_setting(user_id, "voice_speech_style", "conversational")
|
||
|
||
messages, context_meta = await build_context(
|
||
user_id, history_to_use, context_note_id, user_content,
|
||
history_summary=history_summary,
|
||
include_note_ids=include_note_ids,
|
||
excluded_note_ids=excluded_note_ids,
|
||
rag_project_id=rag_project_id,
|
||
workspace_project_id=workspace_project_id,
|
||
user_timezone=user_timezone,
|
||
conv_id=conv_id,
|
||
voice_mode=voice_mode,
|
||
voice_speech_style=voice_speech_style,
|
||
)
|
||
|
||
# Pick the smallest context tier that fits the current messages AND the
|
||
# tool schemas (which can be 6-10K tokens on their own with ~40 tools).
|
||
# Using the minimum needed tier reduces KV cache size and speeds up prefill.
|
||
num_ctx = pick_num_ctx(messages, tools=tools)
|
||
logger.debug("Adaptive num_ctx=%d for conv %d", num_ctx, conv_id)
|
||
|
||
# Emit context event
|
||
buf.append_event("context", {"context": context_meta})
|
||
|
||
# `_should_think` is authoritative — frontend callers pass think=False by
|
||
# default and let this classifier decide based on message content. An
|
||
# explicit think=True still forces on as an override.
|
||
think_requested = think
|
||
think = _should_think(user_content, think_requested)
|
||
|
||
t_start = time.monotonic()
|
||
timing: dict = {
|
||
"think_requested": think_requested,
|
||
"think": think,
|
||
"num_ctx": num_ctx,
|
||
"tools": [],
|
||
"rounds": 0,
|
||
"prompt_tokens": None,
|
||
"output_tokens": None,
|
||
"first_token_ms": None,
|
||
"thinking_ms": None,
|
||
"ttft_ms": None,
|
||
"generation_ms": None,
|
||
"total_ms": None,
|
||
}
|
||
|
||
last_flush = time.monotonic()
|
||
all_tool_calls: list[dict] = []
|
||
new_rag_scope: object = False # sentinel; set to int|None when scope changes
|
||
new_rag_scope_label: str | None = None
|
||
|
||
try:
|
||
cancelled = False
|
||
research_completed = False
|
||
|
||
for _round in range(MAX_TOOL_ROUNDS):
|
||
timing["rounds"] = _round + 1
|
||
round_tool_calls: list[dict] = []
|
||
logger.info("Generation round %d started for conv %d (model=%s)", _round, conv_id, model)
|
||
|
||
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, tools, think, num_ctx=num_ctx):
|
||
if buf.cancel_event.is_set():
|
||
cancelled = True
|
||
break
|
||
|
||
if chunk.type == "thinking":
|
||
if timing["first_token_ms"] is None:
|
||
timing["first_token_ms"] = int((time.monotonic() - t_start) * 1000)
|
||
buf.append_event("thinking_chunk", {"chunk": chunk.content})
|
||
|
||
elif chunk.type == "content":
|
||
if timing["ttft_ms"] is None:
|
||
now_ms = int((time.monotonic() - t_start) * 1000)
|
||
timing["ttft_ms"] = now_ms
|
||
if timing["first_token_ms"] is None:
|
||
# No thinking phase occurred — first token IS content.
|
||
timing["first_token_ms"] = now_ms
|
||
else:
|
||
# Thinking phase duration = gap between first thinking
|
||
# token and first content token.
|
||
timing["thinking_ms"] = now_ms - timing["first_token_ms"]
|
||
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 == "done":
|
||
if chunk.prompt_tokens is not None:
|
||
timing["prompt_tokens"] = (timing["prompt_tokens"] or 0) + chunk.prompt_tokens
|
||
if chunk.output_tokens is not None:
|
||
timing["output_tokens"] = (timing["output_tokens"] or 0) + chunk.output_tokens
|
||
|
||
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()
|
||
if tool_name == "research_topic":
|
||
topic = arguments.get("topic", "")
|
||
try:
|
||
note = await run_research_pipeline(topic, user_id, model, buf, project_id=workspace_project_id)
|
||
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)
|
||
err_msg = str(e) or f"{type(e).__name__}: unexpected error"
|
||
result = {"success": False, "error": err_msg}
|
||
err_text = f"\nResearch failed: {err_msg}"
|
||
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,
|
||
conv_id=conv_id,
|
||
workspace_project_id=workspace_project_id,
|
||
)
|
||
|
||
# Capture RAG scope change for SSE done event
|
||
if result.get("type") == "rag_scope_set" and result.get("success"):
|
||
new_rag_scope = arguments.get("project_id")
|
||
new_rag_scope_label = result.get("scope_label")
|
||
|
||
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:
|
||
logger.info("Generation cancelled for conv %d", conv_id)
|
||
break
|
||
|
||
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))
|
||
|
||
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 = ""
|
||
|
||
# Strip model artifacts from final content
|
||
buf.content_so_far = _TOOL_CALL_MARKER.sub("", buf.content_so_far)
|
||
|
||
# Final save
|
||
logger.info("Generation complete for conv %d: content_length=%d, tool_calls=%d",
|
||
conv_id, len(buf.content_so_far), len(all_tool_calls))
|
||
await _update_message(
|
||
msg_id,
|
||
buf.content_so_far,
|
||
"complete",
|
||
tool_calls=all_tool_calls if all_tool_calls else None,
|
||
)
|
||
|
||
timing["total_ms"] = int((time.monotonic() - t_start) * 1000)
|
||
logger.info(
|
||
"Generation timing for conv %d: total=%dms think=%s(req=%s) first_token=%s "
|
||
"thinking=%s ttft=%s generation=%s tools=%s",
|
||
conv_id, timing["total_ms"],
|
||
timing["think"], timing["think_requested"],
|
||
timing["first_token_ms"], timing["thinking_ms"], timing["ttft_ms"],
|
||
timing["generation_ms"],
|
||
[(t["name"], t["ms"]) for t in timing["tools"]],
|
||
)
|
||
try:
|
||
await log_generation(user_id, conv_id, model, timing)
|
||
except Exception:
|
||
logger.warning("Failed to persist generation timing for conv %d", conv_id, exc_info=True)
|
||
|
||
buf.state = GenerationState.COMPLETED
|
||
buf.finished_at = time.monotonic()
|
||
done_payload: dict = {"done": True, "message_id": msg_id, "timing": timing}
|
||
if new_rag_scope is not False:
|
||
done_payload["new_rag_scope"] = new_rag_scope
|
||
done_payload["new_rag_scope_label"] = new_rag_scope_label
|
||
buf.append_event("done", done_payload)
|
||
|
||
# Fire push notification when complete (non-critical, fire-and-forget)
|
||
try:
|
||
from fabledassistant.services.push import send_push_notification, vapid_enabled
|
||
if vapid_enabled():
|
||
text = buf.content_so_far.strip()
|
||
if text:
|
||
preview = text[:120].rstrip()
|
||
if len(text) > 120:
|
||
preview += "…"
|
||
else:
|
||
# Tool-only response — summarise what was done
|
||
tool_names = [tc.get("function") for tc in all_tool_calls if tc.get("function")]
|
||
if tool_names:
|
||
preview = f"Completed: {', '.join(tool_names[:3])}"
|
||
else:
|
||
preview = "Action completed"
|
||
asyncio.create_task(send_push_notification(
|
||
user_id,
|
||
title="Response ready",
|
||
body=preview,
|
||
url=f"/chat/{conv_id}",
|
||
))
|
||
except Exception:
|
||
logger.warning("Failed to schedule push notification", exc_info=True)
|
||
|
||
# Title generation is non-critical — fire-and-forget so done fires immediately
|
||
non_system = [m for m in messages if m["role"] != "system"]
|
||
msg_count = len(non_system)
|
||
should_gen_title = not conv_title or (msg_count > 0 and msg_count % 10 == 0)
|
||
|
||
if should_gen_title:
|
||
title_messages = messages + [
|
||
{"role": "assistant", "content": buf.content_so_far}
|
||
]
|
||
|
||
async def _bg_title() -> None:
|
||
try:
|
||
title = await _generate_title(title_messages, user_id)
|
||
if title:
|
||
await update_conversation_title(user_id, conv_id, title)
|
||
except Exception:
|
||
logger.warning("Failed to generate title for conversation %d", conv_id, exc_info=True)
|
||
if not conv_title:
|
||
fallback = user_content[:80]
|
||
if len(user_content) > 80:
|
||
fallback += "..."
|
||
await update_conversation_title(user_id, conv_id, fallback)
|
||
|
||
asyncio.create_task(_bg_title())
|
||
|
||
except Exception as e:
|
||
logger.exception("Error in generation task for conversation %d", conv_id)
|
||
# Save partial content with error status
|
||
try:
|
||
await _update_message(msg_id, buf.content_so_far, "error")
|
||
except Exception:
|
||
logger.warning("Failed to save error state for message %d", msg_id, exc_info=True)
|
||
|
||
buf.state = GenerationState.ERRORED
|
||
buf.finished_at = time.monotonic()
|
||
buf.append_event("error", {"error": str(e)})
|
||
|
||
|
||
async def run_assist_generation(
|
||
buf: GenerationBuffer,
|
||
messages: list[dict],
|
||
model: str,
|
||
) -> None:
|
||
"""Stream LLM response for assist into buffer. No DB persistence.
|
||
|
||
Retries up to 3 times on Ollama 500 errors (model still loading).
|
||
On each retry the accumulated content is reset so the done event
|
||
always reflects only the successful generation.
|
||
"""
|
||
from fabledassistant.services.llm import pick_num_ctx
|
||
input_chars = sum(len(m.get("content", "")) for m in messages)
|
||
num_ctx = pick_num_ctx(messages)
|
||
logger.info("Assist generation started: model=%s, input_chars=%d, num_ctx=%d", model, input_chars, num_ctx)
|
||
|
||
last_exc: BaseException | None = None
|
||
for attempt in range(3):
|
||
if attempt > 0:
|
||
delay = 3.0 * attempt
|
||
logger.warning(
|
||
"Ollama assist stream 500 (attempt %d/3), retrying in %.0fs", attempt, delay
|
||
)
|
||
await asyncio.sleep(delay)
|
||
try:
|
||
buf.content_so_far = ""
|
||
async for chunk in stream_chat(messages, model, options={"num_predict": num_ctx}, num_ctx=num_ctx):
|
||
buf.content_so_far += chunk
|
||
buf.append_event("chunk", {"chunk": chunk})
|
||
|
||
output_chars = len(buf.content_so_far)
|
||
logger.info(
|
||
"Assist generation complete: output_chars=%d, events=%d",
|
||
output_chars, len(buf.events),
|
||
)
|
||
buf.state = GenerationState.COMPLETED
|
||
buf.finished_at = time.monotonic()
|
||
buf.append_event("done", {"done": True, "full_text": buf.content_so_far})
|
||
logger.info("Assist done event appended (event index %d)", len(buf.events) - 1)
|
||
return
|
||
|
||
except httpx.HTTPStatusError as exc:
|
||
last_exc = exc
|
||
if exc.response.status_code != 500:
|
||
break
|
||
except Exception as exc:
|
||
last_exc = exc
|
||
break
|
||
|
||
logger.exception("Error in assist generation task")
|
||
buf.state = GenerationState.ERRORED
|
||
buf.finished_at = time.monotonic()
|
||
buf.append_event("error", {"error": str(last_exc)})
|