Skip to content

Core Module API

fuzzy_infer.core

Core inference engine for the FuzzyInfer system.

This module contains the main FuzzyInfer class that implements forward-chaining fuzzy inference with proper type hints and modern Python patterns.

FuzzyInfer 

FuzzyInfer(max_iterations: int = 100)

Production rule system for fuzzy inference.

This class implements a forward-chaining inference engine with fuzzy logic support, allowing for reasoning with uncertain information and degrees of belief.

Attributes:

  • facts (Dict[Tuple[str, Tuple], Fact]) –

    Set of facts in the knowledge base

  • rules (List[Rule]) –

    List of production rules

  • inference_log (List[str]) –

    Log of inference steps for debugging

  • max_iterations –

    Maximum iterations for inference loop

Examples:

>>> # Basic usage
>>> inf = FuzzyInfer()
>>> inf.add_fact(Fact('is-zebra', ['sam'], 0.8))
>>> inf.add_rule(RuleBuilder()
...     .when('is-zebra', ['?x'])
...     .with_degree_greater_than(0.5)
...     .then_add('has-stripes', ['?x'])
...     .with_degree_multiplied_by(0.9)
...     .build())
>>> inf.run()
>>> results = inf.query('has-stripes', ['sam'])

>>> # Fluent API
>>> inf = (FuzzyInfer()
...     .add_fact(Fact('is-person', ['alice'], 1.0))
...     .add_fact(Fact('is-tall', ['alice'], 0.7))
...     .add_rule(tall_person_rule)
...     .run())

>>> # Context manager
>>> with FuzzyInfer.session() as inf:
...     inf.add_facts(facts)
...     inf.add_rules(rules)
...     inf.run()
...     results = inf.query('target-pred', ['arg'])

Initialize the FuzzyInfer inference engine.

Parameters:

  • max_iterations (int, default: 100 ) –

    Maximum number of inference iterations

Source code in fuzzy_infer/core.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
def __init__(self, max_iterations: int = 100):
    """
    Initialize the FuzzyInfer inference engine.

    Args:
        max_iterations: Maximum number of inference iterations
    """
    self.facts: Dict[Tuple[str, Tuple], Fact] = {}
    self.rules: List[Rule] = []
    self.inference_log: List[str] = []
    self.max_iterations = max_iterations
    self._iteration_count = 0
    self._rule_history: Set[Tuple] = set()

session classmethod 

session(**kwargs) -> Generator[FuzzyInfer, None, None]

Create a FuzzyInfer session as a context manager.

Yields:

  • FuzzyInfer ( FuzzyInfer ) –

    Configured inference engine

Examples:

>>> with FuzzyInfer.session(max_iterations=50) as inf:
...     inf.add_facts(facts)
...     results = inf.run()
Source code in fuzzy_infer/core.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@classmethod
@contextmanager
def session(cls, **kwargs) -> Generator['FuzzyInfer', None, None]:
    """
    Create a FuzzyInfer session as a context manager.

    Yields:
        FuzzyInfer: Configured inference engine

    Examples:
        >>> with FuzzyInfer.session(max_iterations=50) as inf:
        ...     inf.add_facts(facts)
        ...     results = inf.run()
    """
    engine = cls(**kwargs)
    try:
        yield engine
    finally:
        logger.info(f"Inference session completed with {len(engine.facts)} facts")

add_fact 

add_fact(fact: Union[Fact, Dict[str, Any]]) -> FuzzyInfer

Add a single fact to the knowledge base.

Parameters:

  • fact (Union[Fact, Dict[str, Any]]) –

    Fact object or dictionary representation

Returns:

Examples:

>>> inf.add_fact(Fact('is-person', ['sam'], 1.0))
>>> inf.add_fact({'pred': 'is-tall', 'args': ['sam'], 'deg': 0.8})
Source code in fuzzy_infer/core.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def add_fact(self, fact: Union[Fact, Dict[str, Any]]) -> "FuzzyInfer":
    """
    Add a single fact to the knowledge base.

    Args:
        fact: Fact object or dictionary representation

    Returns:
        Self for method chaining

    Examples:
        >>> inf.add_fact(Fact('is-person', ['sam'], 1.0))
        >>> inf.add_fact({'pred': 'is-tall', 'args': ['sam'], 'deg': 0.8})
    """
    if isinstance(fact, dict):
        fact = Fact.from_dict(fact)

    key = (fact.predicate, tuple(fact.args))

    # Fuzzy OR: take maximum degree if fact already exists
    if key in self.facts:
        existing_fact = self.facts[key]
        if fact.degree > existing_fact.degree:
            self.facts[key] = fact
            logger.debug(f"Updated fact {key} degree: {existing_fact.degree} -> {fact.degree}")
    else:
        self.facts[key] = fact
        logger.debug(f"Added fact: {fact.predicate}{fact.args} [deg={fact.degree}]")

    return self

