Use a spec in your app

Call the spec you tested from application code, one row at a time, with the same answers and cache as batch runs.

You have tested a spec over hundreds of rows. Now your coding agent is about to run one more command, and your code has to decide before it does. judge() answers with the same spec and the same store as hunch run.

One row

The user changed their mind about a framework, and the agent’s answer is to delete the project:

import hunch

answers = hunch.judge(
    "src/hunch/recipes/agent_commands/command_guard.yml",
    request="forget I even mentioned RemixJS just do SolidJS",
    cwd="D:\\IceBerg",
    description="",
    command='rm -rf Dockerfile README.md app node_modules package-lock.json package.json public '
            'react-router.config.ts tsconfig.json vite.config.ts .gitignore .dockerignore .react-router .claude && echo "done"',
)

{'destroys': {'label': 'yes', 'p': 0.95, 'margin': 0.9, 'route': 'act', 'cached': False},
 'reaches_outside': {'label': 'no', 'p': 0.88, 'margin': 0.76, 'route': '', 'cached': False},
 'sends_out': {'label': 'no', 'p': 0.97, 'margin': 0.94, 'route': '', 'cached': False}}

The row’s columns are keyword arguments, and must include every column in the spec’s state. A row the store has never seen makes one engine call, and its answer is then stored for every later caller, batch or app: ask again and the same answers come back with 'cached': True, in a few milliseconds.

The cache works the other way too. Running hunch run over past rows warms it, so repeats of them in the app cost nothing.

Code that compares a label with a name, like == "lost_or_stolen_card", breaks if the option is renamed. List those names in the spec’s exposures and the rename fails lint instead: see apps depend on answer names.

Act on route, not on label

A label on its own doesn’t tell your code whether to trust it. Give the question an act threshold in the spec:

questions:
  destroys:
    type: noul
    instructions: Would running `command` delete, overwrite or reset files, data, branches or history in a way that would be hard to undo?
    act: 0.90

Then each answer carries a route:

route	Meaning	Typical handling
act	p is at or above act	Use the answer
review	p is below act	Send to a person, a fallback, or a slower model
""	The question has no act	Your code decides

Because act lives in the spec rather than in your code, hunch test can check it. Its dial shows how many rows a threshold automates and how many of those are wrong. Changing act asks nothing new, since it is not part of the cache key.

In async code

Inside FastAPI or any event loop, use ajudge. judge starts its own loop and cannot run inside one. Here the guard is a service an agent’s hook calls before each command: it runs the command only when every answer is a confident no.

from fastapi import FastAPI
import hunch

app = FastAPI()
SPEC = "src/hunch/recipes/agent_commands/command_guard.yml"


@app.post("/before-command")
async def before_command(request: str, cwd: str, description: str, command: str):
    a = await hunch.ajudge(SPEC, request=request, cwd=cwd, description=description, command=command)
    if all(x["label"] == "no" for x in a.values()) and a["destroys"]["route"] == "act":
        return {"decision": "allow"}
    return {"decision": "ask", "why": [q for q, x in a.items() if x["label"] == "yes"]}

Sending uncertain answers to a second engine

A review route doesn’t have to mean a person. A question can hand its uncertain answers to another engine first:

destroys:
  type: noul
  instructions: Would running `command` delete, overwrite or reset files, data, branches or history in a way that would be hard to undo?
  act: 0.90
  escalate: {model: "deepseek:deepseek-flash"}

When the first answer is below act, the same question goes to the escalate model. Its answer replaces the first only if it clears act itself; otherwise the first stays and routes to review. Both are kept in the store, and the batch table’s destroys_by column records which engine answered.

Capping the cost

A burst of new rows shouldn’t be able to run up a bill:

export HUNCH_MAX_COST=0.01   # USD, per judgment per call, estimated before asking

If the missing answers would cost more, nothing is asked and judge raises SystemExit. A spec error raises SystemExit too, and a row missing a state column raises KeyError, so catch both in a long-running service.

judge loads each spec once per process: restart after editing one. To keep rows so that a spec written later can be tried on them, pass log=True; see Change a spec safely.

From a Pydantic AI agent

If the decision already runs in production as a Pydantic AI agent on a decision model, test that agent, not a copy of it. A copy drifts: Pydantic AI builds each question from the field’s name and description, the class docstring, the agent’s instructions, enum member docstrings and BoolCriteria, and a hand-written spec would have to match all of it, release after release.

spec_from_agent asks Pydantic AI itself. It starts one run of the agent on a model that records the request and stops, so nothing is sent anywhere, and writes the questions into a spec word for word:

from typing import Annotated

from pydantic import BaseModel, ConfigDict
from pydantic_ai import Agent, BoolCriteria
import hunch


class Guard(BaseModel):
    """Decide whether a person should see a coding agent's shell command before it runs."""
    model_config = ConfigDict(use_attribute_docstrings=True)
    destroys: Annotated[bool, BoolCriteria(true="It removes or replaces something that has no copy, or rewrites history.",
                                           false="It only reads, builds, tests, or changes things that are easy to put back.")]
    """Would running this command delete, overwrite or reset something in a way that would be hard to undo?"""
    sends_out: bool
    """Would running this command send code, files or data to another machine or service?"""


agent = Agent("typesafe:jev-1.13.0", output_type=Guard, instructions="You guard a coding agent's shell commands.")
spec = hunch.spec_from_agent(agent, state="command", source="commands.csv")
open("guard.yml", "w").write(hunch.spec_yaml(spec))

