bvandeusen
90afbec4c2
Backend: - GET /api/knowledge/ids: returns up to 100 note IDs cheaply (no body parsing), supports same filters as /api/knowledge, includes has_more - GET /api/knowledge/batch?ids=...: fetches full items for given IDs in order; used by frontend to load content in controlled batches Frontend (KnowledgeView): - Fetch 100 IDs upfront, load first 50 as content on mount - IntersectionObserver sentinel (root: null) triggers 24-item content batches as user scrolls - Proactive ID refill when queue drops below 48 unloaded IDs - fetchGen counter invalidates stale in-flight responses on filter reset - IDs claimed before async fetch to prevent double-loading - sentinelVisible ref drives post-load re-check when content doesn't push sentinel off screen Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
269 lines
9.1 KiB
Python
269 lines
9.1 KiB
Python
"""Knowledge service — unified query across notes, people, places, and lists."""
|
||
import logging
|
||
|
||
from sqlalchemy import func, select
|
||
|
||
from fabledassistant.models import async_session
|
||
from fabledassistant.models.note import Note
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_SNIPPET_LEN = 200
|
||
|
||
|
||
def _note_to_item(note: Note) -> dict:
|
||
meta = note.entity_meta or {}
|
||
item: dict = {
|
||
"id": note.id,
|
||
"note_type": note.entity_type,
|
||
"title": note.title,
|
||
"snippet": (note.body or "")[:_SNIPPET_LEN],
|
||
"tags": note.tags or [],
|
||
"project_id": note.project_id,
|
||
"metadata": meta,
|
||
"created_at": note.created_at.isoformat(),
|
||
"updated_at": note.updated_at.isoformat(),
|
||
}
|
||
# Type-specific convenience fields
|
||
if note.entity_type == "person":
|
||
item["relationship"] = meta.get("relationship", "")
|
||
item["email"] = meta.get("email", "")
|
||
item["phone"] = meta.get("phone", "")
|
||
elif note.entity_type == "place":
|
||
item["address"] = meta.get("address", "")
|
||
item["phone"] = meta.get("phone", "")
|
||
item["hours"] = meta.get("hours", "")
|
||
elif note.entity_type == "list":
|
||
# Parse markdown task list syntax into structured items
|
||
body = note.body or ""
|
||
list_items = []
|
||
for line in body.split("\n"):
|
||
stripped = line.strip()
|
||
if stripped.startswith("- [ ] ") or stripped.startswith("- [x] ") or stripped.startswith("- [X] "):
|
||
checked_item = not stripped.startswith("- [ ] ")
|
||
list_items.append({"text": stripped[6:], "checked": checked_item})
|
||
item["list_items"] = list_items
|
||
item["item_count"] = len(list_items)
|
||
item["checked_count"] = sum(1 for i in list_items if i["checked"])
|
||
item["body"] = body
|
||
return item
|
||
|
||
|
||
async def query_knowledge(
|
||
user_id: int,
|
||
note_type: str | None,
|
||
tags: list[str],
|
||
sort: str,
|
||
q: str | None,
|
||
limit: int,
|
||
offset: int,
|
||
) -> tuple[list[dict], int]:
|
||
"""Query knowledge objects (non-task notes) with filters.
|
||
|
||
Returns (items, total_count).
|
||
"""
|
||
# Semantic search path — scores take priority over sort
|
||
if q:
|
||
return await _semantic_knowledge_search(
|
||
user_id, q, note_type=note_type, tags=tags, limit=limit, offset=offset
|
||
)
|
||
|
||
async with async_session() as session:
|
||
base = (
|
||
select(Note)
|
||
.where(Note.user_id == user_id)
|
||
.where(Note.status.is_(None)) # exclude tasks
|
||
)
|
||
|
||
if note_type:
|
||
base = base.where(Note.note_type == note_type)
|
||
else:
|
||
# Exclude tasks — already done above; also exclude any legacy nulls
|
||
base = base.where(Note.note_type.in_(["note", "person", "place", "list"]))
|
||
|
||
for tag in tags:
|
||
base = base.where(Note.tags.contains([tag]))
|
||
|
||
# Count before pagination
|
||
count_stmt = select(func.count()).select_from(base.subquery())
|
||
total: int = (await session.execute(count_stmt)).scalar_one()
|
||
|
||
# Apply sort
|
||
if sort == "created":
|
||
base = base.order_by(Note.created_at.desc())
|
||
elif sort == "alpha":
|
||
base = base.order_by(Note.title.asc())
|
||
elif sort == "type":
|
||
base = base.order_by(Note.note_type.asc(), Note.updated_at.desc())
|
||
else: # modified (default)
|
||
base = base.order_by(Note.updated_at.desc())
|
||
|
||
rows = list((await session.execute(base.limit(limit).offset(offset))).scalars().all())
|
||
|
||
return [_note_to_item(n) for n in rows], total
|
||
|
||
|
||
async def _semantic_knowledge_search(
|
||
user_id: int,
|
||
q: str,
|
||
note_type: str | None,
|
||
tags: list[str],
|
||
limit: int,
|
||
offset: int,
|
||
) -> tuple[list[dict], int]:
|
||
"""Semantic search over knowledge objects, with SQL filters applied post-rank."""
|
||
try:
|
||
from fabledassistant.services.embeddings import semantic_search_notes
|
||
# Fetch a larger candidate set to allow for filtering
|
||
candidates = await semantic_search_notes(
|
||
user_id=user_id,
|
||
query=q,
|
||
limit=min(200, limit * 8),
|
||
threshold=0.3,
|
||
is_task=False,
|
||
)
|
||
except Exception:
|
||
logger.warning("Semantic search unavailable, falling back to SQL", exc_info=True)
|
||
return await query_knowledge(user_id, note_type, tags, "modified", None, limit, offset)
|
||
|
||
results = []
|
||
for _score, note in candidates:
|
||
if note_type and note.entity_type != note_type:
|
||
continue
|
||
if tags and not all(t in (note.tags or []) for t in tags):
|
||
continue
|
||
results.append(note)
|
||
|
||
total = len(results)
|
||
page_items = results[offset: offset + limit]
|
||
return [_note_to_item(n) for n in page_items], total
|
||
|
||
|
||
async def get_knowledge_tags(user_id: int, note_type: str | None = None) -> list[str]:
|
||
"""Return all distinct tags used across knowledge objects for this user."""
|
||
from sqlalchemy.dialects.postgresql import array
|
||
async with async_session() as session:
|
||
stmt = (
|
||
select(func.unnest(Note.tags).label("tag"))
|
||
.where(Note.user_id == user_id)
|
||
.where(Note.status.is_(None))
|
||
)
|
||
if note_type:
|
||
stmt = stmt.where(Note.note_type == note_type)
|
||
else:
|
||
stmt = stmt.where(Note.note_type.in_(["note", "person", "place", "list"]))
|
||
|
||
stmt = (
|
||
select(func.unnest(Note.tags).label("tag"))
|
||
.where(Note.user_id == user_id)
|
||
.where(Note.status.is_(None))
|
||
.distinct()
|
||
.order_by("tag")
|
||
)
|
||
if note_type:
|
||
stmt = stmt.where(Note.note_type == note_type)
|
||
|
||
rows = list((await session.execute(stmt)).scalars().all())
|
||
return [r for r in rows if r]
|
||
|
||
|
||
async def query_knowledge_ids(
|
||
user_id: int,
|
||
note_type: str | None,
|
||
tags: list[str],
|
||
sort: str,
|
||
q: str | None,
|
||
limit: int = 100,
|
||
offset: int = 0,
|
||
) -> tuple[list[int], int]:
|
||
"""Return note IDs only — cheap query for the two-tier pagination feed."""
|
||
if q:
|
||
# Re-use semantic search, extract IDs in rank order
|
||
items, total = await _semantic_knowledge_search(
|
||
user_id, q, note_type=note_type, tags=tags,
|
||
limit=limit, offset=offset,
|
||
)
|
||
return [item["id"] for item in items], total
|
||
|
||
async with async_session() as session:
|
||
base = (
|
||
select(Note.id)
|
||
.where(Note.user_id == user_id)
|
||
.where(Note.status.is_(None))
|
||
)
|
||
if note_type:
|
||
base = base.where(Note.note_type == note_type)
|
||
else:
|
||
base = base.where(Note.note_type.in_(["note", "person", "place", "list"]))
|
||
for tag in tags:
|
||
base = base.where(Note.tags.contains([tag]))
|
||
|
||
count_stmt = select(func.count()).select_from(base.subquery())
|
||
total: int = (await session.execute(count_stmt)).scalar_one()
|
||
|
||
if sort == "created":
|
||
base = base.order_by(Note.created_at.desc())
|
||
elif sort == "alpha":
|
||
base = base.order_by(Note.title.asc())
|
||
elif sort == "type":
|
||
base = base.order_by(Note.note_type.asc(), Note.updated_at.desc())
|
||
else:
|
||
base = base.order_by(Note.updated_at.desc())
|
||
|
||
ids = list((await session.execute(base.limit(limit).offset(offset))).scalars().all())
|
||
|
||
return ids, total
|
||
|
||
|
||
async def get_knowledge_by_ids(user_id: int, ids: list[int]) -> list[dict]:
|
||
"""Fetch full items for the given IDs, preserving the requested order."""
|
||
if not ids:
|
||
return []
|
||
async with async_session() as session:
|
||
stmt = (
|
||
select(Note)
|
||
.where(Note.user_id == user_id)
|
||
.where(Note.id.in_(ids))
|
||
)
|
||
rows = list((await session.execute(stmt)).scalars().all())
|
||
by_id = {n.id: n for n in rows}
|
||
return [_note_to_item(by_id[i]) for i in ids if i in by_id]
|
||
|
||
|
||
async def get_people_and_places_context(user_id: int) -> str:
|
||
"""Return a compact summary of known people and places for LLM system prompt injection."""
|
||
async with async_session() as session:
|
||
stmt = (
|
||
select(Note)
|
||
.where(Note.user_id == user_id)
|
||
.where(Note.note_type.in_(["person", "place"]))
|
||
.where(Note.status.is_(None))
|
||
.order_by(Note.title.asc())
|
||
.limit(50)
|
||
)
|
||
rows = list((await session.execute(stmt)).scalars().all())
|
||
|
||
if not rows:
|
||
return ""
|
||
|
||
people = [n for n in rows if n.entity_type == "person"]
|
||
places = [n for n in rows if n.entity_type == "place"]
|
||
|
||
lines = []
|
||
if people:
|
||
parts = []
|
||
for p in people:
|
||
meta = p.entity_meta or {}
|
||
rel = meta.get("relationship", "")
|
||
parts.append(f"{p.title}" + (f" ({rel})" if rel else ""))
|
||
lines.append("Known people: " + ", ".join(parts))
|
||
if places:
|
||
parts = []
|
||
for p in places:
|
||
meta = p.entity_meta or {}
|
||
addr = meta.get("address", "")
|
||
parts.append(f"{p.title}" + (f" – {addr}" if addr else ""))
|
||
lines.append("Known places: " + "; ".join(parts))
|
||
|
||
return "\n".join(lines)
|