Spaces:
Runtime error
Runtime error
| """ | |
| GraphAnalysisAgent - Network pattern detection | |
| Analyzes account network: peers, cycles, communities | |
| """ | |
| from typing import Dict, List | |
| from .base import Agent, AgentConfig, AgentResult | |
| import time | |
| class GraphAnalysisAgent(Agent): | |
| """ | |
| Analyzes transaction network around an account | |
| Direct graph operations (no LLM needed for retrieval) | |
| """ | |
| def __init__(self, api_pool, app_state=None): | |
| config = AgentConfig( | |
| name="GraphAnalysisAgent", | |
| model="llama-3.1-8b-instant", # llama-3.1-70b-versatile deprecated | |
| temperature=0.2, | |
| max_tokens=1000, | |
| timeout_ms=10000, | |
| ) | |
| super().__init__(config, api_pool) | |
| self.app_state = app_state | |
| def set_app_state(self, app_state): | |
| self.app_state = app_state | |
| def _build_prompt(self, **inputs) -> str: | |
| return "" | |
| def _parse_response(self, response_text: str) -> Dict: | |
| return {} | |
| async def invoke(self, account_id: str = None, hops: int = 2, max_nodes: int = 30, **kwargs) -> AgentResult: | |
| """ | |
| Analyze network around account | |
| Returns: peers, cycles, community info, network metrics | |
| """ | |
| start_time = time.time() | |
| if not self.app_state or not account_id: | |
| return await self._create_result( | |
| success=False, | |
| data={}, | |
| error="No account_id or app_state", | |
| start_time=start_time, | |
| ) | |
| try: | |
| data = { | |
| "account_id": account_id, | |
| "network_metrics": {}, | |
| "peers": [], | |
| "cycles": [], | |
| "community": {}, | |
| "subgraph_summary": {}, | |
| } | |
| if not hasattr(self.app_state, "graph") or self.app_state.graph is None: | |
| return await self._create_result( | |
| success=False, | |
| data=data, | |
| error="Graph not available", | |
| start_time=start_time, | |
| ) | |
| graph = self.app_state.graph | |
| if account_id not in graph.nodes(): | |
| return await self._create_result( | |
| success=True, | |
| data=data, | |
| start_time=start_time, | |
| ) | |
| # Get network metrics | |
| data["network_metrics"] = { | |
| "in_degree": graph.in_degree(account_id), | |
| "out_degree": graph.out_degree(account_id), | |
| "total_degree": graph.degree(account_id), | |
| } | |
| # Add centrality scores if available | |
| if hasattr(self.app_state, "pagerank_scores"): | |
| data["network_metrics"]["pagerank"] = self.app_state.pagerank_scores.get(account_id, 0) | |
| if hasattr(self.app_state, "betweenness_scores"): | |
| data["network_metrics"]["betweenness"] = self.app_state.betweenness_scores.get(account_id, 0) | |
| # Community info | |
| if hasattr(self.app_state, "louvain_partition"): | |
| community_id = self.app_state.louvain_partition.get(account_id) | |
| if community_id is not None: | |
| community_members = [ | |
| node for node, comm in self.app_state.louvain_partition.items() | |
| if comm == community_id | |
| ] | |
| fraud_in_community = sum( | |
| 1 for node in community_members | |
| if self.app_state.features_by_account.get(node, {}).get("fraud_flag", 0) | |
| ) | |
| data["community"] = { | |
| "community_id": community_id, | |
| "size": len(community_members), | |
| "fraud_count": fraud_in_community, | |
| "fraud_rate": fraud_in_community / max(1, len(community_members)), | |
| } | |
| # Get direct peers (1-hop) | |
| successors = list(graph.successors(account_id))[:10] | |
| predecessors = list(graph.predecessors(account_id))[:10] | |
| peers = [] | |
| for peer_id in set(successors + predecessors): | |
| peer_features = self.app_state.features_by_account.get(peer_id, {}) | |
| peer_data = { | |
| "account_id": peer_id, | |
| "relationship": [], | |
| "risk_score": peer_features.get("risk_score", 0), | |
| "is_fraud": peer_features.get("fraud_flag", 0) == 1, | |
| } | |
| if peer_id in successors: | |
| peer_data["relationship"].append("sends_to") | |
| if peer_id in predecessors: | |
| peer_data["relationship"].append("receives_from") | |
| peers.append(peer_data) | |
| # Sort peers by risk | |
| peers.sort(key=lambda x: x["risk_score"], reverse=True) | |
| data["peers"] = peers[:10] | |
| # Detect cycles (simple cycle detection) | |
| cycles = self._detect_cycles(graph, account_id, max_length=5) | |
| data["cycles"] = cycles[:5] | |
| # Subgraph summary | |
| try: | |
| from src.graph_builder import get_subgraph | |
| subgraph = get_subgraph(graph, account_id, hops=hops, max_nodes=max_nodes) | |
| data["subgraph_summary"] = { | |
| "nodes_count": subgraph.number_of_nodes(), | |
| "edges_count": subgraph.number_of_edges(), | |
| "depth": hops, | |
| } | |
| except Exception as e: | |
| self.logger.warning(f"Subgraph extraction failed: {e}") | |
| tokens_estimate = self._estimate_tokens(str(data)) | |
| self.logger.info( | |
| f"[OK] {self.config.name}: Analyzed {account_id} " | |
| f"({len(peers)} peers, {len(cycles)} cycles, ~{tokens_estimate} tokens)" | |
| ) | |
| return await self._create_result( | |
| success=True, | |
| data=data, | |
| tokens_output=tokens_estimate, | |
| start_time=start_time, | |
| ) | |
| except Exception as e: | |
| self.logger.error(f"[FAIL] {self.config.name}: {e}") | |
| return await self._create_result( | |
| success=False, | |
| data={}, | |
| error=str(e), | |
| start_time=start_time, | |
| ) | |
| def _detect_cycles(self, graph, account_id: str, max_length: int = 5) -> List[Dict]: | |
| """Detect cycles involving the account (simplified)""" | |
| cycles = [] | |
| try: | |
| import networkx as nx | |
| # Get small subgraph for cycle detection (performance) | |
| try: | |
| from src.graph_builder import get_subgraph | |
| subgraph = get_subgraph(graph, account_id, hops=3, max_nodes=50) | |
| except Exception: | |
| subgraph = graph | |
| # Find simple cycles containing account_id | |
| simple_cycles = list(nx.simple_cycles(subgraph)) | |
| for cycle in simple_cycles[:5]: | |
| if account_id in cycle and len(cycle) <= max_length: | |
| cycles.append({ | |
| "cycle": cycle + [cycle[0]], # Close the cycle for display | |
| "length": len(cycle), | |
| }) | |
| except Exception as e: | |
| self.logger.warning(f"Cycle detection failed: {e}") | |
| return cycles | |