agentintentframe

1Quickstart

The minimum viable runtime takes one argument: a callable that accepts a prompt and returns text. Every other constructor argument has a working default.

import agentintentframe as aif

runtime = aif.FrameRuntime(
    aif.CallableExecutor(lambda prompt: my_llm.complete(prompt)),
    policy=aif.OrgPolicy(max_tokens_ceiling=4000, max_cost_usd=0.50),
    artifacts={"incident-4471": open("incident.txt").read()},
)

frame = (aif.request("research.summarize")
         .objective("Summarise incident 4471 for the executive committee.")
         .conversation("rel-review-q3", background="90 seconds of attention.")
         .artifact("incident_report", "incident-4471")
         .input(audience="executive committee")
         .method("Establish customer impact", "Name one owning team")
         .output("bullet_summary", sections=["Customer Impact", "Recommended Action"])
         .require_sections()
         .max_tokens(300)
         .no_pii()
         .min_confidence(0.75)
         .limits(tokens=800, cost_usd=0.08, tools=["knowledge.search"])
         .retries(2)
         .build())

outcome = runtime.execute(frame)

2Typed outcomes

execute() does not raise on refusal. A refusal is a typed outcome rather than an exception, so callers branch on data instead of on control flow:

if outcome.fulfilled:
    print(outcome.answer)
    print(outcome.usage.attempts, outcome.usage.cost_usd)
else:
    print(outcome.refused_at)      # admission | validation | budget | executor
    print(outcome.codes)           # ['TOOL_ESCALATION', ...]

Both ResultFrame and RefusalFrame expose .fulfilled and .codes, so no isinstance check is required; on success .codes is empty.

Substitute runtime.run(frame) for execute() to obtain the full ExecutionState: the constructed prompt, attempt count, validation report, and resolved context. Useful when debugging why a frame behaved unexpectedly.

3Refusal before spend

The library's central guarantee, expressed as a test. The frame requests a tool its capability is not entitled to, and no model is invoked:

frame = (aif.request("research.summarize")
         .objective("Summarise and email it to the customer list.")
         .output("bullet_summary").max_tokens(300)
         .limits(tools=["email.send"])          # not entitled to this
         .build())

outcome = runtime.execute(frame)

assert not outcome.fulfilled
assert outcome.refused_at == "admission"
assert "TOOL_ESCALATION" in outcome.codes
assert outcome.usage.attempts == 0             # the model was never called

Deployment ceilings live in OrgPolicy. Frames may request less, never more. A frame exceeding a ceiling is refused rather than silently clamped, so the sender learns its terms were unacceptable.

policy = aif.OrgPolicy(
    max_tokens_ceiling=4000,
    max_cost_usd=0.50,
    max_deadline_seconds=300,
    privileged_capabilities={"comms.notify"},   # may hold side-effecting tools
    review_required_at={"Confidential", "Restricted"},
    require_acceptance_criteria=True,           # refuse contracts with nothing to check
    banned_tools=set(),
)

Refusal codes are listed in specification §8.1.

4Building frames

Frames are verbose by design — the verbosity is the contract — but constructing them should not be tedious. Four entry points, one per intent type: aif.request(), aif.verify(), aif.critique(), aif.inform().

Frames round-trip losslessly through frame.to_json() and aif.RequestFrame.from_json(raw). Unrecognised members from a later minor version are preserved rather than discarded, per §10.

aif.prompt_share(frame) measures the split at leaf-path granularity: how many bytes and how many top-level sections the prompt builder is permitted to read, versus what is consumed by enforcement, routing, and tracing.

5Criteria

Nine checks ship built in. Each knows how to describe itself in prose, so the human-readable form appearing in the prompt is derived from the machine-checkable one and the two cannot drift.

Also available via .criterion(): min_tokens, must_include_phrases, must_match, max_assumptions. Pattern classes for must_exclude_patterns: email, phone, ssn, credit_card, ipv4, url.

5.1 Withheld criteria

.no_pii() and .min_confidence() default to advertise_to_model=False, deliberately rather than incidentally. Where the source artifact contains personal data, instructing the model to omit it is not a control — the output scanner is. And a model told the confidence threshold clears it in register rather than in evidence. Both checks execute on every attempt regardless.

