Spaces:
Sleeping
Sleeping
File size: 3,328 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 | """
LangChain integration for AgentSight.
Allows you to monitor LangChain agents for hallucinations in real-time.
"""
from typing import Any, Dict, List, Optional
try:
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import AgentAction, AgentFinish
except ImportError:
BaseCallbackHandler = object
AgentAction = None
AgentFinish = None
from agentsight_sdk import AgentMonitor
class AgentSightCallback(BaseCallbackHandler):
"""
A LangChain callback handler that streams execution steps to AgentSight
for step-level hallucination detection.
"""
def __init__(self, threshold: float = 0.40):
if BaseCallbackHandler is object:
raise ImportError("Please install langchain to use this callback: pip install langchain")
self.monitor = AgentMonitor()
self.monitor.threshold = threshold
# State for the current trajectory
self._trajectory = []
self._current_step_num = 1
self._current_query = "Unknown query"
def on_chain_start(
self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any
) -> Any:
"""Capture the initial user query if present."""
if "input" in inputs:
self._current_query = str(inputs["input"])
elif "question" in inputs:
self._current_query = str(inputs["question"])
def on_agent_action(self, action: AgentAction, **kwargs: Any) -> Any:
"""Record an action being taken."""
self._trajectory.append({
"step": self._current_step_num,
"content": action.log.split("Action:")[0].strip(),
"tool_calls": [{
"name": action.tool,
"arguments": action.tool_input if isinstance(action.tool_input, dict) else {"input": action.tool_input}
}],
"tool_responses": []
})
def on_tool_end(self, output: str, **kwargs: Any) -> Any:
"""Record the output of the tool."""
if self._trajectory:
self._trajectory[-1]["tool_responses"].append(str(output))
# Evaluate after tool finishes
result = self.monitor.evaluate_trajectory({
"question": self._current_query,
"trajectory": self._trajectory
})
if result.get("is_hallucinated"):
prob = result.get("max_hallucination_prob", 0.0)
root_step = result.get("predicted_root_cause_step", self._current_step_num)
print(f"\n[AgentSight] ⚠️ Hallucination Warning: Step {root_step} flagged (prob={prob:.2f})")
self._current_step_num += 1
def on_agent_finish(self, finish: AgentFinish, **kwargs: Any) -> Any:
"""Final evaluation when the agent finishes."""
result = self.monitor.evaluate_trajectory({
"question": self._current_query,
"trajectory": self._trajectory
})
if result.get("is_hallucinated"):
root_step = result.get("predicted_root_cause_step")
print(f"\n[AgentSight] 🚨 Final Verdict: Hallucination Detected (Root Cause: Step {root_step})")
else:
print("\n[AgentSight] ✅ Final Verdict: Trajectory Clean")
|