#!/usr/bin/env python # coding: utf-8 import json import hashlib import numpy as np from typing import List, Dict, Any, Optional from sklearn.metrics.pairwise import cosine_similarity class Node: def __init__(self, node_id: str, node_type: str, text: str, expected_keys: List[str] = None, runs: List[str] = None, execution_time_ms: float = None): self.id = node_id self.type = node_type.lower() self.text = str(text or "") self.embedding = None self.is_valid_format = True self.validation_error = None self.expected_keys = expected_keys self.runs = runs self.consistency_score = 100.0 self.instability_index = 0.0 self.execution_time_ms = execution_time_ms self.token_count = max(1, len(self.text.split())) def get_signature(self) -> str: content = f"{self.type}:{self.text.strip().lower()}" return hashlib.md5(content.encode('utf-8')).hexdigest() class Edge: def __init__(self, from_node_id: str, to_node_id: str): self.from_node_id = from_node_id self.to_node_id = to_node_id self.drift_score = 0.0 self.z_score = 0.0 class TrajectoryGraph: def __init__(self): self.nodes: Dict[str, Node] = {} self.edges: List[Edge] = [] def add_node(self, node_id: str, node_type: str, text: str, expected_keys: List[str] = None, runs: List[str] = None, execution_time_ms: float = None) -> Node: node = Node(node_id, node_type, text, expected_keys, runs, execution_time_ms) self.nodes[node_id] = node return node def add_edge(self, from_node_id: str, to_node_id: str): if from_node_id in self.nodes and to_node_id in self.nodes: edge = Edge(from_node_id, to_node_id) self.edges.append(edge) else: raise ValueError(f"Both nodes [{from_node_id}, {to_node_id}] must exist before creating an edge.") def validate_tool_calls(self): """Valideaza payload-urile JSON si detecteaza erori de executie ale uneltelor.""" for node in self.nodes.values(): if node.type == 'tool': stripped_text = node.text.strip() if not stripped_text: node.is_valid_format = False node.validation_error = "Empty tool execution response" continue # Verificare erori standard de sistem error_signatures = ["traceback (most recent call last)", "error:", "exception:", "unauthorized", "timed out"] if any(sig in stripped_text.lower() for sig in error_signatures): node.is_valid_format = False node.validation_error = f"Tool Execution Failure: {stripped_text[:100]}" continue if stripped_text.startswith('{') or stripped_text.startswith('['): try: data = json.loads(stripped_text) node.is_valid_format = True if node.expected_keys and isinstance(data, dict): missing_keys = [key for key in node.expected_keys if key not in data] if missing_keys: node.is_valid_format = False node.validation_error = f"Missing required fields in tool output: {missing_keys}" except json.JSONDecodeError as e: node.is_valid_format = False node.validation_error = f"Malformed JSON structure: {str(e)}" else: node.is_valid_format = True def detect_trajectory_cycles(self) -> List[Dict[str, Any]]: """Detecteaza cicluri repetitive de tip Ping-Pong sau bucle infinite consecutive.""" detected_loops = [] if len(self.edges) < 2: return detected_loops # Cautam secvente consecutive repetitive: A -> B urmat din nou de A -> B transition_history = [] for edge in self.edges: transition = f"{self.nodes[edge.from_node_id].get_signature()}->{self.nodes[edge.to_node_id].get_signature()}" if transition in transition_history: detected_loops.append({ 'failure_type': 'INFINITE_EXECUTION_LOOP', 'from_node': edge.from_node_id, 'to_node': edge.to_node_id, 'reason': f"Infinite State Cycle: Repeated transition pattern detected between [{edge.from_node_id}] and [{edge.to_node_id}].", 'details': f"Cycle Signature: {transition[:16]}..." }) transition_history.append(transition) return detected_loops def detect_stagnation(self, min_drift_threshold: float = 1.5) -> List[Dict[str, Any]]: """Detecteaza daca agentul bate pasul pe loc fara progres semantic intre unelte.""" stagnations = [] for edge in self.edges: n_from = self.nodes[edge.from_node_id] n_to = self.nodes[edge.to_node_id] if n_from.type in ['tool', 'thought'] and n_to.type in ['tool', 'thought']: if edge.drift_score < min_drift_threshold: stagnations.append({ 'failure_type': 'AGENT_STAGNATION', 'from_node': edge.from_node_id, 'to_node': edge.to_node_id, 'reason': f"Semantic Stagnation: Minimal cognitive drift ({edge.drift_score:.2f}%) between consecutive tool steps. Redundant execution suspected." }) return stagnations def calculate_node_consistency(self, get_embedding_func): """Calculeaza stabilitatea nodului intre multiple rulari.""" for node in self.nodes.values(): if node.runs and len(node.runs) >= 2: embeddings = [get_embedding_func(run) for run in node.runs] similarities = [] for i in range(len(embeddings)): for j in range(i + 1, len(embeddings)): sim = cosine_similarity([embeddings[i]], [embeddings[j]])[0][0] similarities.append(float(sim)) avg_sim = float(np.mean(similarities)) if similarities else 1.0 node.consistency_score = float(avg_sim * 100.0) node.instability_index = float(100.0 - node.consistency_score) def calculate_drift_scores(self): """Calculeaza si normalizeaza driftul semantic (0 - 100%) pe fiecare tranzitie.""" for edge in self.edges: node_from = self.nodes[edge.from_node_id] node_to = self.nodes[edge.to_node_id] if node_from.embedding is not None and node_to.embedding is not None: dot = np.dot(node_from.embedding, node_to.embedding) norm_a = np.linalg.norm(node_from.embedding) norm_b = np.linalg.norm(node_to.embedding) sim = dot / (norm_a * norm_b) if (norm_a * norm_b) > 0 else 0.0 # Normalizare sigura intre 0% si 100% similarity_percentage = float(np.clip(sim, -1.0, 1.0)) * 100.0 edge.drift_score = float(np.clip(100.0 - similarity_percentage, 0.0, 100.0))