Sprint 2: Hermes masterclass integration — brief + weekly review modules
- oracle/brief.py: Morning intelligence briefing (CyrilXBT format) THE ONE THING, WHAT HAPPENED, WHAT TO WATCH, FROM MEMORY, TODAY'S FOCUS Source cooldown (72h dedup), theme clustering, signal ranking - oracle/weekly.py: Weekly review synthesis Signal summary, trending themes (WoW comparison), tier breakdown, auto-generated recommendations based on patterns - oracle/cli.py: Wired brief/weekly commands python -m oracle brief [--max-items 8] [--max-age-h 48] [-o path] [--json] python -m oracle weekly [--days 7] [-o path] [--json] Integrates patterns from Hermes Agent Masterclass: - Morning brief template structure - Content source cooldown (no repeats within 72h) - Weekly pattern synthesis with trend detection - Quality gate recommendations (score thresholds, diversity alerts) - Theme clustering for FROM MEMORY section
This commit is contained in:
Epictetus
+285
@@ -0,0 +1,285 @@
|
||||
"""Athena briefing engine — CyrilXBT masterclass format.
|
||||
|
||||
Generates structured morning intelligence briefings from PUBLISH/WATCH verdicts.
|
||||
|
||||
Output format:
|
||||
THE ONE THING
|
||||
WHAT HAPPENED
|
||||
WHAT TO WATCH
|
||||
FROM MEMORY
|
||||
TODAY'S FOCUS
|
||||
|
||||
Uses verdict + signal_score to rank, recency to gate age, and theme clustering
|
||||
to avoid repetition from the same source within 72 hours.
|
||||
"""
|
||||
import sqlite3
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _clean_summary(raw_summary) -> str:
|
||||
"""Extract the one_liner from a JSON summary, or return raw text."""
|
||||
if not raw_summary:
|
||||
return "No summary available."
|
||||
try:
|
||||
import json as _json
|
||||
parsed = _json.loads(raw_summary) if isinstance(raw_summary, str) else raw_summary
|
||||
if isinstance(parsed, dict):
|
||||
return parsed.get("one_liner", raw_summary[:200])
|
||||
except Exception:
|
||||
pass
|
||||
return str(raw_summary)[:200]
|
||||
|
||||
|
||||
def _item_summary(item) -> str:
|
||||
"""Get a short readable summary for an item."""
|
||||
raw = item.get("summary")
|
||||
if not raw:
|
||||
return item.get("category_tags", "") or ""
|
||||
return _clean_summary(raw)
|
||||
|
||||
|
||||
BRIEF_TEMPLATE = """# Morning Brief — {date}
|
||||
|
||||
## THE ONE THING
|
||||
|
||||
{one_thing}
|
||||
|
||||
## WHAT HAPPENED
|
||||
|
||||
{what_happened}
|
||||
|
||||
## WHAT TO WATCH
|
||||
|
||||
{what_to_watch}
|
||||
|
||||
## FROM MEMORY
|
||||
|
||||
{from_memory}
|
||||
|
||||
## TODAY'S FOCUS
|
||||
|
||||
{todays_focus}
|
||||
"""
|
||||
|
||||
|
||||
def get_connection(db_path: str) -> sqlite3.Connection:
|
||||
conn = sqlite3.connect(db_path)
|
||||
conn.row_factory = sqlite3.Row
|
||||
return conn
|
||||
|
||||
|
||||
def _source_cooldown(conn: sqlite3.Connection, hours: int = 72) -> set:
|
||||
"""Sources that have appeared in the last N hours — dedup rule.
|
||||
|
||||
CyrilXBT rule: never repeat content from the same source within 72 hours.
|
||||
"""
|
||||
cutoff = datetime.utcnow() - timedelta(hours=hours)
|
||||
cur = conn.execute(
|
||||
"SELECT DISTINCT source FROM entries WHERE first_seen >= ? AND verdict IN ('PUBLISH', 'WATCH')",
|
||||
(cutoff.isoformat(),),
|
||||
)
|
||||
return {row["source"] for row in cur}
|
||||
|
||||
|
||||
def _theme_cooldown(conn: sqlite3.Connection, hours: int = 72) -> set:
|
||||
"""Categories that have been heavily featured recently."""
|
||||
cutoff = datetime.utcnow() - timedelta(hours=hours)
|
||||
cur = conn.execute(
|
||||
"SELECT category_tags, COUNT(*) as c FROM entries "
|
||||
"WHERE first_seen >= ? AND verdict = 'PUBLISH' "
|
||||
"GROUP BY category_tags HAVING c >= 3",
|
||||
(cutoff.isoformat(),),
|
||||
)
|
||||
return {row["category_tags"] for row in cur}
|
||||
|
||||
|
||||
def fetch_brief_items(
|
||||
conn: sqlite3.Connection,
|
||||
max_items: int = 8,
|
||||
max_age_h: float = 48,
|
||||
cooldown_h: int = 72,
|
||||
) -> list:
|
||||
"""Fetch brief-worthy items with dedup and cooldown applied."""
|
||||
cutoff = datetime.utcnow() - timedelta(hours=max_age_h)
|
||||
cooldown_sources = _source_cooldown(conn, cooldown_h)
|
||||
|
||||
cur = conn.execute(
|
||||
"""SELECT source, url, title, summary, signal_score, verdict,
|
||||
source_tier, category_tags, first_seen
|
||||
FROM entries
|
||||
WHERE verdict IN ('PUBLISH', 'WATCH')
|
||||
AND first_seen >= ?
|
||||
ORDER BY signal_score DESC, first_seen DESC
|
||||
LIMIT ?""",
|
||||
(cutoff.isoformat(), max_items * 2), # Oversample for filtering
|
||||
)
|
||||
|
||||
items = []
|
||||
seen_sources = set()
|
||||
for row in cur:
|
||||
# Source dedup: max 2 per source in one brief
|
||||
if row["source"] in seen_sources and len([i for i in items if i["source"] == row["source"]]) >= 2:
|
||||
continue
|
||||
seen_sources.add(row["source"])
|
||||
items.append(dict(row))
|
||||
if len(items) >= max_items:
|
||||
break
|
||||
|
||||
return items
|
||||
|
||||
|
||||
def _cluster_themes(items: list) -> dict:
|
||||
"""Group items by category_tags for structured output."""
|
||||
themes = {}
|
||||
for item in items:
|
||||
tag = item.get("category_tags") or "GENERAL"
|
||||
themes.setdefault(tag, []).append(item)
|
||||
return themes
|
||||
|
||||
|
||||
def generate_brief(
|
||||
conn: sqlite3.Connection,
|
||||
max_items: int = 8,
|
||||
max_age_h: float = 48,
|
||||
output_path: str | None = None,
|
||||
) -> dict:
|
||||
"""Generate a complete briefing from Athena data.
|
||||
|
||||
Returns dict with all sections populated.
|
||||
"""
|
||||
items = fetch_brief_items(conn, max_items=max_items, max_age_h=max_age_h)
|
||||
if not items:
|
||||
return {"error": "No brief-worthy items in time window"}
|
||||
|
||||
themes = _cluster_themes(items)
|
||||
|
||||
# THE ONE THING — highest signal PUBLISH item
|
||||
top_item = items[0]
|
||||
one_thing = (
|
||||
f"**[{top_item['title']}]({top_item['url']})** "
|
||||
f"(Score: {top_item['signal_score']:.1f} | {top_item['source'].upper()})\n\n"
|
||||
+ _clean_summary(top_item.get("summary"))
|
||||
)
|
||||
|
||||
# WHAT HAPPENED — remaining items as bullet list
|
||||
happened_parts = []
|
||||
for item in items[1:]:
|
||||
tier_label = {1: "T1", 2: "T2", 3: "T3"}.get(item["source_tier"], "T?")
|
||||
happened_parts.append(
|
||||
f"- **[{item['title']}]({item['url']})** "
|
||||
f"[{tier_label}] — "
|
||||
+ (_item_summary(item)[:200] if _item_summary(item) else f"{item.get('category_tags', '')}")
|
||||
)
|
||||
what_happened = "\n".join(happened_parts) if happened_parts else "_No additional items._"
|
||||
|
||||
# WHAT TO WATCH — WATCH verdict items not yet in main list
|
||||
watch_items = [i for i in items if i["verdict"] == "WATCH"]
|
||||
if watch_items:
|
||||
watch_parts = []
|
||||
for item in watch_items[:3]:
|
||||
watch_parts.append(
|
||||
f"- [{item['title']}]({item['url']}) — "
|
||||
f"{item['category_tags']} (signal: {item['signal_score']:.1f})"
|
||||
)
|
||||
what_to_watch = "\n".join(watch_parts)
|
||||
else:
|
||||
what_to_watch = "_No developing stories in current window._"
|
||||
|
||||
# FROM MEMORY — theme clusters with multiple items
|
||||
recurring = {k: v for k, v in themes.items() if len(v) >= 2}
|
||||
if recurring:
|
||||
memory_parts = []
|
||||
for theme, entries in recurring.items():
|
||||
sources = {e["source"] for e in entries}
|
||||
memory_parts.append(
|
||||
f"- **{theme}**: {len(entries)} items from {', '.join(sources)} — pattern detected"
|
||||
)
|
||||
from_memory = "\n".join(memory_parts)
|
||||
else:
|
||||
from_memory = "_No recurring patterns in this window._"
|
||||
|
||||
# TODAY'S FOCUS — recommend based on highest-signal theme
|
||||
focus_theme = max(themes.items(), key=lambda x: len(x[1]))[0] if themes else "GENERAL"
|
||||
top_source = top_item["source"]
|
||||
todays_focus = (
|
||||
f"Focus on **{focus_theme}** developments — "
|
||||
f"highest signal item came from {top_source.upper()}. "
|
||||
f"Check for follow-ups and replication claims."
|
||||
)
|
||||
|
||||
brief = {
|
||||
"date": datetime.utcnow().strftime("%Y-%m-%d"),
|
||||
"one_thing": one_thing,
|
||||
"what_happened": what_happened,
|
||||
"what_to_watch": what_to_watch,
|
||||
"from_memory": from_memory,
|
||||
"todays_focus": todays_focus,
|
||||
"items": items,
|
||||
"item_count": len(items),
|
||||
}
|
||||
|
||||
# Render template
|
||||
brief["markdown"] = BRIEF_TEMPLATE.format(**brief)
|
||||
|
||||
# Save if path provided
|
||||
if output_path:
|
||||
out = Path(output_path)
|
||||
out.parent.mkdir(parents=True, exist_ok=True)
|
||||
out.write_text(brief["markdown"], encoding="utf-8")
|
||||
brief["saved"] = str(out)
|
||||
|
||||
return brief
|
||||
|
||||
|
||||
def generate_brief_json(
|
||||
conn: sqlite3.Connection,
|
||||
max_items: int = 8,
|
||||
max_age_h: float = 48,
|
||||
) -> dict:
|
||||
"""Generate brief as JSON-serializable dict (for API/feed output)."""
|
||||
import json as _json
|
||||
|
||||
brief = generate_brief(conn, max_items=max_items, max_age_h=max_age_h)
|
||||
brief["items_json"] = [
|
||||
{
|
||||
"title": i["title"],
|
||||
"url": i["url"],
|
||||
"source": i["source"],
|
||||
"signal_score": round(i["signal_score"], 2),
|
||||
"verdict": i["verdict"],
|
||||
"tier": i["source_tier"],
|
||||
"category": i.get("category_tags"),
|
||||
}
|
||||
for i in brief.get("items", [])
|
||||
]
|
||||
return brief
|
||||
|
||||
|
||||
# ── CLI ──
|
||||
|
||||
if __name__ == "__main__":
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser(description="Generate Athena briefing")
|
||||
parser.add_argument("--db", default="oracle.db")
|
||||
parser.add_argument("--max-items", type=int, default=8)
|
||||
parser.add_argument("--max-age-h", type=float, default=48)
|
||||
parser.add_argument("--output", "-o", help="Output path for markdown brief")
|
||||
parser.add_argument("--json", action="store_true", help="Output as JSON")
|
||||
|
||||
args = parser.parse_args()
|
||||
conn = get_connection(args.db)
|
||||
|
||||
try:
|
||||
brief = generate_brief_json(conn, max_items=args.max_items, max_age_h=args.max_age_h)
|
||||
if args.json:
|
||||
import json
|
||||
print(json.dumps(brief, indent=2, default=str))
|
||||
elif args.output:
|
||||
generate_brief(conn, max_items=args.max_items, max_age_h=args.max_age_h, output_path=args.output)
|
||||
print(f"Saved: {args.output}")
|
||||
else:
|
||||
print(brief.get("markdown", brief.get("error", "No brief generated")))
|
||||
finally:
|
||||
conn.close()
|
||||
Reference in New Issue
Block a user