add_facts 

add_facts(
    facts: List[Union[Fact, Dict[str, Any]]],
) -> FuzzyInfer

Add multiple facts to the knowledge base.

Parameters:

  • facts (List[Union[Fact, Dict[str, Any]]]) –

    List of Fact objects or dictionary representations

Returns:

Source code in fuzzy_infer/core.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
def add_facts(self, facts: List[Union[Fact, Dict[str, Any]]]) -> "FuzzyInfer":
    """
    Add multiple facts to the knowledge base.

    Args:
        facts: List of Fact objects or dictionary representations

    Returns:
        Self for method chaining
    """
    for fact in facts:
        self.add_fact(fact)
    return self

add_rule 

add_rule(rule: Union[Rule, Dict[str, Any]]) -> FuzzyInfer

Add a single rule to the rule base.

Parameters:

  • rule (Union[Rule, Dict[str, Any]]) –

    Rule object or dictionary representation

Returns:

Source code in fuzzy_infer/core.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def add_rule(self, rule: Union[Rule, Dict[str, Any]]) -> "FuzzyInfer":
    """
    Add a single rule to the rule base.

    Args:
        rule: Rule object or dictionary representation

    Returns:
        Self for method chaining
    """
    if isinstance(rule, dict):
        rule = Rule.from_dict(rule)

    self.rules.append(rule)
    # Sort rules by priority (higher first)
    self.rules.sort(key=lambda r: r.priority, reverse=True)
    logger.debug(f"Added rule: {rule.name or f'Rule#{len(self.rules)}'}")

    return self

add_rules 

add_rules(
    rules: List[Union[Rule, Dict[str, Any]]],
) -> FuzzyInfer

Add multiple rules to the rule base.

Parameters:

  • rules (List[Union[Rule, Dict[str, Any]]]) –

    List of Rule objects or dictionary representations

Returns:

Source code in fuzzy_infer/core.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
def add_rules(self, rules: List[Union[Rule, Dict[str, Any]]]) -> "FuzzyInfer":
    """
    Add multiple rules to the rule base.

    Args:
        rules: List of Rule objects or dictionary representations

    Returns:
        Self for method chaining
    """
    for rule in rules:
        self.add_rule(rule)
    return self

run 

run(max_iterations: Optional[int] = None) -> FuzzyInfer

Execute the forward-chaining inference process.

Parameters:

  • max_iterations (Optional[int], default: None ) –

    Override default max iterations

Returns:

Raises:

  • InferenceError –

    If maximum iterations exceeded

Source code in fuzzy_infer/core.py 
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
def run(self, max_iterations: Optional[int] = None) -> "FuzzyInfer":
    """
    Execute the forward-chaining inference process.

    Args:
        max_iterations: Override default max iterations

    Returns:
        Self for method chaining

    Raises:
        InferenceError: If maximum iterations exceeded
    """
    max_iter = max_iterations or self.max_iterations
    self._iteration_count = 0
    facts_before = len(self.facts)

    while self._iteration_count < max_iter:
        self._iteration_count += 1
        changed = False

        for rule in self.rules:
            # Check if conditions are satisfied
            all_bindings = self._satisfies_conditions(rule.conditions)

            for bindings in all_bindings:
                # Create a unique key for this rule application
                rule_key = (id(rule), tuple(sorted(bindings.items())))

                # Skip if we've already applied this exact rule with these bindings
                if rule_key in self._rule_history:
                    continue

                self._rule_history.add(rule_key)

                # Apply actions
                for action in rule.actions:
                    self._apply_action(action, bindings)

                changed = True

                if rule.name:
                    logger.info(f"Fired rule: {rule.name} with bindings {bindings}")

        if not changed:
            break

    if self._iteration_count >= max_iter:
        raise InferenceError(
            f"Inference did not converge after {max_iter} iterations"
        )

    facts_after = len(self.facts)
    logger.info(
        f"Inference completed in {self._iteration_count} iterations. "
        f"Facts: {facts_before} -> {facts_after}"
    )

    return self

