Files
ALwrity/backend/services/intelligence/agents/specialized/link_graph.py

60 lines
2.2 KiB
Python

"""
Link Graph Agent implementation.
"""
from typing import List, Dict, Any, Optional
from datetime import datetime
from loguru import logger
from .base import SIFBaseAgent
from services.intelligence.agents.core_agent_framework import TaskProposal
from services.intelligence.txtai_service import TxtaiIntelligenceService
class LinkGraphAgent(SIFBaseAgent):
"""Agent for internal linking and graph optimization."""
def __init__(self, intelligence_service: TxtaiIntelligenceService, user_id: str, **kwargs):
super().__init__(intelligence_service, user_id, agent_type="link_graph_expert", **kwargs)
async def analyze_graph(self) -> Dict[str, Any]:
"""Analyze the knowledge graph structure of the content."""
if not self.intelligence.is_initialized():
return {}
try:
# Construct a graph from semantic relationships
graph = await self.intelligence.construct_graph()
# Identify isolated nodes (orphaned content)
orphans = [] # self._find_orphans(graph)
# Identify central nodes (pillars)
hubs = [] # self._find_hubs(graph)
return {
"node_count": 0, # graph.number_of_nodes(),
"edge_count": 0, # graph.number_of_edges(),
"orphaned_content": orphans,
"content_hubs": hubs
}
except Exception as e:
logger.error(f"[{self.__class__.__name__}] Graph analysis failed: {e}")
return {}
async def propose_daily_tasks(self, context: Dict[str, Any]) -> List[TaskProposal]:
"""Propose internal linking tasks."""
proposals = []
# 1. Internal Link Opportunity
proposals.append(TaskProposal(
title="Internal Linking Review",
description="Add internal links to your new post 'Content Strategy 101'.",
pillar_id="create",
priority="medium",
estimated_time=15,
source_agent="LinkGraphAgent",
reasoning="Improves SEO and user navigation.",
action_type="navigate",
action_url="/content-planning-dashboard"
))
return proposals