feat: RSS embeddings, semantic news in chat, article-to-chat, richer briefings
- Embed RSS items at fetch time (nomic-embed-text); backfill at startup
- Semantic news search injected into chat system prompt ("Recent News You've Seen")
when items match query above 0.55 cosine threshold (independent of note RAG)
- "Discuss in chat" button on news cards — creates a seeded conversation with
the article title + full content, navigates directly to the new chat
- Briefing compilation now passes 500-char article excerpts (not just headlines)
to the LLM and uses 8192 num_ctx to accommodate the larger prompt
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,28 @@
|
|||||||
|
"""Add rss_item_embeddings table for semantic news search."""
|
||||||
|
|
||||||
|
import sqlalchemy as sa
|
||||||
|
from alembic import op
|
||||||
|
from sqlalchemy.dialects.postgresql import JSONB
|
||||||
|
|
||||||
|
revision = "0035"
|
||||||
|
down_revision = "0034"
|
||||||
|
branch_labels = None
|
||||||
|
depends_on = None
|
||||||
|
|
||||||
|
|
||||||
|
def upgrade() -> None:
|
||||||
|
op.create_table(
|
||||||
|
"rss_item_embeddings",
|
||||||
|
sa.Column("rss_item_id", sa.Integer(), sa.ForeignKey("rss_items.id", ondelete="CASCADE"), primary_key=True),
|
||||||
|
sa.Column("user_id", sa.Integer(), nullable=False),
|
||||||
|
sa.Column("embedding", JSONB(), nullable=False),
|
||||||
|
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
|
||||||
|
)
|
||||||
|
op.create_index("ix_rss_item_embeddings_user_id", "rss_item_embeddings", ["user_id"])
|
||||||
|
op.create_index("ix_rss_item_embeddings_rss_item_id", "rss_item_embeddings", ["rss_item_id"])
|
||||||
|
|
||||||
|
|
||||||
|
def downgrade() -> None:
|
||||||
|
op.drop_index("ix_rss_item_embeddings_rss_item_id", table_name="rss_item_embeddings")
|
||||||
|
op.drop_index("ix_rss_item_embeddings_user_id", table_name="rss_item_embeddings")
|
||||||
|
op.drop_table("rss_item_embeddings")
|
||||||
@@ -426,6 +426,10 @@ export async function deleteRssReaction(rssItemId: number): Promise<void> {
|
|||||||
return apiDelete(`/api/briefing/rss-reactions/${rssItemId}`);
|
return apiDelete(`/api/briefing/rss-reactions/${rssItemId}`);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export async function openArticleInChat(itemId: number): Promise<{ conversation_id: number }> {
|
||||||
|
return apiPost(`/api/chat/from-article/${itemId}`, {});
|
||||||
|
}
|
||||||
|
|
||||||
export async function geocodeAddress(address: string): Promise<{ lat: number; lon: number; display_name: string } | null> {
|
export async function geocodeAddress(address: string): Promise<{ lat: number; lon: number; display_name: string } | null> {
|
||||||
try {
|
try {
|
||||||
const r = await apiPost<{ lat: number; lon: number; label: string }>('/api/briefing/weather/geocode', { query: address });
|
const r = await apiPost<{ lat: number; lon: number; label: string }>('/api/briefing/weather/geocode', { query: address });
|
||||||
|
|||||||
@@ -1,14 +1,18 @@
|
|||||||
<script setup lang="ts">
|
<script setup lang="ts">
|
||||||
import { ref, onMounted } from 'vue'
|
import { ref, onMounted } from 'vue'
|
||||||
|
import { useRouter } from 'vue-router'
|
||||||
import {
|
import {
|
||||||
getBriefingFeeds,
|
getBriefingFeeds,
|
||||||
postRssReaction,
|
postRssReaction,
|
||||||
deleteRssReaction,
|
deleteRssReaction,
|
||||||
getNewsItems,
|
getNewsItems,
|
||||||
|
openArticleInChat,
|
||||||
type BriefingFeed,
|
type BriefingFeed,
|
||||||
} from '@/api/client'
|
} from '@/api/client'
|
||||||
import type { NewsItem } from '@/types/news'
|
import type { NewsItem } from '@/types/news'
|
||||||
|
|
||||||
|
const router = useRouter()
|
||||||
|
|
||||||
const LIMIT = 40
|
const LIMIT = 40
|
||||||
|
|
||||||
const items = ref<NewsItem[]>([])
|
const items = ref<NewsItem[]>([])
|
||||||
@@ -20,6 +24,8 @@ const selectedFeedId = ref<number | null>(null)
|
|||||||
|
|
||||||
// Reactions map: item id → current reaction
|
// Reactions map: item id → current reaction
|
||||||
const reactions = ref<Record<number, 'up' | 'down' | null>>({})
|
const reactions = ref<Record<number, 'up' | 'down' | null>>({})
|
||||||
|
// Track which items are currently being opened in chat
|
||||||
|
const openingChat = ref<Set<number>>(new Set())
|
||||||
|
|
||||||
async function loadMore() {
|
async function loadMore() {
|
||||||
if (loading.value || !hasMore.value) return
|
if (loading.value || !hasMore.value) return
|
||||||
@@ -79,6 +85,19 @@ function formatRelativeDate(iso: string | null): string {
|
|||||||
return d.toLocaleDateString(undefined, { month: 'short', day: 'numeric' })
|
return d.toLocaleDateString(undefined, { month: 'short', day: 'numeric' })
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async function openInChat(itemId: number) {
|
||||||
|
if (openingChat.value.has(itemId)) return
|
||||||
|
openingChat.value.add(itemId)
|
||||||
|
try {
|
||||||
|
const result = await openArticleInChat(itemId)
|
||||||
|
router.push(`/chat/${result.conversation_id}`)
|
||||||
|
} catch {
|
||||||
|
// silently fail — button returns to enabled state
|
||||||
|
} finally {
|
||||||
|
openingChat.value.delete(itemId)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
onMounted(async () => {
|
onMounted(async () => {
|
||||||
feeds.value = await getBriefingFeeds().catch(() => [])
|
feeds.value = await getBriefingFeeds().catch(() => [])
|
||||||
await loadMore()
|
await loadMore()
|
||||||
@@ -145,6 +164,13 @@ onMounted(async () => {
|
|||||||
@click="handleReaction(item.id, 'down')"
|
@click="handleReaction(item.id, 'down')"
|
||||||
title="Not interested"
|
title="Not interested"
|
||||||
>👎</button>
|
>👎</button>
|
||||||
|
<button
|
||||||
|
class="reaction-btn open-chat-btn"
|
||||||
|
:class="{ busy: openingChat.has(item.id) }"
|
||||||
|
:disabled="openingChat.has(item.id)"
|
||||||
|
@click="openInChat(item.id)"
|
||||||
|
title="Discuss in chat"
|
||||||
|
>{{ openingChat.has(item.id) ? '…' : '💬' }}</button>
|
||||||
</div>
|
</div>
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
@@ -335,6 +361,15 @@ a.news-card-title:hover {
|
|||||||
background: color-mix(in srgb, var(--color-primary) 12%, transparent);
|
background: color-mix(in srgb, var(--color-primary) 12%, transparent);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
.open-chat-btn {
|
||||||
|
margin-left: auto;
|
||||||
|
}
|
||||||
|
|
||||||
|
.open-chat-btn.busy {
|
||||||
|
opacity: 0.4;
|
||||||
|
cursor: wait;
|
||||||
|
}
|
||||||
|
|
||||||
.news-footer {
|
.news-footer {
|
||||||
display: flex;
|
display: flex;
|
||||||
justify-content: center;
|
justify-content: center;
|
||||||
|
|||||||
@@ -261,6 +261,11 @@ def create_app() -> Quart:
|
|||||||
await backfill_project_summaries()
|
await backfill_project_summaries()
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.warning("Project summary backfill failed", exc_info=True)
|
logger.warning("Project summary backfill failed", exc_info=True)
|
||||||
|
try:
|
||||||
|
from fabledassistant.services.embeddings import backfill_rss_item_embeddings
|
||||||
|
await backfill_rss_item_embeddings()
|
||||||
|
except Exception:
|
||||||
|
logger.warning("RSS embedding backfill failed", exc_info=True)
|
||||||
|
|
||||||
asyncio.create_task(_delayed_backfill())
|
asyncio.create_task(_delayed_backfill())
|
||||||
|
|
||||||
|
|||||||
@@ -42,3 +42,4 @@ from fabledassistant.models.rss_feed import RssFeed, RssItem # noqa: E402, F401
|
|||||||
from fabledassistant.models.weather_cache import WeatherCache # noqa: E402, F401
|
from fabledassistant.models.weather_cache import WeatherCache # noqa: E402, F401
|
||||||
from fabledassistant.models.api_key import ApiKey # noqa: E402, F401
|
from fabledassistant.models.api_key import ApiKey # noqa: E402, F401
|
||||||
from fabledassistant.models.user_profile import UserProfile # noqa: E402, F401
|
from fabledassistant.models.user_profile import UserProfile # noqa: E402, F401
|
||||||
|
from fabledassistant.models.rss_item_embedding import RssItemEmbedding # noqa: E402, F401
|
||||||
|
|||||||
@@ -0,0 +1,25 @@
|
|||||||
|
from datetime import datetime, timezone
|
||||||
|
|
||||||
|
from sqlalchemy import DateTime, ForeignKey, Integer
|
||||||
|
from sqlalchemy.dialects.postgresql import JSONB
|
||||||
|
from sqlalchemy.orm import Mapped, mapped_column
|
||||||
|
|
||||||
|
from fabledassistant.models import Base
|
||||||
|
|
||||||
|
|
||||||
|
class RssItemEmbedding(Base):
|
||||||
|
"""Stores the embedding vector for an RSS item, used for semantic news search."""
|
||||||
|
|
||||||
|
__tablename__ = "rss_item_embeddings"
|
||||||
|
|
||||||
|
rss_item_id: Mapped[int] = mapped_column(
|
||||||
|
Integer,
|
||||||
|
ForeignKey("rss_items.id", ondelete="CASCADE"),
|
||||||
|
primary_key=True,
|
||||||
|
)
|
||||||
|
user_id: Mapped[int] = mapped_column(Integer, nullable=False, index=True)
|
||||||
|
embedding: Mapped[list] = mapped_column(JSONB, nullable=False)
|
||||||
|
updated_at: Mapped[datetime] = mapped_column(
|
||||||
|
DateTime(timezone=True),
|
||||||
|
default=lambda: datetime.now(timezone.utc),
|
||||||
|
)
|
||||||
@@ -502,3 +502,52 @@ async def delete_model_route():
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.warning("Failed to delete model %s: %s", model_name, e)
|
logger.warning("Failed to delete model %s: %s", model_name, e)
|
||||||
return jsonify({"error": str(e)}), 500
|
return jsonify({"error": str(e)}), 500
|
||||||
|
|
||||||
|
|
||||||
|
@chat_bp.route("/from-article/<int:item_id>", methods=["POST"])
|
||||||
|
@login_required
|
||||||
|
async def create_conversation_from_article(item_id: int):
|
||||||
|
"""Create a chat conversation seeded with an RSS article's content."""
|
||||||
|
from sqlalchemy import select as _select
|
||||||
|
from fabledassistant.models import async_session as _async_session
|
||||||
|
from fabledassistant.models.rss_feed import RssItem, RssFeed
|
||||||
|
|
||||||
|
uid = get_current_user_id()
|
||||||
|
|
||||||
|
async with _async_session() as session:
|
||||||
|
result = await session.execute(
|
||||||
|
_select(RssItem, RssFeed.title.label("feed_title"))
|
||||||
|
.join(RssFeed, RssItem.feed_id == RssFeed.id)
|
||||||
|
.where(RssItem.id == item_id, RssFeed.user_id == uid)
|
||||||
|
)
|
||||||
|
row = result.first()
|
||||||
|
|
||||||
|
if row is None:
|
||||||
|
return jsonify({"error": "Article not found"}), 404
|
||||||
|
|
||||||
|
item, feed_title = row
|
||||||
|
|
||||||
|
conv_title = (item.title or "Article discussion")[:80]
|
||||||
|
conv = await create_conversation(uid, title=conv_title, conversation_type="chat")
|
||||||
|
|
||||||
|
source = feed_title or "News"
|
||||||
|
content_body = (item.content or "").strip()
|
||||||
|
seeded_text = f"**{source}**\n\n**{item.title}**"
|
||||||
|
if content_body:
|
||||||
|
seeded_text += f"\n\n{content_body}"
|
||||||
|
if item.url:
|
||||||
|
seeded_text += f"\n\nSource: {item.url}"
|
||||||
|
|
||||||
|
from fabledassistant.models.conversation import Message
|
||||||
|
from fabledassistant.models import async_session as _session2
|
||||||
|
async with _session2() as session:
|
||||||
|
msg = Message(
|
||||||
|
conversation_id=conv.id,
|
||||||
|
role="assistant",
|
||||||
|
content=seeded_text,
|
||||||
|
msg_metadata={"rss_item_ids": [item_id]},
|
||||||
|
)
|
||||||
|
session.add(msg)
|
||||||
|
await session.commit()
|
||||||
|
|
||||||
|
return jsonify({"conversation_id": conv.id}), 201
|
||||||
|
|||||||
@@ -233,7 +233,7 @@ async def _gather_external(user_id: int) -> dict:
|
|||||||
|
|
||||||
# ── LLM synthesis ─────────────────────────────────────────────────────────────
|
# ── LLM synthesis ─────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
async def _llm_synthesise(system_prompt: str, user_prompt: str, model: str) -> str:
|
async def _llm_synthesise(system_prompt: str, user_prompt: str, model: str, num_ctx: int = 4096) -> str:
|
||||||
"""Single non-streaming LLM call. Returns the assistant's response text."""
|
"""Single non-streaming LLM call. Returns the assistant's response text."""
|
||||||
payload = {
|
payload = {
|
||||||
"model": model,
|
"model": model,
|
||||||
@@ -242,7 +242,7 @@ async def _llm_synthesise(system_prompt: str, user_prompt: str, model: str) -> s
|
|||||||
{"role": "user", "content": user_prompt},
|
{"role": "user", "content": user_prompt},
|
||||||
],
|
],
|
||||||
"stream": False,
|
"stream": False,
|
||||||
"options": {"num_ctx": 4096, "temperature": 0.4},
|
"options": {"num_ctx": num_ctx, "temperature": 0.4},
|
||||||
}
|
}
|
||||||
try:
|
try:
|
||||||
async with httpx.AsyncClient(timeout=120.0) as client:
|
async with httpx.AsyncClient(timeout=120.0) as client:
|
||||||
@@ -309,13 +309,17 @@ def _unified_user_prompt(internal_data: dict, external_data: dict, slot: str, te
|
|||||||
lines.append(f" (and {len(overdue) - 3} more)")
|
lines.append(f" (and {len(overdue) - 3} more)")
|
||||||
lines.append("")
|
lines.append("")
|
||||||
|
|
||||||
# News highlights (top 3 — right panel shows full list)
|
# News highlights (top 3 with excerpts — right panel shows full list)
|
||||||
rss = external_data.get("rss_items") or []
|
rss = external_data.get("rss_items") or []
|
||||||
if rss:
|
if rss:
|
||||||
lines.append("NEWS HIGHLIGHTS (mention 1-2 briefly, the full list is shown separately):")
|
lines.append("NEWS HIGHLIGHTS (weave 1-2 into your briefing naturally; the full list is shown separately):")
|
||||||
for item in rss[:3]:
|
for item in rss[:3]:
|
||||||
source = item.get("feed_title") or item.get("source") or "News"
|
source = item.get("feed_title") or item.get("source") or "News"
|
||||||
lines.append(f" [{source}] {item.get('title', '')}")
|
title = item.get("title", "")
|
||||||
|
excerpt = (item.get("content") or item.get("snippet") or "")[:500].strip()
|
||||||
|
lines.append(f" [{source}] {title}")
|
||||||
|
if excerpt:
|
||||||
|
lines.append(f" {excerpt}")
|
||||||
lines.append("")
|
lines.append("")
|
||||||
|
|
||||||
return "\n".join(lines)
|
return "\n".join(lines)
|
||||||
@@ -427,6 +431,7 @@ async def run_compilation(
|
|||||||
_unified_system_prompt(profile_context),
|
_unified_system_prompt(profile_context),
|
||||||
_unified_user_prompt(internal_data_filtered, external_data_filtered, slot, temp_unit),
|
_unified_user_prompt(internal_data_filtered, external_data_filtered, slot, temp_unit),
|
||||||
model,
|
model,
|
||||||
|
num_ctx=8192,
|
||||||
)
|
)
|
||||||
|
|
||||||
# ── Post-processing ─────────────────────────────────────────────────────────
|
# ── Post-processing ─────────────────────────────────────────────────────────
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
"""Semantic note search via Ollama embedding model (nomic-embed-text).
|
"""Semantic note search via Ollama embedding model (nomic-embed-text).
|
||||||
|
|
||||||
Embeddings are stored in the note_embeddings table (one row per note).
|
Embeddings are stored in the note_embeddings table (one row per note).
|
||||||
|
RSS item embeddings are stored in rss_item_embeddings (one row per item).
|
||||||
All search operations degrade gracefully — if the embedding model is
|
All search operations degrade gracefully — if the embedding model is
|
||||||
unavailable the callers fall back to keyword search.
|
unavailable the callers fall back to keyword search.
|
||||||
"""
|
"""
|
||||||
@@ -8,6 +9,7 @@ unavailable the callers fall back to keyword search.
|
|||||||
import asyncio
|
import asyncio
|
||||||
import logging
|
import logging
|
||||||
import math
|
import math
|
||||||
|
from datetime import datetime, timedelta, timezone
|
||||||
|
|
||||||
import httpx
|
import httpx
|
||||||
from sqlalchemy import delete, select
|
from sqlalchemy import delete, select
|
||||||
@@ -16,6 +18,8 @@ from fabledassistant.config import Config
|
|||||||
from fabledassistant.models import async_session
|
from fabledassistant.models import async_session
|
||||||
from fabledassistant.models.embedding import NoteEmbedding
|
from fabledassistant.models.embedding import NoteEmbedding
|
||||||
from fabledassistant.models.note import Note
|
from fabledassistant.models.note import Note
|
||||||
|
from fabledassistant.models.rss_feed import RssItem
|
||||||
|
from fabledassistant.models.rss_item_embedding import RssItemEmbedding
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -24,6 +28,10 @@ logger = logging.getLogger(__name__)
|
|||||||
# 0.45 keeps only genuinely relevant notes; lower values like 0.30 let in
|
# 0.45 keeps only genuinely relevant notes; lower values like 0.30 let in
|
||||||
# loosely-related results that pad the sidebar without adding real value.
|
# loosely-related results that pad the sidebar without adding real value.
|
||||||
_SIMILARITY_THRESHOLD = 0.45
|
_SIMILARITY_THRESHOLD = 0.45
|
||||||
|
_RSS_SIMILARITY_THRESHOLD = 0.55
|
||||||
|
_RSS_SEARCH_LIMIT = 3
|
||||||
|
_RSS_SEARCH_DAYS = 30
|
||||||
|
_RSS_SNIPPET_CHARS = 500
|
||||||
|
|
||||||
|
|
||||||
async def get_embedding(text: str, model: str | None = None) -> list[float]:
|
async def get_embedding(text: str, model: str | None = None) -> list[float]:
|
||||||
@@ -172,3 +180,120 @@ async def backfill_note_embeddings() -> None:
|
|||||||
await asyncio.sleep(0.05) # gentle pacing
|
await asyncio.sleep(0.05) # gentle pacing
|
||||||
|
|
||||||
logger.info("Embedding backfill complete: %d/%d notes embedded", success, len(notes_to_embed))
|
logger.info("Embedding backfill complete: %d/%d notes embedded", success, len(notes_to_embed))
|
||||||
|
|
||||||
|
|
||||||
|
# ── RSS item embeddings ───────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
async def upsert_rss_item_embedding(item_id: int, user_id: int, title: str, content: str) -> None:
|
||||||
|
"""Generate and persist an embedding for an RSS item. Safe to fire-and-forget."""
|
||||||
|
text = f"{title}\n{content}".strip()
|
||||||
|
if not text:
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
embedding = await get_embedding(text)
|
||||||
|
except Exception:
|
||||||
|
logger.debug("Skipping embedding for RSS item %d — model unavailable", item_id)
|
||||||
|
return
|
||||||
|
|
||||||
|
try:
|
||||||
|
async with async_session() as session:
|
||||||
|
await session.execute(
|
||||||
|
delete(RssItemEmbedding).where(RssItemEmbedding.rss_item_id == item_id)
|
||||||
|
)
|
||||||
|
session.add(RssItemEmbedding(rss_item_id=item_id, user_id=user_id, embedding=embedding))
|
||||||
|
await session.commit()
|
||||||
|
logger.debug("Upserted embedding for RSS item %d", item_id)
|
||||||
|
except Exception:
|
||||||
|
logger.warning("Failed to persist embedding for RSS item %d", item_id, exc_info=True)
|
||||||
|
|
||||||
|
|
||||||
|
async def semantic_search_rss_items(
|
||||||
|
user_id: int,
|
||||||
|
query_vector: list[float],
|
||||||
|
limit: int = _RSS_SEARCH_LIMIT,
|
||||||
|
days: int = _RSS_SEARCH_DAYS,
|
||||||
|
) -> list[tuple[float, RssItem]]:
|
||||||
|
"""Return up to *limit* (score, RssItem) pairs most relevant to *query_vector*.
|
||||||
|
|
||||||
|
Only considers items fetched within the last *days* days.
|
||||||
|
Returns an empty list on any error.
|
||||||
|
"""
|
||||||
|
since = datetime.now(timezone.utc) - timedelta(days=days)
|
||||||
|
try:
|
||||||
|
async with async_session() as session:
|
||||||
|
stmt = (
|
||||||
|
select(RssItemEmbedding, RssItem)
|
||||||
|
.join(RssItem, RssItemEmbedding.rss_item_id == RssItem.id)
|
||||||
|
.where(
|
||||||
|
RssItemEmbedding.user_id == user_id,
|
||||||
|
RssItem.fetched_at >= since,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
rows = list((await session.execute(stmt)).all())
|
||||||
|
except Exception:
|
||||||
|
logger.warning("Failed to query RSS item embeddings", exc_info=True)
|
||||||
|
return []
|
||||||
|
|
||||||
|
if not rows:
|
||||||
|
return []
|
||||||
|
|
||||||
|
scored: list[tuple[float, RssItem]] = []
|
||||||
|
for rie, item in rows:
|
||||||
|
try:
|
||||||
|
sim = _cosine_similarity(query_vector, rie.embedding)
|
||||||
|
except Exception:
|
||||||
|
continue
|
||||||
|
if sim >= _RSS_SIMILARITY_THRESHOLD:
|
||||||
|
scored.append((sim, item))
|
||||||
|
|
||||||
|
scored.sort(key=lambda x: x[0], reverse=True)
|
||||||
|
return scored[:limit]
|
||||||
|
|
||||||
|
|
||||||
|
async def backfill_rss_item_embeddings() -> None:
|
||||||
|
"""Generate embeddings for all RSS items that don't have one yet.
|
||||||
|
|
||||||
|
Runs as a background task at startup. Adds a small sleep between items
|
||||||
|
to avoid overwhelming Ollama.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
async with async_session() as session:
|
||||||
|
existing = {
|
||||||
|
row[0]
|
||||||
|
for row in (
|
||||||
|
await session.execute(select(RssItemEmbedding.rss_item_id))
|
||||||
|
).fetchall()
|
||||||
|
}
|
||||||
|
result = await session.execute(
|
||||||
|
select(RssItem.id, RssItem.feed_id, RssItem.title, RssItem.content)
|
||||||
|
)
|
||||||
|
items_to_embed = [row for row in result.fetchall() if row[0] not in existing]
|
||||||
|
except Exception:
|
||||||
|
logger.warning("RSS embedding backfill: failed to query items", exc_info=True)
|
||||||
|
return
|
||||||
|
|
||||||
|
if not items_to_embed:
|
||||||
|
logger.info("RSS embedding backfill: all items already have embeddings")
|
||||||
|
return
|
||||||
|
|
||||||
|
# Resolve user_id per feed_id
|
||||||
|
try:
|
||||||
|
from fabledassistant.models.rss_feed import RssFeed
|
||||||
|
async with async_session() as session:
|
||||||
|
result = await session.execute(select(RssFeed.id, RssFeed.user_id))
|
||||||
|
feed_user_map = {fid: uid for fid, uid in result.fetchall()}
|
||||||
|
except Exception:
|
||||||
|
logger.warning("RSS embedding backfill: failed to load feed user map", exc_info=True)
|
||||||
|
return
|
||||||
|
|
||||||
|
logger.info("RSS embedding backfill: generating embeddings for %d items", len(items_to_embed))
|
||||||
|
success = 0
|
||||||
|
for item_id, feed_id, title, content in items_to_embed:
|
||||||
|
user_id = feed_user_map.get(feed_id)
|
||||||
|
if user_id is None:
|
||||||
|
continue
|
||||||
|
await upsert_rss_item_embedding(item_id, user_id, title or "", content or "")
|
||||||
|
success += 1
|
||||||
|
await asyncio.sleep(0.05)
|
||||||
|
|
||||||
|
logger.info("RSS embedding backfill complete: %d/%d items embedded", success, len(items_to_embed))
|
||||||
|
|||||||
@@ -668,6 +668,33 @@ async def build_context(
|
|||||||
+ "\n--- End Included Notes ---"
|
+ "\n--- End Included Notes ---"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# Search for semantically relevant recent news items
|
||||||
|
try:
|
||||||
|
from fabledassistant.services.embeddings import get_embedding, semantic_search_rss_items
|
||||||
|
news_query_vec = await get_embedding(user_message)
|
||||||
|
news_hits = await semantic_search_rss_items(user_id, news_query_vec)
|
||||||
|
if news_hits:
|
||||||
|
news_snippets = []
|
||||||
|
for score, rss_item in news_hits:
|
||||||
|
feed_title = getattr(rss_item, "feed_title", "") or ""
|
||||||
|
excerpt = (rss_item.content or "")[:500].strip()
|
||||||
|
news_snippets.append(
|
||||||
|
f"[{feed_title or 'News'}] {rss_item.title} (relevance: {round(score * 100)}%)\n"
|
||||||
|
+ (f"{excerpt}\n" if excerpt else "")
|
||||||
|
+ f"URL: {rss_item.url}"
|
||||||
|
)
|
||||||
|
system_parts.append(
|
||||||
|
"\n\n--- Recent News You've Seen ---\n"
|
||||||
|
+ "\n\n".join(news_snippets)
|
||||||
|
+ "\n--- End Recent News ---"
|
||||||
|
)
|
||||||
|
context_meta["rss_news"] = [
|
||||||
|
{"id": item.id, "title": item.title, "score": round(score, 2)}
|
||||||
|
for score, item in news_hits
|
||||||
|
]
|
||||||
|
except Exception:
|
||||||
|
logger.debug("RSS semantic search skipped", exc_info=True)
|
||||||
|
|
||||||
# Fetch URL content from user message
|
# Fetch URL content from user message
|
||||||
urls = _find_urls(user_message)
|
urls = _find_urls(user_message)
|
||||||
for url in urls[:2]: # Limit to 2 URLs
|
for url in urls[:2]: # Limit to 2 URLs
|
||||||
|
|||||||
@@ -112,14 +112,18 @@ async def fetch_and_cache_feed(feed_id: int, url: str) -> int:
|
|||||||
# only writes to items it successfully classifies, so already-classified items
|
# only writes to items it successfully classifies, so already-classified items
|
||||||
# are not re-processed (they have classified_at set).
|
# are not re-processed (they have classified_at set).
|
||||||
unclassified_ids: list[int] = []
|
unclassified_ids: list[int] = []
|
||||||
|
new_item_data: list[tuple[int, str, str]] = [] # (id, title, content) for embedding
|
||||||
if new_count > 0:
|
if new_count > 0:
|
||||||
result = await session.execute(
|
result = await session.execute(
|
||||||
select(RssItem.id).where(
|
select(RssItem.id, RssItem.title, RssItem.content, RssItem.classified_at).where(
|
||||||
RssItem.feed_id == feed_id,
|
RssItem.feed_id == feed_id,
|
||||||
RssItem.classified_at.is_(None),
|
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
unclassified_ids = list(result.scalars().all())
|
for row in result.fetchall():
|
||||||
|
item_id, title, content, classified_at = row
|
||||||
|
if classified_at is None:
|
||||||
|
unclassified_ids.append(item_id)
|
||||||
|
new_item_data.append((item_id, title or "", content or ""))
|
||||||
|
|
||||||
# Prune old items to keep DB tidy
|
# Prune old items to keep DB tidy
|
||||||
await _prune_old_items(feed_id)
|
await _prune_old_items(feed_id)
|
||||||
@@ -129,6 +133,17 @@ async def fetch_and_cache_feed(feed_id: int, url: str) -> int:
|
|||||||
from fabledassistant.services.rss_classifier import classify_and_store
|
from fabledassistant.services.rss_classifier import classify_and_store
|
||||||
asyncio.create_task(classify_and_store(unclassified_ids, feed_user_id))
|
asyncio.create_task(classify_and_store(unclassified_ids, feed_user_id))
|
||||||
|
|
||||||
|
# Fire-and-forget embedding for new items
|
||||||
|
if new_item_data and feed_user_id is not None:
|
||||||
|
from fabledassistant.services.embeddings import upsert_rss_item_embedding
|
||||||
|
|
||||||
|
async def _embed_new_items() -> None:
|
||||||
|
for item_id, title, content in new_item_data:
|
||||||
|
await upsert_rss_item_embedding(item_id, feed_user_id, title, content)
|
||||||
|
await asyncio.sleep(0.05)
|
||||||
|
|
||||||
|
asyncio.create_task(_embed_new_items())
|
||||||
|
|
||||||
return new_count
|
return new_count
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user