run_incremental 

run_incremental(
    max_iterations: Optional[int] = None,
) -> Iterator[InferenceResult]

Execute inference incrementally, yielding results.

Parameters:

  • max_iterations (Optional[int], default: None ) –

    Override default max iterations

Yields:

Source code in fuzzy_infer/core.py 
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
def run_incremental(self, max_iterations: Optional[int] = None) -> Iterator[InferenceResult]:
    """Execute inference incrementally, yielding results.

    Args:
        max_iterations: Override default max iterations

    Yields:
        InferenceResult for each iteration with new facts
    """
    max_iter = max_iterations or self.max_iterations
    iteration = 0

    while iteration < max_iter:
        initial_count = len(self.facts)
        fired_rules = []
        changed = False

        for rule in self.rules:
            all_bindings = self._satisfies_conditions(rule.conditions)

            for bindings in all_bindings:
                rule_key = (id(rule), tuple(sorted(bindings.items())))

                if rule_key in self._rule_history:
                    continue

                self._rule_history.add(rule_key)

                for action in rule.actions:
                    self._apply_action(action, bindings)

                changed = True
                if rule.name:
                    fired_rules.append(rule.name)

        # Collect new facts
        current_facts = list(self.facts.values())
        if len(current_facts) > initial_count:
            new_facts = current_facts[initial_count:]
            yield InferenceResult(
                new_facts=new_facts,
                fired_rules=list(set(fired_rules)),
                iteration=iteration
            )
        elif not changed:
            break

        iteration += 1

stream_process 

stream_process(
    fact_stream: Iterator[Fact],
    window_size: int = 100,
    max_iterations_per_window: int = 10,
) -> Iterator[Fact]

Process a stream of facts, yielding inferred facts.

Parameters:

  • fact_stream (Iterator[Fact]) –

    Iterator of input facts

  • window_size (int, default: 100 ) –

    Number of facts to accumulate before inference

  • max_iterations_per_window (int, default: 10 ) –

    Max iterations per window

Yields:

  • Fact –

    Newly inferred facts

Source code in fuzzy_infer/core.py 
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
def stream_process(self, fact_stream: Iterator[Fact],
                   window_size: int = 100,
                   max_iterations_per_window: int = 10) -> Iterator[Fact]:
    """Process a stream of facts, yielding inferred facts.

    Args:
        fact_stream: Iterator of input facts
        window_size: Number of facts to accumulate before inference
        max_iterations_per_window: Max iterations per window

    Yields:
        Newly inferred facts
    """
    window = []

    for fact in fact_stream:
        self.add_fact(fact)
        window.append(fact)

        if len(window) >= window_size:
            # Run inference on current KB
            initial_facts = set(self.facts.values())
            self.run(max_iterations=max_iterations_per_window)

            # Yield new facts
            for f in self.facts.values():
                if f not in initial_facts:
                    yield f

            window = []

    # Process remaining window
    if window:
        initial_facts = set(self.facts.values())
        self.run(max_iterations=max_iterations_per_window)
        for f in self.facts.values():
            if f not in initial_facts:
                yield f

query 

query(
    predicate: str,
    args: Optional[List[Any]] = None,
    min_degree: float = 0.0,
) -> List[Fact]

Query the knowledge base for matching facts.

Parameters:

  • predicate (str) –

    Predicate to search for

  • args (Optional[List[Any]], default: None ) –

    Arguments to match (can contain variables like '?x')

  • min_degree (float, default: 0.0 ) –

    Minimum degree threshold

Returns:

  • List[Fact] –

    List of matching facts

Examples:

>>> # Find all zebras
>>> inf.query('is-zebra')

>>> # Find specific fact
>>> inf.query('is-zebra', ['sam'])

