I. The Expensive if/else Statement

LLMs are remarkably good at handling the unknown. Give them an edge case you've never anticipated, and they'll reason through it correctly. That's genuine value.

The problem is that most of your traffic isn't unknown. A payment that failed on the third retry, a VIP customer with a disputed charge, an order flagged at 0.9 risk score — you already know what to do with these. But if your architecture routes everything through the LLM, it pays full inference cost regardless:

from pydantic_ai import Agent
from enum import Enum

class Action(Enum):
    FLAG_FRAUD = "flag_fraud"
    OPEN_DISPUTE = "open_dispute"
    SCHEDULE_RETRY = "schedule_retry"
    ALERT_ACCOUNT_MANAGER = "alert_account_manager"
    STANDARD_PROCESSING = "standard_processing"

agent = Agent(
    "openai:gpt-4o-mini",
    output_type=Action,
    system_prompt="Decide what action to take on this order event.",
)

async def handle(event: dict) -> Action:
    result = await agent.run(str(event))
    return result.output

This works. It also pays LLM prices for decisions that — in a large portion of cases — are already fully determined by the input data.

The obvious fix — an if/elif pre-filter — creates a different problem. It works until you have 15 conditions, three engineers with conflicting opinions, a silent ordering bug, and tests that only verify outputs, not the logic structure itself. The spaghetti grows fast.

What you actually need is a layer that:

• codifies what you already know as explicit, deterministic rules — zero API cost, microsecond latency
• routes only the genuinely unknown cases to the LLM
• stays auditable, testable, and readable as it grows

That's exactly what airules is built for. By the end of this article you'll have a working hybrid system that reserves the LLM for what it's actually good at — and handles everything else for free.

II. The core architecture: Facts, Rules, and Engines

Why type safety matters here

The problem with raw if/elif chains isn't just readability. event["risk_score"] is untyped — a missing key silently raises KeyError at runtime, a misspelled field name goes unnoticed until production, and nothing stops you from accidentally comparing a float to a string.

airules brings static typing to decision logic. Your IDE catches bad field references. Pyright flags type mismatches. The rule set becomes something a tool can inspect, not just a human can read.

Step 1: Define a Fact (your typed input schema)

from typing import Literal
from airules import Fact, Field, NumberField

OrderStatus = Literal["placed", "payment_failed", "fulfilled", "disputed", "refunded"]
CustomerTier = Literal["standard", "vip", "wholesale"]

class OrderEvent(Fact):
    status: Field[OrderStatus]
    amount_usd: NumberField[float]
    customer_tier: Field[CustomerTier]
    retry_count: NumberField[int]
    risk_score: NumberField[float]  # 0.0–1.0 from fraud detection service

Fact is not a Pydantic model or a dataclass — it's a purpose-built descriptor system where each field type (NumberField, Field, ListField) doubles as a predicate builder. That's what lets you write OrderEvent.risk_score.ge(0.85) as a first-class object rather than an inline comparison that lives and dies in one function.

Step 2: Build the engine

from airules import KnowledgeEngine, Rule, Default

class OrderRouter(KnowledgeEngine[OrderEvent, Action]):

    # High risk score is an unconditional block — checked first
    @Rule(OrderEvent.risk_score.ge(0.85))
    def high_risk(self, event: OrderEvent) -> Action:
        return Action.FLAG_FRAUD

    # A disputed charge always opens a case, regardless of other signals
    @Rule(OrderEvent.status.eq("disputed"))
    def dispute(self, event: OrderEvent) -> Action:
        return Action.OPEN_DISPUTE

    # Failed payment with retries remaining → schedule another attempt
    @Rule(
        OrderEvent.status.eq("payment_failed")
        & OrderEvent.retry_count.lt(3)
    )
    def retry(self, event: OrderEvent) -> Action:
        return Action.SCHEDULE_RETRY

    # VIP customer exhausted retries → human touch, not automation
    @Rule(
        OrderEvent.status.eq("payment_failed")
        & OrderEvent.customer_tier.eq("vip")
        & OrderEvent.retry_count.ge(3)
    )
    def vip_payment_exhausted(self, event: OrderEvent) -> Action:
        return Action.ALERT_ACCOUNT_MANAGER

    # Fires only when nothing above matched
    @Default
    def fallback(self, event: OrderEvent) -> Action:
        return Action.STANDARD_PROCESSING

How execution flows

Rules evaluate top-to-bottom. The first match wins — no fall-through, no ambiguity. @Default is always last, regardless of where you declare it.

router = OrderRouter()

router.run(OrderEvent(
    status="payment_failed",
    amount_usd=149.00,
    customer_tier="vip",
    retry_count=3,
    risk_score=0.2,
))
# → Action.ALERT_ACCOUNT_MANAGER  (matched `vip_payment_exhausted`, stopped there)

