#!/usr/bin/env python # coding: utf-8 import os import json import re import hashlib from collections import OrderedDict from typing import Union, List, Dict, Any, Tuple import numpy as np import pandas as pd import onnxruntime as ort from transformers import AutoTokenizer from huggingface_hub import hf_hub_download from graph_engine import TrajectoryGraph _emb_tokenizer = None _model = None _nli_tokenizer = None _cross_encoder = None NLI_ID2LABEL = {0: "contradiction", 1: "entailment", 2: "neutral"} PRICE_INPUT_PER_TOKEN = 2.5 / 1000000 PRICE_OUTPUT_PER_TOKEN = 10.0 / 1000000 MAX_FREE_TIER_STEPS = 20 PROFILES_CONFIG = { "standard": {"multiplier": 1.0, "max_latency_ms": 4000.0, "forbidden_words": ["competitorxyz", "guaranteed refund"], "required_words": []}, "banking": {"multiplier": 0.5, "max_latency_ms": 2000.0, "forbidden_words": ["guaranteed profit", "unlimited cash back"], "required_words": ["disclaimer"]}, "healthcare": {"multiplier": 0.6, "max_latency_ms": 2500.0, "forbidden_words": ["100% cure", "prescribe without doctor"], "required_words": ["medical advice"]}, "customer_support": {"multiplier": 1.0, "max_latency_ms": 3000.0, "forbidden_words": ["fuck", "idiot"], "required_words": []}, "creative": {"multiplier": 1.5, "max_latency_ms": 6000.0, "forbidden_words": [], "required_words": []} } DEFAULT_STRICTNESS_PROFILE = "standard" DEFAULT_FORBIDDEN_PHRASES = ["as an ai", "as a language model", "ignore previous instructions", "system prompt"] PROMPT_INJECTION_PATTERNS = [ r"ignore\s+(all\s+)?previous\s+instructions", r"system\s+override", r"you\s+are\s+now\s+in\s+developer\s+mode", r"disregard\s+prior\s+guidelines", r"jailbreak", r"dan\s+mode" ] EMAIL_REGEX = r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}" API_KEY_REGEX = r"sk-[a-zA-Z0-9]{20,}|gsk_[a-zA-Z0-9]{30,}|limina_live_[a-zA-Z0-9]{24,}" def get_emb_model(): global _emb_tokenizer, _model if _model is None: model_id = "Xenova/all-MiniLM-L6-v2" _emb_tokenizer = AutoTokenizer.from_pretrained(model_id) model_path = hf_hub_download(repo_id=model_id, filename="onnx/model_quantized.onnx") opts = ort.SessionOptions() opts.intra_op_num_threads = 2 opts.inter_op_num_threads = 2 opts.execution_mode = ort.ExecutionMode.ORT_SEQUENTIAL _model = ort.InferenceSession(model_path, sess_options=opts, providers=['CPUExecutionProvider']) return _emb_tokenizer, _model def get_nli_model(): global _nli_tokenizer, _cross_encoder if _cross_encoder is None: model_id = "Xenova/nli-deberta-v3-small" _nli_tokenizer = AutoTokenizer.from_pretrained(model_id) model_path = hf_hub_download(repo_id=model_id, filename="onnx/model_quantized.onnx") opts = ort.SessionOptions() opts.intra_op_num_threads = 2 opts.inter_op_num_threads = 2 opts.execution_mode = ort.ExecutionMode.ORT_SEQUENTIAL _cross_encoder = ort.InferenceSession(model_path, sess_options=opts, providers=['CPUExecutionProvider']) return _nli_tokenizer, _cross_encoder class LRUEmbeddingCache: def __init__(self, capacity: int = 2048): self.cache = OrderedDict() self.capacity = capacity def get(self, key): if key not in self.cache: return None self.cache.move_to_end(key) return self.cache[key] def set(self, key, value): self.cache[key] = value self.cache.move_to_end(key) if len(self.cache) > self.capacity: self.cache.popitem(last=False) EMBEDDING_CACHE = LRUEmbeddingCache(capacity=2048) def get_cached_embedding(text: str) -> np.ndarray: text_hash = hashlib.sha256(text.encode('utf-8')).hexdigest() cached = EMBEDDING_CACHE.get(text_hash) if cached is not None: return cached tokenizer, emb_model = get_emb_model() inputs = tokenizer(text, padding=True, truncation=True, max_length=512, return_tensors='np') input_names = [inp.name for inp in emb_model.get_inputs()] ort_inputs = {k: v for k, v in inputs.items() if k in input_names} outputs = emb_model.run(None, ort_inputs) token_embeddings = outputs[0] input_mask_expanded = np.expand_dims(inputs['attention_mask'], -1) sum_embeddings = np.sum(token_embeddings * input_mask_expanded, axis=1) sum_mask = np.clip(input_mask_expanded.sum(axis=1), a_min=1e-9, a_max=None) embedding = (sum_embeddings / sum_mask)[0] EMBEDDING_CACHE.set(text_hash, embedding) return embedding def detect_prompt_injection(user_text: str) -> Tuple[bool, str]: for pattern in PROMPT_INJECTION_PATTERNS: match = re.search(pattern, user_text, re.IGNORECASE) if match: return True, f"Prompt Injection Attempt: '{match.group(0)}'" return False, "" def detect_pii_leakage(generated_text: str, reference_text: str) -> Tuple[bool, str]: gen_emails = set(re.findall(EMAIL_REGEX, generated_text)) ref_emails = set(re.findall(EMAIL_REGEX, reference_text)) leaked_emails = gen_emails - ref_emails gen_keys = set(re.findall(API_KEY_REGEX, generated_text, re.IGNORECASE)) ref_keys = set(re.findall(API_KEY_REGEX, reference_text, re.IGNORECASE)) leaked_keys = gen_keys - ref_keys if leaked_emails: return True, f"Leaked sensitive emails: {list(leaked_emails)}" if leaked_keys: return True, f"Leaked API Credentials: {list(leaked_keys)}" return False, "" def validate_tone_and_style(text: str, max_sentences: int = 4) -> Tuple[bool, str]: stripped_text = text.strip() if not stripped_text: return True, "" sentences = [s for s in re.split(r'(?<=[.!?])\s+', stripped_text) if len(s.strip()) > 0] if len(sentences) > max_sentences: return False, f"Tone Violation: Agent response verbose ({len(sentences)}/{max_sentences} sentences)." text_lower = stripped_text.lower() found_cliches = [p for p in DEFAULT_FORBIDDEN_PHRASES if p in text_lower] if found_cliches: return False, f"Style Violation: Robotic cliché detected: {found_cliches}" return True, "" def validate_business_rules(text: str, profile_name: str = "standard") -> Tuple[bool, str]: profile = PROFILES_CONFIG.get(profile_name.lower(), PROFILES_CONFIG["standard"]) stripped = text.lower().strip() if not stripped: return True, "" found_forbidden = [w for w in profile.get('forbidden_words', []) if w in stripped] if found_forbidden: return False, f"Business Rule Violation: Forbidden keyword(s): {found_forbidden}" missing_required = [w for w in profile.get('required_words', []) if w not in stripped] if missing_required: return False, f"Business Rule Violation: Missing mandatory phrase(s): {missing_required}" return True, "" def get_edge_tolerances(from_type: str, to_type: str, profile_name: str = "standard") -> Tuple[float, float]: profile = PROFILES_CONFIG.get(profile_name.lower(), PROFILES_CONFIG["standard"]) mult = profile["multiplier"] if from_type == 'user' and to_type == 'thought': return 1.8 * mult, 35.0 if from_type == 'tool' and to_type == 'agent': return 0.8 * mult, 14.0 return 1.2 * mult, 20.0 def verify_goal_completion(initial_user_text: str, final_agent_text: str) -> Tuple[bool, str]: if not initial_user_text.strip() or not final_agent_text.strip(): return True, "" tokenizer, cross_enc = get_nli_model() nli_inputs = tokenizer([initial_user_text], [final_agent_text], padding=True, truncation=True, max_length=512, return_tensors='np') input_names = [inp.name for inp in cross_enc.get_inputs()] ort_inputs = {k: v for k, v in nli_inputs.items() if k in input_names} logits = cross_enc.run(None, ort_inputs)[0] label_idx = int(np.argmax(logits[0])) label = NLI_ID2LABEL.get(label_idx, "unknown").lower() if "contradict" in label: return False, "Goal Abandonment: Final agent output directly contradicts initial user goal." return True, "" def verify_atomic_grounding(agent_text: str, context_text: str) -> Tuple[bool, str]: if not agent_text.strip() or not context_text.strip(): return True, "" tokenizer, cross_enc = get_nli_model() sentences = [s.strip() for s in re.split(r'(?<=[.!?])\s+', agent_text) if len(s.strip()) > 5] if not sentences: return True, "" contexts = [context_text] * len(sentences) nli_inputs = tokenizer(contexts, sentences, padding=True, truncation=True, max_length=512, return_tensors='np') input_names = [inp.name for inp in cross_enc.get_inputs()] ort_inputs = {k: v for k, v in nli_inputs.items() if k in input_names} nli_outputs = cross_enc.run(None, ort_inputs) logits = nli_outputs[0] label_indices = np.argmax(logits, axis=1) ungrounded = [] for idx, label_idx in enumerate(label_indices): label = NLI_ID2LABEL.get(int(label_idx), "unknown").lower() if "contradict" in label or "neutral" in label: ungrounded.append({'sentence': sentences[idx], 'nli_label': label}) if ungrounded: details = " | ".join([f"'{item['sentence']}' ({item['nli_label']})" for item in ungrounded]) return False, f"Ungrounded Hallucination: {len(ungrounded)} claims unsupported by tool context. Details: {details}" return True, "" def generate_executive_summary(reports: list) -> dict: total = len(reports) if total == 0: return {} failed = [r for r in reports if r.get('status') == 'FAILED'] success_rate = ((total - len(failed)) / total) * 100 rating = "A" if success_rate >= 90 else "B" if success_rate >= 75 else "C" if success_rate >= 50 else "F" unique_failures = set(f.get('failure_type') for s in failed for f in s.get('failures', [])) total_nodes = 0 total_errors = 0 for r in reports: total_nodes += len(r.get('enriched_graph', {}).get('nodes', [])) total_errors += len(r.get('failures', [])) return { 'health_rating': rating, 'success_rate_percentage': round(success_rate, 1), 'most_vulnerable_component': ", ".join(list(unique_failures)) if unique_failures else "NONE", 'actionable_advice': "Review failed trajectories and apply prompt patches." if failed else "Optimal trajectory stability verified.", 'total_nodes': total_nodes, 'errors_detected': total_errors } def evaluate_trajectories_batch( input_data: Union[str, List[Dict[str, Any]]], profile: str = None, run_stress_test: bool = False, plan: str = "free" ) -> dict: if isinstance(input_data, str): with open(input_data, 'r', encoding='utf-8') as f: sessions_data = pd.DataFrame(json.load(f)) elif isinstance(input_data, list): sessions_data = pd.DataFrame(input_data) else: raise ValueError("Invalid input format.") # --- VERIFICARE LIMITA STEPS PE FREE TIER --- if plan == "free": for _, session in sessions_data.iterrows(): node_count = len(session.get("nodes", [])) if node_count > MAX_FREE_TIER_STEPS: return { "error": f"Free Tier Limit Exceeded: Session [{session.get('session_id', 'unknown')}] contains {node_count} steps. (Limit is {MAX_FREE_TIER_STEPS}). Upgrade to Pro for unlimited trajectory depth.", "status_code": 429 } active_profile = profile or DEFAULT_STRICTNESS_PROFILE profile_cfg = PROFILES_CONFIG.get(active_profile.lower(), PROFILES_CONFIG["standard"]) max_tool_latency = profile_cfg["max_latency_ms"] tokenizer, cross_enc = get_nli_model() emb_tok, _ = get_emb_model() batch_reports = [] for _, session in sessions_data.iterrows(): # Izolare robusta pe fiecare sesiune individuala try: graph = TrajectoryGraph() for node in session.get('nodes', []): graph.add_node(str(node['id']), str(node['type']), str(node.get('text', '')), execution_time_ms=node.get('execution_time_ms')) for edge in session.get('edges', []): graph.add_edge(str(edge.get('from') or edge.get('from_node')), str(edge.get('to') or edge.get('to_node'))) graph.validate_tool_calls() for node in graph.nodes.values(): node.embedding = get_cached_embedding(node.text) graph.calculate_drift_scores() drifts = [edge.drift_score for edge in graph.edges] mean_drift = float(np.mean(drifts)) if drifts else 0.0 std_drift = float(np.std(drifts)) if drifts else 0.0 max_drift = max(drifts) if drifts else 0.0 failed_transitions = [] # 1. Prompt Injection for n in graph.nodes.values(): if n.type == 'user': has_inj, inj_msg = detect_prompt_injection(n.text) if has_inj: failed_transitions.append({'from_node': n.id, 'to_node': n.id, 'failure_type': 'PROMPT_INJECTION_ATTEMPT', 'reason': inj_msg}) # 2. Cicluri Infinite for loop_fail in graph.detect_trajectory_cycles(): failed_transitions.append(loop_fail) # 3. Stagnare Semantica for stag_fail in graph.detect_stagnation(): failed_transitions.append(stag_fail) # 4. Muchii si Validare Noduri for edge in graph.edges: node_from = graph.nodes[edge.from_node_id] node_to = graph.nodes[edge.to_node_id] if node_to.type == 'tool' and node_to.execution_time_ms and node_to.execution_time_ms > max_tool_latency: failed_transitions.append({ 'from_node': edge.from_node_id, 'to_node': edge.to_node_id, 'failure_type': 'TOOL_TIMEOUT', 'reason': f"Latency Timeout: {node_to.execution_time_ms:.1f}ms > {max_tool_latency}ms" }) if node_to.type == 'tool' and not node_to.is_valid_format: failed_transitions.append({ 'from_node': edge.from_node_id, 'to_node': edge.to_node_id, 'failure_type': 'TOOL_FORMAT_ERROR', 'reason': node_to.validation_error or "Invalid tool format" }) if node_to.type == 'agent': has_leak, leak_details = detect_pii_leakage(node_to.text, node_from.text) if has_leak: failed_transitions.append({ 'from_node': edge.from_node_id, 'to_node': edge.to_node_id, 'failure_type': 'SECURITY_LEAK', 'reason': 'Sensitive Credential Leak', 'details': leak_details }) is_valid_tone, tone_err = validate_tone_and_style(node_to.text) if not is_valid_tone: failed_transitions.append({'from_node': edge.from_node_id, 'to_node': edge.to_node_id, 'failure_type': 'TONE_STYLE_VIOLATION', 'reason': tone_err}) is_valid_biz, biz_err = validate_business_rules(node_to.text, profile_name=active_profile) if not is_valid_biz: failed_transitions.append({'from_node': edge.from_node_id, 'to_node': edge.to_node_id, 'failure_type': 'BUSINESS_RULE_VIOLATION', 'reason': biz_err}) _, drift_threshold = get_edge_tolerances(node_from.type, node_to.type, profile_name=active_profile) if edge.drift_score > drift_threshold: nli_inputs = tokenizer([node_from.text], [node_to.text], padding=True, truncation=True, max_length=512, return_tensors='np') input_names = [inp.name for inp in cross_enc.get_inputs()] ort_inputs = {k: v for k, v in nli_inputs.items() if k in input_names} nli_outputs = cross_enc.run(None, ort_inputs) logits = nli_outputs[0] label_idx = int(np.argmax(logits[0])) label = NLI_ID2LABEL.get(label_idx, "unknown").lower() if "contradict" in label: failed_transitions.append({ 'from_node': edge.from_node_id, 'to_node': edge.to_node_id, 'failure_type': 'GENERATION_CONTRADICTION', 'drift_score': float(edge.drift_score), 'reason': 'Logical Contradiction: Output contradicts previous state.' }) # 5. Atomic Grounding tool_nodes = [n for n in graph.nodes.values() if n.type == 'tool'] agent_nodes = [n for n in graph.nodes.values() if n.type == 'agent'] if tool_nodes and agent_nodes: combined_context = "\n".join([t.text for t in tool_nodes]) for agent_node in agent_nodes: is_grounded, grounding_error = verify_atomic_grounding(agent_node.text, combined_context) if not is_grounded: failed_transitions.append({ 'from_node': tool_nodes[-1].id, 'to_node': agent_node.id, 'failure_type': 'UNGROUNDED_HALLUCINATION', 'reason': grounding_error }) # 6. Goal Completion user_nodes = [n for n in graph.nodes.values() if n.type == 'user'] if user_nodes and agent_nodes: first_user = user_nodes[0] last_agent = agent_nodes[-1] is_goal_met, goal_err = verify_goal_completion(first_user.text, last_agent.text) if not is_goal_met: failed_transitions.append({ 'from_node': first_user.id, 'to_node': last_agent.id, 'failure_type': 'GOAL_ABANDONMENT', 'reason': goal_err }) input_tokens = sum(len(emb_tok.encode(n.text)) for n in graph.nodes.values() if n.type != 'agent') output_tokens = sum(len(emb_tok.encode(n.text)) for n in graph.nodes.values() if n.type == 'agent') batch_reports.append({ 'session_id': session.get('session_id', 'unknown'), 'description': session.get('description', 'Agent Trajectory'), 'status': 'FAILED' if failed_transitions else 'STABLE', 'max_drift_detected': float(max_drift), 'mean_drift': mean_drift, 'std_drift': std_drift, 'failures': failed_transitions, 'total_tokens': input_tokens + output_tokens, 'estimated_cost_usd': (input_tokens * PRICE_INPUT_PER_TOKEN) + (output_tokens * PRICE_OUTPUT_PER_TOKEN), 'enriched_graph': { 'nodes': [{'id': n.id, 'type': n.type, 'text': n.text, 'execution_time_ms': n.execution_time_ms} for n in graph.nodes.values()], 'edges': [{'from': e.from_node_id, 'to': e.to_node_id, 'drift_score': float(e.drift_score / 100.0)} for e in graph.edges] } }) except Exception as sess_err: batch_reports.append({ 'session_id': session.get('session_id', 'unknown'), 'status': 'FAILED', 'failures': [{'failure_type': 'MALFORMED_GRAPH', 'reason': f"Graph parsing error: {str(sess_err)}"}], 'enriched_graph': {'nodes': [], 'edges': []} }) summary = generate_executive_summary(batch_reports) return { 'executive_summary': summary, 'regression_report': {'status': 'STABLE', 'message': 'Verified'}, 'results': batch_reports } # Pre-incarcare in memorie la boot try: print("[Limina Engine]: Pre-loading models into cache at startup...") get_emb_model() get_nli_model() print("[Limina Engine]: Models loaded successfully.") except Exception as e: print(f"[Limina Engine Startup]: Warmup notice: {e}")