Spaces:
Sleeping
Sleeping
File size: 4,853 Bytes
f747401 | 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 | import json
import os
import re
TRACE_FILE = "ai_observe/traces.json"
# Heuristics for prompt injection
INJECTION_KEYWORDS = [
"ignore previous instructions",
"ignore all prior instructions",
"disregard previous",
"you are now",
"system prompt",
"bypassing"
]
def detect_failures(pipeline_result):
"""
Analyzes the output of the RAG pipeline and recent traces
to detect failures in the system.
"""
failures = []
query = pipeline_result.get("query", "").lower()
retrieved = pipeline_result.get("retrieved", [])
judgment = pipeline_result.get("judgment", {})
# 1. Prompt Injection Detection
is_injection = any(keyword in query for keyword in INJECTION_KEYWORDS)
if is_injection or re.search(
r"system.*prompt.*(ignore|reveal)", query, re.IGNORECASE
):
failures.append("prompt_injection")
# 2. Retrieval Failure Detection
# If the top retrieved document has a low semantic similarity score
if not retrieved:
failures.append("retrieval_failure")
else:
top_score = max([doc.get("score", 0.0) for doc in retrieved])
if top_score < 0.15: # Threshold tuned for MVP embeddings
failures.append("retrieval_failure")
# 3. Hallucination Detection
# If the generated answer has low semantic overlap with retrieved context
if judgment.get("verdict") == "fail":
failures.append("hallucination")
# 4. Latency Anomaly Detection
# Read traces.json to find the latest execution of run_pipeline
latency = 0.0
try:
if os.path.exists(TRACE_FILE):
with open(TRACE_FILE, "r") as f:
traces = json.load(f)
# Find the most recent run_pipeline trace
for trace in reversed(traces):
if trace.get("function") == "run_pipeline":
latency = trace.get("latency_ms", 0.0)
break
except Exception:
pass
if latency > 8000.0: # 8 seconds is abnormally slow for local MVP
failures.append("latency_anomaly")
return {
"failures": failures,
"metrics": {
"top_retrieval_score": top_score if retrieved else 0.0,
"grounding_score": judgment.get("score", 0.0),
"pipeline_latency_ms": latency
}
}
def get_root_causes(failures):
"""
Maps detected failures to potential root causes and mitigation suggestions.
"""
diagnostics = {}
if "prompt_injection" in failures:
diagnostics["Prompt Injection"] = [
"User query contained suspicious keywords attempting "
"to override system behavior.",
"Action: Implement an intent-classification "
"guardrail model before the RAG pipeline.",
"Action: Refuse to answer queries matching known "
"injection patterns."
]
if "retrieval_failure" in failures:
diagnostics["Retrieval Failure"] = [
"Vector database returned context with low semantic "
"similarity to the query.",
"Action: Increase chunk overlap or modify chunking sizes.",
"Action: Evaluate upgrading the embedding model (e.g., "
"all-MiniLM-L6-v2 -> text-embedding-ada-002).",
"Action: Implement a Re-ranker to improve top-k relevance."
]
if "hallucination" in failures:
diagnostics["Hallucination (Low Grounding)"] = [
"The generated answer did not align with the retrieved context.",
"Action: Enhance the system prompt to enforce strict adherence "
"to context ('Say I don't know if not found').",
"Action: Check if context length exceeded the model's window, "
"causing it to truncate useful facts.",
"Action: Tune generation parameters (lower temperature)."
]
if "latency_anomaly" in failures:
diagnostics["Latency Anomaly"] = [
"The pipeline execution exceeded the latency threshold (>8000ms).",
"Action: Profile `generate_answer` vs `retrieve`.",
"Action: Scale down the LLM size or deploy it on specialized "
"inferencing hardware (e.g., vLLM or ONNX)."
]
if not failures:
diagnostics["Healthy"] = [
"No failures detected. Trace execution is within "
"expected parameters."
]
return diagnostics
def analyze_trace(pipeline_result):
"""
Executes the full Failure Detection and Root Cause pipeline.
"""
detection = detect_failures(pipeline_result)
root_causes = get_root_causes(detection["failures"])
return {
"metrics": detection["metrics"],
"detected_failures": detection["failures"],
"diagnostics": root_causes
}
|