bvandeusen
9f6608fc8f
Three briefing quality fixes surfaced by reading today's 2026-04-11 compilation output: - **Stale weather**: get_weather was returning 48h-old cache data after a missed scheduler run. Tool now auto-refreshes any cached location older than 6h (fetching fresh data from Open-Meteo), and stamps each location with cache_age_hours + is_stale so the model can hedge instead of faithfully relaying old numbers. - **Cancelled tasks leaking into prose**: briefing loop now defaults list_tasks calls to status=["todo","in_progress"] when the model doesn't specify, so cancelled/done items stop showing up in the summary. Localized to the briefing path — chat still sees full history. - **Overdue in-progress tasks missed by midday check-in**: tightened the check-in prompt to explicitly require two list_tasks calls — one for in_progress (catches items dragging past their due date) and one filtered by due date — so long-running tasks stop getting silently dropped.
418 lines
17 KiB
Python
418 lines
17 KiB
Python
"""
|
|
Briefing pipeline: agentic tool-use loop + UI metadata gather.
|
|
|
|
Slot names: 'compilation' (4am), 'morning' (8am), 'midday' (12pm), 'afternoon' (4pm)
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
|
|
|
from fabledassistant.config import Config
|
|
from fabledassistant.services.settings import get_setting
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SLOT_NAMES = ("compilation", "morning", "midday", "afternoon")
|
|
|
|
|
|
# ── External data gather ──────────────────────────────────────────────────────
|
|
|
|
async def _gather_external(user_id: int) -> dict:
|
|
"""Collect RSS items and weather."""
|
|
from fabledassistant.services.rss import get_recent_items
|
|
from fabledassistant.services.weather import get_cached_weather
|
|
|
|
rss_items, weather = await asyncio.gather(
|
|
get_recent_items(user_id, limit=20),
|
|
get_cached_weather(user_id),
|
|
return_exceptions=True,
|
|
)
|
|
return {
|
|
"rss_items": rss_items if not isinstance(rss_items, Exception) else [],
|
|
"weather": weather if not isinstance(weather, Exception) else [],
|
|
}
|
|
|
|
|
|
# ── Agentic briefing (tool-use loop) ──────────────────────────────────────────
|
|
|
|
_BRIEFING_AGENT_MAX_ROUNDS = 8
|
|
_BRIEFING_AGENT_NUM_CTX = 8192
|
|
|
|
|
|
def _agentic_system_prompt(
|
|
profile_body: str,
|
|
slot: str,
|
|
today_iso: str,
|
|
tz_name: str,
|
|
day_from_iso: str,
|
|
day_to_iso: str,
|
|
) -> str:
|
|
"""System prompt for the agentic briefing path.
|
|
|
|
Pushes the model to ground every factual claim in a tool result and
|
|
to be honest when tools return nothing, rather than fabricating
|
|
content to fill the narrative.
|
|
|
|
Pre-computes today's window in the user's local timezone so the
|
|
model can call date-sensitive tools (list_events, list_tasks filters)
|
|
without having to do any timezone math itself — eliminating a whole
|
|
class of "wrong day" bugs.
|
|
"""
|
|
tz_block = (
|
|
f"Today is {today_iso} ({tz_name}). "
|
|
f"When calling list_events for today, use:\n"
|
|
f" date_from = {day_from_iso}\n"
|
|
f" date_to = {day_to_iso}\n"
|
|
f"These are already the correct local-day boundaries — do not convert "
|
|
f"them to UTC or any other timezone. For other date ranges, compute in "
|
|
f"the same timezone.\n\n"
|
|
)
|
|
|
|
if slot == "compilation":
|
|
base = (
|
|
"You are the user's personal assistant giving their full morning briefing. "
|
|
"Weave real data from tool calls into a warm, natural-sounding summary.\n\n"
|
|
"Tools to call every compilation (skip only if you already know a category is empty):\n"
|
|
"- list_tasks — what's due today, overdue, or in progress\n"
|
|
"- list_events — what's on the calendar today\n"
|
|
"- get_weather — today's forecast\n"
|
|
"- get_rss_items — recent news/blog items from the user's feeds\n"
|
|
"- list_projects (optional) — active project context for narrative continuity\n\n"
|
|
"Rules:\n"
|
|
"- Call tools to see the data. Never assert facts you didn't learn from a tool.\n"
|
|
"- If a tool returns nothing (no events today, no overdue tasks, no news items), "
|
|
"say so honestly. Don't fabricate items to fill space.\n"
|
|
"- For news, pick one or two items worth mentioning — surface the theme, not a laundry list.\n"
|
|
"- Write flowing prose. No markdown, no headers, no bullet points.\n"
|
|
"- Aim for 6 to 10 sentences. Skip topics that have nothing interesting.\n"
|
|
"- Close on one or two concrete, actionable suggestions.\n\n"
|
|
)
|
|
elif slot == "weekly_review":
|
|
base = (
|
|
"You are the user's personal assistant delivering a weekly review. "
|
|
"Use the tools available to see what was accomplished this week, what's still "
|
|
"overdue, how many notes were captured, and what's coming up in the next seven days. "
|
|
"Write a reflective recap that celebrates real progress and gently flags what's stuck.\n\n"
|
|
"Rules:\n"
|
|
"- Call tools to see the data. Never assert facts you didn't learn from a tool.\n"
|
|
"- If a category is empty, say so honestly rather than inventing items.\n"
|
|
"- Write flowing prose. No markdown, no bullet points.\n"
|
|
"- Aim for 5 to 8 sentences. Reflective and encouraging tone.\n\n"
|
|
)
|
|
else: # morning, midday, afternoon check-ins
|
|
base = (
|
|
f"You are the user's personal assistant giving a brief {slot} check-in. "
|
|
"Use tools to see what's changed since this morning. Focus on progress and "
|
|
"what's still unaddressed.\n\n"
|
|
"When checking tasks, call list_tasks at least twice:\n"
|
|
"- once with status=\"in_progress\" to see anything already being worked on "
|
|
"(regardless of due date — these can quietly drag past their due dates)\n"
|
|
"- once filtered by due date for what's coming up or overdue today\n\n"
|
|
"Rules:\n"
|
|
"- Call tools to see current state. Never assert facts without tool results.\n"
|
|
"- If nothing meaningful has changed, say so briefly — don't invent progress.\n"
|
|
"- 3 to 5 sentences, natural prose, no markdown.\n\n"
|
|
)
|
|
|
|
base = tz_block + base
|
|
if profile_body:
|
|
base += f"User profile (tone and preferences):\n{profile_body}\n"
|
|
return base
|
|
|
|
|
|
def _agentic_user_trigger(slot: str, date_str: str) -> str:
|
|
"""Seed user-role message that kicks off the agentic run."""
|
|
labels = {
|
|
"compilation": "morning briefing",
|
|
"morning": "morning check-in",
|
|
"midday": "midday check-in",
|
|
"afternoon": "afternoon wrap-up",
|
|
"weekly_review": "weekly review",
|
|
}
|
|
label = labels.get(slot, f"{slot} briefing")
|
|
return f"Generate my {label} for {date_str}."
|
|
|
|
|
|
async def run_agentic_briefing(
|
|
user_id: int,
|
|
slot: str,
|
|
model: str,
|
|
conv_id: int | None = None,
|
|
rss_override: list[dict] | None = None,
|
|
) -> tuple[str, list[dict]]:
|
|
"""
|
|
Run the agentic briefing loop for a user and slot.
|
|
|
|
Uses the chat pipeline's tool-use loop with a curated read-only tool
|
|
subset and a slot-specific system prompt. Every fact the model states
|
|
is either derived from a tool result visible in the returned message
|
|
list or it's the model hallucinating — so follow-up chat in the same
|
|
conversation can hold the model to what the tool results actually show.
|
|
|
|
Returns ``(final_prose, message_list)`` where ``message_list`` is the
|
|
full sequence including system, user trigger, tool calls, and tool
|
|
results. Callers are expected to persist those intermediate turns
|
|
alongside the final prose so the receipts remain in conversation
|
|
history on follow-up.
|
|
|
|
If the loop fails or the model returns empty prose, returns
|
|
``("", [])`` and the caller should fall back to the legacy path.
|
|
"""
|
|
from fabledassistant.services.llm import stream_chat_with_tools, ChatChunk # noqa: F401
|
|
from fabledassistant.services.tools import execute_tool
|
|
from fabledassistant.services.briefing_tools import get_briefing_tools
|
|
from fabledassistant.services.user_profile import build_profile_context
|
|
|
|
profile_context = await build_profile_context(user_id)
|
|
tools = await get_briefing_tools(user_id)
|
|
|
|
if not tools:
|
|
logger.warning(
|
|
"Agentic briefing for user %d slot %s: no tools available — aborting",
|
|
user_id, slot,
|
|
)
|
|
return "", []
|
|
|
|
# Compute today's window in the user's local timezone so the model
|
|
# receives ready-to-use ISO 8601 boundaries and never has to do its
|
|
# own tz math when calling date-sensitive tools like list_events.
|
|
tz_name = await get_setting(user_id, "user_timezone") or "UTC"
|
|
try:
|
|
user_tz = ZoneInfo(tz_name)
|
|
except ZoneInfoNotFoundError:
|
|
user_tz = ZoneInfo("UTC")
|
|
tz_name = "UTC"
|
|
now_local = datetime.now(user_tz)
|
|
today_iso = now_local.date().isoformat()
|
|
day_start = datetime(now_local.year, now_local.month, now_local.day, 0, 0, 0, tzinfo=user_tz)
|
|
day_end = datetime(now_local.year, now_local.month, now_local.day, 23, 59, 59, tzinfo=user_tz)
|
|
day_from_iso = day_start.isoformat()
|
|
day_to_iso = day_end.isoformat()
|
|
|
|
system_prompt = _agentic_system_prompt(
|
|
profile_context, slot, today_iso, tz_name, day_from_iso, day_to_iso,
|
|
)
|
|
messages: list[dict] = [
|
|
{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": _agentic_user_trigger(slot, today_iso)},
|
|
]
|
|
|
|
final_text = ""
|
|
for round_idx in range(_BRIEFING_AGENT_MAX_ROUNDS):
|
|
accumulated_content = ""
|
|
accumulated_tool_calls: list[dict] = []
|
|
|
|
try:
|
|
async for chunk in stream_chat_with_tools(
|
|
messages, model, tools=tools, think=False,
|
|
num_ctx=_BRIEFING_AGENT_NUM_CTX,
|
|
):
|
|
if chunk.type == "content" and chunk.content:
|
|
accumulated_content += chunk.content
|
|
elif chunk.type == "tool_calls" and chunk.tool_calls:
|
|
accumulated_tool_calls.extend(chunk.tool_calls)
|
|
except Exception:
|
|
logger.warning(
|
|
"Agentic briefing stream failed (user %d, slot %s, round %d)",
|
|
user_id, slot, round_idx, exc_info=True,
|
|
)
|
|
return "", []
|
|
|
|
# Append the assistant turn (content + any tool calls) to history
|
|
assistant_msg: dict = {"role": "assistant", "content": accumulated_content}
|
|
if accumulated_tool_calls:
|
|
assistant_msg["tool_calls"] = accumulated_tool_calls
|
|
messages.append(assistant_msg)
|
|
|
|
# No tool calls → the model is done
|
|
if not accumulated_tool_calls:
|
|
final_text = accumulated_content.strip()
|
|
break
|
|
|
|
# Execute each tool call and append results as tool-role messages
|
|
for tc in accumulated_tool_calls:
|
|
fn = tc.get("function") or {}
|
|
tool_name = fn.get("name", "")
|
|
arguments = fn.get("arguments") or {}
|
|
if isinstance(arguments, str):
|
|
try:
|
|
import json as _json
|
|
arguments = _json.loads(arguments)
|
|
except Exception:
|
|
arguments = {}
|
|
|
|
# Default list_tasks to active statuses only so cancelled/done
|
|
# items don't slip into briefing prose. The model can still
|
|
# pass an explicit status filter when it wants something else.
|
|
if tool_name == "list_tasks" and not arguments.get("status"):
|
|
arguments["status"] = ["todo", "in_progress"]
|
|
|
|
try:
|
|
if tool_name == "get_rss_items" and rss_override is not None:
|
|
# Use topic-scored/filtered items already computed by
|
|
# the briefing pipeline rather than the raw feed dump
|
|
# that execute_tool would return. Keeps the model's
|
|
# view of news aligned with the user's topic prefs
|
|
# and the sidebar's rss_item_ids metadata.
|
|
slim = [
|
|
{
|
|
"id": item.get("id"),
|
|
"title": item.get("title", ""),
|
|
"url": item.get("url", ""),
|
|
"source": item.get("feed_title", ""),
|
|
"summary": (item.get("content") or "")[:400],
|
|
"published_at": item.get("published_at"),
|
|
"topics": item.get("topics") or [],
|
|
}
|
|
for item in rss_override
|
|
]
|
|
result = {"success": True, "data": {"items": slim, "count": len(slim)}}
|
|
else:
|
|
result = await execute_tool(user_id, tool_name, arguments, conv_id=conv_id)
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"Tool %s failed during agentic briefing: %s", tool_name, exc,
|
|
)
|
|
result = {"success": False, "error": str(exc)}
|
|
|
|
# Serialize the result compactly for the model's context
|
|
import json as _json
|
|
try:
|
|
result_str = _json.dumps(result, default=str)[:4000]
|
|
except Exception:
|
|
result_str = str(result)[:4000]
|
|
|
|
messages.append({
|
|
"role": "tool",
|
|
"content": result_str,
|
|
"tool_name": tool_name,
|
|
})
|
|
else:
|
|
logger.warning(
|
|
"Agentic briefing hit max rounds (%d) for user %d slot %s — using last content",
|
|
_BRIEFING_AGENT_MAX_ROUNDS, user_id, slot,
|
|
)
|
|
# Walk back to find the last assistant message with non-empty content
|
|
for m in reversed(messages):
|
|
if m.get("role") == "assistant" and m.get("content"):
|
|
final_text = m["content"].strip()
|
|
break
|
|
|
|
return final_text, messages
|
|
|
|
|
|
# ── Main entry point ───────────────────────────────────────────────────────────
|
|
|
|
async def _get_temp_unit(user_id: int) -> str:
|
|
"""Read the user's preferred temperature unit from briefing_config ('C' or 'F')."""
|
|
import json
|
|
raw = await get_setting(user_id, "briefing_config", "{}")
|
|
try:
|
|
config = json.loads(raw) if isinstance(raw, str) else (raw or {})
|
|
unit = config.get("temp_unit", "C")
|
|
return unit if unit in ("C", "F") else "C"
|
|
except Exception:
|
|
return "C"
|
|
|
|
|
|
async def run_compilation(
|
|
user_id: int,
|
|
slot: str,
|
|
model: str | None = None,
|
|
) -> tuple[str, dict]:
|
|
"""
|
|
Run the agentic briefing loop and gather UI metadata (RSS + weather).
|
|
|
|
Returns ``(briefing_text, metadata)`` where metadata contains
|
|
``rss_item_ids``, ``rss_items``, ``weather`` for frontend rendering,
|
|
and ``agentic_messages`` (the full tool-call sequence) for the
|
|
scheduler to persist as separate conversation rows.
|
|
"""
|
|
if model is None:
|
|
model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL)
|
|
|
|
from fabledassistant.services.briefing_preferences import (
|
|
load_topic_preferences,
|
|
load_topic_reaction_scores,
|
|
score_and_filter_items,
|
|
)
|
|
from fabledassistant.services.weather import parse_weather_card_data, get_cached_weather_rows
|
|
|
|
include_topics, exclude_topics = await load_topic_preferences(user_id)
|
|
topic_scores = await load_topic_reaction_scores(user_id)
|
|
|
|
external_data, weather_rows, temp_unit = await asyncio.gather(
|
|
_gather_external(user_id),
|
|
get_cached_weather_rows(user_id),
|
|
_get_temp_unit(user_id),
|
|
)
|
|
|
|
raw_rss = external_data.get("rss_items") or []
|
|
filtered_rss = score_and_filter_items(
|
|
raw_rss,
|
|
include_topics=include_topics,
|
|
exclude_topics=exclude_topics,
|
|
topic_scores=topic_scores,
|
|
max_items=10,
|
|
)
|
|
rss_item_ids = [item["id"] for item in filtered_rss if item.get("id")]
|
|
rss_items_meta = [
|
|
{
|
|
"id": item["id"],
|
|
"title": item.get("title", ""),
|
|
"url": item.get("url", ""),
|
|
"source": item.get("feed_title", ""),
|
|
"snippet": (item.get("content") or "")[:300],
|
|
"published_at": item.get("published_at"),
|
|
}
|
|
for item in filtered_rss
|
|
if item.get("id")
|
|
]
|
|
|
|
weather_card = parse_weather_card_data(weather_rows[0], temp_unit) if weather_rows else None
|
|
|
|
briefing_text, agentic_messages = await run_agentic_briefing(
|
|
user_id, slot, model, conv_id=None, rss_override=filtered_rss,
|
|
)
|
|
|
|
metadata: dict = {
|
|
"rss_item_ids": rss_item_ids,
|
|
"rss_items": rss_items_meta,
|
|
"weather": weather_card,
|
|
}
|
|
if agentic_messages:
|
|
metadata["agentic_messages"] = agentic_messages
|
|
|
|
if not briefing_text:
|
|
logger.warning("Briefing compilation produced no content for user %d slot %s", user_id, slot)
|
|
return "", metadata
|
|
|
|
return briefing_text, metadata
|
|
|
|
|
|
async def run_slot_injection(
|
|
user_id: int,
|
|
slot: str,
|
|
model: str | None = None,
|
|
) -> tuple[str, dict]:
|
|
"""
|
|
Lighter check-in update for 8am/12pm/4pm slots.
|
|
|
|
Runs the agentic loop with the slot-specific prompt. Returns
|
|
``(text, metadata)`` where metadata contains ``agentic_messages``
|
|
for the scheduler to persist.
|
|
"""
|
|
if model is None:
|
|
model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL)
|
|
|
|
text, agentic_messages = await run_agentic_briefing(
|
|
user_id, slot, model, conv_id=None,
|
|
)
|
|
|
|
metadata: dict = {}
|
|
if agentic_messages:
|
|
metadata["agentic_messages"] = agentic_messages
|
|
return text, metadata
|