Spaces:
Sleeping
Sleeping
File size: 3,839 Bytes
2aed081 | 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 | import os
import sys
from typing import List, Dict, Optional
# Ensure imports work
script_dir = os.path.dirname(os.path.abspath(__file__))
project_root = os.path.join(script_dir, "..")
if project_root not in sys.path:
sys.path.insert(0, project_root)
from product.agentsight_api import AgentSightDetector
class StepResult:
def __init__(self, step_index: int, hallucination_prob: float, risk_score: float, is_flagged: bool):
self.step_index = step_index
self.hallucination_prob = hallucination_prob
self.risk_score = risk_score
self.is_flagged = is_flagged
class Report:
def __init__(self, overall_hallucination: bool, riskiest_step: Optional[StepResult], steps: List[StepResult]):
self.overall_hallucination = overall_hallucination
self.riskiest_step = riskiest_step
self.steps = steps
def summary(self):
print("=== AgentSight Summary ===")
print(f"Overall Hallucination Detected: {self.overall_hallucination}")
if self.riskiest_step:
print(f"Riskiest Step: {self.riskiest_step.step_index} (Prob: {self.riskiest_step.hallucination_prob:.3f})")
for step in self.steps:
flag = "🚨" if step.is_flagged else "✅"
print(f"Step {step.step_index}: {flag} Prob={step.hallucination_prob:.3f}, Risk={step.risk_score:.3f}")
class AgentMonitor:
def __init__(self):
self.detector = AgentSightDetector()
self.query = ""
self.steps = []
self.step_counter = 1
def set_query(self, query: str):
self.query = query
def add_step(self, thought: str, action: str, observation: str) -> StepResult:
step_dict = {
"step": self.step_counter,
"content": thought,
"tool_calls": [],
"tool_responses": [observation]
}
# Simple parse for action if it's structured, else just treat as text
# If action is in a format we can parse, we could, but for generic compat, put it in tool_calls
step_dict["tool_calls"].append({"name": "AgentAction", "arguments": {"action_str": action}})
self.steps.append(step_dict)
# We can run the detector incrementally or just return a dummy result until get_report is called
# But the PDF implies add_step returns a result immediately.
# However, AgentSightModel takes the full sequence up to now.
result = self.detector.detect(self.query, self.steps)
# The latest step is at the end of the details array
latest_detail = result["details"][-1] if result["details"] else {"prob": 0.0, "risk": 0.0}
is_flagged = latest_detail["prob"] > 0.5
step_result = StepResult(
step_index=self.step_counter,
hallucination_prob=latest_detail["prob"],
risk_score=latest_detail["risk"],
is_flagged=is_flagged
)
self.step_counter += 1
return step_result
def get_report(self) -> Report:
result = self.detector.detect(self.query, self.steps)
step_results = []
riskiest_step = None
max_prob = 0.0
for i, detail in enumerate(result["details"]):
sr = StepResult(
step_index=detail["step"],
hallucination_prob=detail["prob"],
risk_score=detail["risk"],
is_flagged=detail["prob"] > 0.5
)
step_results.append(sr)
if detail["prob"] > max_prob:
max_prob = detail["prob"]
riskiest_step = sr
return Report(
overall_hallucination=result["is_hallucinated"],
riskiest_step=riskiest_step,
steps=step_results
)
|