My AI agent kept inventing pollution numbers, so I made it prove every one (Strands Agents + Amazon Bedrock)

@tool
@traced
def get_trip_result(trip_id: str) -> dict:
    """The saved analysis of the trip the user is looking at: recommended route and departure time,
    the fastest option leaving now, every route's exposure, confidence, and the decision rule used."""
    t = trips.get(trip_id)
    if t is None:
        raise ValueError("trip not found; ask the user to plan the trip first")
    return trip_summary(t)

def traced(fn):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        tr = current.get()
        inputs = {**{f"arg{i}": a for i, a in enumerate(args)}, **kwargs}
        # Small models sometimes repeat an identical call; hand back the earlier result and a nudge.
        prev = next((c for c in tr.calls if c["name"] == fn.__name__ and c["input"] == inputs and c["ok"]), None)
        if prev is not None:
            return json.dumps({"note": "You already have this result. Answer the user now.",
                               "result": prev["output"]})
        tr.emit({"type": "tool_start", "name": fn.__name__, "input": inputs})
        try:
            out, ok = fn(*args, **kwargs), True
        except Exception as e:  # the model sees the error text and can say so plainly
            out, ok = {"error": str(e)}, False
        call = {"name": fn.__name__, "input": inputs, "output": out, "ok": ok}
        tr.calls.append(call)
        tr.emit({"type": "tool_end", **call})
        return json.dumps(out)
    return wrapper

TRIP_TOOLS = [get_trip_result, now_ist]
PLANNING_TOOLS = [geocode, best_departure_time, score_routes, get_forecast, now_ist]

def tools_for(trip_id: str | None) -> list:
    return TRIP_TOOLS if trip_id else PLANNING_TOOLS

agent = Agent(model=model, tools=tools_for(trip_id), system_prompt=system_prompt(profile, trip_id),
              messages=_history_messages(history), callback_handler=None)

async for ev in agent.stream_async(prompt):
    if "data" in ev and isinstance(ev["data"], str):
        text_parts.append(ev["data"])
        queue.put_nowait({"type": "token", "text": ev["data"]})
    if "message" in ev and ev["message"].get("role") == "assistant":
        # A new assistant turn after tool use: drop text produced before the tool call.
        if any("toolUse" in c for c in ev["message"].get("content", [])):
            text_parts.clear()
            queue.put_nowait({"type": "reset"})

>>> tool = [{"recommended": {"depart_at": "09:45", "label": "via Ring Road",
...          "exposure_change_pct": -23.4, "arrive_later_min": 20}, "confidence": 0.67}]

>>> guard.check("Leave at 09:45 via Ring Road: about 23% less exposure, "
...             "arriving 20 minutes later. Confidence is 67%.", tool)
(True, [])

>>> guard.check("Leave at 09:45 via Ring Road for roughly 35% less smoke, "
...             "arriving 20 minutes later.", tool)
(False, ['35'])

def _model():
    cfg = settings()
    if cfg.llm_backend == "ollama":
        from strands.models.ollama import OllamaModel
        return OllamaModel(host=cfg.ollama_host, model_id=cfg.ollama_model, temperature=0.2,
                           max_tokens=700, keep_alive="30m", additional_args={"think": False})
    if cfg.llm_backend == "bedrock":
        from strands.models import BedrockModel
        return BedrockModel(model_id=cfg.bedrock_model_id, region_name=cfg.aws_region,
                            temperature=0.2, max_tokens=700)
    return None

import contextvars, functools, json, re
from strands import Agent, tool

calls = contextvars.ContextVar("calls")
NUM = re.compile(r"-?\d+(?:\.\d+)?")

def traced(fn):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        out = fn(*args, **kwargs)
        calls.get().append(out)
        return json.dumps(out)
    return wrapper

def numbers(x):
    text = x if isinstance(x, str) else json.dumps(x)
    return {abs(float(n)) for n in NUM.findall(text)}

def grounded(answer, outputs, tol=0.03):
    sources = numbers(outputs)
    sources |= {round(v * 100, 1) for v in sources if 0 < v <= 1}   # 0.67 -> 67%
    bad = [n for n in numbers(answer)
           if n > 10 and not any(abs(n - s) <= max(0.5, tol * s) for s in sources)]
    return not bad, bad

def ask(agent: Agent, question: str, fallback):
    calls.set([])
    answer = str(agent(question))
    ok, bad = grounded(answer, calls.get())
    return answer if ok else fallback(calls.get(), bad)