feat: briefing pipeline — parallel gather, two-lane LLM synthesis
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -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)
|
||||
Reference in New Issue
Block a user