>>> # Find with minimum certainty
>>> inf.query('is-zebra', min_degree=0.7)
Source code in fuzzy_infer/core.py 
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
def query(
    self, predicate: str, args: Optional[List[Any]] = None, min_degree: float = 0.0
) -> List[Fact]:
    """
    Query the knowledge base for matching facts.

    Args:
        predicate: Predicate to search for
        args: Arguments to match (can contain variables like '?x')
        min_degree: Minimum degree threshold

    Returns:
        List of matching facts

    Examples:
        >>> # Find all zebras
        >>> inf.query('is-zebra')

        >>> # Find specific fact
        >>> inf.query('is-zebra', ['sam'])

        >>> # Find with minimum certainty
        >>> inf.query('is-zebra', min_degree=0.7)
    """
    results = []
    pattern = Fact(predicate, args or [], 1.0)

    for fact_key, fact in self.facts.items():
        if fact.predicate != predicate:
            continue

        if fact.degree < min_degree:
            continue

        if args:
            matched, _ = fact.matches_pattern(pattern)
            if not matched:
                continue

        results.append(fact)

    return results

ask 

ask(
    conditions: List[Union[List, Dict[str, Any]]],
) -> List[Dict[str, Any]]

Ask a question and get all matching variable bindings.

Parameters:

  • conditions (List[Union[List, Dict[str, Any]]]) –

    List of conditions to match

Returns:

  • List[Dict[str, Any]] –

    List of all satisfying variable bindings

Examples:

>>> # Who is a tall person?
>>> inf.ask([
...     {'pred': 'is-person', 'args': ['?x']},
...     {'pred': 'is-tall', 'args': ['?x']}
... ])
Source code in fuzzy_infer/core.py 
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
def ask(self, conditions: List[Union[List, Dict[str, Any]]]) -> List[Dict[str, Any]]:
    """
    Ask a question and get all matching variable bindings.

    Args:
        conditions: List of conditions to match

    Returns:
        List of all satisfying variable bindings

    Examples:
        >>> # Who is a tall person?
        >>> inf.ask([
        ...     {'pred': 'is-person', 'args': ['?x']},
        ...     {'pred': 'is-tall', 'args': ['?x']}
        ... ])
    """
    # Convert list format to Condition objects
    condition_objs = []
    for cond in conditions:
        if isinstance(cond, list):
            # Convert ['pred', ['arg1', 'arg2']] format
            if len(cond) >= 2:
                condition_objs.append(Condition(predicate=cond[0], args=cond[1]))
        elif isinstance(cond, dict):
            condition_objs.append(Condition.from_dict(cond))

    # Run inference first
    self.run()

    # Find all matching bindings
    return self._satisfies_conditions(condition_objs)

explain 

explain(fact: Union[Fact, Dict[str, Any]]) -> List[str]

Explain how a fact was derived.

Parameters:

  • fact (Union[Fact, Dict[str, Any]]) –

    Fact to explain

Returns:

  • List[str] –

    List of explanation strings

Source code in fuzzy_infer/core.py 
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
def explain(self, fact: Union[Fact, Dict[str, Any]]) -> List[str]:
    """
    Explain how a fact was derived.

    Args:
        fact: Fact to explain

    Returns:
        List of explanation strings
    """
    # This would require tracking provenance during inference
    # For now, return a simple message
    if isinstance(fact, dict):
        fact = Fact.from_dict(fact)

    key = (fact.predicate, tuple(fact.args))
    if key in self.facts:
        return [f"Fact {fact.predicate}{fact.args} exists with degree {self.facts[key].degree}"]
    else:
        return [f"Fact {fact.predicate}{fact.args} not found in knowledge base"]

clear 

clear() -> FuzzyInfer

Clear all facts and rules.

Source code in fuzzy_infer/core.py 
726
727
728
729
730
731
732
def clear(self) -> "FuzzyInfer":
    """Clear all facts and rules."""
    self.facts.clear()
    self.rules.clear()
    self._rule_history.clear()
    self.inference_log.clear()
    return self

get_facts 

get_facts() -> List[Fact]

Get all facts in the knowledge base.

Source code in fuzzy_infer/core.py 
734
735
736
def get_facts(self) -> List[Fact]:
    """Get all facts in the knowledge base."""
    return list(self.facts.values())

get_rules 

get_rules() -> List[Rule]

Get all rules in the rule base.

Source code in fuzzy_infer/core.py 
738
739
740
def get_rules(self) -> List[Rule]:
    """Get all rules in the rule base."""
    return self.rules.copy()

InferenceResult dataclass 

InferenceResult(
    new_facts: List[Fact],
    fired_rules: List[str],
    iteration: int,
)

