Install
Guarded Agentic Compaction requires Python 3.11 or newer. The core compiler is framework-neutral; the Agents SDK is an optional integration.
git clone https://github.com/rrahimi-uci/guarded-agentic-compaction.git
cd guarded-agentic-compaction
python -m venv .venv
.venv/bin/pip install -e '.[dev,live,figures]'
For provider-backed examples, copy .env.example to .env and set OPENAI_API_KEY. Credentials are read into the process environment and are not serialized into result artifacts.
1. Estimate feasibility
guarded-agentic-compaction estimate traces.jsonl \
--effects configs/effects.example.yaml \
--entry channel locale product
The estimator reports independent-group support, candidate-region coverage, expected calibration requirements, runtime savings, and break-even assumptions. A decisive “do not compile” is a successful result.
2. Compile and inspect
guarded-agentic-compaction compile traces.jsonl \
--effects configs/effects.example.yaml \
--entry channel locale product \
--out artifacts/v1
guarded-agentic-compaction explain artifacts/v1
Review the emitted pseudocode, argument bindings, effect declarations, group splits, replay results, threshold grid, compatibility key, and every rejection reason. Do not promote an artifact whose effect catalog or registered task contract has not received application review.
Python API
import guarded_agentic_compaction as ac
episodes = ac.read_jsonl("traces.jsonl")
catalog = ac.load_catalog("configs/effects.example.yaml")
job = ac.optimize(
episodes,
catalog,
algorithms=["grc"],
mode="offline",
partition_by=["tenant_partition", "principal", "policy_version"],
entry_schema=["channel", "locale"],
sandbox=make_sandbox,
)
print(job.report())
print(job.explain())
ac.validate(job, suites=["replay", "perturbation"])
ac.promote(job, stage="shadow")
3. Run in shadow
The outer runner owns the entry snapshot and staging boundary. It provides the strongest fallback semantics and is the recommended integration for production-shaped experiments.
runner = ac.CompactingRunner(
dispatcher=ac.Dispatcher(
registry=registry,
catalog=catalog,
mode="shadow",
),
catalog=catalog,
manifest=manifest,
)
decision = runner.execute_pre_model(
entry_state,
executor=execute_tool,
continuation_compatibility_key=continuation_manifest.compatibility_key(),
)
provider_input = decision.observations if decision.compacted else baseline_input
Pre-model guarded composite
GCS is appropriate only when the artifact's verified live-outs support a task-specific projection and the continuation contract is pinned exactly.
decision = runner.execute_pre_model(
entry_state,
executor=execute_tool,
continuation_compatibility_key=continuation_manifest.compatibility_key(),
)
if decision.compacted:
context = decision.observations
else:
context = None # unchanged baseline path
OpenAI Agents SDK
The maintained SDK adapter captures traces through an AgentsTraceProcessor. The optional CompactingModel handles a narrower model-boundary path and rejects unsupported streaming, hosted-tool, MCP, handoff, and loop surfaces.
from agents import Agent
from guarded_agentic_compaction.runtime.model_provider import CompactingModel
agent = Agent(
name="support",
model=CompactingModel(
base_model,
registry=registry,
catalog=catalog,
manifest=manifest,
mode="shadow",
entry_state_fn=entry_state_from_sdk_input,
partition_fn=partition_from_sdk_input,
),
tools=tools,
)
Verify the checkout
.venv/bin/python -m pytest
.venv/bin/python scripts/verify_release.py
.venv/bin/python paper/scripts/validate_artifacts.py
.venv/bin/python -m build
Do not start with live mode. Compare baseline and candidate in shadow, inspect mismatches, test invalidation and rollback, then advance through a reviewed canary.