Harness Engineering

# agent_loop.py — the smallest real loop
MAX_STEPS = 3   # a guardrail (see 1.6)

def recheck_poi(poi):
    for step in range(MAX_STEPS):
        if open("STOP", "r").read().strip() == "1" if __import__("os").path.exists("STOP") else False:
            break  # kill switch
        action = decide_next_action(poi)      # model plans
        if action["type"] == "done":
            return action["poi"]
        result = run_tool(action)             # you act
        poi = observe(poi, result)            # feed result back
    return poi  # ran out of steps — keep best effort

from pydantic import BaseModel

class GeocodeArgs(BaseModel):
    address: str   # the address to look up

# Tool definition the model reads (name + description + schema)
GEOCODE_TOOL = {
    "type": "function",
    "function": {
        "name": "geocode_address",
        "description": "Look up coordinates for an address via OpenStreetMap Nominatim.",
        "parameters": GeocodeArgs.model_json_schema(),
    },
}
# Give tools to Ollama:  ollama.chat(model=..., messages=..., tools=[GEOCODE_TOOL])

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import requests

@retry(stop=stop_after_attempt(3),
       wait=wait_exponential(multiplier=1, min=1, max=10),
       retry=retry_if_exception_type(requests.RequestException))
def geocode_address(address: str) -> dict:
    r = requests.get("https://nominatim.openstreetmap.org/search",
                     params={"q": address, "format": "json", "limit": 1},
                     headers={"User-Agent": "weekend2-poi-agent/1.0 (you@example.com)"},
                     timeout=10)
    r.raise_for_status()
    return r.json()[0] if r.json() else {}

import json, os

CKPT = "checkpoint.json"

def load(): return json.load(open(CKPT)) if os.path.exists(CKPT) else {}
def save(state): json.dump(state, open(CKPT, "w"))

def run(records):
    state = load()                       # resume-aware
    for rec in records:
        if state.get(rec["id"]) == "done":
            continue                     # idempotent skip
        process(rec)                     # your work
        state[rec["id"]] = "done"
        save(state)                      # checkpoint after each unit

import logging, json

logging.basicConfig(filename="agent.log", level=logging.INFO,
                    format="%(message)s")

def log_decision(poi_id, step, decision, confidence, tokens=0):
    logging.info(json.dumps({
        "poi_id": poi_id, "step": step, "decision": decision,
        "confidence": confidence, "tokens": tokens,
    }))
# Later:  grep poi_id agent.log   to replay one POI's whole journey

def eval_gate(pred_csv, golden_csv, threshold=0.85):
    precision, recall, f1 = evaluate(pred_csv, golden_csv)  # from Weekend 1
    passed = f1 >= threshold
    print(f"F1={f1:.2f}  gate={'PASS' if passed else 'FAIL'}")
    return passed   # if False, quarantine the batch instead of publishing

import os

class Guardrails:
    def __init__(self, max_steps=3, cost_budget=1.00):
        self.max_steps, self.cost_budget = max_steps, cost_budget
        self.steps, self.cost = 0, 0.0
    def check(self):
        if os.path.exists("STOP"): raise SystemExit("kill switch")
        if self.steps >= self.max_steps: raise SystemExit("max steps")
        if self.cost  >= self.cost_budget: raise SystemExit("over budget")
    def spend(self, dollars): self.cost += dollars
    def tick(self): self.steps += 1; self.check()