Result of an inference step.

Overview

The fuzzy_infer.core module contains the main inference engine implementing forward-chaining fuzzy logic reasoning.

Quick Reference

from fuzzy_infer import FuzzyInfer, Fact, Rule

# Create engine
inf = FuzzyInfer(max_iterations=100)

# Add facts
inf.add_fact(Fact("is-bird", ["tweety"], 0.9))
inf.add_facts([
    Fact("has-wings", ["tweety"], 1.0),
    Fact("can-fly", ["eagle"], 0.95)
])

# Add rules
inf.add_rule(bird_rule)
inf.add_rules([rule1, rule2, rule3])

# Run inference
result = inf.run()

# Query results
facts = inf.query("can-fly")
tweety = inf.query("can-fly", ["tweety"])

# Context manager
with FuzzyInfer.session() as inf:
    inf.add_facts(facts)
    inf.run()

Class Methods

FuzzyInfer

Production rule system for fuzzy inference.

This class implements a forward-chaining inference engine with fuzzy logic support, allowing for reasoning with uncertain information and degrees of belief.

Attributes:

  • facts (Dict[Tuple[str, Tuple], Fact]) –

    Set of facts in the knowledge base

  • rules (List[Rule]) –

    List of production rules

  • inference_log (List[str]) –

    Log of inference steps for debugging

  • max_iterations –

    Maximum iterations for inference loop

Examples:

>>> # Basic usage
>>> inf = FuzzyInfer()
>>> inf.add_fact(Fact('is-zebra', ['sam'], 0.8))
>>> inf.add_rule(RuleBuilder()
...     .when('is-zebra', ['?x'])
...     .with_degree_greater_than(0.5)
...     .then_add('has-stripes', ['?x'])
...     .with_degree_multiplied_by(0.9)
...     .build())
>>> inf.run()
>>> results = inf.query('has-stripes', ['sam'])

>>> # Fluent API
>>> inf = (FuzzyInfer()
...     .add_fact(Fact('is-person', ['alice'], 1.0))
...     .add_fact(Fact('is-tall', ['alice'], 0.7))
...     .add_rule(tall_person_rule)
...     .run())

>>> # Context manager
>>> with FuzzyInfer.session() as inf:
...     inf.add_facts(facts)
...     inf.add_rules(rules)
...     inf.run()
...     results = inf.query('target-pred', ['arg'])

Initialize the FuzzyInfer inference engine.

Parameters:

  • max_iterations (int, default: 100 ) –

    Maximum number of inference iterations

Source code in fuzzy_infer/core.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
def __init__(self, max_iterations: int = 100):
    """
    Initialize the FuzzyInfer inference engine.

    Args:
        max_iterations: Maximum number of inference iterations
    """
    self.facts: Dict[Tuple[str, Tuple], Fact] = {}
    self.rules: List[Rule] = []
    self.inference_log: List[str] = []
    self.max_iterations = max_iterations
    self._iteration_count = 0
    self._rule_history: Set[Tuple] = set()

session classmethod 

session(**kwargs) -> Generator[FuzzyInfer, None, None]

Create a FuzzyInfer session as a context manager.

Yields:

  • FuzzyInfer ( FuzzyInfer ) –

    Configured inference engine

Examples:

>>> with FuzzyInfer.session(max_iterations=50) as inf:
...     inf.add_facts(facts)
...     results = inf.run()
Source code in fuzzy_infer/core.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@classmethod
@contextmanager
def session(cls, **kwargs) -> Generator['FuzzyInfer', None, None]:
    """
    Create a FuzzyInfer session as a context manager.

    Yields:
        FuzzyInfer: Configured inference engine

    Examples:
        >>> with FuzzyInfer.session(max_iterations=50) as inf:
        ...     inf.add_facts(facts)
        ...     results = inf.run()
    """
    engine = cls(**kwargs)
    try:
        yield engine
    finally:
        logger.info(f"Inference session completed with {len(engine.facts)} facts")

add_fact 

add_fact(fact: Union[Fact, Dict[str, Any]]) -> FuzzyInfer

Add a single fact to the knowledge base.

Parameters:

  • fact (Union[Fact, Dict[str, Any]]) –

    Fact object or dictionary representation

Returns:

Examples:

