Spaces:
Running
Running
File size: 3,232 Bytes
9d0fd45 | 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 | """AssertionObserver — synthesize assertions from evidence on loop end.
Uses cluster-based grouping: evidence cards grouped by query,
each group becomes one assertion. No LLM call needed.
"""
from __future__ import annotations
from collections import defaultdict
from typing import Any
from frontier_agent.core.loop_types import AgentLoopResult, BaseObserver
class AssertionObserver(BaseObserver):
"""Synthesize assertions from evidence_cards at loop end.
critical=True because callers read metadata["assertions"] from the result.
"""
critical: bool = True
async def on_loop_end(self, result: AgentLoopResult) -> None:
evidence_cards: list[dict] = result.metadata.get("evidence_cards", [])
assertions = _cluster_assertions(evidence_cards)
# Fallback: single assertion from final_content if no evidence
if not assertions and result.final_content:
assertions = [{
"id": "as-react-001",
"statement": result.final_content[:500],
"confidence": 0.5,
"supporting_evidence": [],
"counter_evidence": [],
"is_disputed": False,
}]
result.metadata["assertions"] = assertions
def _default_confidence(avg_rel: float, count: int) -> float:
"""Confidence formula for research assertions."""
conf = avg_rel * 0.7 + min(count, 10) * 0.02
return round(max(0.1, min(0.80, conf)), 2)
def _benchmark_confidence(avg_rel: float, count: int) -> float:
"""Simpler confidence formula for benchmark solvers."""
return round(min(0.75, 0.4 + count * 0.05), 2)
def _cluster_assertions(
evidence_cards: list[dict],
confidence_fn: Any = None,
) -> list[dict]:
"""Group evidence by query -> one assertion per cluster."""
if not evidence_cards:
return []
if confidence_fn is None:
confidence_fn = _default_confidence
clusters: dict[str, list[dict]] = defaultdict(list)
for card in evidence_cards:
key = card.get("query", "").strip().lower()[:80] or "general"
clusters[key].append(card)
assertions: list[dict[str, Any]] = []
for i, (query, cards) in enumerate(clusters.items()):
claims = [
c.get("claim", "")[:100]
for c in cards[:3] if c.get("claim")
]
statement = (
f"{query}: " + "; ".join(claims) if claims else query
)
avg_rel = sum(
float(c.get("relevance_score", 0.3) or 0.3)
for c in cards
) / len(cards)
assertions.append({
"id": f"as-ev-{i + 1:03d}",
"statement": statement[:300],
"confidence": confidence_fn(avg_rel, len(cards)),
"supporting_evidence": [c["id"] for c in cards[:15]],
"counter_evidence": [],
"is_disputed": False,
})
return assertions
def extract_assertions_from_response(
response_text: str,
evidence_cards: list[dict],
) -> list[dict]:
"""Lightweight assertion placeholder for benchmark solvers."""
if not response_text or not evidence_cards:
return []
return _cluster_assertions(evidence_cards, _benchmark_confidence)
|