Spaces:
Running
Running
File size: 2,935 Bytes
09801ca | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | """
Memory Agent - Incremental knowledge graph updates
Maintains and enhances the workspace knowledge graph
"""
from agents.base.agent_runner import AgentRunner, Insight
from graph.query import load_graph
import logging
from typing import List
logger = logging.getLogger(__name__)
class MemoryAgent(AgentRunner):
"""Maintains and updates knowledge graph memory"""
def __init__(self):
super().__init__('MemoryAgent')
async def detect_insights(self, workspace_id: str) -> List[Insight]:
"""
Update knowledge graph and detect memory-related insights
This agent focuses on maintaining data quality
"""
insights = []
try:
# Load current graph
graph = load_graph(workspace_id)
if not graph or graph.number_of_nodes() == 0:
self.logger.info(f"No graph data for workspace {workspace_id}")
return []
# Check graph health
num_nodes = graph.number_of_nodes()
num_edges = graph.number_of_edges()
# Detect orphaned nodes
orphaned_nodes = [node for node in graph.nodes() if graph.degree(node) == 0]
if len(orphaned_nodes) > num_nodes * 0.1: # More than 10% orphaned
insight = Insight(
title="🧠 Knowledge Graph Alert",
body=f"Detected {len(orphaned_nodes)} isolated entities in your knowledge graph. This may indicate data quality issues or missing relationships.",
severity='medium',
score=60,
metadata={
'orphaned_count': len(orphaned_nodes),
'total_nodes': num_nodes,
'total_edges': num_edges
}
)
insights.append(insight)
# Positive insight if graph is healthy
if num_nodes > 100 and num_edges > 200:
insight = Insight(
title="✅ Knowledge Graph Healthy",
body=f"Your business knowledge graph is well-connected with {num_nodes} entities and {num_edges} relationships. Good data quality detected!",
severity='low',
score=100,
metadata={
'nodes': num_nodes,
'edges': num_edges,
'density': num_edges / (num_nodes * (num_nodes - 1)) if num_nodes > 1 else 0
}
)
insights.append(insight)
self.logger.info(f"MemoryAgent generated {len(insights)} insights")
return insights
except Exception as e:
self.logger.error(f"MemoryAgent failed: {e}", exc_info=True)
return []
|