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:
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.
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.
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.
from airules import 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
def fallback(self, event: OrderEvent) -> Action:
return Action.STANDARD_PROCESSING
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,
))
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 beforedispute
because a high-risk disputed order should be flagged for fraud, not just opened as a dispute case.
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.
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()
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))
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.
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.
@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:
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)
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
andevaluate_async
handle async@Default
methods automatically. Your synchronous@Rule
methods don't need to change.
Once this is running, the number to watch isn't "how many events did the engine handle." It's "what percentage hit @Default?"
The gap between those numbers is the payoff from getting your rules right.
The engine accepts an observer that receives every evaluation result — without blocking the engine:
from airules import LoggingObserver
router = OrderRouter(observer=LoggingObserver())
await router.run_async(event)
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
Your @Default
hits aren't failures — they're a data-driven roadmap:
@Default
hit with the full outcome.fact
@Rule
for each pattern you can nameThe 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.
Good fit:
Poor fit:
if
statement is the right tool. No ceremony needed.
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.
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.