graph_engine.py

#!/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))