The engine is Generic[OrderEvent, Action] — run() returns Action | None, fully typed. No casting, no Any, no surprises.

Trade-off to know: airules uses declaration order as the default priority. If two rules can match the same input, the one declared first wins. This is explicit and predictable, but it means rule ordering is load-bearing — treat it with the same care you'd treat database index order. In the example above, high_risk must come before dispute because a high-risk disputed order should be flagged for fraud, not just opened as a dispute case.

III. Serializable Rules: Your LLM can read them too

The problem with hardcoded logic

If your rules live only in Python source, the rest of your system is blind to them. Your database can't query them. A code review diff shows syntax changes, not logic changes. And critically — your LLM can't reason about them.

Rules as data

airules predicates are first-class objects that serialize to plain dicts:

p = (
    OrderEvent.status.eq("payment_failed")
    & OrderEvent.retry_count.lt(3)
)

p.to_dict()
# {
#   "op": "and",
#   "operands": [
#     {"op": "eq",  "field": "status",      "value": "payment_failed"},
#     {"op": "lt",  "field": "retry_count", "value": 3}
#   ]
# }

# Round-trips cleanly
restored = Predicate.from_dict(p.to_dict())

One level up, describe() dumps the entire rule set — every rule, its predicate, its priority:

import json
print(json.dumps(OrderRouter.describe(), indent=2))
# {
#   "facts": [{"name": "OrderEvent", "fields": {
#     "status": "Field", "amount_usd": "NumberField",
#     "customer_tier": "Field", "retry_count": "NumberField", "risk_score": "NumberField"
#   }}],
#   "rules": [
#     {"name": "high_risk",            "predicate": {...}, "priority": 5, "is_default": false},
#     {"name": "dispute",              "predicate": {...}, "priority": 4, "is_default": false},
#     {"name": "retry",                "predicate": {...}, "priority": 3, "is_default": false},
#     {"name": "vip_payment_exhausted","predicate": {...}, "priority": 2, "is_default": false},
#     {"name": "fallback",             "predicate": null,  "priority": 0, "is_default": true}
#   ]
# }

Store it. Diff it in PRs. Feed it into a rules-editor UI. Or — and this is the key move — pass it directly into your LLM's system prompt.

IV. Building the hybrid router

The architecture

Incoming order event
        │
        ▼
┌──────────────────┐    match     ┌──────────────────┐
│   OrderRouter    │ ──────────▶  │  Return Action   │  ← ~0ms, $0.00
│   (airules)      │              └──────────────────┘
│                  │    no match
│                  │ ──────────▶  ┌──────────────────┐
└──────────────────┘              │  LLM fallback    │  ← ~800ms, costs tokens
                                  └──────────────────┘

Rules handle the known cases for free. The LLM handles only what genuinely falls through — unusual combinations of signals that no single rule anticipated.

The @Default fallback

The LLM call lives inside @Default — the method that fires only when the engine found no match. Everything from sections II and III stays the same; you're replacing just the fallback method:

import json
from pydantic_ai import Agent
from airules import Fact, Field, NumberField, KnowledgeEngine, Rule, Default

class OrderRouter(KnowledgeEngine[OrderEvent, Action]):

    @Rule(OrderEvent.risk_score.ge(0.85))
    def high_risk(self, event: OrderEvent) -> Action:
        return Action.FLAG_FRAUD

    @Rule(OrderEvent.status.eq("disputed"))
    def dispute(self, event: OrderEvent) -> Action:
        return Action.OPEN_DISPUTE

    @Rule(
        OrderEvent.status.eq("payment_failed")
        & OrderEvent.retry_count.lt(3)
    )
    def retry(self, event: OrderEvent) -> Action:
        return Action.SCHEDULE_RETRY

    @Rule(
        OrderEvent.status.eq("payment_failed")
        & OrderEvent.customer_tier.eq("vip")
        & OrderEvent.retry_count.ge(3)
    )
    def vip_payment_exhausted(self, event: OrderEvent) -> Action:
        return Action.ALERT_ACCOUNT_MANAGER

    @Default
    async def llm_triage(self, event: OrderEvent) -> Action:
        # Only reached for combinations no rule anticipated —
        # e.g. a wholesale customer with moderate risk (0.6) and a fulfilled
        # order that was just disputed for the first time.
        rules_schema = json.dumps(type(self).describe(), indent=2)

        agent = Agent(
            "openai-chat:gpt-5.4-mini",,
            output_type=Action,
            system_prompt=(
                "You are an order event classifier. "
                "The rules below already handle known cases deterministically — "
                "you only receive events that matched none of them. "
                "Decide the best action for this edge case.\n\n"
                f"Existing rules:\n{rules_schema}"
            ),
        )
        result = await agent.run(str(event))
        return result.output


