refactor(briefing)!: remove legacy one-shot synthesis, agentic-only
Deletes ~760 lines of legacy briefing code: format_task, compute_task_hash, upsert_task_snapshots, _gather_internal, _gather_weekly_review, _llm_synthesise, and the unified prompt helpers. run_compilation and run_slot_injection are now agentic-tool-use-loop only. briefing_scheduler and user_profile migrated from the deleted helper to services.llm.generate_completion (retry + keep_alive baked in). routes/briefing.manual_trigger now persists agentic tool-call receipts via _persist_agentic_messages (previously silently dropped them) and adds POST /api/briefing/reset-today to wipe today's briefing messages. BREAKING: briefing_mode setting no longer honored; no legacy fallback. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -308,14 +308,52 @@ async def manual_trigger():
|
|||||||
logger.warning("Pre-trigger refresh failed for user %d", g.user.id, exc_info=True)
|
logger.warning("Pre-trigger refresh failed for user %d", g.user.id, exc_info=True)
|
||||||
|
|
||||||
from fabledassistant.services.briefing_pipeline import run_compilation
|
from fabledassistant.services.briefing_pipeline import run_compilation
|
||||||
|
from fabledassistant.services.briefing_scheduler import _persist_agentic_messages
|
||||||
|
|
||||||
model = await get_setting(g.user.id, "default_model", "")
|
model = await get_setting(g.user.id, "default_model", "")
|
||||||
conv = await get_or_create_today_conversation(g.user.id, model)
|
conv = await get_or_create_today_conversation(g.user.id, model)
|
||||||
text, metadata = await run_compilation(g.user.id, slot, model)
|
text, metadata = await run_compilation(g.user.id, slot, model)
|
||||||
msg = await post_message(conv.id, "assistant", text, metadata=metadata)
|
await _persist_agentic_messages(conv.id, slot, metadata)
|
||||||
|
final_meta = {k: v for k, v in metadata.items() if k != "agentic_messages"}
|
||||||
|
final_meta["briefing_slot"] = slot
|
||||||
|
msg = await post_message(conv.id, "assistant", text, metadata=final_meta)
|
||||||
return jsonify({"conversation_id": conv.id, "message_id": msg.id, "slot": slot})
|
return jsonify({"conversation_id": conv.id, "message_id": msg.id, "slot": slot})
|
||||||
|
|
||||||
|
|
||||||
|
@briefing_bp.route("/reset-today", methods=["POST"])
|
||||||
|
@_REQUIRE
|
||||||
|
async def reset_today_briefing():
|
||||||
|
"""Delete all messages in today's briefing conversation.
|
||||||
|
|
||||||
|
The conversation row itself is kept so its id stays stable for any
|
||||||
|
open UI sessions. Intended for "wipe and start fresh" workflows
|
||||||
|
driven from the MCP when iterating on prompts. Pair with
|
||||||
|
``POST /api/briefing/trigger`` to immediately regenerate.
|
||||||
|
"""
|
||||||
|
from datetime import date as _date
|
||||||
|
from sqlalchemy import delete as _delete
|
||||||
|
|
||||||
|
today = _date.today()
|
||||||
|
async with async_session() as session:
|
||||||
|
result = await session.execute(
|
||||||
|
select(Conversation).where(
|
||||||
|
Conversation.user_id == g.user.id,
|
||||||
|
Conversation.conversation_type == "briefing",
|
||||||
|
Conversation.briefing_date == today,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
conv = result.scalars().first()
|
||||||
|
if conv is None:
|
||||||
|
return jsonify({"deleted": 0, "conversation_id": None})
|
||||||
|
deleted_result = await session.execute(
|
||||||
|
_delete(Message).where(Message.conversation_id == conv.id)
|
||||||
|
)
|
||||||
|
await session.commit()
|
||||||
|
deleted = deleted_result.rowcount or 0
|
||||||
|
|
||||||
|
return jsonify({"deleted": deleted, "conversation_id": conv.id})
|
||||||
|
|
||||||
|
|
||||||
# ── RSS Reactions ──────────────────────────────────────────────────────────────
|
# ── RSS Reactions ──────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
@briefing_bp.route("/rss-reactions", methods=["POST"])
|
@briefing_bp.route("/rss-reactions", methods=["POST"])
|
||||||
|
|||||||
@@ -1,19 +1,14 @@
|
|||||||
"""
|
"""
|
||||||
Briefing pipeline: parallel data gather + two-lane LLM synthesis.
|
Briefing pipeline: agentic tool-use loop + UI metadata gather.
|
||||||
|
|
||||||
Slot names: 'compilation' (4am), 'morning' (8am), 'midday' (12pm), 'afternoon' (4pm)
|
Slot names: 'compilation' (4am), 'morning' (8am), 'midday' (12pm), 'afternoon' (4pm)
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import hashlib
|
|
||||||
import logging
|
import logging
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime
|
||||||
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
||||||
|
|
||||||
import httpx
|
|
||||||
|
|
||||||
from fabledassistant.models import async_session
|
|
||||||
|
|
||||||
from fabledassistant.config import Config
|
from fabledassistant.config import Config
|
||||||
from fabledassistant.services.settings import get_setting
|
from fabledassistant.services.settings import get_setting
|
||||||
|
|
||||||
@@ -22,421 +17,6 @@ logger = logging.getLogger(__name__)
|
|||||||
SLOT_NAMES = ("compilation", "morning", "midday", "afternoon")
|
SLOT_NAMES = ("compilation", "morning", "midday", "afternoon")
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def format_task(task: dict, today: str | None = None) -> str:
|
|
||||||
parts = [task.get("title", "Untitled")]
|
|
||||||
if task.get("status"):
|
|
||||||
parts.append(f"[{task['status'].replace('_', ' ').title()}]")
|
|
||||||
if task.get("due_date"):
|
|
||||||
parts.append(f"due {task['due_date']}")
|
|
||||||
if today and task["due_date"] < today and task.get("status") not in ("done", "cancelled"):
|
|
||||||
from datetime import date
|
|
||||||
try:
|
|
||||||
due = date.fromisoformat(task["due_date"])
|
|
||||||
td = date.fromisoformat(today)
|
|
||||||
days = (td - due).days
|
|
||||||
parts.append(f"({days} day{'s' if days != 1 else ''} overdue)")
|
|
||||||
except ValueError:
|
|
||||||
pass
|
|
||||||
if task.get("priority") and task["priority"] not in ("none", ""):
|
|
||||||
parts.append(f"priority: {task['priority']}")
|
|
||||||
if task.get("project_title"):
|
|
||||||
parts.append(f"project: {task['project_title']}")
|
|
||||||
return " — ".join(parts)
|
|
||||||
|
|
||||||
|
|
||||||
def compute_task_hash(task: dict) -> str:
|
|
||||||
"""Stable SHA-256 of the task's key change-detectable fields."""
|
|
||||||
key = "|".join([
|
|
||||||
str(task.get("status") or ""),
|
|
||||||
str(task.get("priority") or ""),
|
|
||||||
str(task.get("due_date") or ""),
|
|
||||||
str(task.get("title") or ""),
|
|
||||||
])
|
|
||||||
return hashlib.sha256(key.encode()).hexdigest()
|
|
||||||
|
|
||||||
|
|
||||||
async def split_changed_tasks(
|
|
||||||
user_id: int,
|
|
||||||
tasks: list[dict],
|
|
||||||
) -> tuple[list[dict], int]:
|
|
||||||
"""
|
|
||||||
Compare tasks against the briefing_task_snapshot table.
|
|
||||||
Returns (changed_tasks, unchanged_count).
|
|
||||||
changed_tasks includes new tasks (no snapshot row) and tasks whose hash differs.
|
|
||||||
"""
|
|
||||||
from sqlalchemy import text
|
|
||||||
|
|
||||||
if not tasks:
|
|
||||||
return [], 0
|
|
||||||
|
|
||||||
task_ids = [t["task_id"] for t in tasks if t.get("task_id")]
|
|
||||||
|
|
||||||
async with async_session() as session:
|
|
||||||
result = await session.execute(
|
|
||||||
text("""
|
|
||||||
SELECT task_id, snapshot_hash
|
|
||||||
FROM briefing_task_snapshot
|
|
||||||
WHERE user_id = :uid AND task_id = ANY(:ids)
|
|
||||||
""").bindparams(uid=user_id, ids=task_ids)
|
|
||||||
)
|
|
||||||
snapshots = {row.task_id: row.snapshot_hash for row in result}
|
|
||||||
|
|
||||||
changed = []
|
|
||||||
unchanged_count = 0
|
|
||||||
for task in tasks:
|
|
||||||
current_hash = compute_task_hash(task)
|
|
||||||
stored_hash = snapshots.get(task.get("task_id"))
|
|
||||||
if stored_hash is None or stored_hash != current_hash:
|
|
||||||
changed.append(task)
|
|
||||||
else:
|
|
||||||
unchanged_count += 1
|
|
||||||
|
|
||||||
return changed, unchanged_count
|
|
||||||
|
|
||||||
|
|
||||||
async def upsert_task_snapshots(user_id: int, tasks: list[dict]) -> None:
|
|
||||||
"""Upsert snapshot hashes for all tasks included in this briefing."""
|
|
||||||
from sqlalchemy import text
|
|
||||||
|
|
||||||
if not tasks:
|
|
||||||
return
|
|
||||||
|
|
||||||
now = datetime.now(timezone.utc)
|
|
||||||
async with async_session() as session:
|
|
||||||
for task in tasks:
|
|
||||||
task_id = task.get("task_id")
|
|
||||||
if not task_id:
|
|
||||||
continue
|
|
||||||
await session.execute(
|
|
||||||
text("""
|
|
||||||
INSERT INTO briefing_task_snapshot (user_id, task_id, snapshot_hash, last_briefed)
|
|
||||||
VALUES (:uid, :tid, :hash, :now)
|
|
||||||
ON CONFLICT (user_id, task_id)
|
|
||||||
DO UPDATE SET snapshot_hash = EXCLUDED.snapshot_hash,
|
|
||||||
last_briefed = EXCLUDED.last_briefed
|
|
||||||
""").bindparams(
|
|
||||||
uid=user_id,
|
|
||||||
tid=task_id,
|
|
||||||
hash=compute_task_hash(task),
|
|
||||||
now=now,
|
|
||||||
)
|
|
||||||
)
|
|
||||||
await session.commit()
|
|
||||||
|
|
||||||
|
|
||||||
# ── Internal data gather ──────────────────────────────────────────────────────
|
|
||||||
|
|
||||||
async def _gather_internal(user_id: int) -> dict:
|
|
||||||
"""Collect tasks, calendar events, and project data."""
|
|
||||||
from fabledassistant.services.notes import list_notes
|
|
||||||
from fabledassistant.services.projects import list_projects
|
|
||||||
from fabledassistant.services.caldav import is_caldav_configured, 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")
|
|
||||||
|
|
||||||
today = datetime.now(user_tz).date().isoformat()
|
|
||||||
|
|
||||||
# Tasks: overdue, due today, high priority in-progress
|
|
||||||
all_tasks: list[dict] = []
|
|
||||||
try:
|
|
||||||
all_task_objs, _total = await list_notes(user_id, is_task=True, limit=100)
|
|
||||||
# Build project title lookup for enrichment
|
|
||||||
project_titles: dict[int, str] = {}
|
|
||||||
try:
|
|
||||||
for p in await list_projects(user_id):
|
|
||||||
project_titles[p.id] = p.title or "Untitled"
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
all_tasks = [
|
|
||||||
{
|
|
||||||
"task_id": t.id,
|
|
||||||
"title": t.title,
|
|
||||||
"status": t.status,
|
|
||||||
"due_date": t.due_date.isoformat() if t.due_date else None,
|
|
||||||
"priority": t.priority,
|
|
||||||
"project_title": project_titles.get(t.project_id) if t.project_id else None,
|
|
||||||
}
|
|
||||||
for t in all_task_objs
|
|
||||||
]
|
|
||||||
overdue = [
|
|
||||||
format_task(t, today) for t in all_tasks
|
|
||||||
if t.get("due_date") and t["due_date"] < today and t.get("status") not in ("done", "cancelled")
|
|
||||||
]
|
|
||||||
due_today = [
|
|
||||||
format_task(t, today) for t in all_tasks
|
|
||||||
if t.get("due_date") == today and t.get("status") not in ("done", "cancelled")
|
|
||||||
]
|
|
||||||
high_priority = [
|
|
||||||
format_task(t, today) for t in all_tasks
|
|
||||||
if t.get("priority") == "high" and t.get("status") not in ("done", "cancelled") and
|
|
||||||
format_task(t, today) not in overdue and format_task(t, today) not in due_today
|
|
||||||
][:5]
|
|
||||||
except Exception:
|
|
||||||
logger.warning("Failed to gather tasks for briefing", exc_info=True)
|
|
||||||
overdue, due_today, high_priority = [], [], []
|
|
||||||
|
|
||||||
# Calendar events today — internal store
|
|
||||||
calendar_events: list[str] = []
|
|
||||||
internal_events: list[dict] = []
|
|
||||||
try:
|
|
||||||
from fabledassistant.services.events import list_events as list_internal_events
|
|
||||||
today_date = datetime.now(user_tz).date()
|
|
||||||
day_start = datetime(today_date.year, today_date.month, today_date.day, 0, 0, 0, tzinfo=user_tz)
|
|
||||||
day_end = datetime(today_date.year, today_date.month, today_date.day, 23, 59, 59, tzinfo=user_tz)
|
|
||||||
internal_events = await list_internal_events(
|
|
||||||
user_id=user_id, date_from=day_start, date_to=day_end
|
|
||||||
)
|
|
||||||
for e in internal_events:
|
|
||||||
if e.get("all_day"):
|
|
||||||
time_str = "all day"
|
|
||||||
elif e.get("start_dt"):
|
|
||||||
from datetime import datetime as _dt
|
|
||||||
start = e["start_dt"]
|
|
||||||
if isinstance(start, str):
|
|
||||||
start = _dt.fromisoformat(start)
|
|
||||||
local_dt = start.astimezone(user_tz) if start.tzinfo else start.replace(tzinfo=timezone.utc).astimezone(user_tz)
|
|
||||||
time_str = local_dt.strftime("%-I:%M %p")
|
|
||||||
else:
|
|
||||||
time_str = "unknown time"
|
|
||||||
calendar_events.append(f"{e.get('title', 'Event')} at {time_str}")
|
|
||||||
except Exception:
|
|
||||||
logger.warning("Failed to gather internal calendar events for briefing", exc_info=True)
|
|
||||||
# Also pull CalDAV events (deduped)
|
|
||||||
try:
|
|
||||||
if await is_caldav_configured(user_id):
|
|
||||||
caldav_evs = await list_events(user_id, start=today, end=today)
|
|
||||||
for e in (caldav_evs or []):
|
|
||||||
summary = f"{e.get('summary', 'Event')} at {e.get('dtstart', 'unknown time')}"
|
|
||||||
if summary not in calendar_events:
|
|
||||||
calendar_events.append(summary)
|
|
||||||
except Exception:
|
|
||||||
logger.warning("Failed to gather CalDAV calendar events for briefing", exc_info=True)
|
|
||||||
|
|
||||||
# Weather × calendar cross-reference
|
|
||||||
weather_warnings: list[str] = []
|
|
||||||
outdoor_keywords = {"park", "field", "patio", "garden", "outdoor", "trail", "beach",
|
|
||||||
"pool", "stadium", "yard", "lawn", "hike", "walk", "run", "bike",
|
|
||||||
"soccer", "baseball", "football", "tennis", "golf", "playground"}
|
|
||||||
try:
|
|
||||||
import json as _json
|
|
||||||
raw_config = await get_setting(user_id, "briefing_config", "{}")
|
|
||||||
config = _json.loads(raw_config) if isinstance(raw_config, str) else (raw_config or {})
|
|
||||||
locations = config.get("locations", {})
|
|
||||||
loc = locations.get("home") or locations.get("work")
|
|
||||||
if loc and loc.get("lat") and loc.get("lon") and internal_events:
|
|
||||||
from fabledassistant.services.weather import fetch_hourly_precip
|
|
||||||
hourly_precip = await fetch_hourly_precip(loc["lat"], loc["lon"])
|
|
||||||
if hourly_precip:
|
|
||||||
for e in internal_events:
|
|
||||||
if e.get("all_day"):
|
|
||||||
continue
|
|
||||||
# Check if event title or location contains outdoor keywords
|
|
||||||
event_text = f"{e.get('title', '')} {e.get('location', '')}".lower()
|
|
||||||
if not any(kw in event_text for kw in outdoor_keywords):
|
|
||||||
continue
|
|
||||||
# Get the event's start hour
|
|
||||||
start = e.get("start_dt")
|
|
||||||
if not start:
|
|
||||||
continue
|
|
||||||
from datetime import datetime as _dt
|
|
||||||
if isinstance(start, str):
|
|
||||||
start = _dt.fromisoformat(start)
|
|
||||||
local_start = start.astimezone(user_tz) if start.tzinfo else start.replace(tzinfo=timezone.utc).astimezone(user_tz)
|
|
||||||
# Check precip probability around the event time (hour before through event)
|
|
||||||
hour_key = local_start.strftime("%Y-%m-%dT%H:00")
|
|
||||||
precip = hourly_precip.get(hour_key, 0)
|
|
||||||
if precip >= 30:
|
|
||||||
time_str = local_start.strftime("%-I:%M %p")
|
|
||||||
weather_warnings.append(
|
|
||||||
f"Rain likely ({precip}%) around {time_str} when '{e.get('title', 'Event')}' "
|
|
||||||
f"is scheduled — consider an umbrella or backup plan"
|
|
||||||
)
|
|
||||||
except Exception:
|
|
||||||
logger.debug("Weather cross-reference failed", exc_info=True)
|
|
||||||
|
|
||||||
# Recent activity — what was the user working on yesterday?
|
|
||||||
recent_activity: list[str] = []
|
|
||||||
stale_projects: list[str] = []
|
|
||||||
try:
|
|
||||||
from sqlalchemy import select as _sel, func as _func
|
|
||||||
from fabledassistant.models.note import Note
|
|
||||||
from fabledassistant.models.project import Project
|
|
||||||
|
|
||||||
yesterday = datetime.now(user_tz) - __import__('datetime').timedelta(days=1)
|
|
||||||
async with async_session() as session:
|
|
||||||
# Notes/tasks updated in last 24h, grouped by project
|
|
||||||
result = await session.execute(
|
|
||||||
_sel(Project.title, _func.count(Note.id))
|
|
||||||
.join(Note, Note.project_id == Project.id)
|
|
||||||
.where(Note.user_id == user_id)
|
|
||||||
.where(Note.updated_at >= yesterday)
|
|
||||||
.group_by(Project.title)
|
|
||||||
.order_by(_func.count(Note.id).desc())
|
|
||||||
.limit(3)
|
|
||||||
)
|
|
||||||
for proj_title, count in result.all():
|
|
||||||
recent_activity.append(f"{proj_title}: {count} item{'s' if count != 1 else ''} updated")
|
|
||||||
|
|
||||||
# Stale projects — active projects with no note/task activity in 7+ days
|
|
||||||
week_ago = datetime.now(user_tz) - __import__('datetime').timedelta(days=7)
|
|
||||||
active_projects_q = await session.execute(
|
|
||||||
_sel(Project).where(Project.user_id == user_id, Project.status == "active")
|
|
||||||
)
|
|
||||||
for p in active_projects_q.scalars().all():
|
|
||||||
latest = await session.execute(
|
|
||||||
_sel(_func.max(Note.updated_at))
|
|
||||||
.where(Note.project_id == p.id, Note.user_id == user_id)
|
|
||||||
)
|
|
||||||
last_activity = latest.scalar()
|
|
||||||
if last_activity is None or last_activity < week_ago:
|
|
||||||
stale_projects.append(p.title or "Untitled")
|
|
||||||
except Exception:
|
|
||||||
logger.debug("Failed to gather recent activity for briefing", exc_info=True)
|
|
||||||
|
|
||||||
# Projects: active projects
|
|
||||||
projects_summary = []
|
|
||||||
try:
|
|
||||||
projects = await list_projects(user_id)
|
|
||||||
for p in projects[:5]:
|
|
||||||
projects_summary.append(p.title or "Untitled project")
|
|
||||||
except Exception:
|
|
||||||
logger.warning("Failed to gather projects for briefing", exc_info=True)
|
|
||||||
|
|
||||||
# Build prioritized "Your Day" focus list
|
|
||||||
# Priority: overdue > due today > high priority in-progress > other in-progress
|
|
||||||
focus_tasks: list[str] = []
|
|
||||||
seen_titles: set[str] = set()
|
|
||||||
for t in sorted(all_tasks, key=lambda x: (
|
|
||||||
0 if (x.get("due_date") and x["due_date"] < today and x.get("status") not in ("done", "cancelled")) else
|
|
||||||
1 if (x.get("due_date") == today and x.get("status") not in ("done", "cancelled")) else
|
|
||||||
2 if (x.get("priority") == "high" and x.get("status") not in ("done", "cancelled")) else
|
|
||||||
3 if (x.get("status") == "in_progress") else 4
|
|
||||||
)):
|
|
||||||
if t.get("status") in ("done", "cancelled"):
|
|
||||||
continue
|
|
||||||
formatted = format_task(t, today)
|
|
||||||
if formatted not in seen_titles:
|
|
||||||
seen_titles.add(formatted)
|
|
||||||
focus_tasks.append(formatted)
|
|
||||||
if len(focus_tasks) >= 5:
|
|
||||||
break
|
|
||||||
|
|
||||||
# Calendar gap analysis — extract event times for LLM
|
|
||||||
event_time_blocks: list[str] = []
|
|
||||||
for e in internal_events:
|
|
||||||
try:
|
|
||||||
if e.get("all_day"):
|
|
||||||
continue
|
|
||||||
start = e.get("start_dt")
|
|
||||||
end = e.get("end_dt")
|
|
||||||
if not start:
|
|
||||||
continue
|
|
||||||
from datetime import datetime as _dt
|
|
||||||
if isinstance(start, str):
|
|
||||||
start = _dt.fromisoformat(start)
|
|
||||||
if isinstance(end, str):
|
|
||||||
end = _dt.fromisoformat(end)
|
|
||||||
s_local = start.astimezone(user_tz) if start.tzinfo else start.replace(tzinfo=timezone.utc).astimezone(user_tz)
|
|
||||||
s_str = s_local.strftime("%-I:%M %p")
|
|
||||||
if end:
|
|
||||||
e_local = end.astimezone(user_tz) if end.tzinfo else end.replace(tzinfo=timezone.utc).astimezone(user_tz)
|
|
||||||
e_str = e_local.strftime("%-I:%M %p")
|
|
||||||
event_time_blocks.append(f"{e.get('title', 'Event')}: {s_str}–{e_str}")
|
|
||||||
else:
|
|
||||||
event_time_blocks.append(f"{e.get('title', 'Event')}: {s_str}–?")
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
|
|
||||||
return {
|
|
||||||
"date": today,
|
|
||||||
"overdue_tasks": overdue,
|
|
||||||
"due_today": due_today,
|
|
||||||
"high_priority": high_priority,
|
|
||||||
"calendar_events": calendar_events,
|
|
||||||
"active_projects": projects_summary,
|
|
||||||
"all_tasks_raw": all_tasks,
|
|
||||||
"focus_tasks": focus_tasks,
|
|
||||||
"event_time_blocks": event_time_blocks,
|
|
||||||
"weather_warnings": weather_warnings,
|
|
||||||
"recent_activity": recent_activity,
|
|
||||||
"stale_projects": stale_projects,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
async def _gather_weekly_review(user_id: int) -> dict:
|
|
||||||
"""Collect 7-day activity summary for the weekly review."""
|
|
||||||
from fabledassistant.services.notes import list_notes
|
|
||||||
from fabledassistant.services.projects import list_projects
|
|
||||||
from fabledassistant.services.events import list_events as list_internal_events
|
|
||||||
|
|
||||||
tz_name = await get_setting(user_id, "user_timezone") or "UTC"
|
|
||||||
try:
|
|
||||||
user_tz = ZoneInfo(tz_name)
|
|
||||||
except ZoneInfoNotFoundError:
|
|
||||||
user_tz = ZoneInfo("UTC")
|
|
||||||
|
|
||||||
now = datetime.now(user_tz)
|
|
||||||
week_ago = now - __import__('datetime').timedelta(days=7)
|
|
||||||
next_week = now + __import__('datetime').timedelta(days=7)
|
|
||||||
|
|
||||||
# Tasks completed this week
|
|
||||||
completed_tasks = []
|
|
||||||
overdue_tasks = []
|
|
||||||
try:
|
|
||||||
all_task_objs, _ = await list_notes(user_id, is_task=True, limit=200)
|
|
||||||
for t in all_task_objs:
|
|
||||||
if t.status == "done" and t.completed_at and t.completed_at >= week_ago:
|
|
||||||
completed_tasks.append(t.title or "Untitled")
|
|
||||||
elif t.due_date and t.due_date.isoformat() < now.date().isoformat() and t.status not in ("done", "cancelled"):
|
|
||||||
overdue_tasks.append(t.title or "Untitled")
|
|
||||||
except Exception:
|
|
||||||
logger.debug("Weekly review: failed to gather tasks", exc_info=True)
|
|
||||||
|
|
||||||
# Notes created this week
|
|
||||||
notes_created = 0
|
|
||||||
try:
|
|
||||||
all_notes, _ = await list_notes(user_id, is_task=False, limit=200)
|
|
||||||
notes_created = sum(1 for n in all_notes if n.created_at and n.created_at >= week_ago)
|
|
||||||
except Exception:
|
|
||||||
logger.debug("Weekly review: failed to count notes", exc_info=True)
|
|
||||||
|
|
||||||
# Project activity
|
|
||||||
active_projects = []
|
|
||||||
try:
|
|
||||||
projects = await list_projects(user_id)
|
|
||||||
active_projects = [p.title for p in projects if p.status == "active"][:5]
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
|
|
||||||
# Events next week
|
|
||||||
upcoming_events = []
|
|
||||||
try:
|
|
||||||
next_events = await list_internal_events(
|
|
||||||
user_id=user_id,
|
|
||||||
date_from=now,
|
|
||||||
date_to=next_week,
|
|
||||||
)
|
|
||||||
for e in next_events[:8]:
|
|
||||||
upcoming_events.append(f"{e.get('title', 'Event')} — {e.get('start_dt', '')[:10]}")
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
|
|
||||||
return {
|
|
||||||
"completed_count": len(completed_tasks),
|
|
||||||
"completed_highlights": completed_tasks[:5],
|
|
||||||
"overdue_count": len(overdue_tasks),
|
|
||||||
"overdue_items": overdue_tasks[:5],
|
|
||||||
"notes_created": notes_created,
|
|
||||||
"active_projects": active_projects,
|
|
||||||
"upcoming_events": upcoming_events,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
# ── External data gather ──────────────────────────────────────────────────────
|
# ── External data gather ──────────────────────────────────────────────────────
|
||||||
|
|
||||||
async def _gather_external(user_id: int) -> dict:
|
async def _gather_external(user_id: int) -> dict:
|
||||||
@@ -687,265 +267,6 @@ async def run_agentic_briefing(
|
|||||||
return final_text, messages
|
return final_text, messages
|
||||||
|
|
||||||
|
|
||||||
# ── Legacy one-shot synthesis ─────────────────────────────────────────────────
|
|
||||||
|
|
||||||
async def _llm_synthesise(system_prompt: str, user_prompt: str, model: str, num_ctx: int = 4096) -> str:
|
|
||||||
"""Single non-streaming LLM call. Returns the assistant's response text."""
|
|
||||||
payload = {
|
|
||||||
"model": model,
|
|
||||||
"messages": [
|
|
||||||
{"role": "system", "content": system_prompt},
|
|
||||||
{"role": "user", "content": user_prompt},
|
|
||||||
],
|
|
||||||
"stream": False,
|
|
||||||
"options": {"num_ctx": num_ctx, "temperature": 0.4},
|
|
||||||
}
|
|
||||||
try:
|
|
||||||
async with httpx.AsyncClient(timeout=120.0) as client:
|
|
||||||
resp = await client.post(f"{Config.OLLAMA_URL}/api/chat", json=payload)
|
|
||||||
resp.raise_for_status()
|
|
||||||
data = resp.json()
|
|
||||||
return data.get("message", {}).get("content", "").strip()
|
|
||||||
except Exception:
|
|
||||||
logger.warning("LLM synthesis failed", exc_info=True)
|
|
||||||
return ""
|
|
||||||
|
|
||||||
|
|
||||||
def _unified_system_prompt(profile_body: str, slot: str = "compilation") -> str:
|
|
||||||
base = (
|
|
||||||
"You are a personal assistant. "
|
|
||||||
"Speak naturally and conversationally — as if talking to the user, not writing a report. "
|
|
||||||
"Use no markdown: no headers, no bullet points, no bold, no lists. Write in flowing prose. "
|
|
||||||
"Be warm, concise, and human. "
|
|
||||||
"The user can reply to act on your suggestions via tool calls.\n\n"
|
|
||||||
)
|
|
||||||
|
|
||||||
if slot == "compilation":
|
|
||||||
base += (
|
|
||||||
"This is the MORNING BRIEFING — the full daily overview. "
|
|
||||||
"Weave together what matters today: mention the weather, note calendar events and tasks, "
|
|
||||||
"briefly reference one or two noteworthy news themes, and suggest a daily focus. "
|
|
||||||
"Be actionable: for overdue or due-today tasks, suggest a concrete next step. "
|
|
||||||
"Aim for 4 to 8 sentences.\n\n"
|
|
||||||
)
|
|
||||||
elif slot in ("morning", "midday"):
|
|
||||||
base += (
|
|
||||||
"This is a MIDDAY CHECK-IN — brief progress nudge, not a full briefing. "
|
|
||||||
"Focus on what's been done and what hasn't been touched yet today. "
|
|
||||||
"If weather conditions have changed, mention it briefly. "
|
|
||||||
"Keep it to 2-3 sentences. Don't repeat news or the full task list.\n\n"
|
|
||||||
)
|
|
||||||
elif slot == "afternoon":
|
|
||||||
base += (
|
|
||||||
"This is the AFTERNOON WRAP-UP — help the user close out the day. "
|
|
||||||
"Summarize what was accomplished today. Flag anything still overdue. "
|
|
||||||
"Ask if there's anything to capture before tomorrow — notes, thoughts, follow-ups. "
|
|
||||||
"Keep it to 2-4 sentences. Reflective and encouraging tone.\n\n"
|
|
||||||
)
|
|
||||||
elif slot == "weekly_review":
|
|
||||||
base += (
|
|
||||||
"This is the WEEKLY REVIEW — a reflective recap of the past 7 days. "
|
|
||||||
"Summarize accomplishments, flag lingering overdue items, note how many notes were created, "
|
|
||||||
"and preview the upcoming week's key events. "
|
|
||||||
"Be reflective and encouraging — celebrate progress, gently flag what's stuck. "
|
|
||||||
"Aim for 5 to 8 sentences. This wraps up a chapter.\n\n"
|
|
||||||
)
|
|
||||||
|
|
||||||
if profile_body:
|
|
||||||
base += f"User profile:\n{profile_body}\n"
|
|
||||||
return base
|
|
||||||
|
|
||||||
|
|
||||||
def _unified_user_prompt(internal_data: dict, external_data: dict, slot: str, temp_unit: str = "C") -> str:
|
|
||||||
lines = [f"Date: {internal_data['date']}", f"Slot: {slot}", ""]
|
|
||||||
|
|
||||||
# Weekly review — separate data path
|
|
||||||
if slot == "weekly_review":
|
|
||||||
weekly = internal_data.get("weekly_review") or {}
|
|
||||||
lines.append(f"COMPLETED THIS WEEK: {weekly.get('completed_count', 0)} tasks")
|
|
||||||
if weekly.get("completed_highlights"):
|
|
||||||
lines.extend(f" - {t}" for t in weekly["completed_highlights"])
|
|
||||||
lines.append("")
|
|
||||||
if weekly.get("overdue_count", 0) > 0:
|
|
||||||
lines.append(f"STILL OVERDUE: {weekly['overdue_count']} task(s)")
|
|
||||||
lines.extend(f" - {t}" for t in weekly.get("overdue_items", []))
|
|
||||||
lines.append("")
|
|
||||||
lines.append(f"NOTES CREATED: {weekly.get('notes_created', 0)}")
|
|
||||||
lines.append("")
|
|
||||||
if weekly.get("active_projects"):
|
|
||||||
lines.append(f"ACTIVE PROJECTS: {', '.join(weekly['active_projects'])}")
|
|
||||||
lines.append("")
|
|
||||||
if weekly.get("upcoming_events"):
|
|
||||||
lines.append("UPCOMING WEEK:")
|
|
||||||
lines.extend(f" - {e}" for e in weekly["upcoming_events"])
|
|
||||||
lines.append("")
|
|
||||||
return "\n".join(lines)
|
|
||||||
|
|
||||||
# For check-in slots, build a slimmer prompt
|
|
||||||
if slot in ("morning", "midday"):
|
|
||||||
# Just tasks status + any weather changes
|
|
||||||
if internal_data.get("due_today"):
|
|
||||||
lines.append("TASKS DUE TODAY (note which are still untouched):")
|
|
||||||
lines.extend(f" - {t}" for t in internal_data["due_today"])
|
|
||||||
lines.append("")
|
|
||||||
if internal_data.get("overdue_tasks"):
|
|
||||||
lines.append(f"STILL OVERDUE: {len(internal_data['overdue_tasks'])} task(s)")
|
|
||||||
lines.append("")
|
|
||||||
if internal_data.get("calendar_events"):
|
|
||||||
lines.append("REMAINING EVENTS TODAY:")
|
|
||||||
lines.extend(f" - {e}" for e in internal_data["calendar_events"])
|
|
||||||
lines.append("")
|
|
||||||
return "\n".join(lines)
|
|
||||||
|
|
||||||
if slot == "afternoon":
|
|
||||||
if internal_data.get("due_today"):
|
|
||||||
lines.append("TASKS THAT WERE DUE TODAY:")
|
|
||||||
lines.extend(f" - {t}" for t in internal_data["due_today"])
|
|
||||||
lines.append("")
|
|
||||||
if internal_data.get("overdue_tasks"):
|
|
||||||
overdue = internal_data["overdue_tasks"]
|
|
||||||
lines.append(f"OVERDUE ({len(overdue)} remaining):")
|
|
||||||
lines.extend(f" - {t}" for t in overdue[:3])
|
|
||||||
lines.append("")
|
|
||||||
lines.append("Ask the user if there's anything to capture before tomorrow — a note, a thought, a follow-up task.")
|
|
||||||
return "\n".join(lines)
|
|
||||||
|
|
||||||
# ── Full compilation prompt below ──────────────────────────────────────────
|
|
||||||
|
|
||||||
# Weather (brief — card handles detail)
|
|
||||||
weather = external_data.get("weather") or []
|
|
||||||
if weather:
|
|
||||||
loc = weather[0]
|
|
||||||
days = loc.get("days") or []
|
|
||||||
if days:
|
|
||||||
d = days[0]
|
|
||||||
t_min = _format_temp(d["temp_min"], temp_unit)
|
|
||||||
t_max = _format_temp(d["temp_max"], temp_unit)
|
|
||||||
unit_sym = f"°{temp_unit}"
|
|
||||||
lines.append(
|
|
||||||
f"WEATHER: {loc['location_label']} — {d['description']}, "
|
|
||||||
f"{t_min}–{t_max}{unit_sym}"
|
|
||||||
)
|
|
||||||
lines.append("")
|
|
||||||
|
|
||||||
# Today's calendar events
|
|
||||||
if internal_data.get("calendar_events"):
|
|
||||||
lines.append("TODAY'S EVENTS:")
|
|
||||||
lines.extend(f" - {e}" for e in internal_data["calendar_events"])
|
|
||||||
lines.append("")
|
|
||||||
|
|
||||||
# Weather warnings for outdoor events
|
|
||||||
if internal_data.get("weather_warnings"):
|
|
||||||
lines.append("WEATHER ALERTS (mention these naturally when discussing affected events):")
|
|
||||||
lines.extend(f" ⚠ {w}" for w in internal_data["weather_warnings"])
|
|
||||||
lines.append("")
|
|
||||||
|
|
||||||
# Tasks due today
|
|
||||||
if internal_data.get("due_today"):
|
|
||||||
lines.append("DUE TODAY (suggest next steps — offer to mark in progress or help prioritize):")
|
|
||||||
lines.extend(f" - {t}" for t in internal_data["due_today"])
|
|
||||||
lines.append("")
|
|
||||||
|
|
||||||
# Overdue tasks
|
|
||||||
if internal_data.get("overdue_tasks"):
|
|
||||||
overdue = internal_data["overdue_tasks"]
|
|
||||||
lines.append(f"OVERDUE ({len(overdue)} task{'s' if len(overdue) != 1 else ''}) — suggest rescheduling or breaking down:")
|
|
||||||
lines.extend(f" - {t}" for t in overdue[:5])
|
|
||||||
if len(overdue) > 5:
|
|
||||||
lines.append(f" (and {len(overdue) - 5} more)")
|
|
||||||
lines.append("")
|
|
||||||
|
|
||||||
# High priority in-progress
|
|
||||||
if internal_data.get("high_priority"):
|
|
||||||
lines.append("HIGH PRIORITY:")
|
|
||||||
lines.extend(f" - {t}" for t in internal_data["high_priority"])
|
|
||||||
lines.append("")
|
|
||||||
|
|
||||||
# Your Day — prioritized focus + calendar gaps (compilation slot only)
|
|
||||||
if internal_data.get("focus_tasks") and slot == "compilation":
|
|
||||||
lines.append("YOUR DAY — top tasks to focus on (mention these as a suggested plan):")
|
|
||||||
for i, t in enumerate(internal_data["focus_tasks"], 1):
|
|
||||||
lines.append(f" {i}. {t}")
|
|
||||||
if internal_data.get("event_time_blocks"):
|
|
||||||
lines.append("")
|
|
||||||
lines.append(" Scheduled time blocks:")
|
|
||||||
lines.extend(f" {b}" for b in internal_data["event_time_blocks"])
|
|
||||||
lines.append(" (Identify free blocks between events for focused work.)")
|
|
||||||
lines.append("")
|
|
||||||
|
|
||||||
# Project continuity — what was the user working on?
|
|
||||||
if internal_data.get("recent_activity") and slot == "compilation":
|
|
||||||
lines.append("YESTERDAY'S ACTIVITY (mention what the user was working on to help them pick up):")
|
|
||||||
lines.extend(f" - {a}" for a in internal_data["recent_activity"])
|
|
||||||
lines.append("")
|
|
||||||
|
|
||||||
if internal_data.get("stale_projects") and slot == "compilation":
|
|
||||||
lines.append("STALE PROJECTS (no activity in 7+ days — gently flag):")
|
|
||||||
lines.extend(f" - {p}" for p in internal_data["stale_projects"])
|
|
||||||
lines.append("")
|
|
||||||
|
|
||||||
# News — clustered by topic, ranked by preference score
|
|
||||||
rss = external_data.get("rss_items") or []
|
|
||||||
topic_scores = external_data.get("topic_scores") or {}
|
|
||||||
if rss:
|
|
||||||
# Group articles by their primary topic
|
|
||||||
clusters: dict[str, list[dict]] = {}
|
|
||||||
uncategorized: list[dict] = []
|
|
||||||
for item in rss:
|
|
||||||
topics = item.get("topics") or []
|
|
||||||
if topics:
|
|
||||||
primary = topics[0]
|
|
||||||
clusters.setdefault(primary, []).append(item)
|
|
||||||
else:
|
|
||||||
uncategorized.append(item)
|
|
||||||
|
|
||||||
# Score each cluster by aggregate preference (positive = user likes this topic)
|
|
||||||
def cluster_score(topic: str, items: list[dict]) -> float:
|
|
||||||
pref = topic_scores.get(topic, 0.0)
|
|
||||||
return pref * 2.0 + len(items) # preference-weighted, size as tiebreak
|
|
||||||
|
|
||||||
# Filter out clusters where the user has a strongly negative preference
|
|
||||||
filtered_clusters = [
|
|
||||||
(topic, items) for topic, items in clusters.items()
|
|
||||||
if topic_scores.get(topic, 0.0) >= -2.0
|
|
||||||
]
|
|
||||||
|
|
||||||
# Sort by preference-weighted score
|
|
||||||
sorted_clusters = sorted(filtered_clusters, key=lambda x: cluster_score(x[0], x[1]), reverse=True)
|
|
||||||
|
|
||||||
lines.append("NEWS THEMES (synthesize 1-2 themes into your briefing; mention article count per theme):")
|
|
||||||
for i, (topic, items) in enumerate(sorted_clusters[:4]):
|
|
||||||
count = len(items)
|
|
||||||
titles = [it.get("title", "") for it in items[:3]]
|
|
||||||
sources = list({it.get("feed_title") or it.get("source") or "News" for it in items})
|
|
||||||
pref_indicator = ""
|
|
||||||
pref = topic_scores.get(topic, 0.0)
|
|
||||||
if pref >= 2.0:
|
|
||||||
pref_indicator = " ★" # user's preferred topic
|
|
||||||
lines.append(f" [{topic.title()}]{pref_indicator} ({count} article{'s' if count != 1 else ''} from {', '.join(sources[:2])})")
|
|
||||||
for t in titles:
|
|
||||||
lines.append(f" • {t}")
|
|
||||||
# Include excerpt for preferred clusters or the top cluster
|
|
||||||
if pref >= 1.0 or i == 0:
|
|
||||||
excerpt = (items[0].get("content") or items[0].get("snippet") or "")[:400].strip()
|
|
||||||
if excerpt:
|
|
||||||
lines.append(f" > {excerpt}")
|
|
||||||
if uncategorized:
|
|
||||||
lines.append(f" [Other] ({len(uncategorized)} uncategorized article{'s' if len(uncategorized) != 1 else ''})")
|
|
||||||
lines.append("")
|
|
||||||
|
|
||||||
return "\n".join(lines)
|
|
||||||
|
|
||||||
|
|
||||||
def _format_temp(value: float, unit: str) -> str:
|
|
||||||
"""Convert Celsius to the requested unit and format as an integer string."""
|
|
||||||
if unit == "F":
|
|
||||||
return f"{value * 9 / 5 + 32:.0f}"
|
|
||||||
return f"{value:.0f}"
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
# ── Main entry point ───────────────────────────────────────────────────────────
|
# ── Main entry point ───────────────────────────────────────────────────────────
|
||||||
|
|
||||||
async def _get_temp_unit(user_id: int) -> str:
|
async def _get_temp_unit(user_id: int) -> str:
|
||||||
@@ -966,14 +287,16 @@ async def run_compilation(
|
|||||||
model: str | None = None,
|
model: str | None = None,
|
||||||
) -> tuple[str, dict]:
|
) -> tuple[str, dict]:
|
||||||
"""
|
"""
|
||||||
Run the full two-lane briefing pipeline for a user and slot.
|
Run the agentic briefing loop and gather UI metadata (RSS + weather).
|
||||||
Returns (briefing_text, metadata_dict) where metadata contains
|
|
||||||
weather card data and rss_item_ids for frontend rendering.
|
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:
|
if model is None:
|
||||||
model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL)
|
model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL)
|
||||||
|
|
||||||
from fabledassistant.services.user_profile import build_profile_context
|
|
||||||
from fabledassistant.services.briefing_preferences import (
|
from fabledassistant.services.briefing_preferences import (
|
||||||
load_topic_preferences,
|
load_topic_preferences,
|
||||||
load_topic_reaction_scores,
|
load_topic_reaction_scores,
|
||||||
@@ -981,31 +304,15 @@ async def run_compilation(
|
|||||||
)
|
)
|
||||||
from fabledassistant.services.weather import parse_weather_card_data, get_cached_weather_rows
|
from fabledassistant.services.weather import parse_weather_card_data, get_cached_weather_rows
|
||||||
|
|
||||||
profile_context, temp_unit = await asyncio.gather(
|
|
||||||
build_profile_context(user_id),
|
|
||||||
_get_temp_unit(user_id),
|
|
||||||
)
|
|
||||||
|
|
||||||
# ── Pre-processing ──────────────────────────────────────────────────────────
|
|
||||||
include_topics, exclude_topics = await load_topic_preferences(user_id)
|
include_topics, exclude_topics = await load_topic_preferences(user_id)
|
||||||
topic_scores = await load_topic_reaction_scores(user_id)
|
topic_scores = await load_topic_reaction_scores(user_id)
|
||||||
|
|
||||||
# Parallel raw gather — weather rows fetched in same gather to avoid extra DB round-trip
|
external_data, weather_rows, temp_unit = await asyncio.gather(
|
||||||
internal_data, external_data, weather_rows = await asyncio.gather(
|
|
||||||
_gather_internal(user_id),
|
|
||||||
_gather_external(user_id),
|
_gather_external(user_id),
|
||||||
get_cached_weather_rows(user_id),
|
get_cached_weather_rows(user_id),
|
||||||
|
_get_temp_unit(user_id),
|
||||||
)
|
)
|
||||||
|
|
||||||
# Weekly review: gather 7-day summary data
|
|
||||||
if slot == "weekly_review":
|
|
||||||
internal_data["weekly_review"] = await _gather_weekly_review(user_id)
|
|
||||||
|
|
||||||
# Task change detection
|
|
||||||
all_tasks = internal_data.get("all_tasks_raw", [])
|
|
||||||
changed_tasks, unchanged_count = await split_changed_tasks(user_id, all_tasks)
|
|
||||||
|
|
||||||
# RSS filtering
|
|
||||||
raw_rss = external_data.get("rss_items") or []
|
raw_rss = external_data.get("rss_items") or []
|
||||||
filtered_rss = score_and_filter_items(
|
filtered_rss = score_and_filter_items(
|
||||||
raw_rss,
|
raw_rss,
|
||||||
@@ -1028,48 +335,17 @@ async def run_compilation(
|
|||||||
if item.get("id")
|
if item.get("id")
|
||||||
]
|
]
|
||||||
|
|
||||||
# Weather staleness gate — returns None if data is >24h old
|
|
||||||
weather_card = parse_weather_card_data(weather_rows[0], temp_unit) if weather_rows else None
|
weather_card = parse_weather_card_data(weather_rows[0], temp_unit) if weather_rows else None
|
||||||
|
|
||||||
# ── LLM Synthesis ──────────────────────────────────────────────────────────
|
briefing_text, agentic_messages = await run_agentic_briefing(
|
||||||
# Build filtered internal data with only changed tasks
|
user_id, slot, model, conv_id=None,
|
||||||
internal_data_filtered = dict(internal_data)
|
)
|
||||||
internal_data_filtered["unchanged_task_count"] = unchanged_count
|
|
||||||
internal_data_filtered["changed_tasks"] = [format_task(t) for t in changed_tasks]
|
|
||||||
|
|
||||||
# Build filtered external data (suppress weather prose — card handles it)
|
metadata: dict = {
|
||||||
external_data_filtered = {
|
"rss_item_ids": rss_item_ids,
|
||||||
"rss_items": filtered_rss,
|
"rss_items": rss_items_meta,
|
||||||
"weather": [],
|
"weather": weather_card,
|
||||||
"topic_scores": topic_scores,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
briefing_mode = await get_setting(user_id, "briefing_mode", "legacy")
|
|
||||||
agentic_messages: list[dict] = []
|
|
||||||
briefing_text = ""
|
|
||||||
|
|
||||||
if briefing_mode == "agentic":
|
|
||||||
briefing_text, agentic_messages = await run_agentic_briefing(
|
|
||||||
user_id, slot, model, conv_id=None,
|
|
||||||
)
|
|
||||||
if not briefing_text:
|
|
||||||
logger.warning(
|
|
||||||
"Agentic briefing returned empty for user %d slot %s — falling back to legacy path",
|
|
||||||
user_id, slot,
|
|
||||||
)
|
|
||||||
|
|
||||||
if not briefing_text:
|
|
||||||
briefing_text = await _llm_synthesise(
|
|
||||||
_unified_system_prompt(profile_context, slot),
|
|
||||||
_unified_user_prompt(internal_data_filtered, external_data_filtered, slot, temp_unit),
|
|
||||||
model,
|
|
||||||
num_ctx=8192,
|
|
||||||
)
|
|
||||||
|
|
||||||
# ── Post-processing ─────────────────────────────────────────────────────────
|
|
||||||
await upsert_task_snapshots(user_id, all_tasks)
|
|
||||||
|
|
||||||
metadata: dict = {"rss_item_ids": rss_item_ids, "rss_items": rss_items_meta, "weather": weather_card}
|
|
||||||
if agentic_messages:
|
if agentic_messages:
|
||||||
metadata["agentic_messages"] = agentic_messages
|
metadata["agentic_messages"] = agentic_messages
|
||||||
|
|
||||||
@@ -1086,50 +362,18 @@ async def run_slot_injection(
|
|||||||
model: str | None = None,
|
model: str | None = None,
|
||||||
) -> tuple[str, dict]:
|
) -> tuple[str, dict]:
|
||||||
"""
|
"""
|
||||||
Lighter update for 8am/12pm/4pm — generates a slot-specific check-in.
|
Lighter check-in update for 8am/12pm/4pm slots.
|
||||||
|
|
||||||
Honors the ``briefing_mode`` setting just like ``run_compilation``:
|
Runs the agentic loop with the slot-specific prompt. Returns
|
||||||
agentic mode routes through the tool-use loop, legacy mode runs the
|
``(text, metadata)`` where metadata contains ``agentic_messages``
|
||||||
one-shot synthesis path. Falls back to legacy automatically if
|
for the scheduler to persist.
|
||||||
agentic returns empty.
|
|
||||||
|
|
||||||
Returns ``(text, metadata)`` where metadata may contain
|
|
||||||
``agentic_messages`` (the full tool-call sequence) if the agentic
|
|
||||||
path was used.
|
|
||||||
"""
|
"""
|
||||||
if model is None:
|
if model is None:
|
||||||
model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL)
|
model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL)
|
||||||
|
|
||||||
briefing_mode = await get_setting(user_id, "briefing_mode", "legacy")
|
text, agentic_messages = await run_agentic_briefing(
|
||||||
agentic_messages: list[dict] = []
|
user_id, slot, model, conv_id=None,
|
||||||
text = ""
|
)
|
||||||
|
|
||||||
if briefing_mode == "agentic":
|
|
||||||
text, agentic_messages = await run_agentic_briefing(
|
|
||||||
user_id, slot, model, conv_id=None,
|
|
||||||
)
|
|
||||||
if not text:
|
|
||||||
logger.warning(
|
|
||||||
"Agentic slot injection returned empty for user %d slot %s — falling back to legacy",
|
|
||||||
user_id, slot,
|
|
||||||
)
|
|
||||||
|
|
||||||
if not text:
|
|
||||||
internal_data, external_data, temp_unit = await asyncio.gather(
|
|
||||||
_gather_internal(user_id),
|
|
||||||
_gather_external(user_id),
|
|
||||||
_get_temp_unit(user_id),
|
|
||||||
)
|
|
||||||
system = (
|
|
||||||
f"You are a personal assistant giving a brief {slot} check-in. "
|
|
||||||
"The user already had their morning briefing — focus only on what's changed or newly relevant. "
|
|
||||||
"Speak naturally in 2-3 sentences, no markdown formatting, no headers or bullet points."
|
|
||||||
)
|
|
||||||
text = await _llm_synthesise(
|
|
||||||
system,
|
|
||||||
_unified_user_prompt(internal_data, external_data, slot, temp_unit),
|
|
||||||
model,
|
|
||||||
)
|
|
||||||
|
|
||||||
metadata: dict = {}
|
metadata: dict = {}
|
||||||
if agentic_messages:
|
if agentic_messages:
|
||||||
|
|||||||
@@ -444,7 +444,7 @@ async def _run_profile_closeout(user_id: int, model: str) -> None:
|
|||||||
and append them to the briefing profile note.
|
and append them to the briefing profile note.
|
||||||
"""
|
"""
|
||||||
from fabledassistant.services.user_profile import append_observations
|
from fabledassistant.services.user_profile import append_observations
|
||||||
from fabledassistant.services.briefing_pipeline import _llm_synthesise
|
from fabledassistant.services.llm import generate_completion
|
||||||
from fabledassistant.models.conversation import Conversation, Message
|
from fabledassistant.models.conversation import Conversation, Message
|
||||||
|
|
||||||
yesterday = (date.today() - timedelta(days=1))
|
yesterday = (date.today() - timedelta(days=1))
|
||||||
@@ -477,7 +477,17 @@ async def _run_profile_closeout(user_id: int, model: str) -> None:
|
|||||||
"Example: '- User skipped news about finance', '- Prefers weather for home location first'. "
|
"Example: '- User skipped news about finance', '- Prefers weather for home location first'. "
|
||||||
"If nothing notable, output only: (nothing to note)"
|
"If nothing notable, output only: (nothing to note)"
|
||||||
)
|
)
|
||||||
observations = await _llm_synthesise(system, transcript, model)
|
try:
|
||||||
|
observations = (await generate_completion(
|
||||||
|
[
|
||||||
|
{"role": "system", "content": system},
|
||||||
|
{"role": "user", "content": transcript},
|
||||||
|
],
|
||||||
|
model,
|
||||||
|
)).strip()
|
||||||
|
except Exception:
|
||||||
|
logger.warning("Profile closeout synthesis failed for user %d", user_id, exc_info=True)
|
||||||
|
observations = ""
|
||||||
if observations and "(nothing to note)" not in observations.lower():
|
if observations and "(nothing to note)" not in observations.lower():
|
||||||
await append_observations(user_id, observations)
|
await append_observations(user_id, observations)
|
||||||
|
|
||||||
|
|||||||
@@ -114,7 +114,7 @@ async def _consolidate_observations(user_id: int) -> str:
|
|||||||
learned_summary paragraph. Prunes raw entries older than 30 days afterwards.
|
learned_summary paragraph. Prunes raw entries older than 30 days afterwards.
|
||||||
"""
|
"""
|
||||||
from fabledassistant.config import Config
|
from fabledassistant.config import Config
|
||||||
from fabledassistant.services.briefing_pipeline import _llm_synthesise
|
from fabledassistant.services.llm import generate_completion
|
||||||
from fabledassistant.services.settings import get_setting
|
from fabledassistant.services.settings import get_setting
|
||||||
|
|
||||||
async with async_session() as session:
|
async with async_session() as session:
|
||||||
@@ -145,7 +145,17 @@ async def _consolidate_observations(user_id: int) -> str:
|
|||||||
user_prompt += f"New observations:\n{obs_text}"
|
user_prompt += f"New observations:\n{obs_text}"
|
||||||
|
|
||||||
model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL)
|
model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL)
|
||||||
new_summary = await _llm_synthesise(system, user_prompt, model)
|
try:
|
||||||
|
new_summary = (await generate_completion(
|
||||||
|
[
|
||||||
|
{"role": "system", "content": system},
|
||||||
|
{"role": "user", "content": user_prompt},
|
||||||
|
],
|
||||||
|
model,
|
||||||
|
)).strip()
|
||||||
|
except Exception:
|
||||||
|
logger.warning("Observation consolidation failed for user %d", user_id, exc_info=True)
|
||||||
|
new_summary = ""
|
||||||
|
|
||||||
if new_summary:
|
if new_summary:
|
||||||
cutoff = (date.today() - timedelta(days=30)).isoformat()
|
cutoff = (date.today() - timedelta(days=30)).isoformat()
|
||||||
|
|||||||
Reference in New Issue
Block a user