bvandeusen
ac188c40a5
- services/tools/journal.py — record_moment + search_journal tool handlers - services/tools/_registry.py: add `journal` flag on ToolDef + tool() decorator - get_tools_for_user(user_id, conversation_type='chat'|'journal') — exclude journal-only tools from chat sessions; exclude set_rag_scope from journal sessions - services/tools/__init__.py: register the new journal module; drop the unused get_briefing_tools export - services/llm.py build_context: short-circuit for journal conversations, using journal_pipeline.build_journal_system_prompt and skipping all notes-RAG injection (preserves the journal/notes isolation invariant) - services/generation_task.py: pass conversation_type into get_tools_for_user Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
978 lines
41 KiB
Python
978 lines
41 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 keep_alive_for(model: str) -> str:
|
||
"""Return the Ollama keep_alive duration for *model*.
|
||
|
||
Background models get a shorter window because they're called
|
||
sporadically; the main interactive model gets a longer one so it
|
||
stays warm between user messages.
|
||
"""
|
||
if model == Config.OLLAMA_BACKGROUND_MODEL:
|
||
return Config.OLLAMA_KEEP_ALIVE_BACKGROUND
|
||
return Config.OLLAMA_KEEP_ALIVE_MAIN
|
||
|
||
|
||
def pick_num_ctx(messages: list[dict], tools: list[dict] | None = None) -> int:
|
||
"""Return the smallest context tier that fits *messages* + *tools* with 25% headroom.
|
||
|
||
The ``tools`` JSON schemas are a large, often-overlooked chunk of the prompt.
|
||
With ~40 tools in the registry the schemas alone can be 6-10K tokens — enough
|
||
that omitting them from the estimate causes silent prompt truncation.
|
||
|
||
Stays at or below Config.OLLAMA_NUM_CTX (the configured ceiling).
|
||
"""
|
||
total_chars = sum(len(m.get("content") or "") for m in messages)
|
||
if tools:
|
||
# Serialize the same way Ollama will see them. json.dumps gives us a
|
||
# faithful char count for the schema payload without any guesswork.
|
||
total_chars += len(json.dumps(tools))
|
||
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).
|
||
|
||
Tags are normalized to lowercase. Ollama stores whatever casing was used at
|
||
pull time (``ollama pull gemma3:12B`` keeps the ``B``), but inference calls
|
||
against the capitalized tag can 400 — the two code paths are inconsistent.
|
||
Lowercasing on read avoids exposing that mixed-case name in the UI and
|
||
keeps the validation check below from accepting a form Ollama will reject.
|
||
"""
|
||
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"].lower()
|
||
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."""
|
||
# Match the lowercase normalization in get_installed_models so a legacy
|
||
# mixed-case setting doesn't force a spurious re-pull at startup.
|
||
model = model.lower()
|
||
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 _raise_ollama_error(resp: httpx.Response, model: str) -> None:
|
||
"""On a non-2xx Ollama response, log its body before raising.
|
||
|
||
``resp.raise_for_status()`` alone throws away Ollama's error body, which
|
||
carries the actual reason (e.g. ``"model does not support tools"``,
|
||
``"context length exceeded"``). Reading the body first means failures
|
||
surface a usable message instead of a bare HTTPStatusError.
|
||
"""
|
||
if resp.status_code < 400:
|
||
return
|
||
body = (await resp.aread()).decode("utf-8", errors="replace").strip()
|
||
logger.error(
|
||
"Ollama /api/chat %d for model=%s: %s",
|
||
resp.status_code, model, body[:500],
|
||
)
|
||
resp.raise_for_status()
|
||
|
||
|
||
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": keep_alive_for(model)}
|
||
# 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:
|
||
await _raise_ollama_error(resp, model)
|
||
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": keep_alive_for(model),
|
||
}
|
||
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:
|
||
await _raise_ollama_error(resp, model)
|
||
accumulated_tool_calls: list[dict] = []
|
||
# Silent-generation diagnostic: if Ollama reports non-zero eval_count
|
||
# but we never yielded any thinking/content/tool_calls, something
|
||
# in the frames isn't landing in a field we read. Capture the last
|
||
# few frames so we can see what Ollama actually sent.
|
||
yielded_anything = False
|
||
recent_frames: list[str] = []
|
||
async for line in resp.aiter_lines():
|
||
if not line.strip():
|
||
continue
|
||
if len(recent_frames) >= 5:
|
||
recent_frames.pop(0)
|
||
recent_frames.append(line[:500])
|
||
data = json.loads(line)
|
||
msg = data.get("message", {})
|
||
|
||
# Thinking chunks (qwen3 chain-of-thought, only when think=True)
|
||
thinking = msg.get("thinking", "")
|
||
if thinking:
|
||
yielded_anything = True
|
||
yield ChatChunk(type="thinking", content=thinking)
|
||
|
||
# Content chunks
|
||
chunk = msg.get("content", "")
|
||
if chunk:
|
||
yielded_anything = True
|
||
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],
|
||
)
|
||
yielded_anything = True
|
||
yield ChatChunk(type="tool_calls", tool_calls=accumulated_tool_calls)
|
||
else:
|
||
logger.debug("Ollama done with no tool calls")
|
||
eval_count = data.get("eval_count") or 0
|
||
if not yielded_anything and eval_count > 0:
|
||
logger.warning(
|
||
"Ollama silent generation: model=%s eval_count=%d but no "
|
||
"thinking/content/tool_calls were yielded. Last frames: %s",
|
||
model, eval_count, recent_frames,
|
||
)
|
||
yield ChatChunk(
|
||
type="done",
|
||
prompt_tokens=data.get("prompt_eval_count"),
|
||
output_tokens=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
|
||
# Default num_ctx to Config.OLLAMA_NUM_CTX (matching stream_chat /
|
||
# stream_chat_with_tools). Without this, Ollama silently uses the model's
|
||
# default window (~4k on qwen3) and truncates anything longer. That is
|
||
# how the research pipeline's outline step kept falling back to a single
|
||
# monolith note: its 12-source prompt is ~6k tokens and was being chopped
|
||
# before the model ever saw it. Non-streaming callers must not inherit
|
||
# that footgun — if you truly want the model default, pass num_ctx=0.
|
||
options: dict = {
|
||
"num_predict": max_tokens,
|
||
"num_ctx": num_ctx if num_ctx is not None else Config.OLLAMA_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": keep_alive_for(model),
|
||
},
|
||
)
|
||
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
|
||
|
||
# --- Journal short-circuit ---
|
||
# Journal conversations get a different persona, calibration, and an
|
||
# ambient-moments context block. CRUCIALLY, no notes-RAG injection here
|
||
# (preserves the notes/journal isolation invariant).
|
||
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) == "journal":
|
||
from fabledassistant.services.journal_pipeline import build_journal_system_prompt
|
||
day_date = _conv.day_date or date_type.today()
|
||
system_content = await build_journal_system_prompt(
|
||
user_id=user_id,
|
||
day_date=day_date,
|
||
user_timezone=user_timezone or "UTC",
|
||
)
|
||
messages_out: list[dict] = [{"role": "system", "content": system_content}]
|
||
messages_out.extend(history)
|
||
messages_out.append({"role": "user", "content": user_message})
|
||
return messages_out, {
|
||
"context_note_id": None,
|
||
"context_note_title": None,
|
||
"auto_notes": [],
|
||
"auto_injected_notes": [],
|
||
}
|
||
|
||
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.",
|
||
"GROUNDING: When the user asks about their own data — tasks, notes, events, projects, news, anything stored in this system — call the relevant tool to see what actually exists before answering. Never assert facts about the user's data from memory, prior context, or assumption. If you are unsure whether something exists, check with a tool.",
|
||
"HONESTY WHEN EMPTY: If a tool returns empty results (no matching tasks, no events in the date range, no search hits, no notes found), tell the user plainly that nothing matched. Do not fabricate example items, do not invent plausible-sounding meetings or deadlines to fill the response, and do not hedge with generic suggestions dressed up as real data. A direct 'you don't have anything on your calendar today' is always better than an invented event.",
|
||
]
|
||
actions = [
|
||
"create_note (also creates tasks — set status='todo')", "update_note", "delete_note",
|
||
"read_note", "list_notes", "list_tasks", "log_work", "search_notes",
|
||
"create_project", "list_projects", "get_project", "update_project",
|
||
"search_projects", "create_milestone", "update_milestone", "list_milestones",
|
||
"save_person", "save_place", "create_list", "add_to_list", "clear_checked_items",
|
||
"set_rag_scope", "get_profile", "update_profile", "get_weather", "calculate",
|
||
"get_rss_items", "add_rss_feed", "read_article",
|
||
]
|
||
if has_caldav:
|
||
actions.extend(["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.")
|
||
actions.append("lookup")
|
||
if Config.searxng_enabled():
|
||
actions.extend(["research_topic", "search_images"])
|
||
tool_lines.append(f"Available actions: {', '.join(actions)}.")
|
||
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_lines.append(
|
||
"IMPORTANT: When creating tasks or notes, NEVER infer or guess a project name. "
|
||
"Only set the project parameter if the user explicitly names a project. "
|
||
"If the user says 'create a task to buy milk', do NOT assign it to a project."
|
||
)
|
||
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 Scribe. "
|
||
"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.\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
|
||
|
||
# Skip RAG auto-injection on the first turn of a seeded article discussion.
|
||
# The article body is already the sole context the user wants — pulling in
|
||
# unrelated orphan notes tricks the model into summarizing those instead.
|
||
# Follow-up turns keep RAG on because by then the user's own messages drive
|
||
# the query rather than the generic seed prompt.
|
||
from fabledassistant.services.article_context import ARTICLE_DISCUSS_SEED
|
||
_skip_rag_for_article_seed = user_message.strip() == ARTICLE_DISCUSS_SEED
|
||
|
||
if not _skip_rag_for_article_seed:
|
||
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 and not _skip_rag_for_article_seed:
|
||
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(
|
||
"[The following are reference excerpts from the user's personal notes, "
|
||
"not part of their message. Use them only if relevant to answering.]\n"
|
||
"--- 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 ---"
|
||
)
|
||
|
||
# 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)
|