bvandeusen
bf7a29e8a0
Adds an empirical surface for evaluating model swaps. One row per assistant turn captures: model, think_enabled, tools_available, tools_attempted, tools_succeeded, tools_failed (with error details as JSONB). Without this, judging whether a new model "actually fires record_moment when it should" relies on anecdote across user-reported sessions. With it, the data is queryable directly. Pieces: - Migration 0046: generation_tool_log table with user_created and per-conversation indexes. - Model: SQLAlchemy GenerationToolLog with to_dict() for plain-dict consumption outside session scope. - Service: log_tool_outcomes() normalizes the in-app tool-call shape (function/result/status) into the split buckets and persists. It catches its own exceptions — telemetry failure must NEVER affect the user-facing generation flow. recent_logs() helper for read. - Integration in run_generation: called once per turn right after log_generation, fire-and-forget. - Tests: pure-normalization unit tests using a stub session — no DB needed in CI. Cover the success/error split, the empty-tool-calls case, the exception-swallowing contract, and the success=False edge case where status incorrectly says "success". No UI for the telemetry yet — internal infrastructure (the operator is the consumer, not the journal user), which the FabledRulebook "no UI no ship" explicitly excepts. Query via psql or extend the Fable MCP later if direct shell access gets tiresome. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
637 lines
27 KiB
Python
637 lines
27 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
|
|
|
|
|
|
# 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, scoped to conversation type.
|
|
from fabledassistant.models import async_session as _async_session
|
|
from fabledassistant.models.conversation import Conversation as _Conversation
|
|
async with _async_session() as _sess:
|
|
_conv = await _sess.get(_Conversation, conv_id)
|
|
_conversation_type = (
|
|
_conv.conversation_type if _conv and _conv.conversation_type else "chat"
|
|
)
|
|
tools = await get_tools_for_user(user_id, conversation_type=_conversation_type)
|
|
|
|
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})
|
|
|
|
# Think mode is per-user-setting (default off). Historically forced on for
|
|
# qwen3 because content-gated thinking (87fcaa6) exposed silent-generation
|
|
# failures on short tool-intent prompts. After moving to a model-family
|
|
# decoupled architecture, the bench data (May 2026) showed think costs
|
|
# 1-2 min/turn for unclear quality benefit; default off, opt in via
|
|
# the Settings UI. The generation_tool_log captures per-turn outcomes
|
|
# so reliability regressions surface empirically.
|
|
think_requested = think
|
|
think = (await get_setting(user_id, "think_enabled", "false")).lower() == "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)
|
|
|
|
# Per-turn tool-call telemetry. Empirical surface for evaluating
|
|
# model swaps without needing user reports — answers "did model X
|
|
# actually fire record_moment when it should have?" The helper is
|
|
# internally try/except so this never affects the user-facing flow.
|
|
from fabledassistant.services.generation_log import log_tool_outcomes
|
|
await log_tool_outcomes(
|
|
user_id=user_id,
|
|
conv_id=conv_id,
|
|
assistant_message_id=msg_id,
|
|
model=model,
|
|
think_enabled=think,
|
|
tools_available=[
|
|
(t.get("function") or {}).get("name") for t in tools
|
|
],
|
|
tool_calls=all_tool_calls,
|
|
)
|
|
|
|
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:
|
|
# Feed the title model the *raw* conversation turns only — never
|
|
# the post-build_context ``messages`` list. ``build_context``
|
|
# prepends RAG snippets and URL content 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)})
|