Skip to main content

The Engine

A KnowledgeEngine collects @Rule(...) methods defined on the class and evaluates them top-to-bottom against an input fact. The first matching rule wins; its method is called with the matched fact and its return value is returned from run(...).

from airules import KnowledgeEngine, Rule, Default

class TrafficAdvice(KnowledgeEngine[Light, str]):
    @Rule(Light.color.eq("green"))
    def green(self, light: Light):
        return "go"

    @Rule(Light.color.eq("yellow") & Light.remaining_time.gt(5))
    def yellow_safe(self, light: Light):
        return f"still {light.remaining_time}s"

    @Rule(Light.color.eq("yellow") | Light.color.eq("red"))
    def stop(self, light: Light):
        return "stop"

    @Default
    def fallback(self, light: Light):
        return "unknown signal"

Priority​

Rules are ordered by an explicit priority= argument when provided, otherwise by declaration order (earlier declarations win). @Default-decorated methods are always tried last, after every regular rule has failed, regardless of where they appear in the class body.

@Rule(some_predicate, priority=100)   # checked before priority=10
def high_priority(self, fact): ...

If no rule matches and there is no @Default, run returns None.

Async support​

The engine handles async @Default (or @Rule) methods automatically. Use run_async and evaluate_async when your action needs to await:

@Default
async def llm_fallback(self, fact: MyFact) -> str:
    result = await some_llm_call(fact)
    return result

# Synchronous @Rule methods don't need to change
result = await engine.run_async(fact)

Union fact types​

The engine type parameter accepts a union, which is useful when several fact shapes can be fed to the same rule set:

class MyEngine(KnowledgeEngine[Light | Car, str]):
    ...