700 lines
30 KiB
Python
700 lines
30 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
|
|
|
|
|
|
async def _maybe_save_article_discussion_note(
|
|
user_id: int, conv_id: int, reply_content: str,
|
|
) -> None:
|
|
"""Persist a seeded article-discussion's first reply as a Note.
|
|
|
|
Fires after ``run_generation`` completes. Looks for a synthetic
|
|
read_article seed message on the conversation; if found AND the linked
|
|
``rss_items`` row has no ``discussion_note_id`` yet, saves ``reply_content``
|
|
as a Note, tags it, and writes the backlink. Subsequent discuss clicks on
|
|
the same article are a no-op (already linked).
|
|
|
|
Failures are logged and swallowed — the chat UI should never break because
|
|
Note persistence hit a snag.
|
|
"""
|
|
try:
|
|
if not reply_content or not reply_content.strip():
|
|
return
|
|
from sqlalchemy import select as _select
|
|
from fabledassistant.models.conversation import Message as _Message
|
|
from fabledassistant.models.rss_feed import RssItem as _RssItem
|
|
from fabledassistant.services.notes import create_note
|
|
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
_select(_Message)
|
|
.where(_Message.conversation_id == conv_id)
|
|
.order_by(_Message.id.asc())
|
|
)
|
|
messages = result.scalars().all()
|
|
seed_meta = None
|
|
for m in messages:
|
|
meta = m.msg_metadata or {}
|
|
if meta.get("article_seed") and meta.get("rss_item_id"):
|
|
seed_meta = meta
|
|
break
|
|
if seed_meta is None:
|
|
return
|
|
item_id = int(seed_meta["rss_item_id"])
|
|
item = await session.get(_RssItem, item_id)
|
|
if item is None or item.discussion_note_id is not None:
|
|
return
|
|
article_title = (item.title or "Untitled article").strip()
|
|
article_url = item.url
|
|
article_topics = list(item.topics or [])
|
|
|
|
note_title = f"Article: {article_title}"[:200]
|
|
body_parts = [f"**Source:** {article_url}"] if article_url else []
|
|
body_parts.append(reply_content.strip())
|
|
note_body = "\n\n".join(body_parts)
|
|
tags = ["article-summary"] + [t for t in article_topics if t]
|
|
note = await create_note(
|
|
user_id=user_id,
|
|
title=note_title,
|
|
body=note_body,
|
|
tags=tags,
|
|
entity_meta={
|
|
"source": "article_discussion",
|
|
"rss_item_id": item_id,
|
|
"url": article_url,
|
|
"conversation_id": conv_id,
|
|
},
|
|
)
|
|
|
|
async with async_session() as session:
|
|
fresh = await session.get(_RssItem, item_id)
|
|
if fresh is not None and fresh.discussion_note_id is None:
|
|
fresh.discussion_note_id = note.id
|
|
await session.commit()
|
|
logger.info(
|
|
"Saved article-discussion summary as note %d for rss_item %d (conv %d)",
|
|
note.id, item_id, conv_id,
|
|
)
|
|
except Exception:
|
|
logger.warning(
|
|
"Failed to persist article-discussion note for conv %d",
|
|
conv_id, exc_info=True,
|
|
)
|
|
|
|
# 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",
|
|
"lookup": "Looking up information",
|
|
"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})
|
|
|
|
# Always think on qwen3-class models: reasoning mode is the only reliable
|
|
# path for the tool-call template. Content-based gating was tried in 87fcaa6
|
|
# but exposed silent-generation failures on short tool-intent prompts, since
|
|
# the classifier had no way to tell that "create a task" needs a tool call.
|
|
think_requested = think
|
|
think = True
|
|
|
|
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()
|
|
|
|
approx_msg_chars = sum(len(str(m.get("content", ""))) for m in messages)
|
|
round_content_start = len(buf.content_so_far)
|
|
round_output_tokens_start = timing.get("output_tokens") or 0
|
|
round_prompt_tokens_start = timing.get("prompt_tokens") or 0
|
|
logger.info(
|
|
"CTX_DIAG round_start conv=%d round=%d num_ctx=%d msgs=%d approx_chars=%d think=%s",
|
|
conv_id, _round, num_ctx, len(messages), approx_msg_chars, think,
|
|
)
|
|
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})
|
|
|
|
round_content_added = len(buf.content_so_far) - round_content_start
|
|
round_output_tokens_added = (timing.get("output_tokens") or 0) - round_output_tokens_start
|
|
round_prompt_tokens = (timing.get("prompt_tokens") or 0) - round_prompt_tokens_start
|
|
headroom = num_ctx - round_prompt_tokens if round_prompt_tokens else None
|
|
is_silent = (
|
|
not round_tool_calls
|
|
and round_content_added == 0
|
|
and round_output_tokens_added > 0
|
|
)
|
|
logger.info(
|
|
"CTX_DIAG round_end conv=%d round=%d think=%s prompt_tokens=%d output_tokens=%d headroom=%s content_added=%d tool_calls=%d silent=%s",
|
|
conv_id, _round, think, round_prompt_tokens, round_output_tokens_added,
|
|
headroom, round_content_added, len(round_tool_calls), is_silent,
|
|
)
|
|
|
|
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)
|
|
|
|
# Silent-generation safety net: the model burned output tokens but
|
|
# nothing landed in content or tool_calls (seen with qwen3:14b when
|
|
# its tool-call emission doesn't parse). Show a visible fallback so
|
|
# the user isn't staring at an empty bubble.
|
|
if (
|
|
not cancelled
|
|
and not buf.content_so_far.strip()
|
|
and not all_tool_calls
|
|
and (timing.get("output_tokens") or 0) > 0
|
|
):
|
|
logger.warning(
|
|
"Silent generation for conv %d: output_tokens=%s but empty content "
|
|
"and no tool calls (model=%s)",
|
|
conv_id, timing.get("output_tokens"), model,
|
|
)
|
|
fallback = (
|
|
"I wasn't able to produce a usable response — the model generated "
|
|
"tokens that couldn't be parsed as content or a tool call. "
|
|
"Please try rephrasing, or try again."
|
|
)
|
|
buf.content_so_far = fallback
|
|
buf.append_event("chunk", {"chunk": fallback})
|
|
|
|
# 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)
|
|
|
|
# Persist article-discussion seed conversations as a Note on their
|
|
# first assistant reply. This makes "Discuss" summaries part of RAG
|
|
# so the knowledge base stops being amnesiac about articles the user
|
|
# has already engaged with. The hook detects a seeded conversation by
|
|
# finding a synthetic read_article assistant message whose
|
|
# msg_metadata carries ``article_seed: True`` and whose rss_items row
|
|
# has no discussion_note_id yet. Fire-and-forget so the done event
|
|
# lands immediately.
|
|
asyncio.create_task(_maybe_save_article_discussion_note(
|
|
user_id, conv_id, buf.content_so_far,
|
|
))
|
|
|
|
if should_gen_title:
|
|
# Feed the title model the *raw* conversation turns only — never
|
|
# the post-build_context ``messages`` list. ``build_context``
|
|
# prepends RAG snippets, RSS excerpts, URL content, and briefing
|
|
# article dumps INTO the user message string itself, so filtering
|
|
# by role="user" downstream still surfaces that noise as the
|
|
# "user's message". That pollution caused wildly-wrong titles
|
|
# (bug #109) — the small background model was staring at article
|
|
# excerpts instead of what the user actually typed. Pass the
|
|
# original history + the raw user_content + the assistant reply.
|
|
title_messages: list[dict] = [
|
|
{"role": m["role"], "content": m.get("content") or ""}
|
|
for m in history
|
|
if m.get("role") in ("user", "assistant")
|
|
]
|
|
title_messages.append({"role": "user", "content": user_content})
|
|
title_messages.append({"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)})
|