jpmschweitzer/portainer-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(library-desk): add taxonomy-aware classification to consolidation

Browse Source 
- Fetch existing wiki taxonomy structure before LLM analysis
- Include existing paths in prompt to prefer existing categories
- Add _format_taxonomy_for_prompt helper
- Mark searches as processed even when skipped/errored (prevents buildup)

🤖 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-10 21:30:58 +01:00
co-authored by Claude Opus 4.5
parent 9baea7ee6c
commit 6ddf2a1bd9
1 changed files with 235 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/library-desk/src/services/consolidation_service.py
+235 -5
View File
@@ -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,