Spaces:
Running on Zero
Running on Zero
File size: 7,624 Bytes
31f2a28 e547cb9 31f2a28 e547cb9 31f2a28 e547cb9 31f2a28 e547cb9 31f2a28 e547cb9 31f2a28 e547cb9 31f2a28 e547cb9 31f2a28 e547cb9 31f2a28 e547cb9 31f2a28 e547cb9 31f2a28 e547cb9 31f2a28 e547cb9 31f2a28 e547cb9 31f2a28 e547cb9 31f2a28 e547cb9 31f2a28 e547cb9 31f2a28 e547cb9 31f2a28 e547cb9 31f2a28 e547cb9 | 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 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 | #!/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)) |