feat(briefing): add rss_classifier service for LLM-based topic tagging
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,147 @@
|
||||
"""
|
||||
RSS item topic classifier.
|
||||
|
||||
Classifies RSS items into topic tags using a fast non-streaming LLM call.
|
||||
Called from rss.py after new items are stored — fire-and-forget.
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import httpx
|
||||
|
||||
from fabledassistant.config import Config
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
STANDARD_TOPICS = [
|
||||
"technology", "science", "politics", "business",
|
||||
"health", "environment", "local", "entertainment", "sports", "other",
|
||||
]
|
||||
|
||||
_CLASSIFY_PROMPT = """\
|
||||
Classify each news item into 1-3 topics. Use only topics from this list: {vocab}.
|
||||
Return ONLY a JSON object mapping item_id (as string) to a list of topics.
|
||||
Example: {{"1": ["technology", "ai"], "2": ["politics"]}}
|
||||
|
||||
Items:
|
||||
{items_block}"""
|
||||
|
||||
|
||||
async def _llm_classify(prompt: str, model: str) -> str:
|
||||
"""Make a fast non-streaming LLM call and return the raw text response."""
|
||||
payload = {
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": prompt}],
|
||||
"stream": False,
|
||||
"options": {"num_ctx": 2048, "temperature": 0.0},
|
||||
}
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
resp = await client.post(f"{Config.OLLAMA_URL}/api/chat", json=payload)
|
||||
resp.raise_for_status()
|
||||
return resp.json().get("message", {}).get("content", "")
|
||||
|
||||
|
||||
async def classify_items_batch(
|
||||
items: list[dict],
|
||||
user_include_topics: list[str],
|
||||
model: str | None = None,
|
||||
) -> dict[int, list[str]]:
|
||||
"""
|
||||
Classify a batch of RSS items into topic tags.
|
||||
|
||||
Args:
|
||||
items: list of dicts with 'id', 'title', 'content'
|
||||
user_include_topics: extra topics from user preferences to add to vocabulary
|
||||
model: Ollama model name; defaults to Config.OLLAMA_MODEL
|
||||
|
||||
Returns:
|
||||
dict mapping item_id (int) -> list of topic strings.
|
||||
Items not returned had classification fail; callers should leave classified_at=NULL.
|
||||
"""
|
||||
if not items:
|
||||
return {}
|
||||
|
||||
if model is None:
|
||||
model = Config.OLLAMA_MODEL
|
||||
|
||||
vocab = STANDARD_TOPICS + [t for t in user_include_topics if t not in STANDARD_TOPICS]
|
||||
items_block = "\n".join(
|
||||
f"[{item['id']}] {item['title']} — {item.get('content', '')[:300]}"
|
||||
for item in items
|
||||
)
|
||||
prompt = _CLASSIFY_PROMPT.format(vocab=", ".join(vocab), items_block=items_block)
|
||||
|
||||
try:
|
||||
raw = await _llm_classify(prompt, model)
|
||||
# Extract JSON from response (LLM may wrap it in markdown)
|
||||
raw = raw.strip()
|
||||
if raw.startswith("```"):
|
||||
raw = raw.split("```")[1]
|
||||
if raw.startswith("json"):
|
||||
raw = raw[4:]
|
||||
parsed = json.loads(raw)
|
||||
return {int(k): v for k, v in parsed.items() if isinstance(v, list)}
|
||||
except Exception:
|
||||
logger.warning("RSS classification failed", exc_info=True)
|
||||
return {}
|
||||
|
||||
|
||||
async def classify_and_store(
|
||||
item_ids: list[int],
|
||||
user_id: int,
|
||||
) -> None:
|
||||
"""
|
||||
Classify unclassified RSS items and write results to DB.
|
||||
Called as a fire-and-forget task from rss.py.
|
||||
"""
|
||||
from sqlalchemy import select
|
||||
|
||||
from fabledassistant.models import async_session
|
||||
from fabledassistant.models.rss_feed import RssItem
|
||||
from fabledassistant.services.settings import get_setting
|
||||
|
||||
if not item_ids:
|
||||
return
|
||||
|
||||
# Load the items
|
||||
async with async_session() as session:
|
||||
result = await session.execute(
|
||||
select(RssItem).where(RssItem.id.in_(item_ids))
|
||||
)
|
||||
items = list(result.scalars().all())
|
||||
|
||||
if not items:
|
||||
return
|
||||
|
||||
# Get user's include topics to extend vocabulary
|
||||
raw_include = await get_setting(user_id, "briefing_include_topics", "[]")
|
||||
try:
|
||||
include_topics = json.loads(raw_include) if isinstance(raw_include, str) else []
|
||||
except Exception:
|
||||
include_topics = []
|
||||
|
||||
model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL)
|
||||
|
||||
# Classify in batches of 10
|
||||
batch_size = 10
|
||||
all_results: dict[int, list[str]] = {}
|
||||
for i in range(0, len(items), batch_size):
|
||||
batch = items[i: i + batch_size]
|
||||
batch_dicts = [{"id": it.id, "title": it.title, "content": it.content} for it in batch]
|
||||
results = await classify_items_batch(batch_dicts, include_topics, model=model)
|
||||
all_results.update(results)
|
||||
|
||||
# Write back to DB
|
||||
now = datetime.now(timezone.utc)
|
||||
async with async_session() as session:
|
||||
for item in items:
|
||||
item_db = await session.get(RssItem, item.id)
|
||||
if item_db is None:
|
||||
continue
|
||||
topics = all_results.get(item.id)
|
||||
if topics is not None:
|
||||
item_db.topics = topics
|
||||
item_db.classified_at = now
|
||||
await session.commit()
|
||||
@@ -40,3 +40,91 @@ def test_extract_item_prefers_content_over_summary():
|
||||
entry.published_parsed = None
|
||||
item = extract_item(entry)
|
||||
assert item["content"] == "Full content here"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_classify_items_batch_returns_topic_map():
|
||||
"""classify_items_batch should return a dict mapping item_id to topic list."""
|
||||
from unittest.mock import AsyncMock
|
||||
from fabledassistant.services.rss_classifier import classify_items_batch
|
||||
|
||||
fake_response = '{"1": ["technology", "ai"], "2": ["politics"]}'
|
||||
|
||||
with patch(
|
||||
"fabledassistant.services.rss_classifier._llm_classify",
|
||||
new_callable=AsyncMock,
|
||||
return_value=fake_response,
|
||||
):
|
||||
items = [
|
||||
{"id": 1, "title": "OpenAI releases GPT-5", "content": "..."},
|
||||
{"id": 2, "title": "EU passes new law", "content": "..."},
|
||||
]
|
||||
result = await classify_items_batch(items, user_include_topics=[])
|
||||
assert result[1] == ["technology", "ai"]
|
||||
assert result[2] == ["politics"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_classify_items_batch_handles_llm_failure():
|
||||
"""classify_items_batch should return empty dict on LLM error."""
|
||||
from unittest.mock import AsyncMock
|
||||
from fabledassistant.services.rss_classifier import classify_items_batch
|
||||
|
||||
with patch(
|
||||
"fabledassistant.services.rss_classifier._llm_classify",
|
||||
new_callable=AsyncMock,
|
||||
side_effect=Exception("LLM unavailable"),
|
||||
):
|
||||
items = [{"id": 5, "title": "Some news", "content": ""}]
|
||||
result = await classify_items_batch(items, user_include_topics=[])
|
||||
assert result == {}
|
||||
|
||||
|
||||
def test_score_rss_items_excludes_blacklisted_topics():
|
||||
"""Items with excluded topics should be removed."""
|
||||
from fabledassistant.services.briefing_preferences import score_and_filter_items
|
||||
|
||||
items = [
|
||||
{"id": 1, "title": "Tech news", "topics": ["technology"], "published_at": "2026-03-25T08:00:00"},
|
||||
{"id": 2, "title": "Sports score", "topics": ["sports"], "published_at": "2026-03-25T08:00:00"},
|
||||
]
|
||||
result = score_and_filter_items(
|
||||
items,
|
||||
include_topics=["technology"],
|
||||
exclude_topics=["sports"],
|
||||
topic_scores={},
|
||||
max_items=10,
|
||||
)
|
||||
ids = [r["id"] for r in result]
|
||||
assert 1 in ids
|
||||
assert 2 not in ids
|
||||
|
||||
|
||||
def test_score_rss_items_boosts_included_topics():
|
||||
"""Items matching include_topics should rank higher than neutral items."""
|
||||
from fabledassistant.services.briefing_preferences import score_and_filter_items
|
||||
|
||||
items = [
|
||||
{"id": 1, "title": "Random news", "topics": ["other"], "published_at": "2026-03-25T07:00:00"},
|
||||
{"id": 2, "title": "Tech news", "topics": ["technology"], "published_at": "2026-03-25T06:00:00"},
|
||||
]
|
||||
result = score_and_filter_items(
|
||||
items,
|
||||
include_topics=["technology"],
|
||||
exclude_topics=[],
|
||||
topic_scores={},
|
||||
max_items=10,
|
||||
)
|
||||
assert result[0]["id"] == 2
|
||||
|
||||
|
||||
def test_score_rss_items_no_preferences_returns_all():
|
||||
"""With no preferences, all items should be returned sorted by recency."""
|
||||
from fabledassistant.services.briefing_preferences import score_and_filter_items
|
||||
|
||||
items = [
|
||||
{"id": 1, "title": "A", "topics": [], "published_at": "2026-03-24T10:00:00"},
|
||||
{"id": 2, "title": "B", "topics": [], "published_at": "2026-03-25T10:00:00"},
|
||||
]
|
||||
result = score_and_filter_items(items, [], [], {}, max_items=10)
|
||||
assert result[0]["id"] == 2 # Newer first
|
||||
|
||||
Reference in New Issue
Block a user