guard.yml

judgment: guard
model: jev-1.13.0
key: id
state: command
questions:
  destroys:
    type: noul
    instructions:
      field: destroys
      question: Would running this command delete, overwrite or reset something in a way that would be hard to undo?
      goal: Decide whether a person should see a coding agent's shell command before it runs.
      background: You guard a coding agent's shell commands.
    criteria:
      'true': It removes or replaces something that has no copy, or rewrites history.
      'false': It only reads, builds, tests, or changes things that are easy to put back.
  sends_out:
    type: noul
    instructions:
      field: sends_out
      question: Would running this command send code, files or data to another machine or service?
      goal: Decide whether a person should see a coding agent's shell command before it runs.
      background: You guard a coding agent's shell commands.
source: commands.csv

The agent’s prompt is the text it judges, so state names the one column that holds it, here command. A single column name, rather than a list, is sent as a bare string, which is how the agent sends its prompt. For each row, hunch then sends Jev the same state and the same questions as the agent; only the question names, which the model never sees, are written with __ instead of .. On three real commands (a find script, a git push and the rm -rf above), the agent and hunch gave the same six answers, with margins no further apart than asking Jev the same thing twice.

From here it is an ordinary spec: hunch test, diff, review and docs work on it. When the agent changes, record it again and hunch diff guard.yml --against git:HEAD shows what the new wording does. It needs Pydantic AI 2.50 or later, which your agent already has; hunch doesn’t install it.

This holds when the agent’s request is a function of one text column. spec_from_agent records the agent twice with different prompts and refuses, rather than write a spec that asks something else, when:

• it chooses between several output types or tools, which asks a route question first;
• it has a system_prompt, which turns the state into a conversation (put that text in instructions);
• its instructions change with the prompt. Instructions computed from deps are fine: pass the same deps as in production.

Call it with the prompt as a single string; an agent run on message history or several prompt parts sends a different state.

Measure the decisions production made

A spec on a CSV tells you how the agent does on rows you collected. What the agent decided last week in production is in its traces. With instrumentation on, Pydantic AI records every decision request as a decide span holding the prompt, the questions and the answers (it records content unless include_content=False).

Export the spans as OTLP/JSON lines, which is what the OpenTelemetry Collector’s file exporter writes, and read them with a py() source. This function yields one row per decision, keeping production’s answers beside the prompt:

decide_spans.py

import glob
import json
from pathlib import Path

SPANS = Path(__file__).parent / "traces" / "*.jsonl"


def rows():
    for path in glob.glob(str(SPANS)):
        for line in open(path):
            for rs in json.loads(line)["resourceSpans"]:
                for ss in rs["scopeSpans"]:
                    for span in ss["spans"]:
                        attrs = {a["key"]: next(iter(a["value"].values())) for a in span.get("attributes", [])}
                        if attrs.get("gen_ai.operation.name") != "decide" or "pydantic_ai.decision.state" not in attrs:
                            continue  # not a decision, or recorded without content
                        answers = json.loads(attrs.get("pydantic_ai.decision.answers", "{}"))
                        yield {"id": span["spanId"], "prompt": attrs["pydantic_ai.decision.state"],
                               **{f"live_{q}": ("yes" if a["noul"] >= 0.5 else "no") if a["type"] == "noul"
                                  else a.get("choice") for q, a in answers.items()}}

Point the recorded spec at it, with the prompt as the state:

spec = hunch.spec_from_agent(agent, judgment="guard", state="prompt", source="py(decide_spans.py:rows)")

Each production decision is now a row. hunch run asks it again on the same engine, which gives production’s answers within Jev’s run-to-run noise, at about $0.00002 a row. hunch review then builds gold from real traffic, hunch test says how often the agent is right on it, and hunch diff shows which of last week’s decisions a new wording would change. The live_ columns keep what production answered, for a where clause or to compare by hand.

From a Pydantic class

If the decision is a Pydantic class but not an agent, for example an output_type you haven’t wired up yet, the class itself can be the spec.

The types map as in Pydantic AI, but the wording does not. Pydantic AI 2.50 and later also sends the field’s name, the class docstring, the agent’s instructions and any BoolCriteria with each question; spec_from_model sends the field’s description alone, so the same class can get different answers here and in an agent. For the agent’s own wording, use spec_from_agent.

Install hunch with its pydantic extra:

uv add "hunch-ai[pydantic]"

Then:

from pydantic import BaseModel, Field
import hunch


class Guard(BaseModel):
    destroys: bool = Field(
        description="Would running `command` delete, overwrite or reset something in a way that would be hard to undo?",
        json_schema_extra={"hunch": {"act": 0.9}})
    sends_out: bool = Field(
        description="Would running `command` send code, files or data from this machine to another machine or service?")


spec = hunch.spec_from_model(Guard, source="commands.csv", state=["request", "command"])
verdict = hunch.judge_model(Guard, spec, base="src/hunch/recipes/agent_commands",
                            request="ship it", command="git push --force origin main")
# Guard(destroys=True, sends_out=True)

spec_from_model produces an ordinary spec dict; hunch.spec_yaml(spec) prints it as YAML. Save it and commit it to use hunch test, diff and review on it.

Each field becomes a question: bool a yes/no, Literal or Enum a choice, IntEnum a score. The Python reference has the full mapping and where descriptions and act go.

judge_model returns labels only. For p and route, save the spec, call hunch.judge on the file, and convert with hunch.to_model(Guard, answers).