Agents: bounded, audited tool use#

A first run#

A tool is a Python function with a JSON Schema of its keyword arguments. The planner receives a copy of the run’s state and the tools’ descriptions, and returns a Decision.

from scikitplot.agent import Agent, Decision, Limits, Policy, Tool, verify_audit

PAGES = {"spans": "A chunk's text is a slice of its source.", "ids": "64 hex digits."}

lookup = Tool(
    "lookup",
    lambda topic: PAGES.get(topic, ""),
    description="Return the page about a topic.",
    parameters={
        "type": "object",
        "properties": {"topic": {"type": "string", "enum": sorted(PAGES)}},
        "required": ["topic"],
        "additionalProperties": False,
    },
    scopes={"docs:read"},
)

def planner(state, tools):
    if not state.steps:
        return Decision.act("lookup", {"topic": "spans"})
    return Decision.finish(state.steps[-1].result)

agent = Agent(
    [lookup],
    planner,
    policy=Policy(granted_scopes={"docs:read"}),
    limits=Limits(max_steps=4, max_tool_calls=2),
)
state = agent.run("what is a chunk's text?")
assert state.status.value == "completed"
assert state.answer == "A chunk's text is a slice of its source."
assert verify_audit(agent.audit.records) == []

Without granted_scopes={"docs:read"} the same run completes too, but its first step is denied with scope_denied and the tool never runs. Policy() is least privilege: read tools that need no scope, and nothing else.

Limits and how a run ends#

Limits holds the stopping conditions. Each is checked before the call it bounds, so a call already running is not interrupted; a tool’s own timeout bounds one call.

A run ends completed (the planner finished and the done criterion, if given, accepted the answer), stopped (a limit; stop_reason names it), failed (the planner or the done criterion raised or returned something invalid), or pauses awaiting_approval.

Approval, pause and resume#

write and external tools run only under a policy that allows their effect, and then, by default, only with a person’s approval of each call. An approval is bound to the digest of the exact action (run, step, tool, arguments, effect and scopes) and of the run’s history up to the request; at resume the digest is recomputed from the tools as they are then and every gate runs again, so an approval of an action that has since changed authorises nothing (approval_stale). Each run’s first audit record holds a random nonce, so two runs never share a history, and a run id already in the store or the audit trail is refused. Resuming a copy older than the stored state is refused, and so is resuming a paused copy that the audit trail shows was resumed already: an approval runs its action once. (A caller with neither a store nor a shared audit sink holds every copy itself, and replays what it resumes twice.)

from scikitplot.agent import Approval, MemoryStore

written = []
save = Tool(
    "save",
    lambda text: written.append(text) or "saved",
    parameters={
        "type": "object",
        "properties": {"text": {"type": "string"}},
        "required": ["text"],
        "additionalProperties": False,
    },
    effect="write",
)

def writer(state, tools):
    if not state.steps:
        return Decision.act("save", {"text": "the report"})
    return Decision.finish(state.steps[-1].outcome.value)

store = MemoryStore()
agent = Agent([save], writer, policy=Policy(allow_effects={"read", "write"}), store=store)
paused = agent.run("save the report")
assert paused.status.value == "awaiting_approval" and written == []

request = paused.pending  # what is asked, and why
approval = Approval(request.digest, granted=True, approver="allen")
done = agent.resume(paused.run_id, approvals=[approval])  # read back from the store
assert written == ["the report"] and done.answer == "executed"

An approver= callable answers at once instead (an automated policy, a prompt to a person); an answer of None pauses the run. A refusal (granted=False) is the step’s outcome, approval_denied.

If a run stops while a tool call is in flight (the process is killed), the saved state says so; the call was charged against the limits when it started. At resume an idempotent call (every read tool, by default) starts again, through every gate, at most three times in all; a call of a non-idempotent tool is not repeated, because whether it took effect is unknown: the step is interrupted.

The audit trail#

Each record names its run, a sequence number from 0, the event and its data, and the digest of the record before it. verify_audit finds a record that was changed, removed or reordered. JsonlAuditSink(path) appends one line per record (fsync=True to survive a machine crash), and read_audit reads the file back. The chain cannot show a trail cut after its last record, but the state can: it keeps audit_head, the last record’s number and digest, and a resume whose trail does not hold that record is refused with StateError.

Measured#

On the corpus user guide (recursive (500) chunks, BM25 search) with 82 tasks, each the first half of a guide sentence to be found, three rule planners using search (cost 1) and read (cost 0.25); the benchmark is maintenances/agent/_maintenance/evidence/bench_r5a.py, its output bench_r5a.out beside it:

one shot answers the first search hit, which is exactly plain search’s recall@1. verify reads the hits in order and answers the first whose text holds the query; reading at most three gives the same success at two thirds of the cost. In every run the audit chain verified and every tool call had its audit record. Without the corpus:read scope every action of every run was denied and no tool ran; with a write tool and no approver every run paused before writing.

Writing a planner#

A planner is any callable planner(state, tools) -> Decision:

• state is a copy of the RunState, rebuilt from its saved form: state.goal, state.steps (each with its decision, outcome, result or error) and state.observations() (the same as plain JSON, ready for a prompt); changing it changes nothing;

• tools is Toolset.describe: names, descriptions, JSON Schemas, effects, scopes and costs, ready for a model’s tool-calling interface;

• return Decision.act or Decision.finish, with usage=Usage(tokens, cost) when a model produced it.

A planner that raises, or returns anything else, fails the run with planner_failed or invalid_decision.