Fluent API
Fluent API for FuzzyInfer - Beautiful, Pythonic inference pipelines.
This module provides a modern, chainable API for fuzzy inference with:
- Method chaining for elegant rule and fact composition
- Operator overloading for natural syntax
- Context managers for sessions
- Functional/lambda programming support
- Builder patterns for complex rules
- Query DSL for powerful pattern matching
Examples:
>>> # Beautiful inference pipeline
>>> with FuzzySession() as session:
... session.facts(
... ("is-bird", ["robin"], 0.9),
... ("is-bird", ["eagle"], 1.0)
... ).rule(
... when=("is-bird", "?x"),
... then=("can-fly", "?x", 0.95)
... ).infer()
... flying = session.query("can-fly")
>>> # Functional style
>>> results = (
... FuzzyInfer()
... .fact("is-mammal", ["dog"])
... .fact("is-mammal", ["cat"])
... .rule(lambda kb: kb["is-mammal"]["?x"] >> kb.add("warm-blooded", "?x"))
... .run()
... .where("warm-blooded")
... )
FluentQueryResult(facts: List[Fact], kb: 'FluentKB')
Fluent wrapper for query results with chainable operations.
Source code in fuzzy_infer/fluent.py
| def __init__(self, facts: List[Fact], kb: 'FluentKB'):
self._facts = facts
self._kb = kb
|
property
max_degree: Optional[float]
property
min_degree: Optional[float]
where(
condition: Callable[[Fact], bool],
) -> "FluentQueryResult"
Filter facts by condition.
Source code in fuzzy_infer/fluent.py
| def where(self, condition: Callable[[Fact], bool]) -> 'FluentQueryResult':
"""Filter facts by condition."""
filtered = [f for f in self._facts if condition(f)]
return FluentQueryResult(filtered, self._kb)
|
having_degree(op: str, value: float) -> 'FluentQueryResult'
Filter by degree comparison.
Source code in fuzzy_infer/fluent.py
| def having_degree(self, op: str, value: float) -> 'FluentQueryResult':
"""Filter by degree comparison."""
ops = {
'>': lambda d: d > value,
'>=': lambda d: d >= value,
'<': lambda d: d < value,
'<=': lambda d: d <= value,
'==': lambda d: d == value,
'!=': lambda d: d != value,
}
if op not in ops:
raise ValueError(f"Invalid operator: {op}")
return self.where(lambda f: ops[op](f.degree))
|
with_args(*args: Any) -> 'FluentQueryResult'
Filter by argument values.
Source code in fuzzy_infer/fluent.py
| def with_args(self, *args: Any) -> 'FluentQueryResult':
"""Filter by argument values."""
return self.where(lambda f: f.args == list(args))
|
map(func: Callable[[Fact], Any]) -> List[Any]
Map function over facts.
Source code in fuzzy_infer/fluent.py
| def map(self, func: Callable[[Fact], Any]) -> List[Any]:
"""Map function over facts."""
return [func(f) for f in self._facts]
|
to_dict() -> List[Dict[str, Any]]
Convert to list of dicts.
Source code in fuzzy_infer/fluent.py
| def to_dict(self) -> List[Dict[str, Any]]:
"""Convert to list of dicts."""
return [f.to_dict() for f in self._facts]
|
Convert to JSON string.
Source code in fuzzy_infer/fluent.py
| def to_json(self) -> str:
"""Convert to JSON string."""
return json.dumps(self.to_dict(), indent=2)
|
PatternMatcher(
predicate: str, kb: Optional["FluentKB"] = None
)
DSL for pattern matching in queries and rules.
Source code in fuzzy_infer/fluent.py
| def __init__(self, predicate: str, kb: Optional['FluentKB'] = None):
self.predicate = predicate
self.args: List[Any] = []
self.constraints: List[Callable] = []
self.kb = kb
|
to_condition() -> Condition
Convert to Condition object.
Source code in fuzzy_infer/fluent.py
| def to_condition(self) -> Condition:
"""Convert to Condition object."""
cond = Condition(self.predicate, self.args)
# Add degree constraints
for constraint in self.constraints:
# This is simplified - would need proper Condition support
pass
return cond
|
FluentRule(
condition: Union[PatternMatcher, Tuple, List],
action: Union[PatternMatcher, Tuple, Callable],
kb: Optional["FluentKB"] = None,
)
Fluent rule representation.
Source code in fuzzy_infer/fluent.py
| def __init__(
self,
condition: Union[PatternMatcher, Tuple, List],
action: Union[PatternMatcher, Tuple, Callable],
kb: Optional['FluentKB'] = None
):
self.conditions = [condition] if not isinstance(condition, list) else condition
self.actions = [action] if not isinstance(action, list) else action
self.kb = kb
self.name: Optional[str] = None
self.priority: int = 50
|
named(name: str) -> 'FluentRule'
Set rule name.
Source code in fuzzy_infer/fluent.py
| def named(self, name: str) -> 'FluentRule':
"""Set rule name."""
self.name = name
return self
|
with_priority(priority: int) -> 'FluentRule'
Set rule priority.
Source code in fuzzy_infer/fluent.py
| def with_priority(self, priority: int) -> 'FluentRule':
"""Set rule priority."""
self.priority = priority
return self
|
Build Rule object.
Source code in fuzzy_infer/fluent.py
| def build(self) -> Rule:
"""Build Rule object."""
# Convert conditions and actions to proper format
conditions = []
actions = []
for cond in self.conditions:
if isinstance(cond, PatternMatcher):
conditions.append(cond.to_condition())
elif isinstance(cond, tuple):
pred, *args = cond
conditions.append(Condition(pred, args))
else:
conditions.append(cond)
for action in self.actions:
if isinstance(action, PatternMatcher):
actions.append(Action(
action_type="add",
fact={
"pred": action.predicate,
"args": action.args,
"deg": 1.0
}
))
elif isinstance(action, tuple):
if len(action) == 2:
pred, args = action
degree = 1.0
else:
pred, args, degree = action
if not isinstance(args, list):
args = [args]
actions.append(Action(
action_type="add",
fact={
"pred": pred,
"args": args,
"deg": degree
}
))
elif callable(action):
# Lambda action - need special handling
pass
else:
actions.append(action)
return Rule(
name=self.name,
conditions=conditions,
actions=actions,
priority=self.priority
)
|
FluentKB(engine: Optional[FuzzyInfer] = None)
Fluent knowledge base interface.
Source code in fuzzy_infer/fluent.py
| def __init__(self, engine: Optional[FuzzyInfer] = None):
self._engine = engine or FuzzyInfer()
self._patterns: Dict[str, PatternMatcher] = {}
|
fact(
predicate: str,
args: Union[List[Any], Any],
degree: float = 1.0,
) -> "FluentKB"
Add a single fact.
Source code in fuzzy_infer/fluent.py
| def fact(self, predicate: str, args: Union[List[Any], Any], degree: float = 1.0) -> 'FluentKB':
"""Add a single fact."""
if not isinstance(args, list):
args = [args]
self._engine.add_fact(Fact(predicate, args, degree))
return self
|
facts(
*facts: Union[FactTuple, SimpleFact, Tuple]
) -> "FluentKB"
Add multiple facts.
Source code in fuzzy_infer/fluent.py
| def facts(self, *facts: Union[FactTuple, SimpleFact, Tuple]) -> 'FluentKB':
"""Add multiple facts."""
for f in facts:
if len(f) == 2:
pred, args = f
degree = 1.0
elif len(f) == 3:
# Could be (pred, args, degree) or (pred, arg1, arg2) for multi-arg facts
if isinstance(f[2], (int, float)) and 0 <= f[2] <= 1:
# It's a degree
pred, args, degree = f
else:
# It's a multi-arg fact like ("species", ["rex"], "dog")
pred = f[0]
args = list(f[1:])
degree = 1.0
elif len(f) == 4:
# Multi-arg fact with degree like ("species", ["rex"], "dog", 1.0)
pred = f[0]
args = list(f[1:-1])
degree = f[-1]
else:
pred, args, degree = f[0], list(f[1:]), 1.0
self.fact(pred, args, degree)
return self
|
rule(
when: Optional[
Union[PatternMatcher, Tuple, Callable, List]
] = None,
then: Optional[
Union[PatternMatcher, Tuple, Callable, List]
] = None,
**kwargs
) -> "FluentKB"
Add a rule using keyword arguments or builder.
Source code in fuzzy_infer/fluent.py
| def rule(
self,
when: Optional[Union[PatternMatcher, Tuple, Callable, List]] = None,
then: Optional[Union[PatternMatcher, Tuple, Callable, List]] = None,
**kwargs
) -> 'FluentKB':
"""Add a rule using keyword arguments or builder."""
if when is not None and then is not None:
# Handle list of conditions
if isinstance(when, list):
conditions = when
else:
conditions = [when]
# Handle list of actions
if isinstance(then, list):
actions = then
else:
actions = [then]
# Simple rule creation
rule = FluentRule(conditions, actions, self)
if 'name' in kwargs:
rule.named(kwargs['name'])
if 'priority' in kwargs:
rule.with_priority(kwargs['priority'])
self._engine.add_rule(rule.build())
elif callable(when):
# Lambda-style rule - skip for now since it needs special handling
pass
elif isinstance(when, dict):
# Direct dict rule
self._engine.add_rule(when)
return self
|
rules(*rules: Union[Rule, FluentRule, Dict]) -> 'FluentKB'
Add multiple rules.
Source code in fuzzy_infer/fluent.py
| def rules(self, *rules: Union[Rule, FluentRule, Dict]) -> 'FluentKB':
"""Add multiple rules."""
for r in rules:
if isinstance(r, FluentRule):
self._engine.add_rule(r.build())
elif isinstance(r, Rule):
self._engine.add_rule(r)
elif isinstance(r, dict):
self._engine.add_rule(r)
return self
|
infer(max_iterations: int = 100) -> 'FluentKB'
Run inference.
Source code in fuzzy_infer/fluent.py
| def infer(self, max_iterations: int = 100) -> 'FluentKB':
"""Run inference."""
self._engine.run(max_iterations)
return self
|
run(max_iterations: int = 100) -> 'FluentKB'
Alias for infer.
Source code in fuzzy_infer/fluent.py
| def run(self, max_iterations: int = 100) -> 'FluentKB':
"""Alias for infer."""
return self.infer(max_iterations)
|
query(
predicate: str,
args: Optional[Union[List[Any], Any]] = None,
) -> FluentQueryResult
Query facts.
Source code in fuzzy_infer/fluent.py
| def query(
self,
predicate: str,
args: Optional[Union[List[Any], Any]] = None
) -> FluentQueryResult:
"""Query facts."""
if args is not None:
if not isinstance(args, list):
args = [args]
facts = self._engine.query(predicate, args)
else:
facts = self._engine.query(predicate)
return FluentQueryResult(facts, self)
|
where(predicate: str) -> FluentQueryResult
Query all facts with predicate.
Source code in fuzzy_infer/fluent.py
| def where(self, predicate: str) -> FluentQueryResult:
"""Query all facts with predicate."""
facts = self._engine.query(predicate)
return FluentQueryResult(facts, self)
|
all_facts() -> List[Fact]
Get all facts.
Source code in fuzzy_infer/fluent.py
| def all_facts(self) -> List[Fact]:
"""Get all facts."""
return list(self._engine.facts.values())
|
clear_facts() -> 'FluentKB'
Clear all facts.
Source code in fuzzy_infer/fluent.py
| def clear_facts(self) -> 'FluentKB':
"""Clear all facts."""
self._engine.facts.clear()
return self
|
clear_rules() -> 'FluentKB'
Clear all rules.
Source code in fuzzy_infer/fluent.py
| def clear_rules(self) -> 'FluentKB':
"""Clear all rules."""
self._engine.rules.clear()
return self
|
Clear everything.
Source code in fuzzy_infer/fluent.py
| def clear(self) -> 'FluentKB':
"""Clear everything."""
return self.clear_facts().clear_rules()
|
save(path: Union[str, Path]) -> 'FluentKB'
Save to file.
Source code in fuzzy_infer/fluent.py
| def save(self, path: Union[str, Path]) -> 'FluentKB':
"""Save to file."""
from fuzzy_infer.jsonl import KnowledgeBaseIO
KnowledgeBaseIO.save(self._engine, path)
return self
|
load(path: Union[str, Path]) -> 'FluentKB'
Load from file.
Source code in fuzzy_infer/fluent.py
| def load(self, path: Union[str, Path]) -> 'FluentKB':
"""Load from file."""
from fuzzy_infer.jsonl import KnowledgeBaseIO
loaded = KnowledgeBaseIO.load(path)
self._engine = loaded
return self
|
FuzzySession(max_iterations: int = 100)
Context manager for inference sessions.
Source code in fuzzy_infer/fluent.py
| def __init__(self, max_iterations: int = 100):
self.max_iterations = max_iterations
self.kb: Optional[FluentKB] = None
|
Inference pipeline builder for complex workflows.
Source code in fuzzy_infer/fluent.py
| def __init__(self):
self.steps: List[Callable] = []
self.kb = FluentKB()
|
step(func: Callable) -> 'Pipeline'
Add a step to the pipeline.
Source code in fuzzy_infer/fluent.py
| def step(self, func: Callable) -> 'Pipeline':
"""Add a step to the pipeline."""
self.steps.append(func)
return self
|
facts_from(
source: Union[str, Path, Iterator],
) -> "Pipeline"
Load facts from source.
Source code in fuzzy_infer/fluent.py
| def facts_from(self, source: Union[str, Path, Iterator]) -> 'Pipeline':
"""Load facts from source."""
def load_facts(kb: FluentKB):
if isinstance(source, (str, Path)):
kb.load(source)
else:
for fact in source:
if isinstance(fact, Fact):
kb._engine.add_fact(fact)
elif isinstance(fact, tuple):
kb.fact(*fact)
self.steps.append(load_facts)
return self
|
rules_from(source: Union[str, Path, List]) -> 'Pipeline'
Load rules from source.
Source code in fuzzy_infer/fluent.py
| def rules_from(self, source: Union[str, Path, List]) -> 'Pipeline':
"""Load rules from source."""
def load_rules(kb: FluentKB):
if isinstance(source, (str, Path)):
# Load from file
pass
else:
kb.rules(*source)
self.steps.append(load_rules)
return self
|
transform(func: Callable[[FluentKB], None]) -> 'Pipeline'
Apply transformation to KB.
Source code in fuzzy_infer/fluent.py
| def transform(self, func: Callable[[FluentKB], None]) -> 'Pipeline':
"""Apply transformation to KB."""
self.steps.append(func)
return self
|
infer(max_iterations: int = 100) -> 'Pipeline'
Add inference step.
Source code in fuzzy_infer/fluent.py
| def infer(self, max_iterations: int = 100) -> 'Pipeline':
"""Add inference step."""
self.steps.append(lambda kb: kb.infer(max_iterations))
return self
|
filter_facts(
predicate: Callable[[Fact], bool],
) -> "Pipeline"
Filter facts.
Source code in fuzzy_infer/fluent.py
| def filter_facts(self, predicate: Callable[[Fact], bool]) -> 'Pipeline':
"""Filter facts."""
def filter_step(kb: FluentKB):
filtered = {k: v for k, v in kb._engine.facts.items()
if predicate(v)}
kb._engine.facts = filtered
self.steps.append(filter_step)
return self
|
Execute pipeline.
Source code in fuzzy_infer/fluent.py
| def run(self) -> FluentKB:
"""Execute pipeline."""
for step in self.steps:
step(self.kb)
return self.kb
|
stream(input_stream: Iterator[Fact]) -> Iterator[Fact]
Stream processing mode.
Source code in fuzzy_infer/fluent.py
| def stream(self, input_stream: Iterator[Fact]) -> Iterator[Fact]:
"""Stream processing mode."""
for fact in input_stream:
self.kb._engine.add_fact(fact)
# Run inference after each fact
self.kb.infer(1)
# Yield new facts
for f in self.kb._engine.facts.values():
yield f
|
Expression builder for complex conditions.
Source code in fuzzy_infer/fluent.py
| def __init__(self, expr):
self.expr = expr
|
fuzzy(*facts: Union[FactTuple, SimpleFact]) -> FluentKB
Quick fuzzy KB creation.
Source code in fuzzy_infer/fluent.py
| def fuzzy(*facts: Union[FactTuple, SimpleFact]) -> FluentKB:
"""Quick fuzzy KB creation."""
kb = FluentKB()
return kb.facts(*facts)
|
when(predicate: str, *args: Any) -> PatternMatcher
Create pattern matcher for conditions.
Source code in fuzzy_infer/fluent.py
| def when(predicate: str, *args: Any) -> PatternMatcher:
"""Create pattern matcher for conditions."""
pattern = PatternMatcher(predicate)
if args:
pattern.args = list(args)
return pattern
|
then(
predicate: str, *args: Any, degree: float = 1.0
) -> Tuple
Create action tuple.
Source code in fuzzy_infer/fluent.py
| def then(predicate: str, *args: Any, degree: float = 1.0) -> Tuple:
"""Create action tuple."""
# If degree is passed as last positional argument
if args and isinstance(args[-1], (int, float)) and 0 <= args[-1] <= 1:
actual_args = list(args[:-1])
actual_degree = args[-1]
return (predicate, actual_args, actual_degree)
return (predicate, list(args), degree)
|
stream_session(
input_file: Optional[Path] = None,
output_file: Optional[Path] = None,
)
Context manager for stream processing.
Source code in fuzzy_infer/fluent.py
| @contextmanager
def stream_session(input_file: Optional[Path] = None, output_file: Optional[Path] = None):
"""Context manager for stream processing."""
import sys
old_stdin = sys.stdin
old_stdout = sys.stdout
try:
if input_file:
sys.stdin = open(input_file, 'r')
if output_file:
sys.stdout = open(output_file, 'w')
kb = FluentKB()
yield kb
finally:
if input_file and sys.stdin != old_stdin:
sys.stdin.close()
sys.stdin = old_stdin
if output_file and sys.stdout != old_stdout:
sys.stdout.close()
sys.stdout = old_stdout
|
Overview
The fuzzy_infer.fluent module provides a modern, Pythonic interface for fuzzy inference with method chaining, operator overloading, and functional programming patterns.
Quick Reference
from fuzzy_infer.fluent import FluentKB, FuzzySession, fuzzy, when, then
# Beautiful chaining
results = (
FluentKB()
.fact("is-bird", ["robin"], 0.9)
.rule(when=("is-bird", "?x"), then=("can-fly", "?x", 0.85))
.infer()
.where("can-fly")
)
# Context manager
with FuzzySession() as session:
session.facts(("is-bird", ["robin"], 0.9))
session.rule(when=("is-bird", "?x"), then=("can-fly", "?x"))
results = session.query("can-fly")
# Quick creation
kb = fuzzy(
("is-bird", ["robin"], 0.9),
("is-bird", ["eagle"], 1.0)
)
Core Classes
FluentKB
Fluent knowledge base interface.
Source code in fuzzy_infer/fluent.py
| def __init__(self, engine: Optional[FuzzyInfer] = None):
self._engine = engine or FuzzyInfer()
self._patterns: Dict[str, PatternMatcher] = {}
|
fact(
predicate: str,
args: Union[List[Any], Any],
degree: float = 1.0,
) -> "FluentKB"
Add a single fact.
Source code in fuzzy_infer/fluent.py
| def fact(self, predicate: str, args: Union[List[Any], Any], degree: float = 1.0) -> 'FluentKB':
"""Add a single fact."""
if not isinstance(args, list):
args = [args]
self._engine.add_fact(Fact(predicate, args, degree))
return self
|
facts(
*facts: Union[FactTuple, SimpleFact, Tuple]
) -> "FluentKB"
Add multiple facts.
Source code in fuzzy_infer/fluent.py
| def facts(self, *facts: Union[FactTuple, SimpleFact, Tuple]) -> 'FluentKB':
"""Add multiple facts."""
for f in facts:
if len(f) == 2:
pred, args = f
degree = 1.0
elif len(f) == 3:
# Could be (pred, args, degree) or (pred, arg1, arg2) for multi-arg facts
if isinstance(f[2], (int, float)) and 0 <= f[2] <= 1:
# It's a degree
pred, args, degree = f
else:
# It's a multi-arg fact like ("species", ["rex"], "dog")
pred = f[0]
args = list(f[1:])
degree = 1.0
elif len(f) == 4:
# Multi-arg fact with degree like ("species", ["rex"], "dog", 1.0)
pred = f[0]
args = list(f[1:-1])
degree = f[-1]
else:
pred, args, degree = f[0], list(f[1:]), 1.0
self.fact(pred, args, degree)
return self
|
rule(
when: Optional[
Union[PatternMatcher, Tuple, Callable, List]
] = None,
then: Optional[
Union[PatternMatcher, Tuple, Callable, List]
] = None,
**kwargs
) -> "FluentKB"
Add a rule using keyword arguments or builder.
Source code in fuzzy_infer/fluent.py
| def rule(
self,
when: Optional[Union[PatternMatcher, Tuple, Callable, List]] = None,
then: Optional[Union[PatternMatcher, Tuple, Callable, List]] = None,
**kwargs
) -> 'FluentKB':
"""Add a rule using keyword arguments or builder."""
if when is not None and then is not None:
# Handle list of conditions
if isinstance(when, list):
conditions = when
else:
conditions = [when]
# Handle list of actions
if isinstance(then, list):
actions = then
else:
actions = [then]
# Simple rule creation
rule = FluentRule(conditions, actions, self)
if 'name' in kwargs:
rule.named(kwargs['name'])
if 'priority' in kwargs:
rule.with_priority(kwargs['priority'])
self._engine.add_rule(rule.build())
elif callable(when):
# Lambda-style rule - skip for now since it needs special handling
pass
elif isinstance(when, dict):
# Direct dict rule
self._engine.add_rule(when)
return self
|
rules(*rules: Union[Rule, FluentRule, Dict]) -> 'FluentKB'
Add multiple rules.
Source code in fuzzy_infer/fluent.py
| def rules(self, *rules: Union[Rule, FluentRule, Dict]) -> 'FluentKB':
"""Add multiple rules."""
for r in rules:
if isinstance(r, FluentRule):
self._engine.add_rule(r.build())
elif isinstance(r, Rule):
self._engine.add_rule(r)
elif isinstance(r, dict):
self._engine.add_rule(r)
return self
|
infer(max_iterations: int = 100) -> 'FluentKB'
Run inference.
Source code in fuzzy_infer/fluent.py
| def infer(self, max_iterations: int = 100) -> 'FluentKB':
"""Run inference."""
self._engine.run(max_iterations)
return self
|
query(
predicate: str,
args: Optional[Union[List[Any], Any]] = None,
) -> FluentQueryResult
Query facts.
Source code in fuzzy_infer/fluent.py
| def query(
self,
predicate: str,
args: Optional[Union[List[Any], Any]] = None
) -> FluentQueryResult:
"""Query facts."""
if args is not None:
if not isinstance(args, list):
args = [args]
facts = self._engine.query(predicate, args)
else:
facts = self._engine.query(predicate)
return FluentQueryResult(facts, self)
|
where(predicate: str) -> FluentQueryResult
Query all facts with predicate.
Source code in fuzzy_infer/fluent.py
| def where(self, predicate: str) -> FluentQueryResult:
"""Query all facts with predicate."""
facts = self._engine.query(predicate)
return FluentQueryResult(facts, self)
|
Clear everything.
Source code in fuzzy_infer/fluent.py
| def clear(self) -> 'FluentKB':
"""Clear everything."""
return self.clear_facts().clear_rules()
|
save(path: Union[str, Path]) -> 'FluentKB'
Save to file.
Source code in fuzzy_infer/fluent.py
| def save(self, path: Union[str, Path]) -> 'FluentKB':
"""Save to file."""
from fuzzy_infer.jsonl import KnowledgeBaseIO
KnowledgeBaseIO.save(self._engine, path)
return self
|
load(path: Union[str, Path]) -> 'FluentKB'
Load from file.
Source code in fuzzy_infer/fluent.py
| def load(self, path: Union[str, Path]) -> 'FluentKB':
"""Load from file."""
from fuzzy_infer.jsonl import KnowledgeBaseIO
loaded = KnowledgeBaseIO.load(path)
self._engine = loaded
return self
|
Examples
from fuzzy_infer.fluent import FluentKB
# Method chaining
kb = (
FluentKB()
.fact("is-mammal", ["dog"])
.fact("is-mammal", ["cat"])
.rule(
when=("is-mammal", "?x"),
then=("warm-blooded", "?x", 0.99)
)
.infer()
)
# Query results
warm_blooded = kb.where("warm-blooded")
for fact in warm_blooded:
print(f"{fact.args[0]} is warm-blooded ({fact.degree:.0%})")
# Pattern matching with subscript
kb = FluentKB()
kb.fact("temperature", ["room-1"], 0.8)
temp_pattern = kb["temperature"]["?room"] >= 0.7
kb.rule(when=temp_pattern, then=("ac-on", "?room", 0.95))
# Save/load
kb.save("knowledge.jsonl")
loaded_kb = FluentKB().load("knowledge.jsonl")
FuzzySession
Context manager for inference sessions.
Source code in fuzzy_infer/fluent.py
| def __init__(self, max_iterations: int = 100):
self.max_iterations = max_iterations
self.kb: Optional[FluentKB] = None
|
Enter session.
Source code in fuzzy_infer/fluent.py
| def __enter__(self) -> FluentKB:
"""Enter session."""
self.kb = FluentKB()
return self.kb
|
__exit__(exc_type, exc_val, exc_tb)
Exit session - auto-run inference if not already done.
Source code in fuzzy_infer/fluent.py
| def __exit__(self, exc_type, exc_val, exc_tb):
"""Exit session - auto-run inference if not already done."""
if self.kb and not exc_type:
# Check if inference was already run
if not self.kb._engine.inference_log:
self.kb.infer(self.max_iterations)
|
Examples
from fuzzy_infer.fluent import FuzzySession
# Auto-inference on context exit
with FuzzySession() as session:
session.facts(
("is-bird", ["robin"], 0.9),
("is-bird", ["eagle"], 1.0)
)
session.rule(
when=("is-bird", "?x"),
then=("can-fly", "?x", 0.95)
)
# Inference runs automatically here
flying = session.query("can-fly")
# Manual inference control
with FuzzySession(auto_infer=False) as session:
session.fact("is-bird", ["robin"], 0.9)
session.infer() # Explicit
results = session.query("can-fly")
FluentQueryResult
Fluent wrapper for query results with chainable operations.
Source code in fuzzy_infer/fluent.py
| def __init__(self, facts: List[Fact], kb: 'FluentKB'):
self._facts = facts
self._kb = kb
|
property
max_degree: Optional[float]
property
min_degree: Optional[float]
where(
condition: Callable[[Fact], bool],
) -> "FluentQueryResult"
Filter facts by condition.
Source code in fuzzy_infer/fluent.py
| def where(self, condition: Callable[[Fact], bool]) -> 'FluentQueryResult':
"""Filter facts by condition."""
filtered = [f for f in self._facts if condition(f)]
return FluentQueryResult(filtered, self._kb)
|
having_degree(op: str, value: float) -> 'FluentQueryResult'
Filter by degree comparison.
Source code in fuzzy_infer/fluent.py
| def having_degree(self, op: str, value: float) -> 'FluentQueryResult':
"""Filter by degree comparison."""
ops = {
'>': lambda d: d > value,
'>=': lambda d: d >= value,
'<': lambda d: d < value,
'<=': lambda d: d <= value,
'==': lambda d: d == value,
'!=': lambda d: d != value,
}
if op not in ops:
raise ValueError(f"Invalid operator: {op}")
return self.where(lambda f: ops[op](f.degree))
|
with_args(*args: Any) -> 'FluentQueryResult'
Filter by argument values.
Source code in fuzzy_infer/fluent.py
| def with_args(self, *args: Any) -> 'FluentQueryResult':
"""Filter by argument values."""
return self.where(lambda f: f.args == list(args))
|
map(func: Callable[[Fact], Any]) -> List[Any]
Map function over facts.
Source code in fuzzy_infer/fluent.py
| def map(self, func: Callable[[Fact], Any]) -> List[Any]:
"""Map function over facts."""
return [func(f) for f in self._facts]
|
to_dict() -> List[Dict[str, Any]]
Convert to list of dicts.
Source code in fuzzy_infer/fluent.py
| def to_dict(self) -> List[Dict[str, Any]]:
"""Convert to list of dicts."""
return [f.to_dict() for f in self._facts]
|
Convert to JSON string.
Source code in fuzzy_infer/fluent.py
| def to_json(self) -> str:
"""Convert to JSON string."""
return json.dumps(self.to_dict(), indent=2)
|
Examples
# Query with chaining
results = (
kb.where("species")
.with_args("dog")
.having_degree('>', 0.8)
)
# Functional operations
names = kb.where("person").map(lambda f: f.args[0])
# Degree filtering
high_conf = kb.query("is-mammal").having_degree('>=', 0.9)
# Get first result
first = kb.query("can-fly", ["tweety"]).first
if first:
print(f"Confidence: {first.degree}")
# Get statistics
results = kb.where("temperature")
print(f"Max degree: {results.max_degree}")
print(f"Min degree: {results.min_degree}")
print(f"All degrees: {results.degrees}")
# Convert to JSON
json_str = results.to_json()
dict_list = results.to_dict()
Pipeline
Inference pipeline builder for complex workflows.
Source code in fuzzy_infer/fluent.py
| def __init__(self):
self.steps: List[Callable] = []
self.kb = FluentKB()
|
facts_from(
source: Union[str, Path, Iterator],
) -> "Pipeline"
Load facts from source.
Source code in fuzzy_infer/fluent.py
| def facts_from(self, source: Union[str, Path, Iterator]) -> 'Pipeline':
"""Load facts from source."""
def load_facts(kb: FluentKB):
if isinstance(source, (str, Path)):
kb.load(source)
else:
for fact in source:
if isinstance(fact, Fact):
kb._engine.add_fact(fact)
elif isinstance(fact, tuple):
kb.fact(*fact)
self.steps.append(load_facts)
return self
|
rules_from(source: Union[str, Path, List]) -> 'Pipeline'
Load rules from source.
Source code in fuzzy_infer/fluent.py
| def rules_from(self, source: Union[str, Path, List]) -> 'Pipeline':
"""Load rules from source."""
def load_rules(kb: FluentKB):
if isinstance(source, (str, Path)):
# Load from file
pass
else:
kb.rules(*source)
self.steps.append(load_rules)
return self
|
transform(func: Callable[[FluentKB], None]) -> 'Pipeline'
Apply transformation to KB.
Source code in fuzzy_infer/fluent.py
| def transform(self, func: Callable[[FluentKB], None]) -> 'Pipeline':
"""Apply transformation to KB."""
self.steps.append(func)
return self
|
filter_facts(
predicate: Callable[[Fact], bool],
) -> "Pipeline"
Filter facts.
Source code in fuzzy_infer/fluent.py
| def filter_facts(self, predicate: Callable[[Fact], bool]) -> 'Pipeline':
"""Filter facts."""
def filter_step(kb: FluentKB):
filtered = {k: v for k, v in kb._engine.facts.items()
if predicate(v)}
kb._engine.facts = filtered
self.steps.append(filter_step)
return self
|
infer(max_iterations: int = 100) -> 'Pipeline'
Add inference step.
Source code in fuzzy_infer/fluent.py
| def infer(self, max_iterations: int = 100) -> 'Pipeline':
"""Add inference step."""
self.steps.append(lambda kb: kb.infer(max_iterations))
return self
|
Execute pipeline.
Source code in fuzzy_infer/fluent.py
| def run(self) -> FluentKB:
"""Execute pipeline."""
for step in self.steps:
step(self.kb)
return self.kb
|
Examples
from fuzzy_infer.fluent import Pipeline
# Complex workflow
pipeline = (
Pipeline()
.facts_from("sensors.jsonl")
.rules_from("rules.jsonl")
.transform(lambda kb: kb.fact("extra", ["data"], 0.5))
.filter_facts(lambda f: f.degree > 0.6)
.infer()
)
result_kb = pipeline.run()
# Multi-stage processing
pipeline = (
Pipeline()
.facts_from([
("temperature", ["room-1"], 0.8),
("humidity", ["room-1"], 0.6)
])
.transform(lambda kb: kb.rule(
when=("temperature", "?r"),
then=("hot", "?r", 0.9)
))
.infer()
)
Helper Functions
fuzzy()
Quick fuzzy KB creation.
Source code in fuzzy_infer/fluent.py
| def fuzzy(*facts: Union[FactTuple, SimpleFact]) -> FluentKB:
"""Quick fuzzy KB creation."""
kb = FluentKB()
return kb.facts(*facts)
|
from fuzzy_infer.fluent import fuzzy
# Quick KB creation
kb = fuzzy(
("is-cat", ["fluffy"], 1.0),
("has-fur", ["fluffy"], 0.95),
("likes-fish", ["fluffy"], 0.8)
)
# Returns FluentKB ready for chaining
results = fuzzy(
("is-bird", ["robin"], 0.9)
).rule(
when=("is-bird", "?x"),
then=("can-fly", "?x", 0.85)
).infer().where("can-fly")
when() and then()
Create pattern matcher for conditions.
Source code in fuzzy_infer/fluent.py
| def when(predicate: str, *args: Any) -> PatternMatcher:
"""Create pattern matcher for conditions."""
pattern = PatternMatcher(predicate)
if args:
pattern.args = list(args)
return pattern
|
Create action tuple.
Source code in fuzzy_infer/fluent.py
| def then(predicate: str, *args: Any, degree: float = 1.0) -> Tuple:
"""Create action tuple."""
# If degree is passed as last positional argument
if args and isinstance(args[-1], (int, float)) and 0 <= args[-1] <= 1:
actual_args = list(args[:-1])
actual_degree = args[-1]
return (predicate, actual_args, actual_degree)
return (predicate, list(args), degree)
|
from fuzzy_infer.fluent import when, then
# Create pattern helpers
condition = when("is-mammal", "?x")
action = then("warm-blooded", "?x", 0.99)
kb.rule(when=condition, then=action)
# Multiple conditions
kb.rule(
when=[
when("is-bird", "?x"),
when("has-wings", "?x")
],
then=then("can-fly", "?x", 0.9)
)
stream_session()
Context manager for stream processing.
Source code in fuzzy_infer/fluent.py
| @contextmanager
def stream_session(input_file: Optional[Path] = None, output_file: Optional[Path] = None):
"""Context manager for stream processing."""
import sys
old_stdin = sys.stdin
old_stdout = sys.stdout
try:
if input_file:
sys.stdin = open(input_file, 'r')
if output_file:
sys.stdout = open(output_file, 'w')
kb = FluentKB()
yield kb
finally:
if input_file and sys.stdin != old_stdin:
sys.stdin.close()
sys.stdin = old_stdin
if output_file and sys.stdout != old_stdout:
sys.stdout.close()
sys.stdout = old_stdout
|
from fuzzy_infer.fluent import stream_session
# Streaming inference
def sensor_stream():
import random
while True:
yield ("sensor", [f"s{random.randint(1,10)}"], random.random())
with stream_session() as kb:
for fact in sensor_stream():
kb.fact(*fact)
if fact[2] > 0.9:
kb.fact("alert", fact[1], 0.95)
kb.infer(max_iterations=1)
alerts = kb.where("alert")
if alerts:
print(f"Alert: {alerts.first}")
Operator Overloading
Subscript Access
# Pattern creation
kb = FluentKB()
pattern = kb["temperature"]["?room"] # Matches temperature(?room)
# With arguments
pattern = kb["has-skill"]["?person", "python"]
# Attribute access (underscores → hyphens)
pattern = kb.is_mammal["?x"] # Matches is-mammal(?x)
Comparison Operators
# Degree constraints
high_temp = kb["temperature"]["?x"] > 0.8
low_temp = kb["temperature"]["?x"] <= 0.3
exact = kb["confidence"]["?x"] == 0.95
# Use in rules
kb.rule(when=high_temp, then=("ac-on", "?x", 0.95))
Rule Creation with >>
# Natural rule syntax
rule = kb["is-bird"]["?x"] >> ("can-fly", "?x", 0.9)
# Equivalent to:
kb.rule(
when=("is-bird", "?x"),
then=("can-fly", "?x", 0.9)
)
Logical Operators
# AND conditions
cold = kb["temperature"]["?x"] < 0.3
dry = kb["humidity"]["?x"] < 0.4
desert = cold & dry
# NOT operation
not_hot = ~kb["temperature"]["?x"]
Complete Examples
Animal Classification
from fuzzy_infer.fluent import FluentKB
kb = (
FluentKB()
.facts(
("has-fur", ["rex"], 1.0),
("barks", ["rex"], 0.95),
("has-wings", ["tweety"], 1.0),
("tweets", ["tweety"], 0.9)
)
.rule(
when=[("has-fur", "?x"), ("barks", "?x")],
then=("is-dog", "?x", 0.95)
)
.rule(
when=[("has-wings", "?x"), ("tweets", "?x")],
then=("is-bird", "?x", 0.9)
)
.infer()
)
# Check classifications
for animal in ["rex", "tweety"]:
dog = kb.query("is-dog", [animal]).first
bird = kb.query("is-bird", [animal]).first
if dog:
print(f"{animal} is a dog ({dog.degree:.0%})")
elif bird:
print(f"{animal} is a bird ({bird.degree:.0%})")
IoT Sensor Management
from fuzzy_infer.fluent import FuzzySession
with FuzzySession() as kb:
kb.facts(
("sensor", ["temp-1"], 0.75),
("sensor", ["humidity-1"], 0.60),
("sensor", ["pressure-1"], 0.45)
)
kb.rule(
when=kb["sensor"]["?s"] > 0.7,
then=("alert", "?s", 0.9),
name="high-reading-alert"
).rule(
when=kb["sensor"]["?s"] < 0.3,
then=("maintenance", "?s", 0.8),
name="low-reading-check"
)
alerts = kb.infer().where("alert")
for alert in alerts:
print(f"Alert: {alert.args[0]} at {alert.degree:.0%}")
Functional Style
from fuzzy_infer.fluent import fuzzy
kb = fuzzy(
("age", ["alice"], 0.3), # Young
("age", ["bob"], 0.7), # Middle-aged
("age", ["charlie"], 0.95) # Elderly
)
kb.rule(
when=("age", "?person"),
then=lambda facts: [
("life-stage", facts["?person"], "young", 1.0 - facts.degree),
("life-stage", facts["?person"], "old", facts.degree)
]
)
kb.infer()
young_people = (
kb.query("life-stage")
.where(lambda f: f.args[1] == "young" and f.degree > 0.5)
.map(lambda f: f.args[0])
)
print(f"Young people: {young_people}")
Best Practices
1. Use Context Managers
# Good - automatic cleanup
with FuzzySession() as kb:
kb.facts(data)
results = kb.query("target")
# Avoid - manual management
kb = FluentKB()
kb.facts(data)
# ... might forget to clean up
2. Chain Methods for Readability
# Good - clear flow
results = (
kb.facts(data)
.rules(rules)
.infer()
.where("target")
)
# Avoid - harder to follow
kb.facts(data)
kb.rules(rules)
kb.infer()
results = kb.where("target")
3. Use Helper Functions
# Good - clear intent
kb.rule(
when=when("is-bird", "?x"),
then=then("can-fly", "?x", 0.9)
)
# Also good but more verbose
kb.rule(
when=("is-bird", "?x"),
then=("can-fly", "?x", 0.9)
)
4. Leverage Query Chaining
# Powerful filtering
experts = (
kb.where("has-skill")
.with_args("python")
.having_degree('>', 0.8)
.map(lambda f: {"name": f.args[0], "level": f.degree})
)
Type Hints
from fuzzy_infer.fluent import FluentKB, FluentQueryResult, FuzzySession
from typing import List
def process_kb(kb: FluentKB) -> FluentQueryResult:
return kb.infer().where("result")
def with_session() -> List[Fact]:
with FuzzySession() as session:
session.fact("test", ["data"])
return session.query("test").facts