stalwart_agents.compute

A plan's compute step, run on the customer's host.

client.compute(step, sources, model=write)  ->  Computation

A calculation no tool answers -- what is the average price of my orders -- is a plan's compute step, over the results of the read steps before it. It runs here, on every tier, in four parts:

  1. the prompt -- the runtime's compute_prompt, the toolkit's own Compute.Build, from the prompt_pack bundle the lease holds: the instruction, the result type and the sources' schema, which is their field names and JSON types and never a value of theirs;
  2. the function -- the customer's own model writes it from that prompt. The SDK holds no model credential and calls no provider: model is the caller's;
  3. admission -- the runtime's compute_admit, the toolkit's Compute.Admit. A function that opens a socket, reads a file or imports outside the pack's modules is refused here, and nothing runs;
  4. the run -- sandbox.launch.run, one fresh qls-sandbox container with no network, a read-only root, no file and no variable of the host's, and the result checked against the type asked for.

Nothing of a compute step leaves the host: not the sources, not the prompt, not the function and not the result. What does is two counts, compute_runs (a run that reached the sandbox) and compute_seconds (its wall time, rounded up), counted by the runtime into the meter record the usage report carries.

@dataclass(frozen=True)
class Computation:

What a compute step came to.

state is succeeded (result holds the checked value), failed (the sandbox ran it, or could not start, and failure says which) or not_admitted (admission refused the function and nothing ran; admission names the rule and the line). sandbox is the run's measurements, where it reached the sandbox. code_sha256 is the function's digest as admitted, and prompt_pack_digest the pack the prompt was built from.

Returns: What run gives: its fields above. It has no truth value, so a check reads state.

Example:

done = Computation(state="not_admitted", result=None, result_type="number", ... admission={"admitted": False, "rule": "import: socket", "line": 1}, ... failure={"code": "code_not_admitted"}, sandbox=None, prompt_pack_digest="", ... code_sha256="", prompt=ComputePrompt({"user": "the average price"})) done.state, done.admission["rule"] ('not_admitted', 'import: socket')

Computation( state: str, result: Any, result_type: str, admission: Mapping[str, Any], failure: Mapping[str, Any] | None, sandbox: Mapping[str, Any] | None, prompt_pack_digest: str, code_sha256: str, prompt: ComputePrompt)
state: str
result: Any
result_type: str
admission: Mapping[str, Any]
failure: Mapping[str, Any] | None
sandbox: Mapping[str, Any] | None
prompt_pack_digest: str
code_sha256: str
prompt: ComputePrompt
class ComputePrompt(builtins.str):

The prompt the runtime built for a calculation: the text a model is given, and its parts.

A str, so a model callable that takes one string works as it is: it is the system part and the user part, in that order. system, user and schema are there for a callable that sends roles. The schema is the sources' shape -- field names and JSON types -- and never a value of theirs.

system: str
user: str
schema: Any
Runner = collections.abc.Callable[[str, typing.Any, str], stalwart_agents.sandbox.launch.Outcome]

A function's run in the sandbox: (code, data, result_type) -> launch.Outcome.

def run( step: Any, sources: Sequence[Mapping[str, Any]], *, model: Callable[[ComputePrompt], str], runtime: stalwart_agents.compute.Runtime, runner: Callable[[str, Any, str], stalwart_agents.sandbox.launch.Outcome] | None = None) -> Computation:

One compute step, on this host: the prompt, the caller's model, admission, the sandbox.

Args: step: a plan's compute step (a PlanItem, or its document), or the instruction itself. sources: the results it reads, each {"tool", "result"} (or "result_json", the text a tool returned), with "name" for the key the result sits under in data; keyed by position (source_1, ...) where none is given, as the toolkit keys them. model: any callable taking the prompt (a ComputePrompt, a str) and returning the function's source. It is called here, in the caller's process. runtime: the licensed runtime, whose lease the prompt, the admission and the meters are under. runner: how an admitted function is run; one fresh container from the sandbox image where not given.

Returns: A Computation. A function admission refuses never runs and counts nothing; a run that could not reach the sandbox -- no engine, no image -- is failed and counts nothing.

Raises: QuotaExceededError: this month's compute_runs or compute_seconds are spent. Raised before the model is asked anything, so nothing is written and nothing runs. A run that crosses a cap is counted in full, and the next one is refused.

Example: Client.compute passes its licensed runtime; here a stand-in for it, and for the sandbox:

>>> class Licensed:
...     def call(self, operation, request):
...         if operation == "compute_prompt":
...             return {"user": request["instruction"], "result_type": "number", "prompt_pack_digest": "p"}
...         return {"admitted": True, "sha256": "c"}
...     def count(self, meter, count=1, *, done=False): pass
...     def room(self, meter, count=1): pass
>>> def sandbox(code, data, result_type):
...     return launch.Outcome(ok=True, result=4.5, refusal=None, reason=None, detail="", wall_seconds=0.1,
...                           peak_memory_bytes=None, exit_code=0, output_bytes=3)
>>> done = run("the average price", [{"tool": "list_orders", "result": [{"price": 4}, {"price": 5}]}],
...            model=lambda prompt: "def compute(data):\n    return 4.5\n", runtime=Licensed(), runner=sandbox)
>>> done.state, done.result
('succeeded', 4.5)
@functools.lru_cache(maxsize=1)
def sandbox_ready() -> bool:

Whether this host can run a compute step: a container engine answers, and the sandbox image is on it. Asked once a process; sandbox_ready.cache_clear() asks again.

Returns: True where both answer; a plan then asks compute: run.

Example: After docker pull or python -m stalwart_agents.sandbox, the next process sees the image, or this one once it asks again:

>>> sandbox_ready.cache_clear()
def strip_fences(text: str) -> str:

The function a model wrote, without the code fence it is often wrapped in.

Args: text: the model's reply.

Returns: The function's source, ending in one newline, with any surrounding code fence removed.

Example:

strip_fences("python\ndef compute(data):\n return 1\n") 'def compute(data):\n return 1\n'