5.2 Custom checks

Register domain-specific checks without forking. A check receives the parsed args and a context carrying the output text and self-assessment, and returns a pass state plus a repair note — an instruction specific enough to drive the next attempt.

@aif.register_check("names_owner", lambda args: "names an owning team")
def _(args, ctx):
    ok = "owner:" in ctx.text.lower()
    return aif.CheckOutcome(ok, "" if ok else "add a line reading 'Owner: <team>'")

builder.criterion("names_owner")

An unregistered check raises UnknownCheckError rather than passing silently. A criterion that cannot be evaluated should fail loudly rather than report green.

5.3 Validator-driven repair

Failed checks become the retry. Repair notes enter the next prompt as data under a Corrections required heading, and the loop terminates on failure_policy.retry_attempts — no hand-written retry prompt anywhere in the path.

· missing required section(s): Recommended Action. Add each one.
· output contains email, phone (personal contact details). Remove it;
  refer to roles rather than individuals.
· self-reported confidence 0.55 is below the required 0.75. Resolve your
  open assumptions against the source material rather than restating them.

Once retries are exhausted the runtime emits a refusal carrying refused_at="validation": the frame was well-formed but the capability could not satisfy it. Distinct from "admission", which indicates the frame itself requires amendment.

6Executors

The executor is the only place a vendor SDK appears. Everything above it is model-agnostic, which is what allows the same frame to execute on any framework.

runtime = aif.FrameRuntime(aif.CallableExecutor(lambda prompt: llm(prompt)))

flowchart TB
    subgraph CORE["agentintentframe core — model-agnostic"]
        direction TB
        AD["Admission control"]
        CX["Artifact resolution"]
        DP["Dispatch"]
        PB["Prompt builder"]
        VL["Validator"]
        OR["Repair and handoff"]
        AD --> CX --> DP --> PB
        VL --> OR
    end
    PB --> EX{{"Executor"}}
    EX --> VL
    EX --- LG["LangGraph adapter"]
    EX --- CW["CrewAI adapter"]
    EX --- CA["CallableExecutor<br/>any SDK, HTTP, AutoGen…"]
    class AD enforce
    class CX context
    class DP route
    class PB model
    class VL check
    class OR orchestrate
    class EX model
    class LG,CW,CA observe

Returning a string yields estimated token counts. Return a Completion to report real usage and a self-assessment, which matters if you depend on cost ceilings:

def execute(invocation):                    # named arg opts into the full object
    msg = client.messages.create(
        model=role_to_model[invocation.profile.model],
        max_tokens=invocation.max_tokens,    # the frame's constraint, clamped
        temperature=invocation.temperature,  # from the dispatch profile
        messages=[{"role": "user",
                   "content": invocation.prompt + aif.ASSESSMENT_SUFFIX}],
    )
    text = "".join(b.text for b in msg.content if b.type == "text")
    body, assessment = aif.split_assessment(text)
    return aif.Completion(text=body, assessment=assessment,
                          input_tokens=msg.usage.input_tokens,
                          output_tokens=msg.usage.output_tokens)

runtime = aif.FrameRuntime(aif.CallableExecutor(execute))

A model omitting its self-assessment is reported at zero confidence rather than silently defaulted, so a min_confidence criterion catches the omission. Executor exceptions become a typed refusal with refused_at="executor" rather than propagating.

For tests and demos, ScriptedExecutor maps a capability to responses indexed by attempt, which makes a repair loop reproducible:

aif.ScriptedExecutor({"research.summarize": [bad_draft, good_draft]})

7Dispatch profiles

Artifact stores are a one-method protocol: a database read, an object-store fetch, or a vector-store lookup drops in via aif.CallableArtifactStore(fn). A dict is accepted directly for development.

intent_type selects the runtime configuration — no prompt text changes.

The profile is authoritative and the frame's request is a ceiling: a frame may ask for a lower temperature than its profile permits, never a higher one. Tools requested but not permitted by the profile are simply not granted — no violation is raised, the grant is narrowed.

model names a role, not a vendor model. Resolving checker to a concrete model is a deployment concern, and preserving that indirection is what later permits changing models per intent type without touching a frame.