>>> inf.add_fact(Fact('is-person', ['sam'], 1.0))
>>> inf.add_fact({'pred': 'is-tall', 'args': ['sam'], 'deg': 0.8})
Source code in fuzzy_infer/core.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def add_fact(self, fact: Union[Fact, Dict[str, Any]]) -> "FuzzyInfer":
    """
    Add a single fact to the knowledge base.

    Args:
        fact: Fact object or dictionary representation

    Returns:
        Self for method chaining

    Examples:
        >>> inf.add_fact(Fact('is-person', ['sam'], 1.0))
        >>> inf.add_fact({'pred': 'is-tall', 'args': ['sam'], 'deg': 0.8})
    """
    if isinstance(fact, dict):
        fact = Fact.from_dict(fact)

    key = (fact.predicate, tuple(fact.args))

    # Fuzzy OR: take maximum degree if fact already exists
    if key in self.facts:
        existing_fact = self.facts[key]
        if fact.degree > existing_fact.degree:
            self.facts[key] = fact
            logger.debug(f"Updated fact {key} degree: {existing_fact.degree} -> {fact.degree}")
    else:
        self.facts[key] = fact
        logger.debug(f"Added fact: {fact.predicate}{fact.args} [deg={fact.degree}]")

    return self

add_facts 

add_facts(
    facts: List[Union[Fact, Dict[str, Any]]],
) -> FuzzyInfer

Add multiple facts to the knowledge base.

Parameters:

  • facts (List[Union[Fact, Dict[str, Any]]]) –

    List of Fact objects or dictionary representations

Returns:

Source code in fuzzy_infer/core.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
def add_facts(self, facts: List[Union[Fact, Dict[str, Any]]]) -> "FuzzyInfer":
    """
    Add multiple facts to the knowledge base.

    Args:
        facts: List of Fact objects or dictionary representations

    Returns:
        Self for method chaining
    """
    for fact in facts:
        self.add_fact(fact)
    return self

add_rule 

add_rule(rule: Union[Rule, Dict[str, Any]]) -> FuzzyInfer

Add a single rule to the rule base.

Parameters:

  • rule (Union[Rule, Dict[str, Any]]) –

    Rule object or dictionary representation

Returns:

Source code in fuzzy_infer/core.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def add_rule(self, rule: Union[Rule, Dict[str, Any]]) -> "FuzzyInfer":
    """
    Add a single rule to the rule base.

    Args:
        rule: Rule object or dictionary representation

    Returns:
        Self for method chaining
    """
    if isinstance(rule, dict):
        rule = Rule.from_dict(rule)

    self.rules.append(rule)
    # Sort rules by priority (higher first)
    self.rules.sort(key=lambda r: r.priority, reverse=True)
    logger.debug(f"Added rule: {rule.name or f'Rule#{len(self.rules)}'}")

    return self

add_rules 

add_rules(
    rules: List[Union[Rule, Dict[str, Any]]],
) -> FuzzyInfer

Add multiple rules to the rule base.

Parameters:

  • rules (List[Union[Rule, Dict[str, Any]]]) –

    List of Rule objects or dictionary representations

Returns:

Source code in fuzzy_infer/core.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
def add_rules(self, rules: List[Union[Rule, Dict[str, Any]]]) -> "FuzzyInfer":
    """
    Add multiple rules to the rule base.

    Args:
        rules: List of Rule objects or dictionary representations

    Returns:
        Self for method chaining
    """
    for rule in rules:
        self.add_rule(rule)
    return self

query 

query(
    predicate: str,
    args: Optional[List[Any]] = None,
    min_degree: float = 0.0,
) -> List[Fact]

Query the knowledge base for matching facts.

Parameters:

  • predicate (str) –

    Predicate to search for

  • args (Optional[List[Any]], default: None ) –

    Arguments to match (can contain variables like '?x')

  • min_degree (float, default: 0.0 ) –

    Minimum degree threshold

Returns:

  • List[Fact] –

    List of matching facts

Examples:

>>> # Find all zebras
>>> inf.query('is-zebra')

>>> # Find specific fact
>>> inf.query('is-zebra', ['sam'])

