Epictetus
67c002b665
Source-controlled baseline before Phase 5 cron. Excludes oracle.db, logs/, and __pycache__ via .gitignore. Pipeline verified running clean end-to-end (run_log write confirmed before conn.close()).
553 lines
18 KiB
Python
553 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
AI Research Oracle — Summarization Engine (v1).
|
|
|
|
Generates structured summaries for entries where summary IS NULL.
|
|
Uses source-specific extraction logic (no LLM required — eliminates hallucination).
|
|
|
|
Output schema: {one_liner, key_technical_point, potential_use_case, confidence}
|
|
|
|
Architecture note: This v1 uses deterministic extraction rules to avoid
|
|
hallucination. When a local LLM becomes available (Ollama GPU, Hermes API),
|
|
swap in LLM mode via --llm flag. The DB schema is identical.
|
|
|
|
Usage:
|
|
python3 summarize.py # summarize all pending
|
|
python3 summarize.py --source github # specific source
|
|
python3 summarize.py --limit 10 # max entries
|
|
python3 summarize.py --verify # spot-check 2-3 summaries
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timezone
|
|
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
|
|
|
|
def extract_github_summary(title: str, content: str) -> dict:
|
|
"""Extract summary from GitHub README content.
|
|
|
|
Strategy: Clean HTML, find the first substantive paragraph that
|
|
describes the project (usually below the badges), extract the
|
|
"what it does" sentence.
|
|
"""
|
|
# Aggressive HTML cleaning
|
|
text = re.sub(r'<p[^>]*>', '\n', content)
|
|
text = re.sub(r'</p>', '\n', content)
|
|
text = re.sub(r'<h[1-6][^>]*>', '\n## ', text)
|
|
text = re.sub(r'</h[1-6]>', '\n', text)
|
|
text = re.sub(r'<[^>]+>', '', text)
|
|
text = re.sub(r'&', '&', text)
|
|
text = re.sub(r'—', '—', text)
|
|
text = re.sub(r''', "'", text)
|
|
text = re.sub(r'·', '·', text)
|
|
# Remove code blocks (``` ... ```) — often ASCII art
|
|
text = re.sub(r'```[\s\S]*?```', '', text)
|
|
text = re.sub(r'\n\s*\n+', '\n\n', text)
|
|
text = text.strip()
|
|
|
|
# Confidence starts from source content quality
|
|
source_confidence = "low"
|
|
if len(text) > 2000:
|
|
source_confidence = "high"
|
|
elif len(text) > 500:
|
|
source_confidence = "medium"
|
|
|
|
# Find the one-liner: look for project description paragraph
|
|
one_liner = _find_project_description(text, title)
|
|
if not one_liner:
|
|
one_liner = title[:200]
|
|
|
|
# Key technical point
|
|
key_tech = _extract_technical_point(text, source_confidence)
|
|
|
|
# Use case
|
|
use_case = _extract_use_case(text, title)
|
|
|
|
# Quality-gate confidence on extraction signals, not raw length
|
|
confidence = _assess_extraction_quality(one_liner, key_tech, use_case, source_confidence)
|
|
|
|
# Tag security tooling if detected
|
|
if _is_security_tooling(title, one_liner, key_tech):
|
|
use_case = use_case + " [security:dual-use]"
|
|
|
|
return {
|
|
"one_liner": one_liner[:200],
|
|
"key_technical_point": key_tech[:200],
|
|
"potential_use_case": use_case[:200],
|
|
"confidence": confidence,
|
|
}
|
|
|
|
|
|
def extract_arxiv_summary(title: str, content: str) -> dict:
|
|
"""Extract summary from arXiv abstract.
|
|
|
|
Strategy: arXiv abstracts have a predictable structure:
|
|
1. Background/motivation
|
|
2. "In this paper we propose..."
|
|
3. Results
|
|
4. Implications
|
|
|
|
We extract the contribution statement and key finding.
|
|
"""
|
|
text = re.sub(r'<[^>]+>', ' ', content)
|
|
text = re.sub(r'\s+', ' ', text).strip()
|
|
|
|
# Confidence based on abstract clarity
|
|
confidence = "high" if len(text) > 300 else "medium"
|
|
|
|
# One-liner: find the contribution statement
|
|
one_liner = _find_contribution(text)
|
|
if not one_liner:
|
|
# Fallback: use title as base
|
|
one_liner = f"This paper presents {title.lower()}"
|
|
|
|
# Key technical point: look for method description
|
|
key_tech = _extract_method(text)
|
|
|
|
# Use case: look for application statements
|
|
use_case = _extract_application(text)
|
|
|
|
return {
|
|
"one_liner": one_liner[:200],
|
|
"key_technical_point": key_tech[:200],
|
|
"potential_use_case": use_case[:200],
|
|
"confidence": confidence,
|
|
}
|
|
|
|
|
|
def extract_reddit_summary(title: str, content: str) -> dict:
|
|
"""Extract summary from Reddit post.
|
|
|
|
Strategy: Reddit posts vary wildly in quality. Extract the core
|
|
question or claim, note if it's discussion vs announcement.
|
|
"""
|
|
text = re.sub(r'<[^>]+>', ' ', content)
|
|
text = re.sub(r'\s+', ' ', text).strip()
|
|
|
|
# Confidence based on content length
|
|
if len(text) > 500:
|
|
confidence = "high"
|
|
elif len(text) > 100:
|
|
confidence = "medium"
|
|
else:
|
|
confidence = "low"
|
|
|
|
# One-liner from title (Reddit titles are usually the summary)
|
|
one_liner = title[:200] if title else text[:150]
|
|
|
|
# Key technical point from content
|
|
key_tech = text[:200] if text else "No additional content in post"
|
|
|
|
# Use case: community relevance
|
|
use_case = "AI community discussion"
|
|
|
|
return {
|
|
"one_liner": one_liner,
|
|
"key_technical_point": key_tech,
|
|
"potential_use_case": use_case,
|
|
"confidence": confidence,
|
|
}
|
|
|
|
|
|
# --- Extraction helpers ---
|
|
|
|
def _assess_extraction_quality(one_liner: str, key_tech: str, use_case: str, source_confidence: str) -> str:
|
|
"""Assess extraction quality based on output signals, not source length.
|
|
|
|
A short-but-complete Reddit title should score higher confidence
|
|
than a long README that yielded a fragment.
|
|
"""
|
|
score = 0
|
|
penalties = 0
|
|
|
|
# One-liner quality
|
|
ol = one_liner.strip()
|
|
ol_len = len(ol)
|
|
|
|
# Length window: 40-200 chars is a reasonable sentence
|
|
if 40 <= ol_len <= 200:
|
|
score += 2
|
|
elif 20 <= ol_len < 40:
|
|
score += 1
|
|
elif ol_len > 200:
|
|
penalties += 1 # too long, likely grabbed too much
|
|
|
|
# Ends with terminal punctuation
|
|
if ol.endswith(('.', '!', '?', '…')):
|
|
score += 1
|
|
else:
|
|
penalties += 1
|
|
|
|
# Contains subject-verb pattern (basic heuristic)
|
|
if re.search(r'\b(?:is|are|provides|enables|implements|makes|allows|builds|creates|runs|uses)\b', ol, re.I):
|
|
score += 1
|
|
# Or starts with a proper noun/capitalized phrase
|
|
elif re.match(r'^[A-Z]\w+', ol) and ol_len > 30:
|
|
score += 0.5
|
|
|
|
# No unmatched brackets (artifact from markdown/HTML)
|
|
open_brackets = ol.count('[') + ol.count('(')
|
|
close_brackets = ol.count(']') + ol.count(')')
|
|
if abs(open_brackets - close_brackets) > 0:
|
|
penalties += 1
|
|
if open_brackets > 2:
|
|
penalties += 1 # likely grabbed markdown link syntax
|
|
|
|
# Key technical point quality
|
|
kt = key_tech.strip()
|
|
if kt and len(kt) > 20 and not kt.startswith('See '):
|
|
score += 1
|
|
else:
|
|
penalties += 0.5
|
|
|
|
# Use case quality
|
|
uc = use_case.strip()
|
|
if uc and len(uc) > 10 and not uc.startswith('Relevant for'):
|
|
score += 1
|
|
else:
|
|
penalties += 0.5
|
|
|
|
# Final confidence based on score - penalties
|
|
net = score - penalties
|
|
if net >= 3:
|
|
return source_confidence # extraction is good, trust source quality
|
|
elif net >= 1:
|
|
return "medium"
|
|
else:
|
|
return "low"
|
|
|
|
|
|
def _is_security_tooling(title: str, one_liner: str, key_tech: str) -> bool:
|
|
"""Detect if a project is security/offensive tooling."""
|
|
combined = f"{title} {one_liner} {key_tech}".lower()
|
|
security_signals = [
|
|
"offensive", "pentest", "red team", "exploit", "kill chain",
|
|
"attack surface", "vulnerability scan", "zero-day",
|
|
"reverse engineer", "c2", "command and control",
|
|
]
|
|
return any(sig in combined for sig in security_signals)
|
|
|
|
|
|
def _find_project_description(text: str, title: str) -> str | None:
|
|
"""Find the project description paragraph in a README."""
|
|
paras = text.split('\n\n')
|
|
proj_name = title.split(':')[0].split('/')[0].strip().lower()
|
|
|
|
for para in paras:
|
|
para = para.strip()
|
|
if not para or para.startswith('##') or len(para) < 20:
|
|
continue
|
|
# Skip badges, stats lines, separator lines
|
|
if 'img' in para.lower() or 'badge' in para.lower() or 'shields' in para.lower():
|
|
continue
|
|
# Skip lines that start with stats (~54%, etc.)
|
|
if re.match(r'^[~$#€£¥*»\d]', para):
|
|
continue
|
|
# Skip ASCII art (high ratio of special chars)
|
|
special_chars = sum(1 for c in para if not c.isalnum() and not c.isspace() and c not in ',.!?;:\'"-()[]')
|
|
if special_chars / max(len(para), 1) > 0.4:
|
|
continue
|
|
if len(para) < 40:
|
|
continue
|
|
# Good paragraph — extract first sentence
|
|
sentence = re.split(r'[.!?]', para)[0].strip()
|
|
if len(sentence) > 30:
|
|
return sentence + '.'
|
|
|
|
# Fallback: look for "is a" pattern anywhere
|
|
patterns = [
|
|
rf'{re.escape(proj_name[:20])}\s+(?:is|enables|provides|implements)\s+[^.]+\.?',
|
|
r'(?:This\s+)?(?:project|library|framework|tool|package)\s+(?:is|enables|provides)\s+[^.]+\.?',
|
|
]
|
|
for pattern in patterns:
|
|
match = re.search(pattern, text, re.I)
|
|
if match:
|
|
return match.group(0)[:200]
|
|
|
|
return None
|
|
|
|
|
|
def _find_what_sentence(text: str, title: str) -> str | None:
|
|
"""Find the 'X is a...' sentence that describes what the project does."""
|
|
patterns = [
|
|
rf'{re.escape(title[:30])}\s+(?:is|enables|provides|implements)\s+[^.]+\.?',
|
|
r'(?:This\s+)?(?:project|library|framework|tool|package)\s+(?:is|enables|provides|implements)\s+[^.]+\.?',
|
|
r'(?:makes|allows)\s+[^\s]+\s+(?:to|can)\s+[^.]+\.?',
|
|
r'(?:\w+\s+(?:is|provides|enables|implements|delivers))\s+[a-z].{10,100}\.',
|
|
]
|
|
|
|
for pattern in patterns:
|
|
match = re.search(pattern, text, re.I)
|
|
if match:
|
|
return match.group(0)[:200]
|
|
|
|
# Fallback: first meaningful paragraph
|
|
for para in text.split('\n\n'):
|
|
para = para.strip()
|
|
if len(para) > 30 and not para.startswith('#'):
|
|
return para[:200]
|
|
return None
|
|
|
|
|
|
def _find_contribution(text: str) -> str | None:
|
|
"""Find the 'we propose/introduce/present' statement in an abstract."""
|
|
patterns = [
|
|
r'(?:we|this\s+paper)\s+(?:propose|introduce|present|propose and evaluate)\s+[^.]{10,150}\.',
|
|
r'(?:we\s+(?:show|demonstrate|find|discover|observe))\s+[^.]{10,150}\.',
|
|
r'(?:we\s+(?:introduce|present|propose))\s+(?:a|an|our)\s+\w+\s+[^.]{5,150}\.',
|
|
]
|
|
|
|
for pattern in patterns:
|
|
match = re.search(pattern, text, re.I)
|
|
if match:
|
|
return match.group(0)[:200]
|
|
|
|
# Fallback: first sentence
|
|
first = re.split(r'[.!?]', text)[0].strip()
|
|
return first if first else None
|
|
|
|
|
|
def _extract_technical_point(text: str, confidence: str) -> str:
|
|
"""Extract the main technical approach or innovation."""
|
|
patterns = [
|
|
r'architecture(?:\s+designed)?\s+(?:for|to|that)\s+[^.]+\.?',
|
|
r'(?:using|via|based\s+on|through)\s+[a-z][^.]{10,100}\.',
|
|
r'(?:novel|new|unique|innovative)\s+\w+\s+[^.]{5,80}\.',
|
|
]
|
|
|
|
for pattern in patterns:
|
|
match = re.search(pattern, text, re.I)
|
|
if match:
|
|
return match.group(0)[:200]
|
|
|
|
# Fallback: confidence-based
|
|
if confidence == "low":
|
|
return "Technical details not available in extracted content"
|
|
return "See README for technical details"
|
|
|
|
|
|
def _extract_method(text: str) -> str:
|
|
"""Extract the method/approach from an arXiv abstract."""
|
|
patterns = [
|
|
r'(?:method|approach|framework|technique|model|system)\s+(?:based|using|via|through|with)\s+[a-z][^.]{10,120}\.',
|
|
r'(?:combining|leveraging|exploiting)\s+[a-z][^.]{10,120}\.',
|
|
r'(?:learn|train|optimize|generate)\s+[a-z][^.]{10,120}\.',
|
|
]
|
|
|
|
for pattern in patterns:
|
|
match = re.search(pattern, text, re.I)
|
|
if match:
|
|
return match.group(0)[:200]
|
|
|
|
# Fallback: core contribution
|
|
for pattern in [
|
|
r'(?:propose|introduce)\s+(?:a|an)\s+[^.]{10,100}\.',
|
|
]:
|
|
match = re.search(pattern, text, re.I)
|
|
if match:
|
|
return match.group(0)[:200]
|
|
|
|
return "See full paper for methodology"
|
|
|
|
|
|
def _extract_use_case(text: str, title: str) -> str:
|
|
"""Extract potential use case from README content."""
|
|
patterns = [
|
|
r'(?:for|to)\s+(?:developers|engineers|researchers|teams)\s+who?\s+[^.]{5,80}\.',
|
|
r'(?:enables|allows|helps)\s+[^\s]+\s+to\s+[^.]{10,80}\.',
|
|
r'(?:use\s+case|application|target\s+user)\s*:\s*[^.]{10,80}\.',
|
|
]
|
|
|
|
for pattern in patterns:
|
|
match = re.search(pattern, text, re.I)
|
|
if match:
|
|
return match.group(0)[:200]
|
|
|
|
return f"Relevant for {title.lower()[:50]} developers and users"
|
|
|
|
|
|
def _extract_application(text: str) -> str:
|
|
"""Extract application/use case from arXiv abstract."""
|
|
patterns = [
|
|
r'(?:application|use\s+case|can\s+be\s+used|could\s+be\s+applied)\s+(?:for|in|to)\s+[a-z][^.]{10,80}\.',
|
|
r'(?:improve|enhance|advance)\s+[a-z][^.]{10,80}\.',
|
|
]
|
|
|
|
for pattern in patterns:
|
|
match = re.search(pattern, text, re.I)
|
|
if match:
|
|
return match.group(0)[:200]
|
|
|
|
# Generic fallback based on title keywords
|
|
title_lower = text[:200].lower()
|
|
if any(k in title_lower for k in ["agent", "agentic"]):
|
|
return "Building AI agent systems"
|
|
elif any(k in title_lower for k in ["verification", "verify"]):
|
|
return "LLM output verification and reliability"
|
|
elif any(k in title_lower for k in ["embodied", "robot"]):
|
|
return "Embodied AI and robotics applications"
|
|
elif any(k in title_lower for k in ["distill"]):
|
|
return "Model distillation and knowledge transfer"
|
|
return "See paper for specific applications"
|
|
|
|
|
|
def summarize_entry(entry: dict, conn: sqlite3.Connection) -> bool:
|
|
"""Summarize a single entry using rule-based extraction."""
|
|
source = entry["source"]
|
|
title = entry["title"]
|
|
content = entry.get("extracted_text", "")
|
|
eid = entry["id"]
|
|
|
|
if not content or len(content) < 50:
|
|
return False
|
|
|
|
# Source-specific extraction
|
|
if source == "github":
|
|
summary = extract_github_summary(title, content)
|
|
elif source == "arxiv":
|
|
summary = extract_arxiv_summary(title, content)
|
|
elif source == "reddit":
|
|
summary = extract_reddit_summary(title, content)
|
|
else:
|
|
summary = extract_reddit_summary(title, content) # fallback
|
|
|
|
# Store
|
|
cur = conn.cursor()
|
|
cur.execute("UPDATE entries SET summary = ? WHERE id = ?",
|
|
(json.dumps(summary), eid))
|
|
conn.commit()
|
|
return True
|
|
|
|
|
|
def verify_summaries(conn: sqlite3.Connection, source: str, sample_size: int = 3):
|
|
"""Spot-check summaries against source text.
|
|
|
|
Look for hallucinated specifics: numbers, claims, features not
|
|
present in the original extracted_text.
|
|
|
|
NOTE: Rule-based extraction v1 is inherently lower-risk for
|
|
hallucination since it extracts actual text, not generates new claims.
|
|
But we still verify the extraction logic is working correctly.
|
|
"""
|
|
cur = conn.cursor()
|
|
cur.execute("""
|
|
SELECT id, title, extracted_text, summary
|
|
FROM entries WHERE source = ? AND summary IS NOT NULL
|
|
ORDER BY RANDOM()
|
|
LIMIT ?
|
|
""", (source, sample_size))
|
|
|
|
rows = cur.fetchall()
|
|
if not rows:
|
|
print(f" No summaries to verify for {source}")
|
|
return
|
|
|
|
for eid, title, source_text, summary_json in rows:
|
|
summary = json.loads(summary_json)
|
|
one_liner = summary.get("one_liner", "")
|
|
confidence = summary.get("confidence", "?")
|
|
|
|
issues = []
|
|
|
|
# Check: does the one-liner contain text actually present in source?
|
|
# (For rule-based extraction, this should always be true)
|
|
words = one_liner.split()[:5]
|
|
found = sum(1 for w in words if w.lower() in source_text.lower())
|
|
if found < 3:
|
|
issues.append(f"Low overlap: {found}/5 words from source")
|
|
|
|
# Check: confidence matches content length
|
|
if confidence == "high" and len(source_text) < 500:
|
|
issues.append("High confidence on short source")
|
|
elif confidence == "low" and len(source_text) > 2000:
|
|
issues.append("Low confidence on long source")
|
|
|
|
if issues:
|
|
print(f" ⚠ [{eid}] {title[:50]}... issues: {'; '.join(issues)}")
|
|
print(f" Summary: {one_liner[:80]}...")
|
|
else:
|
|
print(f" ✓ [{eid}] {title[:50]}... confidence={confidence}")
|
|
|
|
time.sleep(0.3)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="AI Research Oracle — Summarization")
|
|
parser.add_argument("--source", default=None, help="Filter by source (github/arxiv/reddit)")
|
|
parser.add_argument("--limit", type=int, default=0, help="Max entries (0=all)")
|
|
parser.add_argument("--verify", action="store_true", help="Spot-check summaries")
|
|
args = parser.parse_args()
|
|
|
|
db_path = os.path.join(os.path.dirname(__file__), "oracle.db")
|
|
conn = sqlite3.connect(db_path)
|
|
cur = conn.cursor()
|
|
|
|
# Find pending entries
|
|
where = "summary IS NULL"
|
|
params = []
|
|
if args.source:
|
|
where += " AND source = ?"
|
|
params.append(args.source)
|
|
|
|
cur.execute(f"SELECT COUNT(*) FROM entries WHERE {where}", params)
|
|
total_pending = cur.fetchone()[0]
|
|
print(f"=== Summarization Engine (Rule-based v1) ===")
|
|
print(f" Pending entries: {total_pending}")
|
|
|
|
if total_pending == 0:
|
|
print(" Nothing to summarize.")
|
|
conn.close()
|
|
return
|
|
|
|
# Fetch entries
|
|
limit_clause = " LIMIT ?" if args.limit > 0 else ""
|
|
limit_params = params + [args.limit] if args.limit > 0 else params
|
|
|
|
cur.execute(f"""
|
|
SELECT id, source, title, extracted_text
|
|
FROM entries WHERE {where}
|
|
ORDER BY signal_score DESC
|
|
{limit_clause}
|
|
""", limit_params)
|
|
|
|
entries = [{"id": r[0], "source": r[1], "title": r[2], "extracted_text": r[3]} for r in cur.fetchall()]
|
|
print(f" Processing: {len(entries)} entries")
|
|
print()
|
|
|
|
success = 0
|
|
failed = 0
|
|
for entry in entries:
|
|
try:
|
|
if summarize_entry(entry, conn):
|
|
success += 1
|
|
print(f" ✓ [{entry['id']}] {entry['title'][:60]}... ({entry['source']})")
|
|
else:
|
|
failed += 1
|
|
print(f" ⚠ [{entry['id']}] Skipped: {entry['title'][:40]}... (too short)")
|
|
except Exception as e:
|
|
print(f" ✗ [{entry['id']}] Error: {e}")
|
|
failed += 1
|
|
|
|
if args.verify:
|
|
print(f"\n [Verification]")
|
|
sources = [args.source] if args.source else ["github", "arxiv", "reddit"]
|
|
for src in sources:
|
|
print(f" Checking {src}...")
|
|
verify_summaries(conn, src)
|
|
print()
|
|
|
|
print(f" Results: {success} summarized, {failed} failed")
|
|
conn.close()
|
|
print(f"\n Done.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|