router = OrderRouter()
action = await router.run_async(event)

Why type(self).describe() is the key line

Without it, your LLM and your rules engine are two separate systems with no shared understanding. The model might return SCHEDULE_RETRY for an event that should have been caught by the high_risk rule — creating inconsistencies that are maddening to debug.

With it, the LLM receives a precise, machine-readable description of every rule that already exists. It knows exactly which cases are handled upstream, so it can't contradict them — and it focuses only on the genuine gap.

The model isn't duplicating your rules. It's extending them.

run_async and evaluate_async handle async @Default methods automatically. Your synchronous @Rule methods don't need to change.

V. Observability: turning defaults into a roadmap

The metric that matters

Once this is running, the number to watch isn't "how many events did the engine handle." It's "what percentage hit @Default?"

• 10% default rate → rules handle 90% of traffic; the LLM sees only genuine edge cases
• 40% default rate → rules are missing major patterns; significant known traffic still hitting the API
• 80% default rate → back to nearly the original problem; your rules aren't covering enough

The gap between those numbers is the payoff from getting your rules right.

Injecting an observer

The engine accepts an observer that receives every evaluation result — without blocking the engine:

from airules import LoggingObserver

# Built-in: logs every evaluation, warns loudly on @Default hits
router = OrderRouter(observer=LoggingObserver())
await router.run_async(event)
# INFO     OrderRouter: rule='retry' result=<Action.SCHEDULE_RETRY: 'schedule_retry'>
# WARNING  OrderRouter: default fired — status='fulfilled', risk_score=0.62, customer_tier='wholesale'

For custom telemetry — Prometheus counters, Datadog traces, a Slack alert when the default rate spikes — implement OutcomeObserver:

import logging
from airules import OutcomeObserver, Outcome

class DefaultRateObserver(OutcomeObserver[OrderEvent, Action]):
    def __init__(self) -> None:
        self.total = 0
        self.defaults = 0

    def observe(
        self,
        outcome: Outcome[OrderEvent, Action],
        engine: KnowledgeEngine[OrderEvent, Action],
    ) -> None:
        self.total += 1
        if outcome.is_default:
            self.defaults += 1
            logging.warning("no rule matched: %s", outcome.fact)

    @property
    def default_rate(self) -> float:
        return self.defaults / self.total if self.total else 0.0

The iterative loop

Your @Default hits aren't failures — they're a data-driven roadmap:

1. Observe — log every @Default hit with the full outcome.fact
2. Analyze — what field combinations keep appearing? Wholesale customers with moderate risk scores? Fulfilled orders that get disputed?
3. Add a rule — write one new @Rule for each pattern you can name
4. Repeat — default rate drops, LLM calls drop, and the rules become a living record of your domain knowledge

The LLM is effectively labeling the gaps in your rule set. Every correct @Default classification is a signal that says "this combination needs a rule." Over time, the engine handles more and more traffic — the LLM handles less and less, reserved for inputs that genuinely benefit from its reasoning.

VI. When to use it — and when not to

Good fit:

• Structured event routing — order events, webhooks, payment signals, API callbacks where inputs have typed, discrete fields
• Business rule enforcement — eligibility checks, rate limiting, tier-based access where the logic is known and articulable
• LLM cost guardrails — deterministically handle the known cases before they ever reach the API
• Auditable decision logic — anywhere you need to explain why a specific input was handled a specific way

Poor fit:

• Unstructured text classification — if your inputs are raw free text with no structured signals, keyword matching creates false positives. Extract structured fields first (even with a lightweight model), then apply rules to those.
• 2–3 branch logic that won't grow — if you have two cases and a clear stopping condition, a plain if statement is the right tool. No ceremony needed.
• Purely fuzzy matching — if you genuinely can't articulate a rule as a typed condition, the engine can't express it. ML classifiers or embedding-based retrieval handle that better.

Production note: airules is experimental at v0.1.1. The core API is stable in spirit, but details may still change. Pin your version and read the changelog before upgrading in production.

VII. Conclusion

Optimizing LLM usage isn't just about picking a cheaper model or adding a cache. It's about being honest about which parts of your pipeline actually need intelligence — and handling everything else deterministically.

A typed rules engine that encodes what you already know is the correct complement to an LLM. airules makes that layer explicit, auditable, and — via describe() — self-documenting enough to feed back into the LLM itself as context.

Get started:

pip install ai-rules-engine

Full working examples | Documentation

Building a similar hybrid system? Share your default rate in the comments — curious how different domains compare. And if airules is missing something you need, open an issue. The roadmap is driven by real use cases.