adaptive-utility-agent

AUA Framework

A production framework for self-correcting, multi-specialist LLM systems.

Full site: https://praneethtota.github.io/Adaptive-Utility-Agent

---

What it does

AUA sits between your application and your language models. It routes prompts to specialist models, scores responses with a utility function, catches contradictions, injects prior verified corrections into future queries, enforces policies in real-time, and self-corrects across sessions.

The core idea: a model that makes a wrong answer on Tuesday should not make the same wrong answer on Thursday. AUA closes that loop without waiting for a new model release.

pip install adaptive-utility-agent
aua init my-project --preset coding --tier macbook
cd my-project && aua serve

---

Sister project: AUA Veritas

AUA Veritas applies the framework ideas in a consumer-facing desktop app — compare multiple frontier models, remember corrections, return one answer with a confidence signal.

👉 AUA Veritas

---

Documentation

---

Quickstart

Install

pip install adaptive-utility-agent

# With GPU serving backend (Linux + CUDA)
pip install "adaptive-utility-agent[vllm]"

# With development tools
pip install "adaptive-utility-agent[dev]"

Scaffold and serve

# Mac / Apple Silicon — uses Ollama (brew install ollama first)
aua init my-project --preset coding --tier macbook
cd my-project
aua doctor        # pre-flight check: config, deps, hardware, compat matrix
aua serve         # start specialists + router on :8000

Send a query

curl -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"query": "Write binary search in Python. State time complexity."}'

from aua import Router
from aua.config import load_config

config = load_config("aua_config.yaml")
router = Router.from_config(config)
result = await router.query("Write bubble sort. What is its O complexity?")
print(result.response)
print(f"U={result.u_score:.3f}  mode={result.routing_mode}  degraded={result.degraded_mode}")

Chat UI

# Terminal 1
aua serve --tier macbook

# Terminal 2
aua ui   # starts on http://localhost:3001 (admin / aua-admin)

---

Hardware tiers

Aliases: gaming → gaming-pc, h100 → h100-cluster, a100 → a100-cluster, rtx4090 → single-4090.

Check compatibility before serving:

aua doctor --compat-matrix              # full model × hardware × backend matrix
aua doctor --compat-matrix-format json  # machine-readable

---

What ships in v1.2

---

Plugin system — 15 interfaces, 13 wired

Every major decision point is replaceable via a single YAML line. No forking required.

plugins:
  routing_strategy:
    import_path: my_plugins:TenantRouter
  full_utility_scorer:
    import_path: my_plugins:SurgeryAwareScorer
  full_promotion_policy:
    import_path: my_plugins:CIGatePromoter

All plugins are validated against their Protocol interface at startup — a misconfigured plugin fails fast, never silently at query time. Every wired plugin has a safe fallback: an exception at query time logs at DEBUG and falls back to the built-in.

---

The utility function

U = w_e(f) · E  +  w_c(f) · C  +  w_k(f) · K

E — Efficacy:    EMA-accumulated task performance                        [0, 1]
C — Confidence:  Kalman-filtered internal consistency after contradiction penalty  [0, 1]
K — Curiosity:   UCB-style exploration bonus (K_base + gap_bonus)       [0, 1]
f — field        (software_engineering, mathematics, surgery, law, ...)

The additive weighted structure is not a convenience — it is the unique functional form satisfying five behavioral axioms, proved via Debreu's representation theorem (Theorem B.1, Appendix B).

Replace it entirely with a full_utility_scorer plugin:

class SurgeryAwareScorer:
    def score(self, response, field, prior_u, confidence, metadata):
        return prior_u  # fallback

    def score_full(self, field, efficacy, confidence, curiosity, weights, metadata):
        if field == "surgery":
            return min(1.0, efficacy * (confidence ** 2))  # non-linear — C is load-bearing
        return weights["w_e"]*efficacy + weights["w_c"]*confidence + weights["w_k"]*curiosity

---

Policies — teaching the framework what good looks like

from aua.guard import assertion, AssertionLevel
from aua.policy import Policy

@assertion(name="PythonSyntaxCheck", level=AssertionLevel.BLOCKING)
def validate_syntax(output: str, context: dict) -> tuple[bool, str | None]:
    import ast, re
    for block in re.findall(r"```python(.*?)```", output, re.DOTALL):
        try:
            ast.parse(block)
        except SyntaxError as e:
            return False, f"Syntax error at line {e.lineno}"
    return True, None

@assertion(name="AnalogyBonus", level=AssertionLevel.INFO, bonus=0.10)
def reward_analogy(output: str, context: dict) -> tuple[bool, str | None]:
    if any(p in output.lower() for p in ["like a", "similar to", "imagine"]):
        return True, "Positive: analogy used"
    return True, None

policy = Policy(name="SafeCoding", max_total_bonus=0.30)
policy.add(validate_syntax)
policy.add(reward_analogy)

