bvandeusen
9af8ab8f70
- build_context: when conversation_type is 'briefing', inject a system prompt instruction telling the model to answer from conversation history and article context instead of searching the web - Consolidate briefing conversation type detection to one DB query (was being checked twice — once for the system prompt addition, once for article context injection) - ChatPanel: render a visual 'New Briefing Update' separator line before 2nd+ briefing slot messages (identified by metadata.rss_item_ids) - types/chat.ts: add metadata field to Message interface Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
875 lines
35 KiB
Python
875 lines
35 KiB
Python
import asyncio
|
||
import ipaddress
|
||
import json
|
||
import logging
|
||
import re
|
||
import socket
|
||
import time
|
||
from collections.abc import AsyncGenerator
|
||
from dataclasses import dataclass, field
|
||
from typing import Literal
|
||
from urllib.parse import urlparse
|
||
|
||
import httpx
|
||
|
||
from fabledassistant.config import Config
|
||
from fabledassistant.services.caldav import is_caldav_configured
|
||
from fabledassistant.services.notes import get_note, search_notes_for_context
|
||
from fabledassistant.services.settings import get_setting
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Context window tiers. The smallest tier that fits the current input is used
|
||
# so Ollama allocates a smaller KV cache, reducing prefill time and VRAM usage.
|
||
# Requests using the same tier hit Ollama's prefix cache; a tier upgrade causes
|
||
# a one-time model reload but then the larger cache stays warm.
|
||
_CTX_TIERS = (8192, 16384, 32768)
|
||
|
||
|
||
def pick_num_ctx(messages: list[dict]) -> int:
|
||
"""Return the smallest context tier that fits *messages* with 25% headroom.
|
||
|
||
Stays at or below Config.OLLAMA_NUM_CTX (the configured ceiling).
|
||
"""
|
||
total_chars = sum(len(m.get("content") or "") for m in messages)
|
||
estimated_tokens = int(total_chars / 3.5)
|
||
needed = int(estimated_tokens * 1.25) + 256 # 25% headroom + output buffer
|
||
cap = Config.OLLAMA_NUM_CTX
|
||
for tier in _CTX_TIERS:
|
||
if tier >= needed and tier <= cap:
|
||
return tier
|
||
return cap
|
||
|
||
|
||
STOP_WORDS = frozenset({
|
||
"a", "an", "the", "is", "it", "to", "in", "for", "of", "and", "or",
|
||
"on", "at", "by", "with", "from", "as", "be", "was", "were", "been",
|
||
"are", "am", "do", "does", "did", "have", "has", "had", "will", "would",
|
||
"can", "could", "shall", "should", "may", "might", "must", "that",
|
||
"this", "these", "those", "i", "me", "my", "you", "your", "he", "she",
|
||
"we", "they", "them", "his", "her", "its", "our", "their", "what",
|
||
"which", "who", "whom", "how", "when", "where", "why", "not", "no",
|
||
"but", "if", "so", "than", "too", "very", "just", "about", "up",
|
||
})
|
||
|
||
RAG_AUTO_THRESHOLD = 0.60
|
||
RAG_AUTO_LIMIT = 3
|
||
RAG_AUTO_SNIPPET = 4000
|
||
|
||
|
||
async def get_installed_models() -> set[str]:
|
||
"""Return set of installed Ollama model names (with and without :latest)."""
|
||
try:
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
resp = await client.get(f"{Config.OLLAMA_URL}/api/tags")
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
names: set[str] = set()
|
||
for m in data.get("models", []):
|
||
name = m["name"]
|
||
names.add(name)
|
||
if name.endswith(":latest"):
|
||
names.add(name.removesuffix(":latest"))
|
||
return names
|
||
except Exception:
|
||
logger.warning("Failed to fetch installed Ollama models")
|
||
return set()
|
||
|
||
|
||
async def ensure_model(model: str) -> None:
|
||
"""Check if model exists in Ollama, pull if missing."""
|
||
try:
|
||
installed = await get_installed_models()
|
||
if model in installed or f"{model}:latest" in installed:
|
||
logger.info("Model '%s' already available", model)
|
||
return
|
||
except Exception:
|
||
logger.warning("Failed to check Ollama models, attempting pull anyway")
|
||
|
||
logger.info("Pulling model '%s' from Ollama...", model)
|
||
try:
|
||
async with httpx.AsyncClient(timeout=1800.0) as client:
|
||
async with client.stream(
|
||
"POST",
|
||
f"{Config.OLLAMA_URL}/api/pull",
|
||
json={"name": model},
|
||
) as resp:
|
||
resp.raise_for_status()
|
||
async for line in resp.aiter_lines():
|
||
if line.strip():
|
||
status = json.loads(line)
|
||
if "status" in status:
|
||
logger.info("Pull %s: %s", model, status["status"])
|
||
logger.info("Model '%s' pulled successfully", model)
|
||
except Exception:
|
||
logger.warning("Failed to pull model '%s' — chat may not work", model, exc_info=True)
|
||
|
||
|
||
async def wait_for_model_loaded(model: str, timeout: float = 90.0) -> bool:
|
||
"""Poll /api/ps every 2s until the model appears in Ollama's loaded-model list.
|
||
|
||
Returns True when the model is loaded, False if timeout elapses first.
|
||
Used before generation to avoid streaming 500s during cold model loads.
|
||
"""
|
||
base = model.removesuffix(":latest")
|
||
deadline = time.monotonic() + timeout
|
||
while True:
|
||
try:
|
||
async with httpx.AsyncClient(timeout=5.0) as client:
|
||
resp = await client.get(f"{Config.OLLAMA_URL}/api/ps")
|
||
resp.raise_for_status()
|
||
loaded = {m["name"] for m in resp.json().get("models", [])}
|
||
if model in loaded or f"{base}:latest" in loaded or base in loaded:
|
||
return True
|
||
except Exception:
|
||
pass # Ollama may still be starting up
|
||
remaining = deadline - time.monotonic()
|
||
if remaining <= 0:
|
||
return False
|
||
await asyncio.sleep(min(2.0, remaining))
|
||
|
||
|
||
async def stream_chat(
|
||
messages: list[dict],
|
||
model: str,
|
||
options: dict | None = None,
|
||
think: bool = False,
|
||
num_ctx: int | None = None,
|
||
) -> AsyncGenerator[str, None]:
|
||
"""Stream chat completion from Ollama, yielding content chunks.
|
||
|
||
Set think=False (default) to disable chain-of-thought on qwen3+ models.
|
||
Thinking tokens are silently discarded anyway, but disabling avoids the
|
||
multi-minute delay before the first content token arrives.
|
||
"""
|
||
merged_options = {"num_ctx": num_ctx or Config.OLLAMA_NUM_CTX}
|
||
if options:
|
||
merged_options.update(options)
|
||
payload: dict = {"model": model, "messages": messages, "stream": True, "options": merged_options, "think": think, "keep_alive": "2h"}
|
||
# read=None: no per-chunk timeout — Ollama may pause for any duration while
|
||
# processing a large input context before the first token arrives.
|
||
async with httpx.AsyncClient(timeout=httpx.Timeout(connect=30.0, read=None, write=None, pool=30.0)) as client:
|
||
async with client.stream(
|
||
"POST",
|
||
f"{Config.OLLAMA_URL}/api/chat",
|
||
json=payload,
|
||
) as resp:
|
||
resp.raise_for_status()
|
||
async for line in resp.aiter_lines():
|
||
if not line.strip():
|
||
continue
|
||
data = json.loads(line)
|
||
chunk = data.get("message", {}).get("content", "")
|
||
if chunk:
|
||
yield chunk
|
||
if data.get("done"):
|
||
break
|
||
|
||
|
||
@dataclass
|
||
class ChatChunk:
|
||
"""A chunk yielded by stream_chat_with_tools."""
|
||
type: Literal["content", "thinking", "tool_calls", "done"]
|
||
content: str = ""
|
||
tool_calls: list[dict] | None = None
|
||
# Token counts from the Ollama done event (only set on type="done")
|
||
prompt_tokens: int | None = None
|
||
output_tokens: int | None = None
|
||
|
||
|
||
async def stream_chat_with_tools(
|
||
messages: list[dict],
|
||
model: str,
|
||
tools: list[dict] | None = None,
|
||
think: bool = False,
|
||
num_ctx: int | None = None,
|
||
) -> AsyncGenerator[ChatChunk, None]:
|
||
"""Stream chat completion from Ollama with tool support.
|
||
|
||
Yields ChatChunk objects. If the model returns tool_calls, a
|
||
ChatChunk(type="tool_calls") is yielded. Always ends with
|
||
ChatChunk(type="done").
|
||
|
||
Set think=True to enable the model's chain-of-thought reasoning (qwen3+).
|
||
Thinking tokens are consumed by Ollama and not forwarded to the caller;
|
||
only the final response content is yielded. Expect higher TTFT when enabled.
|
||
"""
|
||
resolved_ctx = num_ctx or Config.OLLAMA_NUM_CTX
|
||
options: dict = {"num_ctx": resolved_ctx}
|
||
if tools:
|
||
options["num_predict"] = 8192
|
||
payload: dict = {
|
||
"model": model,
|
||
"messages": messages,
|
||
"stream": True,
|
||
"options": options,
|
||
"think": think,
|
||
"keep_alive": "2h",
|
||
}
|
||
if tools:
|
||
payload["tools"] = tools
|
||
# read=None: no per-chunk timeout for the same reason as stream_chat.
|
||
async with httpx.AsyncClient(timeout=httpx.Timeout(connect=30.0, read=None, write=None, pool=30.0)) as client:
|
||
async with client.stream(
|
||
"POST",
|
||
f"{Config.OLLAMA_URL}/api/chat",
|
||
json=payload,
|
||
) as resp:
|
||
resp.raise_for_status()
|
||
accumulated_tool_calls: list[dict] = []
|
||
async for line in resp.aiter_lines():
|
||
if not line.strip():
|
||
continue
|
||
data = json.loads(line)
|
||
msg = data.get("message", {})
|
||
|
||
# Thinking chunks (qwen3 chain-of-thought, only when think=True)
|
||
thinking = msg.get("thinking", "")
|
||
if thinking:
|
||
yield ChatChunk(type="thinking", content=thinking)
|
||
|
||
# Content chunks
|
||
chunk = msg.get("content", "")
|
||
if chunk:
|
||
yield ChatChunk(type="content", content=chunk)
|
||
|
||
# Collect tool calls from any message (some models
|
||
# emit them before the done flag)
|
||
tc = msg.get("tool_calls")
|
||
if tc:
|
||
accumulated_tool_calls.extend(tc)
|
||
|
||
if data.get("done"):
|
||
if accumulated_tool_calls:
|
||
logger.info(
|
||
"Ollama returned %d tool call(s): %s",
|
||
len(accumulated_tool_calls),
|
||
json.dumps(accumulated_tool_calls)[:500],
|
||
)
|
||
yield ChatChunk(type="tool_calls", tool_calls=accumulated_tool_calls)
|
||
else:
|
||
logger.debug("Ollama done with no tool calls")
|
||
yield ChatChunk(
|
||
type="done",
|
||
prompt_tokens=data.get("prompt_eval_count"),
|
||
output_tokens=data.get("eval_count"),
|
||
)
|
||
break
|
||
|
||
|
||
async def generate_completion(
|
||
messages: list[dict],
|
||
model: str,
|
||
max_tokens: int = 4096,
|
||
num_ctx: int | None = None,
|
||
) -> str:
|
||
"""Non-streaming chat completion, returns full response text.
|
||
|
||
Retries up to 2 times on Ollama 500 errors (cold model loading race).
|
||
num_ctx overrides the model's context window for this call only.
|
||
"""
|
||
last_exc: Exception | None = None
|
||
options: dict = {"num_predict": max_tokens}
|
||
if num_ctx is not None:
|
||
options["num_ctx"] = num_ctx
|
||
for attempt in range(3):
|
||
if attempt > 0:
|
||
delay = 3.0 * attempt
|
||
logger.warning(
|
||
"generate_completion 500 (attempt %d/3), retrying in %.0fs", attempt, delay
|
||
)
|
||
await asyncio.sleep(delay)
|
||
try:
|
||
async with httpx.AsyncClient(timeout=httpx.Timeout(1800.0, connect=30.0, read=300.0)) as client:
|
||
resp = await client.post(
|
||
f"{Config.OLLAMA_URL}/api/chat",
|
||
json={
|
||
"model": model,
|
||
"messages": messages,
|
||
"stream": False,
|
||
"think": False,
|
||
"options": options,
|
||
"keep_alive": "2h",
|
||
},
|
||
)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
return data.get("message", {}).get("content", "")
|
||
except httpx.HTTPStatusError as exc:
|
||
last_exc = exc
|
||
if exc.response.status_code != 500:
|
||
break
|
||
except Exception as exc:
|
||
last_exc = exc
|
||
break
|
||
raise last_exc
|
||
|
||
|
||
def _is_private_url(url: str) -> bool:
|
||
"""Return True if the URL resolves to a private/loopback/link-local address (SSRF guard)."""
|
||
try:
|
||
host = urlparse(url).hostname
|
||
if not host:
|
||
return True
|
||
if host.lower() in ("localhost", "::1"):
|
||
return True
|
||
addr_info = socket.getaddrinfo(host, None, proto=socket.IPPROTO_TCP)
|
||
for entry in addr_info:
|
||
ip = ipaddress.ip_address(entry[4][0])
|
||
if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_multicast:
|
||
return True
|
||
return False
|
||
except Exception:
|
||
return True # Block on resolution failure
|
||
|
||
|
||
async def fetch_url_content(url: str) -> str:
|
||
"""Fetch a URL and return text content (HTML tags stripped)."""
|
||
if _is_private_url(url):
|
||
logger.warning("Blocked fetch of private/internal URL: %s", url)
|
||
return "[URL blocked: internal network access not permitted]"
|
||
try:
|
||
async with httpx.AsyncClient(
|
||
timeout=15.0, follow_redirects=False, headers={"User-Agent": "FabledAssistant/1.0"}
|
||
) as client:
|
||
resp = await client.get(url)
|
||
resp.raise_for_status()
|
||
text = resp.text
|
||
# Strip HTML tags
|
||
text = re.sub(r"<script[^>]*>.*?</script>", "", text, flags=re.DOTALL)
|
||
text = re.sub(r"<style[^>]*>.*?</style>", "", text, flags=re.DOTALL)
|
||
text = re.sub(r"<[^>]+>", " ", text)
|
||
# Collapse whitespace
|
||
text = re.sub(r"\s+", " ", text).strip()
|
||
# Truncate to reasonable size
|
||
if len(text) > 4000:
|
||
text = text[:4000] + "..."
|
||
return text
|
||
except Exception as e:
|
||
logger.warning("Failed to fetch URL %s: %s", url, e)
|
||
return f"[Failed to fetch URL: {url}]"
|
||
|
||
|
||
def _extract_keywords(text: str) -> list[str]:
|
||
"""Extract meaningful keywords from text for note search."""
|
||
words = re.findall(r"\b[a-zA-Z]{3,}\b", text.lower())
|
||
keywords = [w for w in words if w not in STOP_WORDS]
|
||
# Deduplicate while preserving order
|
||
seen: set[str] = set()
|
||
unique = []
|
||
for w in keywords:
|
||
if w not in seen:
|
||
seen.add(w)
|
||
unique.append(w)
|
||
return unique[:5]
|
||
|
||
|
||
def _find_urls(text: str) -> list[str]:
|
||
"""Find URLs in text."""
|
||
return re.findall(r"https?://[^\s<>\"')\]]+", text)
|
||
|
||
|
||
# History summarization thresholds
|
||
_HISTORY_SUMMARY_THRESHOLD = 30 # total messages before summarizing
|
||
_HISTORY_KEEP_RECENT = 8 # verbatim tail to preserve (4 exchanges)
|
||
|
||
|
||
async def summarize_history_for_context(
|
||
history: list[dict],
|
||
model: str,
|
||
) -> tuple[list[dict], str | None]:
|
||
"""Summarize old conversation history when it exceeds the threshold.
|
||
|
||
Returns (recent_history, summary_text | None).
|
||
recent_history is the verbatim tail passed to the model.
|
||
summary_text (when not None) should be injected into the system prompt
|
||
so the model retains the gist of earlier exchanges without the full tokens.
|
||
For short conversations, returns (history, None) immediately with no LLM call.
|
||
"""
|
||
if len(history) <= _HISTORY_SUMMARY_THRESHOLD:
|
||
return history, None
|
||
|
||
to_summarize = history[:-_HISTORY_KEEP_RECENT]
|
||
recent = history[-_HISTORY_KEEP_RECENT:]
|
||
|
||
# Two-pass for very long histories: summarize first half, combine with second half
|
||
if len(to_summarize) > 50:
|
||
mid = len(to_summarize) // 2
|
||
first_half = to_summarize[:mid]
|
||
second_half = to_summarize[mid:]
|
||
|
||
# Summarize first half
|
||
first_lines = []
|
||
for m in first_half:
|
||
role = m.get("role", "")
|
||
content = (m.get("content") or "").strip()
|
||
if role in ("user", "assistant") and content:
|
||
label = "User" if role == "user" else "Assistant"
|
||
first_lines.append(f"{label}: {content[:400]}")
|
||
|
||
if first_lines:
|
||
try:
|
||
first_summary_messages = [
|
||
{"role": "system", "content": "Summarize this conversation in 3-4 sentences covering topics, notes/tasks created, and key decisions."},
|
||
{"role": "user", "content": "\n".join(first_lines)},
|
||
]
|
||
summary_a = await generate_completion(first_summary_messages, model, max_tokens=300)
|
||
summary_a = summary_a.strip()
|
||
except Exception:
|
||
summary_a = ""
|
||
else:
|
||
summary_a = ""
|
||
|
||
# Build lines for final pass from second half
|
||
second_lines = []
|
||
for m in second_half:
|
||
role = m.get("role", "")
|
||
content = (m.get("content") or "").strip()
|
||
if role in ("user", "assistant") and content:
|
||
label = "User" if role == "user" else "Assistant"
|
||
second_lines.append(f"{label}: {content[:400]}")
|
||
|
||
if summary_a:
|
||
lines = [f"[Earlier summary: {summary_a}]"] + second_lines
|
||
else:
|
||
lines = second_lines
|
||
else:
|
||
lines: list[str] = []
|
||
for m in to_summarize:
|
||
role = m.get("role", "")
|
||
content = (m.get("content") or "").strip()
|
||
if role in ("user", "assistant") and content:
|
||
label = "User" if role == "user" else "Assistant"
|
||
lines.append(f"{label}: {content[:400]}")
|
||
|
||
if not lines:
|
||
return history, None
|
||
|
||
prompt_messages = [
|
||
{
|
||
"role": "system",
|
||
"content": (
|
||
"Summarize this conversation history. Capture: "
|
||
"(1) All notes, tasks, and projects created or modified — include their exact names. "
|
||
"(2) Key decisions made and conclusions reached. "
|
||
"(3) Open questions and next steps mentioned. "
|
||
"(4) The overall topic arc so the conversation can continue naturally. "
|
||
"Be specific and factual. Output 4-8 concise sentences. Nothing else."
|
||
),
|
||
},
|
||
{"role": "user", "content": "\n".join(lines)},
|
||
]
|
||
|
||
try:
|
||
summary = await generate_completion(prompt_messages, model, max_tokens=400)
|
||
summary = summary.strip()
|
||
if summary:
|
||
logger.info(
|
||
"Summarized %d history messages (%d chars) for context",
|
||
len(to_summarize), len(summary),
|
||
)
|
||
return recent, summary
|
||
except Exception:
|
||
logger.warning("Failed to summarize conversation history", exc_info=True)
|
||
|
||
return history, None
|
||
|
||
|
||
async def build_context(
|
||
user_id: int,
|
||
history: list[dict],
|
||
current_note_id: int | None,
|
||
user_message: str,
|
||
exclude_note_ids: list[int] | None = None,
|
||
history_summary: str | None = None,
|
||
include_note_ids: list[int] | None = None,
|
||
excluded_note_ids: list[int] | None = None,
|
||
rag_project_id: int | None = None,
|
||
workspace_project_id: int | None = None,
|
||
user_timezone: str | None = None,
|
||
conv_id: int | None = None,
|
||
voice_mode: bool = False,
|
||
voice_speech_style: str = "conversational",
|
||
) -> tuple[list[dict], dict]:
|
||
"""Build messages array for Ollama with system prompt and context.
|
||
|
||
Returns (messages, context_meta) where context_meta contains info about
|
||
which notes were included as context.
|
||
"""
|
||
exclude_set = set(exclude_note_ids or [])
|
||
from datetime import date as date_type
|
||
assistant_name = await get_setting(user_id, "assistant_name", "Fable")
|
||
today = date_type.today().isoformat()
|
||
has_caldav = await is_caldav_configured(user_id)
|
||
|
||
# Build tool usage guidance based on available integrations
|
||
# --- Static block (Ollama KV-cache prefix) ---
|
||
# Everything here must be byte-for-byte identical across requests for the same
|
||
# user so Ollama can reuse the cached KV state. No dates, timezones, RAG notes,
|
||
# or user-profile data here — those go in the dynamic tail below.
|
||
tool_lines = [
|
||
"You have access to tool functions. You MUST use them when the user asks you to create, add, find, schedule, or search for anything.",
|
||
"CRITICAL: Call the tool functions directly. NEVER write out function calls as text or code. NEVER describe what you would do — just do it.",
|
||
"Available actions: create_task, create_note, update_note, delete_note, delete_task, get_note, list_notes, list_tasks, search_notes.",
|
||
]
|
||
if has_caldav:
|
||
tool_lines[-1] = (
|
||
"Available actions: create_task, create_note, update_note, delete_note, delete_task, get_note, list_notes, list_tasks, search_notes, "
|
||
"create_event, list_events, search_events, update_event, delete_event, list_calendars."
|
||
)
|
||
tool_lines.append(
|
||
"For calendar events, use ISO 8601 datetime format with the user's timezone offset (stated in context below). "
|
||
"Always include the UTC offset in datetime strings (e.g. 2026-09-30T14:00:00+01:00)."
|
||
)
|
||
tool_lines.append("When the user says 'remind me' with a time before an event, use the reminder_minutes parameter.")
|
||
tool_lines.append(
|
||
"For relative dates like 'Friday' or 'next week', resolve them to YYYY-MM-DD format. "
|
||
"Always include the UTC offset when creating events (user's timezone is stated in context below)."
|
||
)
|
||
tool_lines.append("When creating notes, use the `tags` parameter — do not embed #tag text in the note body.")
|
||
tool_lines.append(
|
||
"When search_images returns results, embed each image directly in your response by writing "
|
||
"the 'embed' field verbatim (e.g. ), then the 'citation' field on the "
|
||
"next line. Never describe images as text or list their URLs — always render them as markdown images."
|
||
)
|
||
tool_lines.append(
|
||
"Use update_note for existing notes/tasks; use create_note only for new content. "
|
||
"Use search_notes for semantic/conceptual queries. "
|
||
"Delete tools require an explicit user request. "
|
||
"Never proactively search notes or comment on absent context."
|
||
)
|
||
tool_guidance = "\n".join(tool_lines)
|
||
|
||
static_block = (
|
||
f"You are a helpful assistant named {assistant_name}, integrated into a note-taking and task-tracking app called Fabled Assistant. "
|
||
"Help users with their notes, tasks, and general questions. "
|
||
"When note context is provided, use it to give relevant answers.\n\n"
|
||
f"{tool_guidance}"
|
||
)
|
||
|
||
# --- Dynamic tail (appended after static block, evaluated every request) ---
|
||
# Date, timezone, user profile, and entities change per-day or per-user.
|
||
# Keeping these at the end preserves the static prefix for KV-cache reuse.
|
||
tz_line = f" The user's timezone is {user_timezone}." if user_timezone else ""
|
||
|
||
from fabledassistant.services.user_profile import build_profile_context
|
||
from fabledassistant.services.knowledge import get_people_and_places_context
|
||
profile_context = await build_profile_context(user_id)
|
||
profile_section = f"\n\n{profile_context}" if profile_context else ""
|
||
entities_context = await get_people_and_places_context(user_id)
|
||
entities_section = f"\n\n{entities_context}" if entities_context else ""
|
||
|
||
dynamic_tail = f"\n\nToday's date is {today}.{tz_line}{profile_section}{entities_section}"
|
||
|
||
# --- System message: stable content only ---
|
||
# Workspace context and history summary stay here because they carry
|
||
# behavioural instructions / conversational state, not retrieved content.
|
||
# Everything retrieval-based (RAG notes, RSS, URL content, current note,
|
||
# briefing articles) goes into the user turn below so the system message
|
||
# prefix stays byte-for-byte identical across requests, enabling Ollama's
|
||
# KV prefix cache to fire reliably.
|
||
|
||
if voice_mode:
|
||
_style_hints = {
|
||
"conversational": "Be warm, natural, and conversational — like speaking to a friend.",
|
||
"concise": "Be brief and to the point. One or two sentences maximum unless detail is essential.",
|
||
"detailed": "Give thorough, informative responses as if narrating an explanation aloud.",
|
||
}
|
||
style_hint = _style_hints.get(voice_speech_style, _style_hints["conversational"])
|
||
voice_preamble = (
|
||
"VOICE MODE: Respond naturally as if speaking aloud. "
|
||
"No markdown, bullet points, headers, or code blocks. Complete sentences only. "
|
||
f"{style_hint}\n\n"
|
||
)
|
||
system_content = voice_preamble + static_block + dynamic_tail
|
||
else:
|
||
system_content = static_block + dynamic_tail
|
||
|
||
# Inject workspace context (behavioural — must stay in system)
|
||
if workspace_project_id is not None:
|
||
from fabledassistant.services.projects import get_project
|
||
try:
|
||
wp = await get_project(user_id, workspace_project_id)
|
||
if wp:
|
||
system_content += (
|
||
f"\n\n--- Active Workspace ---\n"
|
||
f"You are in the \"{wp.title}\" project workspace.\n"
|
||
f"All notes and tasks you create or update MUST belong to this project.\n"
|
||
f"Always pass project=\"{wp.title}\" when calling create_note or create_task.\n"
|
||
f"--- End Active Workspace ---"
|
||
)
|
||
except Exception:
|
||
logger.warning("Failed to fetch workspace project %d", workspace_project_id)
|
||
|
||
# Inject compressed history summary (conversational state — stays in system)
|
||
if history_summary:
|
||
system_content += (
|
||
f"\n\n--- Earlier Conversation ---\n{history_summary}\n--- End Earlier Conversation ---"
|
||
)
|
||
|
||
# Detect briefing conversation — used for both system prompt instruction and article injection
|
||
_is_briefing_conv = False
|
||
if conv_id is not None:
|
||
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)
|
||
if _conv and getattr(_conv, "conversation_type", None) == "briefing":
|
||
_is_briefing_conv = True
|
||
|
||
if _is_briefing_conv:
|
||
system_content += (
|
||
"\n\nYou are in a briefing conversation. "
|
||
"The conversation history contains today's briefing — news stories, weather, and tasks. "
|
||
"When the user asks about a topic, person, or event from the briefing, answer directly "
|
||
"from the conversation history and the article context that follows. "
|
||
"Do NOT search the web for information that is already present in the briefing."
|
||
)
|
||
|
||
context_meta: dict = {
|
||
"context_note_id": None,
|
||
"context_note_title": None,
|
||
"auto_notes": [],
|
||
"auto_injected_notes": [],
|
||
}
|
||
|
||
# --- User turn context prefix: retrieval-based content ---
|
||
# Collected here and prepended to the user message so the system message
|
||
# stays stable and the KV prefix cache can fire on every request.
|
||
user_context_parts: list[str] = []
|
||
|
||
# Current note being viewed (full body, no truncation)
|
||
if current_note_id:
|
||
note = await get_note(user_id, current_note_id)
|
||
if note:
|
||
context_meta["context_note_id"] = note.id
|
||
context_meta["context_note_title"] = note.title
|
||
user_context_parts.append(
|
||
f"--- Current Note ---\n"
|
||
f"Title: {note.title}\n"
|
||
f"Content:\n{note.body}\n"
|
||
f"--- End Note ---"
|
||
)
|
||
|
||
# Semantic / keyword note search
|
||
search_exclude = set(exclude_set)
|
||
if current_note_id:
|
||
search_exclude.add(current_note_id)
|
||
|
||
found_scored: list[tuple[float | None, object]] = []
|
||
|
||
orphan_only = rag_project_id is None
|
||
effective_project_id = rag_project_id if (rag_project_id is not None and rag_project_id != -1) else None
|
||
|
||
try:
|
||
from fabledassistant.services.embeddings import semantic_search_notes
|
||
for score, note in await semantic_search_notes(
|
||
user_id, user_message, exclude_ids=search_exclude or None, limit=8,
|
||
project_id=effective_project_id,
|
||
orphan_only=orphan_only,
|
||
):
|
||
found_scored.append((score, note))
|
||
except Exception:
|
||
logger.warning("Semantic note search failed, falling back to keyword search", exc_info=True)
|
||
|
||
if not found_scored:
|
||
keywords = _extract_keywords(user_message)
|
||
if keywords:
|
||
try:
|
||
for note in await search_notes_for_context(
|
||
user_id, keywords, exclude_ids=search_exclude or None, limit=8,
|
||
project_id=effective_project_id,
|
||
orphan_only=orphan_only,
|
||
):
|
||
found_scored.append((None, note))
|
||
except Exception:
|
||
logger.warning("Failed to search notes for context", exc_info=True)
|
||
|
||
excluded_inject_set = set(excluded_note_ids or [])
|
||
auto_inject: list[tuple[float, object]] = []
|
||
sidebar_only: list[tuple[float | None, object]] = []
|
||
|
||
for score, n in found_scored:
|
||
if (
|
||
score is not None
|
||
and score >= RAG_AUTO_THRESHOLD
|
||
and len(auto_inject) < RAG_AUTO_LIMIT
|
||
and n.id not in excluded_inject_set
|
||
):
|
||
auto_inject.append((score, n))
|
||
else:
|
||
sidebar_only.append((score, n))
|
||
|
||
if auto_inject:
|
||
snippets = []
|
||
for score, n in auto_inject:
|
||
body_snippet = (n.body or "")[:RAG_AUTO_SNIPPET]
|
||
snippets.append(f"**{n.title}** (relevance: {round(score * 100)}%)\n{body_snippet}")
|
||
context_meta["auto_injected_notes"].append({
|
||
"id": n.id,
|
||
"title": n.title,
|
||
"score": round(score, 2),
|
||
})
|
||
user_context_parts.append(
|
||
"--- Relevant Notes ---\n"
|
||
+ "\n\n".join(snippets)
|
||
+ "\n--- End Relevant Notes ---"
|
||
)
|
||
|
||
for score, n in auto_inject:
|
||
context_meta["auto_notes"].append({
|
||
"id": n.id,
|
||
"title": n.title,
|
||
"score": round(score, 2) if score is not None else None,
|
||
"auto_injected": True,
|
||
})
|
||
for score, n in sidebar_only:
|
||
context_meta["auto_notes"].append({
|
||
"id": n.id,
|
||
"title": n.title,
|
||
"score": round(score, 2) if score is not None else None,
|
||
"auto_injected": False,
|
||
})
|
||
context_meta["auto_note_ids"] = [n.id for _, n in found_scored]
|
||
|
||
# Explicitly included notes (user opted in via sidebar)
|
||
if include_note_ids:
|
||
from fabledassistant.services.notes import get_note as _get_note
|
||
included_snippets: list[str] = []
|
||
for nid in include_note_ids:
|
||
try:
|
||
n = await _get_note(user_id, nid)
|
||
if n:
|
||
body_preview = n.body or ""
|
||
included_snippets.append(f"- {n.title}: {body_preview}")
|
||
except Exception:
|
||
logger.warning("Failed to load included note %d for context", nid, exc_info=True)
|
||
if included_snippets:
|
||
user_context_parts.append(
|
||
"--- Included Notes ---\n"
|
||
+ "\n".join(included_snippets)
|
||
+ "\n--- End Included Notes ---"
|
||
)
|
||
|
||
# Semantically relevant RSS news items
|
||
try:
|
||
from fabledassistant.services.embeddings import get_embedding, semantic_search_rss_items
|
||
news_query_vec = await get_embedding(user_message)
|
||
news_hits = await semantic_search_rss_items(user_id, news_query_vec)
|
||
if news_hits:
|
||
news_snippets = []
|
||
for score, rss_item in news_hits:
|
||
feed_title = getattr(rss_item, "feed_title", "") or ""
|
||
excerpt = (rss_item.content or "")[:500].strip()
|
||
news_snippets.append(
|
||
f"[{feed_title or 'News'}] {rss_item.title} (relevance: {round(score * 100)}%)\n"
|
||
+ (f"{excerpt}\n" if excerpt else "")
|
||
+ f"URL: {rss_item.url}"
|
||
)
|
||
user_context_parts.append(
|
||
"--- Recent News You've Seen ---\n"
|
||
+ "\n\n".join(news_snippets)
|
||
+ "\n--- End Recent News ---"
|
||
)
|
||
context_meta["rss_news"] = [
|
||
{"id": item.id, "title": item.title, "score": round(score, 2)}
|
||
for score, item in news_hits
|
||
]
|
||
except Exception:
|
||
logger.debug("RSS semantic search skipped", exc_info=True)
|
||
|
||
# URL content fetched from links in the user message
|
||
urls = _find_urls(user_message)
|
||
for url in urls[:2]:
|
||
content = await fetch_url_content(url)
|
||
if content and not content.startswith("[Failed"):
|
||
user_context_parts.append(
|
||
f"--- Content from {url} ---\n{content}\n--- End URL Content ---"
|
||
)
|
||
|
||
# Briefing article context for follow-up Q&A
|
||
if _is_briefing_conv:
|
||
article_context = await _build_briefing_article_context(conv_id) # type: ignore[arg-type]
|
||
if article_context:
|
||
user_context_parts.append(article_context.strip())
|
||
|
||
# Build final user message — context prefix (if any) followed by the actual message
|
||
if user_context_parts:
|
||
user_turn = "\n\n".join(user_context_parts) + "\n\n" + user_message
|
||
else:
|
||
user_turn = user_message
|
||
|
||
messages = [{"role": "system", "content": system_content}]
|
||
messages.extend(history)
|
||
messages.append({"role": "user", "content": user_turn})
|
||
return messages, context_meta
|
||
|
||
|
||
async def _build_briefing_article_context(conv_id: int) -> str:
|
||
"""Fetch article content from today's briefing message and return a
|
||
formatted context block for injection into the system prompt.
|
||
|
||
Looks at the most recent assistant briefing messages for rss_item_ids
|
||
in their metadata, then loads those items from the DB.
|
||
Capped at 10 articles × 500 chars to keep token use reasonable.
|
||
"""
|
||
import json as _json
|
||
|
||
from sqlalchemy import select, text as _text
|
||
|
||
from fabledassistant.models import async_session as _async_session
|
||
from fabledassistant.models.conversation import Message
|
||
|
||
async with _async_session() as session:
|
||
result = await session.execute(
|
||
select(Message)
|
||
.where(
|
||
Message.conversation_id == conv_id,
|
||
Message.role == "assistant",
|
||
)
|
||
.order_by(Message.created_at.desc())
|
||
.limit(10)
|
||
)
|
||
messages = result.scalars().all()
|
||
|
||
rss_item_ids: list[int] = []
|
||
for msg in messages:
|
||
meta = msg.msg_metadata or {}
|
||
if isinstance(meta, str):
|
||
try:
|
||
meta = _json.loads(meta)
|
||
except Exception:
|
||
continue
|
||
ids = meta.get("rss_item_ids") or []
|
||
if ids:
|
||
rss_item_ids = ids
|
||
break
|
||
|
||
if not rss_item_ids:
|
||
return ""
|
||
|
||
async with _async_session() as session:
|
||
result = await session.execute(
|
||
_text("""
|
||
SELECT i.title, i.url, i.content, f.title AS feed_title
|
||
FROM rss_items i
|
||
JOIN rss_feeds f ON f.id = i.feed_id
|
||
WHERE i.id = ANY(:ids)
|
||
ORDER BY i.published_at DESC NULLS LAST
|
||
LIMIT 10
|
||
""").bindparams(ids=rss_item_ids[:10])
|
||
)
|
||
rows = result.mappings().all()
|
||
|
||
if not rows:
|
||
return ""
|
||
|
||
lines = ["\n\nARTICLE CONTEXT (source articles from today's briefing):"]
|
||
for row in rows:
|
||
lines.append(f"\n[{row['feed_title']}] {row['title']}")
|
||
if row["url"]:
|
||
lines.append(f"URL: {row['url']}")
|
||
if row["content"]:
|
||
lines.append(row["content"][:500])
|
||
return "\n".join(lines)
|