File size: 1,599 Bytes
f983375 | 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 | from __future__ import annotations
import json
import re
from collections.abc import Callable
from pydantic import ValidationError
from core.schemas import SimulationResult
RetryFn = Callable[[str, str], str]
def extract_json(raw: str) -> str:
stripped = raw.strip()
fence_match = re.search(r"```(?:json)?\s*(.*?)\s*```", stripped, re.DOTALL | re.IGNORECASE)
if fence_match:
return fence_match.group(1).strip()
start = stripped.find("{")
end = stripped.rfind("}")
if start == -1 or end == -1 or end < start:
raise json.JSONDecodeError("No JSON object found", stripped, 0)
return stripped[start : end + 1]
def validate_result(
raw_json: str,
persona_id: str,
retry_fn: RetryFn | None = None,
max_retries: int = 2,
) -> SimulationResult | None:
raw = raw_json
for attempt in range(max_retries + 1):
try:
return SimulationResult.model_validate_json(extract_json(raw))
except (json.JSONDecodeError, ValidationError) as exc:
if retry_fn is None or attempt >= max_retries:
return None
raw = retry_fn(persona_id, str(exc))
return None
def validate_swarm(
raw_results: list[tuple[str, str]],
retry_fn: RetryFn | None = None,
) -> tuple[list[SimulationResult], int]:
valid: list[SimulationResult] = []
failures = 0
for persona_id, raw in raw_results:
result = validate_result(raw, persona_id, retry_fn)
if result is None:
failures += 1
else:
valid.append(result)
return valid, failures
|