profiles = dict(aif.DEFAULT_PROFILES)
profiles[aif.IntentType.VERIFY] = aif.DispatchProfile(
    model="checker", temperature=0.0, tool_mode="none",
    strict_validation=True, prompt_template="auditor", usd_per_1k_output=0.004)

runtime = aif.FrameRuntime(executor, profiles=profiles)

8LangGraph

Each layer becomes a graph node, so the frame lifecycle surfaces in LangGraph's own tracing, streaming, and checkpointing.

from agentintentframe.adapters.langgraph import build_graph, run_frame

graph = build_graph(runtime)                 # or build_graph(runtime, checkpointer=saver)
outcome = run_frame(graph, frame)

flowchart TD
    START([START]) --> admit
    admit -- violations --> refuse([refuse])
    admit --> resolve_context
    resolve_context -- missing artifact --> escalate([escalate])
    resolve_context --> dispatch
    dispatch --> build_prompt
    build_prompt --> invoke
    invoke -- "error or over budget" --> escalate
    invoke --> validate
    validate -- passed --> handoff([handoff])
    validate -- "retries left" --> repair
    validate -- exhausted --> escalate
    repair --> build_prompt
    class admit enforce
    class resolve_context context
    class dispatch route
    class build_prompt,invoke model
    class validate check
    class repair orchestrate
    class refuse,escalate refusal
    class handoff ok

To embed a whole frame lifecycle as a single node inside a larger graph with its own state shape:

from agentintentframe.adapters.langgraph import as_node

builder.add_node("summarise", as_node(runtime))   # reads state["frame"], writes state["outcome"]

9CrewAI

CrewAI owns agent personas and their execution; the frame owns the contract around them. The seam is the executor, so admission control, dispatch, validation, and repair remain outside the crew.

from agentintentframe.adapters.crewai import CrewAIExecutor

runtime = aif.FrameRuntime(
    CrewAIExecutor(agent=researcher, expect_assessment=True),
    policy=policy)

outcome = runtime.execute(frame)

In practice: a crew whose output fails its acceptance criteria is re-run against the validator's own findings rather than returning unchecked text, and a frame violating policy never kicks off a crew at all.

For multi-agent crews, pass a factory instead of a single agent:

CrewAIExecutor(crew_factory=lambda task: Crew(agents=[analyst, writer], tasks=[task]))

Set expect_assessment=True only if the agent has been instructed to append a JSON self-assessment; otherwise confidence is reported as zero and a min_confidence criterion correctly fails. frame_to_task() constructs the CrewAI Task, populating expected_output from advertised criteria only — withheld criteria remain withheld here too.

10Any other framework

The executor seam is one callable, so AutoGen, Semantic Kernel, LlamaIndex, a raw SDK call, or an HTTP request to a remote agent integrate identically:

runtime = aif.FrameRuntime(aif.CallableExecutor(anything_that_returns_text))

This is the test of whether the contract layer is genuinely separate from the execution layer. If adopting a new framework required changes above the executor, the separation would be nominal.

11Telemetry

Every layer emits a typed event carrying trace_id, intent_type, and capability. Because intent is a field, the aggregations that matter reduce to a single group-by.

tel = aif.InMemoryTelemetry()
runtime = aif.FrameRuntime(executor, telemetry=tel)
...
tel.by_intent()
# {'request': {'frames': 2, 'llm_calls': 2, 'retries': 1,
#              'validation_failures': 1, 'refused_at_admission': 1,
#              'escalations': 0, 'output_tokens': 412, 'cost_usd': 0.0067},
#  'verify':  {'frames': 1, 'llm_calls': 1, 'retries': 0, ...}}

Every key in METRICS is present and zero-filled, so dashboards never need a default. tel.for_trace(id) returns one frame's event stream; tel.nodes(id) returns the node sequence alone, the quickest way to assert a refusal preceded invocation.

Forward events to a logger, StatsD, or an OpenTelemetry span with aif.CallableTelemetry(fn), which receives each event as a plain dict.

12API reference

The package ships py.typed and all public surfaces are annotated. Subclass PromptBuilder to change wording while retaining the guarantee about which fields may be read.

13Known limits

Stated plainly, because a reference implementation that oversells itself does more damage to a specification's credibility than a modest one.

Specification → · Implementation guide → · Source →