diff --git a/services/library-desk/src/services/consolidation_service.py b/services/library-desk/src/services/consolidation_service.py index ebcdb9a..6cee204 100644 --- a/services/library-desk/src/services/consolidation_service.py +++ b/services/library-desk/src/services/consolidation_service.py @@ -113,9 +113,10 @@ class ConsolidationService: total_pages_updated += result.pages_updated total_entities_added += result.entities_added - # Mark as processed if not dry run - if not dry_run and not result.error: - await self._mark_search_processed(search['id']) + # Mark as processed if not dry run (even if skipped) + # This prevents searches from accumulating when they don't meet criteria + if not dry_run: + await self._mark_search_processed(search['id']) except Exception as e: error_msg = f"Search {search['id'][:8]}: {str(e)}" @@ -127,6 +128,10 @@ class ConsolidationService: error=str(e) )) + # Mark as processed even on error (to avoid retrying failed searches forever) + if not dry_run: + await self._mark_search_processed(search['id']) + # Build response processed_count = len([r for r in results if not r.error]) @@ -233,7 +238,8 @@ class ConsolidationService: analysis = await self._analyze_web_results( query=query, web_results=web_results, - keywords=search.get('keywords', []) + keywords=search.get('keywords', []), + user=user ) if not analysis or not analysis.get('has_novel_info'): @@ -357,13 +363,23 @@ class ConsolidationService: self, query: str, web_results: List[Dict[str, Any]], - keywords: List[str] + keywords: List[str], + user: str = "jpmschweitzer" ) -> Optional[Dict[str, Any]]: """ Analyze web results with Ollama for novel information. Returns analysis with has_novel_info, new_pages, update_pages, new_entities. """ + # Fetch existing taxonomy structure for this user + 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 analysis prompt web_summary = "\n\n".join([ f"[{i+1}] {r['title']}\n{r['url']}\n{r['content'][:300]}..." @@ -427,6 +443,8 @@ Personal information is just as valuable as technical information. - Keep paths 2-3 levels deep maximum - Be consistent with existing paths when possible +{existing_paths_info} + Return ONLY valid JSON: {{ "has_novel_info": true, @@ -473,6 +491,34 @@ JSON:""" logger.error(f"Analysis failed: {e}", exc_info=True) return None + def _format_taxonomy_for_prompt(self, taxonomy: Dict[str, List[str]]) -> str: + """ + Format taxonomy structure for inclusion in LLM prompt. + + Args: + taxonomy: Dict mapping categories to subcategories + + Returns: + Formatted string showing existing paths + """ + if not taxonomy: + return "" + + lines = ["**Existing paths in your wiki (PREFER these over creating new ones):**"] + for category, subcategories in taxonomy.items(): + if subcategories: + lines.append(f"- {category}/") + for sub in subcategories: + lines.append(f" - {category}/{sub}/") + else: + lines.append(f"- {category}/") + + lines.append("") + lines.append("**IMPORTANT:** If a suitable existing path exists, use it instead of creating a new category.") + lines.append("Example: NATO should go in `reference/political-entities/` not a new `reference/military-alliances/`") + + return "\n".join(lines) + async def _mark_search_processed(self, search_id: str): """Mark SearchQuery node as processed.""" query = """ @@ -488,6 +534,140 @@ JSON:""" except Exception as e: logger.error(f"Failed to mark search as processed: {e}") + async def _apply_bidirectional_entity_linking( + self, + page_id: int, + page_title: str, + user: str + ) -> Dict[str, int]: + """ + Apply bidirectional entity linking after page creation/update. + + This runs AFTER ingestion so entities are extracted and in the graph. + + Steps: + 1. Link entities in the new page (forward links to existing entities) + 2. Find pages that mention the new entity (reverse references) + 3. Link entities in those pages (backward links to the new entity) + + Args: + page_id: Wiki page ID + page_title: Page title (used to find reverse references) + user: User identifier + + Returns: + Dict with link counts: { + "forward_links": int, # Links added to the new page + "backward_links": int, # Links added to other pages pointing to new page + "pages_updated": int # Number of other pages updated + } + """ + from src.core.multi_tenancy import get_neo4j_user_base_label + + forward_links = 0 + backward_links = 0 + pages_updated = 0 + + try: + # Import here to avoid circular dependency + from src.routers.entity_linking import link_entities_in_page, EntityLinkingRequest + from src.core.dependencies import get_wiki_service, get_graph_service + + wiki_service = get_wiki_service() + graph_service = get_graph_service() + + # STEP 1: Forward linking - link entities in the new page + logger.info(f"Step 1/3: Linking entities in page {page_id} ('{page_title}')") + try: + forward_result = await link_entities_in_page( + request=EntityLinkingRequest( + user=user, + page_id=page_id, + create_relationships=True, + re_index_if_changed=False # Already indexed, no need to re-index + ), + wiki_service=wiki_service, + graph_service=graph_service, + ingestion_service=self.ingestion_service, + api_key="" # Internal call, no auth needed + ) + forward_links = forward_result.content_links_added + logger.info(f"Added {forward_links} forward links in page {page_id}") + except Exception as e: + logger.error(f"Failed to add forward links: {e}") + + # STEP 2: Find reverse references - which pages mention this new entity? + logger.info(f"Step 2/3: Finding pages that mention '{page_title}'") + user_base_label = get_neo4j_user_base_label(user) + + # Query to find documents that mention entities with this page's title + reverse_query = f""" + // Find entities with the same name as the page title + MATCH (e:{user_base_label}) + WHERE toLower(e.name) = toLower($title) + AND NOT e:Document + + // Find documents that mention those entities + MATCH (d:Document)-[r:MENTIONS]->(e) + WHERE d.page_id <> $page_id // Exclude the page itself + + RETURN DISTINCT d.page_id as page_id, d.title as title + LIMIT 50 + """ + + try: + reverse_refs = await self.neo4j.execute_query( + reverse_query, + {"title": page_title, "page_id": page_id} + ) + logger.info(f"Found {len(reverse_refs)} pages that mention '{page_title}'") + except Exception as e: + logger.error(f"Failed to find reverse references: {e}") + reverse_refs = [] + + # STEP 3: Backward linking - add links in those pages to the new entity + if reverse_refs: + logger.info(f"Step 3/3: Adding backward links in {len(reverse_refs)} pages") + for ref in reverse_refs: + try: + backward_result = await link_entities_in_page( + request=EntityLinkingRequest( + user=user, + page_id=ref['page_id'], + create_relationships=False, # Relationships already exist + re_index_if_changed=False # Don't re-index for link updates + ), + wiki_service=wiki_service, + graph_service=graph_service, + ingestion_service=self.ingestion_service, + api_key="" + ) + if backward_result.content_links_added > 0: + backward_links += backward_result.content_links_added + pages_updated += 1 + logger.info( + f"Added {backward_result.content_links_added} links " + f"in page {ref['page_id']} ('{ref['title']}')" + ) + except Exception as e: + logger.error(f"Failed to add backward links in page {ref['page_id']}: {e}") + else: + logger.info("Step 3/3: No reverse references found, skipping backward linking") + + return { + "forward_links": forward_links, + "backward_links": backward_links, + "pages_updated": pages_updated + } + + except Exception as e: + logger.error(f"Bidirectional entity linking failed: {e}", exc_info=True) + return { + "forward_links": 0, + "backward_links": 0, + "pages_updated": 0 + } + async def _create_or_consolidate_page( self, user: str, @@ -564,6 +744,18 @@ JSON:""" force_refresh=True ) logger.info(f"Ingested updated page {page_id} into knowledge base") + + # Apply bidirectional entity linking after ingestion + link_stats = await self._apply_bidirectional_entity_linking( + page_id=page_id, + page_title=title, + user=user + ) + logger.info( + f"Entity linking complete: {link_stats['forward_links']} forward links, " + f"{link_stats['backward_links']} backward links " + f"({link_stats['pages_updated']} pages updated)" + ) except Exception as e: logger.error(f"Failed to ingest updated page {page_id}: {e}") @@ -606,6 +798,18 @@ JSON:""" force_refresh=False # New page, no need to force ) logger.info(f"Ingested new page {page_id} into knowledge base") + + # Apply bidirectional entity linking after ingestion + link_stats = await self._apply_bidirectional_entity_linking( + page_id=page_id, + page_title=title, + user=user + ) + logger.info( + f"Entity linking complete: {link_stats['forward_links']} forward links, " + f"{link_stats['backward_links']} backward links " + f"({link_stats['pages_updated']} pages updated)" + ) except Exception as e: logger.error(f"Failed to ingest new page {page_id}: {e}") @@ -663,6 +867,32 @@ JSON:""" content=reconstructed_content ) + # Trigger ingestion to update vectors and graph + logger.debug(f"ingestion_service available: {self.ingestion_service is not None}") + if self.ingestion_service: + try: + logger.info(f"Starting ingestion for updated page {page_id}") + await self.ingestion_service.ingest_page( + page_id=page_id, + user=user, + force_refresh=True + ) + logger.info(f"Ingested updated page {page_id} into knowledge base") + + # Apply bidirectional entity linking after ingestion + link_stats = await self._apply_bidirectional_entity_linking( + page_id=page_id, + page_title=title, + user=user + ) + logger.info( + f"Entity linking complete: {link_stats['forward_links']} forward links, " + f"{link_stats['backward_links']} backward links " + f"({link_stats['pages_updated']} pages updated)" + ) + except Exception as e: + logger.error(f"Failed to ingest updated page {page_id}: {e}") + async def _add_entity_to_graph( self, user: str,