From ebbf1a91f25b6faebe591e908a54b68619e08281 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 11 Mar 2026 20:07:10 -0400 Subject: [PATCH] =?UTF-8?q?feat:=20briefing=20pipeline=20=E2=80=94=20paral?= =?UTF-8?q?lel=20gather,=20two-lane=20LLM=20synthesis?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- .../services/briefing_pipeline.py | 286 ++++++++++++++++++ 1 file changed, 286 insertions(+) create mode 100644 src/fabledassistant/services/briefing_pipeline.py diff --git a/src/fabledassistant/services/briefing_pipeline.py b/src/fabledassistant/services/briefing_pipeline.py new file mode 100644 index 0000000..5b56d76 --- /dev/null +++ b/src/fabledassistant/services/briefing_pipeline.py @@ -0,0 +1,286 @@ +""" +Briefing pipeline: parallel data gather + two-lane LLM synthesis. + +Slot names: 'compilation' (4am), 'morning' (8am), 'midday' (12pm), 'afternoon' (4pm) +""" + +import asyncio +import logging +from datetime import date + +import httpx + +from fabledassistant.config import Config +from fabledassistant.services.settings import get_setting + +logger = logging.getLogger(__name__) + +SLOT_NAMES = ("compilation", "morning", "midday", "afternoon") + + +def slot_greeting(slot: str) -> str: + return { + "compilation": "Good morning", + "morning": "Good morning — you're at the office", + "midday": "Midday check-in", + "afternoon": "End of day wrap-up", + }.get(slot, "Update") + + +def format_task(task: dict) -> 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 task.get("priority") and task["priority"] not in ("none", ""): + parts.append(f"priority: {task['priority']}") + return " — ".join(parts) + + +# ── 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 + + today = date.today().isoformat() + + # Tasks: overdue, due today, high priority in-progress + try: + all_task_objs, _total = await list_notes(user_id, is_task=True, limit=100) + all_tasks = [ + { + "title": t.title, + "status": t.status, + "due_date": t.due_date.isoformat() if t.due_date else None, + "priority": t.priority, + } + for t in all_task_objs + ] + overdue = [ + format_task(t) for t in all_tasks + if t.get("due_date") and t["due_date"] < today and t.get("status") != "done" + ] + due_today = [ + format_task(t) for t in all_tasks + if t.get("due_date") == today and t.get("status") != "done" + ] + high_priority = [ + format_task(t) for t in all_tasks + if t.get("priority") == "high" and t.get("status") not in ("done",) and + format_task(t) not in overdue and format_task(t) 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 + calendar_events = [] + try: + if is_caldav_configured(): + events = await list_events(user_id, start=today, end=today) + calendar_events = [ + f"{e.get('summary', 'Event')} at {e.get('dtstart', 'unknown time')}" + for e in (events or []) + ] + except Exception: + logger.warning("Failed to gather calendar events 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.get("title", "Untitled project")) + except Exception: + logger.warning("Failed to gather projects for briefing", exc_info=True) + + return { + "date": today, + "overdue_tasks": overdue, + "due_today": due_today, + "high_priority": high_priority, + "calendar_events": calendar_events, + "active_projects": projects_summary, + } + + +# ── 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 [], + } + + +# ── LLM synthesis ───────────────────────────────────────────────────────────── + +async def _llm_synthesise(system_prompt: str, user_prompt: str, model: str) -> 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": 4096, "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 _internal_system_prompt(profile_body: str) -> str: + return ( + "You are a personal briefing assistant. Your job is to give the user a clear, " + "concise summary of their internal workload: tasks, calendar, and projects. " + "Be direct and prioritised — lead with what's urgent. Use plain text with light " + "markdown. Do not include weather or news.\n\n" + + (f"User profile:\n{profile_body}\n" if profile_body else "") + ) + + +def _external_system_prompt() -> str: + return ( + "You are a briefing assistant for external information. Your job is to summarise " + "the user's RSS feed digest and weather forecast into a concise, engaging update. " + "Group related news items. Note any significant weather changes. " + "Be informative but brief. Do not discuss tasks, calendar, or work items." + ) + + +def _internal_user_prompt(data: dict, slot: str) -> str: + lines = [f"Briefing slot: {slot}", f"Date: {data['date']}", ""] + if data["overdue_tasks"]: + lines.append(f"OVERDUE ({len(data['overdue_tasks'])}):") + lines.extend(f" - {t}" for t in data["overdue_tasks"]) + lines.append("") + if data["due_today"]: + lines.append(f"DUE TODAY ({len(data['due_today'])}):") + lines.extend(f" - {t}" for t in data["due_today"]) + lines.append("") + if data["high_priority"]: + lines.append("HIGH PRIORITY (in progress):") + lines.extend(f" - {t}" for t in data["high_priority"]) + lines.append("") + if data["calendar_events"]: + lines.append("CALENDAR TODAY:") + lines.extend(f" - {e}" for e in data["calendar_events"]) + lines.append("") + if data["active_projects"]: + lines.append(f"ACTIVE PROJECTS: {', '.join(data['active_projects'])}") + return "\n".join(lines) + + +def _external_user_prompt(data: dict, slot: str) -> str: + lines = [f"Briefing slot: {slot}", ""] + if data["weather"]: + lines.append("WEATHER:") + for loc in data["weather"]: + lines.append(f" {loc['location_label']}:") + for day in loc["days"][:3]: + lines.append( + f" {day['date']}: {day['description']}, " + f"{day['temp_min']}–{day['temp_max']}°C, {day['precip_mm']}mm rain" + ) + if loc["changes_since_last_fetch"]: + lines.append(" FORECAST CHANGES:") + lines.extend(f" - {c}" for c in loc["changes_since_last_fetch"]) + lines.append("") + if data["rss_items"]: + lines.append(f"RSS DIGEST ({len(data['rss_items'])} items):") + for item in data["rss_items"][:15]: + lines.append(f" [{item.get('feed_title', 'Feed')}] {item['title']}") + if item.get("content"): + lines.append(f" {item['content'][:200]}") + return "\n".join(lines) + + +# ── Main entry point ─────────────────────────────────────────────────────────── + +async def run_compilation(user_id: int, slot: str, model: str | None = None) -> str: + """ + Run the full two-lane briefing pipeline for a user and slot. + Returns the combined briefing text to be posted as the opening assistant message. + """ + if model is None: + model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL) + + from fabledassistant.services.briefing_profile import get_profile_body + profile_body = await get_profile_body(user_id) + + # Parallel gather + internal_data, external_data = await asyncio.gather( + _gather_internal(user_id), + _gather_external(user_id), + ) + + # Two-lane LLM synthesis (both calls run concurrently) + internal_text, external_text = await asyncio.gather( + _llm_synthesise( + _internal_system_prompt(profile_body), + _internal_user_prompt(internal_data, slot), + model, + ), + _llm_synthesise( + _external_system_prompt(), + _external_user_prompt(external_data, slot), + model, + ), + ) + + greeting = slot_greeting(slot) + today = internal_data["date"] + parts = [f"**{greeting} — {today}**", ""] + if internal_text: + parts += ["## Your Day", "", internal_text, ""] + if external_text: + parts += ["## The World", "", external_text] + + return "\n".join(parts).strip() + + +async def run_slot_injection(user_id: int, slot: str, model: str | None = None) -> str: + """ + Lighter update for 8am/12pm/4pm — gathers fresh data and produces a slot-specific + update prompt. Returns the text to inject as a new user→assistant exchange. + """ + if model is None: + model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL) + + internal_data, external_data = await asyncio.gather( + _gather_internal(user_id), + _gather_external(user_id), + ) + + system = ( + f"You are a briefing assistant providing a {slot} update. Be brief — " + "the user has already seen the morning briefing. Focus on what's changed or new." + ) + user_prompt = ( + f"Slot: {slot}\n\n" + + _internal_user_prompt(internal_data, slot) + + "\n\n" + + _external_user_prompt(external_data, slot) + ) + return await _llm_synthesise(system, user_prompt, model)