jpmschweitzer/library-desk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

feat: add unified memory routing to consolidation service

Browse Source 
- Add MemoryRouteClassification and MemoryRoutingResult models
- Implement unified classifier (_classify_web_results_unified) that routes
  web results to: wiki, volatile, file (Paperless), prefetch, or skip
- Add routing methods: _route_to_volatile, _route_to_files, _register_prefetch
- Update _process_search to use unified classifier instead of separate analysis
- Add get_volatile_cache_service factory to dependencies
- Wire volatile_service and settings_client into ConsolidationService
- Update ConsolidationResult/Response with new routing counters

Test fixes:
- Fix WikiJSClient fixtures to use api_token instead of username/password
- Fix entity linking test assertions to expect full user-namespaced paths
- Add sample_unified_classification fixture for new classifier format

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
jpmschweitzer
2025-12-28 17:19:20 +01:00
co-authored by Claude Opus 4.5
parent ab892745fa
commit 1f47b052d8
9 changed files with 533 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/dependencies.py
+14 -1
View File
@@ -597,7 +597,9 @@ def get_consolidation_service() -> "ConsolidationService":
ollama=get_ollama_client(),
wiki=get_wikijs_client(),
settings=get_settings(),
ingestion_service=get_ingestion_service()
ingestion_service=get_ingestion_service(),
volatile_service=get_volatile_cache_service(),
settings_client=get_settings_client(),
)
@@ -638,6 +640,17 @@ def get_rag_search_service() -> "RAGSearchService":
)
@lru_cache
def get_volatile_cache_service() -> "VolatileCacheService":
"""Get VolatileCacheService singleton."""
from src.services.volatile_service import VolatileCacheService
return VolatileCacheService(
qdrant_client=get_qdrant_client(),
ollama_client=get_ollama_client(),
settings=get_settings()
)
# Authentication
from fastapi import Security, HTTPException
from fastapi.security import HTTPBearer
src/models/consolidation.py
+50
View File
@@ -34,6 +34,9 @@ class ConsolidationResult(BaseModel):
pages_created: int = 0
pages_updated: int = 0
entities_added: int = 0
volatile_cached: int = 0
files_queued: int = 0
prefetch_registered: int = 0
error: Optional[str] = None
@@ -44,6 +47,53 @@ class ConsolidationResponse(BaseModel):
pages_created: int = Field(description="New wiki pages created")
pages_updated: int = Field(description="Existing pages updated")
entities_added: int = Field(description="New entities added to graph")
volatile_cached: int = Field(default=0, description="Items cached to volatile storage")
files_queued: int = Field(default=0, description="Files queued for Paperless")
prefetch_registered: int = Field(default=0, description="Prefetch patterns registered")
errors: List[str] = Field(default=[], description="Error messages")
results: List[ConsolidationResult] = Field(description="Per-search results")
dry_run: bool = Field(description="Whether this was a dry run")
class MemoryRouteClassification(BaseModel):
"""
Unified classification of a web result for memory routing.
Route types:
- wiki: Stable reference content → wiki page creation/update
- volatile: Ephemeral data (weather, news, prices) → volatile cache
- file: Downloadable file (PDF, doc, xls, images) → Paperless ingestion
- prefetch: Regularly updated source → scheduler registration
- skip: Low value, ads, errors → discard
"""
url: str
title: str
route_type: str = Field(description="One of: wiki, volatile, file, prefetch, skip")
# Wiki routing fields
wiki_action: Optional[str] = Field(default=None, description="create or update")
wiki_path: Optional[str] = Field(default=None, description="Wiki path for page")
wiki_summary: Optional[str] = Field(default=None, description="Summary for wiki page")
# Volatile routing fields
volatile_namespace: Optional[str] = Field(default=None, description="weather, news, financial, etc.")
volatile_key: Optional[str] = Field(default=None, description="Cache key")
volatile_ttl_hours: Optional[int] = Field(default=None, description="TTL in hours")
# Prefetch routing fields
prefetch_cron: Optional[str] = Field(default=None, description="Cron expression for refresh")
prefetch_endpoint: Optional[str] = Field(default=None, description="API endpoint to call")
# Classification metadata
confidence: float = Field(default=0.0, ge=0.0, le=1.0)
reason: str = Field(default="")
class MemoryRoutingResult(BaseModel):
"""Aggregated result of memory routing for a search."""
wiki_routed: int = 0
volatile_cached: int = 0
files_queued: int = 0
prefetch_registered: int = 0
skipped: int = 0
classifications: List[MemoryRouteClassification] = []
src/routers/consolidation.py
+5 -2
View File
@@ -12,7 +12,8 @@ from src.models.consolidation import ConsolidationRequest, ConsolidationResponse
from src.services.consolidation_service import ConsolidationService
from src.core.dependencies import (
Neo4jDep, OllamaDep, WikiJSDep,
verify_api_key, get_settings, get_ingestion_service
verify_api_key, get_settings, get_ingestion_service,
get_volatile_cache_service, get_settings_client,
)
from src.config import Settings
@@ -34,7 +35,9 @@ def get_consolidation_service(
ollama=ollama_client,
wiki=wiki_client,
settings=settings,
ingestion_service=get_ingestion_service()
ingestion_service=get_ingestion_service(),
volatile_service=get_volatile_cache_service(),
settings_client=get_settings_client(),
)
src/services/consolidation_service.py
+396 -62
View File
@@ -23,7 +23,9 @@ from src.services.wiki_page_writer import WikiPageWriter
from src.models.consolidation import (
SearchQueryInfo,
ConsolidationResult,
ConsolidationResponse
ConsolidationResponse,
MemoryRouteClassification,
MemoryRoutingResult,
)
from src.config import Settings
@@ -41,7 +43,9 @@ class ConsolidationService:
ollama: OllamaClient,
wiki: WikiJSClient,
settings: Settings,
ingestion_service: Optional["IngestionService"] = None
ingestion_service: Optional["IngestionService"] = None,
volatile_service: Optional["VolatileCacheService"] = None,
settings_client: Optional["SettingsClient"] = None,
):
self.neo4j = neo4j
self.ollama = ollama
@@ -49,6 +53,8 @@ class ConsolidationService:
self.settings = settings
self.wiki_page_writer = WikiPageWriter(ollama_client=ollama, settings=settings)
self.ingestion_service = ingestion_service # Optional to avoid circular dependency
self.volatile_service = volatile_service # For ephemeral data caching
self.settings_client = settings_client # For prefetch registration
async def consolidate_knowledge(
self,
@@ -97,6 +103,9 @@ class ConsolidationService:
total_pages_created = 0
total_pages_updated = 0
total_entities_added = 0
total_volatile_cached = 0
total_files_queued = 0
total_prefetch_registered = 0
errors: List[str] = []
for search in unprocessed:
@@ -112,6 +121,9 @@ class ConsolidationService:
total_pages_created += result.pages_created
total_pages_updated += result.pages_updated
total_entities_added += result.entities_added
total_volatile_cached += result.volatile_cached
total_files_queued += result.files_queued
total_prefetch_registered += result.prefetch_registered
# Mark as processed if not dry run (even if skipped)
# This prevents searches from accumulating when they don't meet criteria
@@ -141,6 +153,9 @@ class ConsolidationService:
pages_created=total_pages_created,
pages_updated=total_pages_updated,
entities_added=total_entities_added,
volatile_cached=total_volatile_cached,
files_queued=total_files_queued,
prefetch_registered=total_prefetch_registered,
errors=errors,
results=results,
dry_run=dry_run
@@ -149,7 +164,8 @@ class ConsolidationService:
logger.info(
f"Consolidation complete: {processed_count}/{len(unprocessed)} searches, "
f"{total_pages_created} pages created, {total_pages_updated} updated, "
f"{total_entities_added} entities added"
f"{total_entities_added} entities, {total_volatile_cached} volatile, "
f"{total_files_queued} files, {total_prefetch_registered} prefetch"
)
return response
@@ -213,6 +229,13 @@ class ConsolidationService:
) -> Optional[ConsolidationResult]:
"""
Process a single search query for knowledge consolidation.
Uses unified memory routing to classify each web result and route to:
- wiki: Stable reference content → wiki page creation/update
- volatile: Ephemeral data → volatile cache
- file: Downloadable documents → Paperless queue
- prefetch: Regular updates → scheduler registration
- skip: Low value content → discard
"""
search_id = search['id']
query = search['query']
@@ -234,97 +257,106 @@ class ConsolidationService:
logger.info(f"Retrieved {len(web_results)} web results")
# Analyze web results with Ollama for novel information
analysis = await self._analyze_web_results(
# Unified classification of all web results
routing_result = await self._classify_web_results_unified(
query=query,
web_results=web_results,
keywords=search.get('keywords', []),
user=user
)
if not analysis or not analysis.get('has_novel_info'):
logger.info("No novel information found")
if not routing_result.classifications:
logger.info("No classifications returned")
return ConsolidationResult(
search_id=search_id,
query=query
)
# Extract consolidation actions
pages_to_create = analysis.get('new_pages', [])
pages_to_update = analysis.get('update_pages', [])
new_entities = analysis.get('new_entities', [])
logger.info(
f"Analysis: {len(pages_to_create)} new pages, "
f"{len(pages_to_update)} updates, {len(new_entities)} entities"
f"Routing: {routing_result.wiki_routed} wiki, "
f"{routing_result.volatile_cached} volatile, "
f"{routing_result.files_queued} files, "
f"{routing_result.prefetch_registered} prefetch, "
f"{routing_result.skipped} skipped"
)
if dry_run:
logger.info("[DRY RUN] Would create/update pages and entities")
logger.info("[DRY RUN] Would route results to destinations")
return ConsolidationResult(
search_id=search_id,
query=query,
pages_created=len(pages_to_create),
pages_updated=len(pages_to_update),
entities_added=len(new_entities)
pages_created=routing_result.wiki_routed,
volatile_cached=routing_result.volatile_cached,
files_queued=routing_result.files_queued,
prefetch_registered=routing_result.prefetch_registered,
)
# Create/update wiki pages
# Process each classification
pages_created = 0
pages_updated = 0
entities_added = 0
volatile_cached = 0
files_queued = 0
prefetch_registered = 0
# Create new pages
for page_data in pages_to_create:
try:
await self._create_or_consolidate_page(
user=user,
title=page_data.get('title'),
path=page_data.get('path'),
summary=page_data.get('summary'),
source_query=query,
web_results=web_results
)
pages_created += 1
logger.info(f"Created page: {page_data.get('title')}")
except Exception as e:
logger.error(f"Failed to create page {page_data.get('title')}: {e}")
# Create URL-to-web_result lookup
url_to_result = {r['url']: r for r in web_results}
# Update existing pages
for page_data in pages_to_update:
try:
await self._update_page_with_facts(
title=page_data.get('title'),
new_facts=page_data.get('new_facts', []),
source_url=page_data.get('source_url'),
user=user
)
pages_updated += 1
logger.info(f"Updated page: {page_data.get('title')}")
except Exception as e:
logger.error(f"Failed to update page {page_data.get('title')}: {e}")
for classification in routing_result.classifications:
web_result = url_to_result.get(classification.url, {})
# Add new entities to graph
for entity_data in new_entities:
try:
await self._add_entity_to_graph(
user=user,
entity_name=entity_data.get('name'),
entity_type=entity_data.get('type'),
description=entity_data.get('description'),
source_search_id=search_id
)
entities_added += 1
logger.info(f"Added entity: {entity_data.get('name')}")
except Exception as e:
logger.error(f"Failed to add entity {entity_data.get('name')}: {e}")
if classification.route_type == 'wiki':
# Route to wiki page creation/update
try:
if classification.wiki_action == 'create':
await self._create_or_consolidate_page(
user=user,
title=classification.title,
path=classification.wiki_path or f"reference/{classification.title.lower().replace(' ', '-')}",
summary=classification.wiki_summary or '',
source_query=query,
web_results=[web_result] if web_result else web_results[:3]
)
pages_created += 1
logger.info(f"Created wiki page: {classification.title}")
elif classification.wiki_action == 'update':
await self._update_page_with_facts(
title=classification.title,
new_facts=[classification.wiki_summary] if classification.wiki_summary else [],
source_url=classification.url,
user=user
)
pages_updated += 1
logger.info(f"Updated wiki page: {classification.title}")
except Exception as e:
logger.error(f"Failed wiki routing for {classification.title}: {e}")
elif classification.route_type == 'volatile':
# Route to volatile cache
if await self._route_to_volatile(classification, web_result, user):
volatile_cached += 1
elif classification.route_type == 'file':
# Route to Paperless queue
if await self._route_to_files(classification, web_result, user):
files_queued += 1
elif classification.route_type == 'prefetch':
# Register prefetch pattern
if await self._register_prefetch(classification, web_result, user):
prefetch_registered += 1
# 'skip' route type - do nothing
return ConsolidationResult(
search_id=search_id,
query=query,
pages_created=pages_created,
pages_updated=pages_updated,
entities_added=entities_added
entities_added=entities_added,
volatile_cached=volatile_cached,
files_queued=files_queued,
prefetch_registered=prefetch_registered,
)
async def _get_web_results(self, search_id: str) -> List[Dict[str, Any]]:
@@ -937,3 +969,305 @@ JSON:"""
logger.debug(f"Added entity to graph: {entity_name} ({entity_type})")
except Exception as e:
logger.error(f"Failed to add entity to graph: {e}")
async def _classify_web_results_unified(
self,
query: str,
web_results: List[Dict[str, Any]],
keywords: List[str],
user: str = "jpmschweitzer"
) -> MemoryRoutingResult:
"""
Unified classification of web results for memory routing.
Each web result is classified into exactly one destination:
- wiki: Stable reference content → wiki page creation/update
- volatile: Ephemeral data (weather, news, prices) → volatile cache
- file: Downloadable file (PDF, doc, xls, images) → Paperless
- prefetch: Regularly updated source → scheduler registration
- skip: Low value, ads, errors → discard
Returns:
MemoryRoutingResult with classifications for each web result
"""
# Fetch existing taxonomy structure for wiki path suggestions
try:
taxonomy_structure = await self.wiki.get_taxonomy_structure(f"users/{user}")
existing_paths_info = self._format_taxonomy_for_prompt(taxonomy_structure)
logger.info(f"Fetched taxonomy with {len(taxonomy_structure)} categories for user {user}")
except Exception as e:
logger.warning(f"Failed to fetch taxonomy structure: {e}")
existing_paths_info = ""
# Build classification prompt
web_summary = "\n\n".join([
f"[{i+1}] Title: {r['title']}\n URL: {r['url']}\n Content: {r['content'][:400]}..."
for i, r in enumerate(web_results[:10])
])
prompt = f"""You are a Memory Router for a personal knowledge system. Classify each web result into ONE destination.
Query: "{query}"
Keywords: {', '.join(keywords) if keywords else 'none'}
Web Results:
{web_summary}
CLASSIFICATION RULES:
**wiki** - Stable reference content worth documenting permanently:
- Factual information about people, places, companies, products
- How-to guides, tutorials, technical documentation
- Historical facts, biographies, definitions
- Content that won't change frequently
**volatile** - Ephemeral data that changes frequently:
- Current weather conditions or forecasts
- Latest news headlines or breaking news
- Stock prices, exchange rates, crypto prices
- Sports scores, live results
- Traffic conditions, transit delays
- Social media trends, notifications
Use namespaces: weather, news, financial, transit, traffic, sports, social, system
**file** - Downloadable documents:
- PDF files (URLs ending in .pdf or containing /pdf/)
- Office documents (.doc, .docx, .xls, .xlsx, .ppt)
- Images (.jpg, .png, .gif when they're primary content)
- CSV/data files
- Any direct download link
**prefetch** - Sources worth checking regularly:
- News feeds or RSS sources
- API endpoints with live data
- Dashboards or status pages
- Only if not already captured by volatile
**skip** - Low value content:
- Ads, paywalled content
- Error pages, 404s
- Duplicate or redundant results
- Content not answering the query
{existing_paths_info}
Return ONLY valid JSON array:
[
{{
"url": "...",
"title": "...",
"route_type": "wiki|volatile|file|prefetch|skip",
"wiki_action": "create|update",
"wiki_path": "category/subcategory/page-name",
"wiki_summary": "What to document",
"volatile_namespace": "weather|news|financial|...",
"volatile_key": "cache-key",
"volatile_ttl_hours": 1,
"prefetch_cron": "0 * * * *",
"prefetch_endpoint": "/volatile/fetch/...",
"confidence": 0.9,
"reason": "Why this classification"
}}
]
Only include fields relevant to the route_type. Set irrelevant fields to null.
JSON:"""
try:
response = await self.ollama.generate_text(
prompt=prompt,
model=self.settings.ollama_model,
stream=False,
temperature=0.0
)
if not response:
logger.warning("Empty response from Ollama for classification")
return MemoryRoutingResult()
# Extract JSON array from response
response_clean = response.strip()
if '[' in response_clean:
json_start = response_clean.find('[')
json_end = response_clean.rfind(']') + 1
response_clean = response_clean[json_start:json_end]
classifications_raw = json.loads(response_clean)
# Parse into MemoryRouteClassification objects
result = MemoryRoutingResult()
for item in classifications_raw:
try:
classification = MemoryRouteClassification(
url=item.get('url', ''),
title=item.get('title', ''),
route_type=item.get('route_type', 'skip'),
wiki_action=item.get('wiki_action'),
wiki_path=item.get('wiki_path'),
wiki_summary=item.get('wiki_summary'),
volatile_namespace=item.get('volatile_namespace'),
volatile_key=item.get('volatile_key'),
volatile_ttl_hours=item.get('volatile_ttl_hours'),
prefetch_cron=item.get('prefetch_cron'),
prefetch_endpoint=item.get('prefetch_endpoint'),
confidence=item.get('confidence', 0.5),
reason=item.get('reason', ''),
)
result.classifications.append(classification)
# Count by route type
if classification.route_type == 'wiki':
result.wiki_routed += 1
elif classification.route_type == 'volatile':
result.volatile_cached += 1
elif classification.route_type == 'file':
result.files_queued += 1
elif classification.route_type == 'prefetch':
result.prefetch_registered += 1
else:
result.skipped += 1
except Exception as e:
logger.warning(f"Failed to parse classification item: {e}")
logger.info(
f"Classification complete: {result.wiki_routed} wiki, "
f"{result.volatile_cached} volatile, {result.files_queued} files, "
f"{result.prefetch_registered} prefetch, {result.skipped} skipped"
)
return result
except json.JSONDecodeError as e:
logger.error(f"Failed to parse classification response as JSON: {e}")
return MemoryRoutingResult()
except Exception as e:
logger.error(f"Classification failed: {e}", exc_info=True)
return MemoryRoutingResult()
async def _route_to_volatile(
self,
classification: MemoryRouteClassification,
web_result: Dict[str, Any],
user: str,
) -> bool:
"""
Route a web result to volatile cache.
Args:
classification: The classification with volatile routing info
web_result: The original web result data
user: User identifier
Returns:
True if successfully cached, False otherwise
"""
if not self.volatile_service:
logger.warning("Volatile service not configured, skipping volatile routing")
return False
namespace = classification.volatile_namespace or "custom"
key = classification.volatile_key or web_result['url'].split('/')[-1]
ttl = (classification.volatile_ttl_hours or 1) * 3600 # Convert hours to seconds
try:
# Store the web result content in volatile cache
data = {
"title": web_result.get('title', ''),
"content": web_result.get('content', ''),
"url": web_result.get('url', ''),
"text": f"{web_result.get('title', '')}: {web_result.get('content', '')[:500]}",
}
await self.volatile_service.store(
user=user,
namespace=namespace,
key=key,
data=data,
source=web_result.get('url', 'web_search'),
ttl=ttl,
)
logger.info(f"Cached to volatile: {namespace}/{key} (ttl={ttl}s)")
return True
except Exception as e:
logger.error(f"Failed to cache to volatile: {e}")
return False
async def _route_to_files(
self,
classification: MemoryRouteClassification,
web_result: Dict[str, Any],
user: str,
) -> bool:
"""
Queue a file for Paperless ingestion.
Args:
classification: The classification with file info
web_result: The original web result data
user: User identifier
Returns:
True if successfully queued, False otherwise
"""
# For now, log the file for manual review or future Paperless integration
url = web_result.get('url', '')
title = web_result.get('title', '')
logger.info(f"File detected for Paperless: {title} ({url})")
# TODO: Implement actual Paperless file upload
# This would involve:
# 1. Download the file
# 2. Upload to Paperless via API
# 3. Add tags based on classification
return True # Placeholder - count as queued
async def _register_prefetch(
self,
classification: MemoryRouteClassification,
web_result: Dict[str, Any],
user: str,
) -> bool:
"""
Register a prefetch pattern with the scheduler.
Args:
classification: The classification with prefetch info
web_result: The original web result data
user: User identifier
Returns:
True if successfully registered, False otherwise
"""
if not self.settings_client:
logger.warning("Settings client not configured, skipping prefetch registration")
return False
cron = classification.prefetch_cron or "0 * * * *" # Default: hourly
endpoint = classification.prefetch_endpoint or ""
if not endpoint:
logger.warning(f"No prefetch endpoint specified for {web_result.get('url')}")
return False
try:
# Store prefetch configuration in settings
prefetch_key = f"prefetch.{user}.{classification.volatile_key or 'auto'}"
prefetch_config = {
"cron": cron,
"endpoint": endpoint,
"source_url": web_result.get('url', ''),
"enabled": True,
}
await self.settings_client.set(prefetch_key, prefetch_config, user=user)
logger.info(f"Registered prefetch: {prefetch_key} ({cron})")
return True
except Exception as e:
logger.error(f"Failed to register prefetch: {e}")
return False