Agent#

class scikitplot.agent.Agent(tools, planner, *, policy=None, limits=None, approver=None, done=None, audit=None, store=None, clock=None, now=None)[source]#

Run a goal with a planner and tools, within a policy and limits.

Parameters:
toolsToolset or iterable of Tool

What the agent may call.

plannercallable

planner(state, tools) -> Decision: state is a copy of the RunState and tools the toolset’s describe. A model adapter or a rule.

policyPolicy or None, default=None

None is Policy(): read tools without scopes only.

limitsLimits or None, default=None

None is Limits(): at most 16 decisions.

approvercallable or None, default=None

approver(request) -> Approval or None, asked when an action needs approval. None (or an answer of None) pauses the run with status awaiting_approval; resume it with an Approval.

donecallable or None, default=None

done(answer, state) -> bool or str: the task’s completion criterion. True accepts the planner’s answer; False or a reason rejects it, and the run goes on. None accepts any answer.

auditAuditSink or None, default=None

Where audit records go; None is a new MemoryAuditSink, kept as audit.

storeStateStore or None, default=None

Where the state is saved after each step; None keeps it only in the returned state.

clockcallable or None, default=None

Seconds, for limits and step times; None is time.monotonic.

nowcallable or None, default=None

Wall-clock time written in audit records; None is time.time.

Raises:
TypeError

If an argument has the wrong type.

Parameters:

Examples

>>> from scikitplot.agent import Agent, Decision, Tool
>>> def planner(state, tools):
...     if not state.steps:
...         return Decision.act("ping")
...     return Decision.finish(state.steps[-1].result)
>>> agent = Agent([Tool("ping", lambda: "pong")], planner)
>>> state = agent.run("check the line")
>>> state.status.value, state.answer, len(agent.audit.records) > 0
('completed', 'pong', True)
resume(state, *, approvals=())[source]#

Continue a paused or interrupted run.

Parameters:
stateRunState or str

The saved state, or its run id (read from store).

approvalsiterable of Approval, default=()

Answers to the pending ApprovalRequest. The one whose digest names the request is used; at resume the action’s digest is recomputed from the tools as they are now and every gate is run again, so an approval of an action that changed authorises nothing (approval_stale).

Returns:
RunState

The last state. A run that has ended is returned unchanged; a paused run with no matching approval stays paused.

Raises:
StateError

If a store is set and holds a newer state of the run than the one given: resuming an old copy again would repeat what the newer state already records (an approved call, for one).

ValueError

If two approvals name the request and disagree, or a run id is given without a store.

Parameters:
Return type:

RunState

Notes

User. With a store, give the run id: the latest state is read from it, and a second resume of a finished run is a no-op. Without a store, the caller holds the states, and resuming the same copy twice is the caller’s own replay. Every resume writes audit records, so the state it returns replaces the one given, even when a paused run stays paused: resume that one next.

run(goal, *, run_id=None)[source]#

Start a run toward goal and drive it until it ends or pauses.

Parameters:
goalstr

What the run is for; the planner reads it from the state.

run_idstr or None, default=None

64 hexadecimal digits; None draws a random one.

Returns:
RunState

The last state: completed, stopped, failed, or awaiting_approval.

Raises:
ValueError

If run_id names a run the store or the audit sink already holds: a run id is used once, so an approval or a trail of one run can never be taken for another’s.

Parameters:

  • goal (str)

  • run_id (str | None)

Return type:

RunState