>>> # Find with minimum certainty
>>> inf.query('is-zebra', min_degree=0.7)
Source code in fuzzy_infer/core.py 
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
def query(
    self, predicate: str, args: Optional[List[Any]] = None, min_degree: float = 0.0
) -> List[Fact]:
    """
    Query the knowledge base for matching facts.

    Args:
        predicate: Predicate to search for
        args: Arguments to match (can contain variables like '?x')
        min_degree: Minimum degree threshold

    Returns:
        List of matching facts

    Examples:
        >>> # Find all zebras
        >>> inf.query('is-zebra')

        >>> # Find specific fact
        >>> inf.query('is-zebra', ['sam'])

        >>> # Find with minimum certainty
        >>> inf.query('is-zebra', min_degree=0.7)
    """
    results = []
    pattern = Fact(predicate, args or [], 1.0)

    for fact_key, fact in self.facts.items():
        if fact.predicate != predicate:
            continue

        if fact.degree < min_degree:
            continue

        if args:
            matched, _ = fact.matches_pattern(pattern)
            if not matched:
                continue

        results.append(fact)

    return results

run 

run(max_iterations: Optional[int] = None) -> FuzzyInfer

Execute the forward-chaining inference process.

Parameters:

  • max_iterations (Optional[int], default: None ) –

    Override default max iterations

Returns:

Raises:

  • InferenceError –

    If maximum iterations exceeded

Source code in fuzzy_infer/core.py 
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
def run(self, max_iterations: Optional[int] = None) -> "FuzzyInfer":
    """
    Execute the forward-chaining inference process.

    Args:
        max_iterations: Override default max iterations

    Returns:
        Self for method chaining

    Raises:
        InferenceError: If maximum iterations exceeded
    """
    max_iter = max_iterations or self.max_iterations
    self._iteration_count = 0
    facts_before = len(self.facts)

    while self._iteration_count < max_iter:
        self._iteration_count += 1
        changed = False

        for rule in self.rules:
            # Check if conditions are satisfied
            all_bindings = self._satisfies_conditions(rule.conditions)

            for bindings in all_bindings:
                # Create a unique key for this rule application
                rule_key = (id(rule), tuple(sorted(bindings.items())))

                # Skip if we've already applied this exact rule with these bindings
                if rule_key in self._rule_history:
                    continue

                self._rule_history.add(rule_key)

                # Apply actions
                for action in rule.actions:
                    self._apply_action(action, bindings)

                changed = True

                if rule.name:
                    logger.info(f"Fired rule: {rule.name} with bindings {bindings}")

        if not changed:
            break

    if self._iteration_count >= max_iter:
        raise InferenceError(
            f"Inference did not converge after {max_iter} iterations"
        )

    facts_after = len(self.facts)
    logger.info(
        f"Inference completed in {self._iteration_count} iterations. "
        f"Facts: {facts_before} -> {facts_after}"
    )

    return self

clear 

clear() -> FuzzyInfer

Clear all facts and rules.

Source code in fuzzy_infer/core.py 
726
727
728
729
730
731
732
def clear(self) -> "FuzzyInfer":
    """Clear all facts and rules."""
    self.facts.clear()
    self.rules.clear()
    self._rule_history.clear()
    self.inference_log.clear()
    return self

InferenceResult

Result of an inference step.

Examples

Basic Usage

from fuzzy_infer import FuzzyInfer, Fact, Rule

inf = FuzzyInfer()

# Add facts
inf.add_fact(Fact("is-mammal", ["dog"], 0.95))
inf.add_fact(Fact("has-fur", ["dog"], 1.0))

# Add rule
inf.add_rule(Rule(
    name="mammals-warm-blooded",
    conditions=[{"pred": "is-mammal", "args": ["?x"]}],
    actions=[{"action": "add", "fact": {"pred": "warm-blooded", "args": ["?x"], "deg": 0.99}}]
))

# Run and query
inf.run()
print(inf.query("warm-blooded", ["dog"]))

With Context Manager

with FuzzyInfer.session(max_iterations=50) as inf:
    inf.add_facts([
        Fact("is-bird", ["robin"], 0.9),
        Fact("has-wings", ["robin"], 1.0)
    ])
    inf.add_rule(birds_fly_rule)
    result = inf.run()
    print(f"Completed in {result.iteration} iterations")

Fluent API

results = (
    FuzzyInfer()
    .add_fact(Fact("is-bird", ["eagle"], 1.0))
    .add_rule(bird_rule)
    .run()
    .query("can-fly")
)