Over time: BLOCKING assertions reduce failures → passing sessions become gold-standard DPO data → aua calibrate --layer 3 exports them → fine-tune → repeat.

---

Resilience — retry and circuit breaker

router:
  retry:
    max_retries: 3          # 0 to disable
    base_delay_ms: 200      # doubles per attempt, capped at max_delay_ms
    max_delay_ms: 5000
    jitter: true            # ±25% — prevents thundering-herd
    retryable_status_codes: [429, 502, 503, 504]

  circuit_breaker:
    enabled: true
    failure_threshold: 5    # failures within window before opening
    failure_window_s: 60.0
    recovery_timeout_s: 30.0
    success_threshold: 2    # consecutive successes in HALF_OPEN → CLOSED

When a circuit is open, responses include degraded_mode: true and degraded_specialists: ["mathematics"]. The router continues serving via the arbiter or remaining healthy specialists — zero additional latency for end users once the circuit opens.

---

Security

security:
  auth_enabled: true
  token_secret_env: AUA_TOKEN_SECRET   # export AUA_TOKEN_SECRET=$(python3 -c "import secrets; print(secrets.token_hex(32))")
  token_expiry_days: 30
  mtls:
    key_file: certs/server.key
    cert_file: certs/server.crt
    ca_file: certs/ca.crt    # omit for server-TLS-only

aua token create --scope aua:query --expires 30d
curl -H "Authorization: Bearer aua_tk_..." http://localhost:8000/query ...
aua certs generate   # self-signed dev certs

---

Project structure

aua/
├── router.py               # Request routing + 50+ REST endpoints
├── arbiter.py              # 4-check arbitration pipeline (logical, math, cross-session, empirical)
├── utility_scorer.py       # U = w_e·E + w_c·C + w_k·K
├── field_classifier.py     # Probabilistic domain routing
├── assertions_store.py     # Cross-session corrections with decay classes A–D
├── retry.py                # Transport-level retry with exponential backoff (#39)
├── circuit_breaker.py      # Per-specialist CLOSED/OPEN/HALF_OPEN state machine (#37)
├── middleware.py           # Extended pipeline: on_chunk, before/after_batch, on_error (#52)
├── auth.py                 # HMAC-SHA256 token auth, 15 scopes, revocation
├── auth_middleware.py      # FastAPI middleware wiring auth into the request path
├── shadow.py               # Shadow mode — real-traffic GREEN evaluation (#48)
├── experiment_tracker.py   # MLflow + W&B lazy integration (#47)
├── batch_queue.py          # Persistent batch queue, priority lanes (#56)
├── model_registry.py       # HF @revision pinning, MLflow models:/ resolution (#46)
├── compat.py               # Model × hardware × backend compatibility matrix (#55)
├── empirical.py            # SymPy / arXiv / PubMed cross-check for arbiter Stage 4 (#61)
├── keywords.py             # Async full-text keyword search index
├── tenancy.py              # Per-tenant contextvar isolation (#44)
├── loadtest.py             # aua loadtest engine (#50)
├── test_harness.py         # aua test built-in fixture suites (#54)
├── blue_green.py           # Utility-deviation-triggered promotion, T_min gate, tau routing
├── guard.py                # @assertion decorator, AssertionLevel, Policy.run()
├── policy.py               # Policy dataclass + YAML loader
├── hooks.py                # HookRunner — 11 lifecycle hook points
├── metrics.py              # 18 Prometheus metrics
├── otel.py                 # OpenTelemetry tracing
├── state.py                # SQLite state store (sessions, corrections, audit log)
├── cli.py                  # aua CLI — 24 command groups
├── config.py               # AUAConfig, RetryConfig, CircuitBreakerConfig, tier loader
└── plugins/
    ├── interfaces.py       # 15 Protocol interfaces
    ├── registry.py         # Plugin load + contract validation
    └── prebuilt/           # OpenAI, Anthropic, Google backends (wired when #74 ships)

apps/
└── aua_chat/               # Next.js 14 Chat UI

tests/                      # 759 tests across Python 3.10 / 3.11 / 3.12

---

Validated results (v1.0 baseline, RTX 4090)

Full record: docs/v1_validation_report.md

---

Roadmap

Tracked in full at aua_roadmap.html.

Recent completions (#37–#55 block):

---

License

Code: GNU General Public License v3.0 — see LICENSE
Whitepaper: Creative Commons Attribution 4.0 — see LICENSE-CC-BY-4.0

If you build on this work, please cite:

Tota, P. (2026). AUA Framework v1.2: A Production Framework for Self-Correcting Multi-Specialist AI Systems. GitHub. https://github.com/praneethtota/Adaptive-Utility-Agent

---

📖 Full documentation, tutorial, and domain deep-dives:
https://praneethtota.github.io/Adaptive-Utility-Agent