Spaces:
Running on Zero
Running on Zero
| #!/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}") |