Spaces:
Paused
Paused
File size: 4,762 Bytes
5d10875 | 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 | """BaseAgent: the shared contract for every advisory agent.
Design (mirrors sensorflow.evaluation.copilot):
* `_analyze(...)` is the deterministic rule-based core — always runs, always
testable offline.
* `maybe_llm_rationale(...)` optionally asks a local Ollama endpoint for a
natural-language rationale; any failure is swallowed and the deterministic
output stands alone. The LLM can only ever ADD prose; it cannot change
structured outputs, confidences, or verdicts.
* Exceptions inside `_analyze` are converted to a failed AgentResult with
escalation required — an agent crash can therefore never silently pass a
stage (the pipeline treats it as missing evidence -> INDETERMINATE-ward).
"""
from __future__ import annotations
import json
from typing import Any, Dict, List, Optional, Tuple
import httpx
from sensorflow.agentic.models import AgentEscalation, AgentResult, now_iso
# Local-only by default; kept short so offline runs fail fast.
OLLAMA_ENDPOINTS = [
{"url": "http://localhost:11434/api/chat", "model": "gemma3:latest"},
]
LLM_TIMEOUT_S = 4.0
class BaseAgent:
name: str = "base"
version: str = "1.0.0"
epistemic_status: str = "DERIVED"
def __init__(self, use_llm: bool = False):
self.use_llm = use_llm
# -- to be provided by subclasses -----------------------------------
def _analyze(self, failure_id: str, **inputs) -> Tuple[Dict[str, Any], float, str,
AgentEscalation]:
"""Return (output, confidence, confidence_basis, escalation)."""
raise NotImplementedError
def _llm_prompt(self, failure_id: str, output: Dict[str, Any],
**inputs) -> Optional[str]:
return None
# -- uniform entry point ---------------------------------------------
def run(self, failure_id: str, **inputs) -> AgentResult:
started = now_iso()
try:
output, confidence, basis, escalation = self._analyze(failure_id, **inputs)
status = "escalated" if escalation.required else "ok"
result = AgentResult(
agent=self.name, agent_version=self.version,
failure_id=failure_id, status=status, output=output,
confidence=confidence, confidence_basis=basis,
epistemic_status=self.epistemic_status,
escalation=escalation, started_at=started)
except Exception as e:
result = AgentResult(
agent=self.name, agent_version=self.version,
failure_id=failure_id, status="failed",
output={"error": str(e)}, confidence=0.0,
confidence_basis="agent raised an exception",
escalation=AgentEscalation(
required=True,
reasons=[f"{self.name} failed: {e}"],
human_review_triggers=["agent_failure"]),
failure_handling=("exception captured; stage marked failed; "
"downstream policy treats this as missing "
"evidence (fail-safe)"),
started_at=started)
result.finished_at = now_iso()
return result
if self.use_llm:
prompt = self._llm_prompt(failure_id, result.output, **inputs)
if prompt:
text, provider = _try_ollama(prompt)
if text:
result.llm_used = True
result.llm_provider = provider
result.llm_rationale = (
"LLM RATIONALE (advisory prose only; structured "
"outputs above are deterministic): " + text)
result.finished_at = now_iso()
return result
def _try_ollama(prompt: str) -> Tuple[Optional[str], Optional[str]]:
for ep in OLLAMA_ENDPOINTS:
try:
res = httpx.post(ep["url"], json={
"model": ep["model"],
"messages": [{"role": "user", "content": prompt}],
"stream": False,
}, timeout=LLM_TIMEOUT_S)
if res.status_code == 200:
text = res.json().get("message", {}).get("content", "")
if text:
return text, ep["url"]
except Exception:
continue
return None, None
def compact(obj: Any, limit: int = 3000) -> str:
return json.dumps(obj, default=str)[:limit]
def no_escalation() -> AgentEscalation:
return AgentEscalation()
def escalate(reasons: List[str], triggers: List[str]) -> AgentEscalation:
return AgentEscalation(required=True, reasons=reasons,
human_review_triggers=triggers)
|