# How to Use the Policy Engine for Compliance Rule Enforcement in Semantica

> Learn how to enforce compliance rules with the Semantica Policy Engine. Discover its backend-agnostic knowledge graph, audit trails, and exception handling for robust rule enforcement.

- Repository: [Semantica /semantica](https://github.com/semantica-agi/semantica)
- Tags: how-to-guide
- Published: 2026-08-14

---

**The Semantica Policy Engine enforces compliance rules by evaluating decision metadata against versioned policies stored in a backend-agnostic knowledge graph, with full audit trails for every policy application and exception.**

The **Policy Engine** is the core enforcement layer in [semantica-agi/semantica](https://github.com/semantica-agi/semantica) that bridges decision-making and organizational compliance. Built on a flexible graph architecture, it enables teams to define rules, check decisions against those rules, track policy versions, and analyze the impact of proposed changes—all while maintaining complete auditability.

## Policy Engine Architecture

The engine in [`semantica/context/policy_engine.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/context/policy_engine.py) is designed for maximum backend flexibility. At initialization, it detects whether your **graph store** supports Cypher queries by checking for an `execute_query` method. If available, the engine sets `self._supports_cypher = True` and uses optimized Cypher operations; otherwise, it falls back to a generic node/edge API (`add_node`, `add_edge`, `find_nodes`).

### Policy Lifecycle and Versioning

Policies are **versioned objects** that persist as nodes in your knowledge graph. The engine maintains a complete version history:

- **`add_policy(policy)`** – Creates a new `Policy` node in the graph
- **`update_policy(policy_id, rules)`** – Creates a new policy version and links the previous version with a `VERSION_OF` edge

This versioning enables rollback and historical analysis without duplicating data.

## Core Compliance Operations

### Evaluating Decisions Against Policies

The `check_compliance(decision, policy_id)` method implements the central enforcement logic. It retrieves the latest policy version and evaluates the decision using `_evaluate_compliance`. Rules follow intuitive naming conventions:

| Prefix | Purpose | Example |
|--------|---------|---------|
| `min_` | Minimum numeric thresholds | `min_credit_score` |
| `max_` | Maximum numeric thresholds | `max_loan_amount` |
| `required_` | Mandatory field presence | `required_collateral` |
| `min_confidence` | Threshold for decision confidence | `min_confidence: 0.85` |
| `allowed_outcomes` | Whitelist of valid outcomes | `["approved", "rejected"]` |

### Recording Policy Applications

Every enforcement action is auditable. The `record_policy_application(decision_id, policy_id, version)` method creates an `APPLIED_POLICY` relationship in the graph, linking the decision node to the specific policy version that governed it.

For cases requiring rule violations, `record_exception(decision_id, policy_id, reason, approver, justification)` creates an `Exception` node with two critical relationships:
- `GRANTED_EXCEPTION` → links to the decision
- `OVERRIDDEN_POLICY` → links to the overridden policy

## Impact Analysis and History

Before deploying policy changes, `analyze_policy_impact(policy_id, proposed_rules)` runs **"what-if" queries** against historical decisions. The method returns:
- Count of affected past decisions
- Quantified `risk_increase` metrics
- Classification of decisions that would change compliance status

For audit and governance, `get_policy_history(policy_id)` traverses the `VERSION_OF` chain, and `get_affected_decisions(policy_id, version)` identifies all decisions that used a specific policy version.

## Complete Implementation Example

```python
from datetime import datetime
from semantica.context.policy_engine import PolicyEngine
from semantica.context.decision_models import Policy, Decision

# 1️⃣ Initialize with any graph backend (Neo4j, FalkorDB, or custom)

engine = PolicyEngine(graph_store=my_neo4j_driver)

# 2️⃣ Define a versioned compliance policy

lending_policy = Policy(
    policy_id="loan_approval",
    name="Retail Mortgage Policy",
    description="Standard underwriting criteria for retail mortgages",
    rules={
        "min_credit_score": 650,
        "max_loan_amount": 500000,
        "max_debt_to_income": 0.43,
        "min_confidence": 0.80,
        "allowed_outcomes": ["approved", "rejected", "manual_review"]
    },
    category="lending",
    version="1.0.0",
    created_at=datetime.utcnow(),
    updated_at=datetime.utcnow(),
    metadata={"region": "north_america", "product": "retail_mortgage"}
)
engine.add_policy(lending_policy)

# 3️⃣ Enforce compliance on a production decision

decision = Decision(
    decision_id="dec_2024_001847",
    scenario="mortgage_application",
    category="lending",
    outcome="approved",
    confidence=0.92,
    metadata={
        "credit_score": 720,
        "loan_amount": 300000,
        "debt_to_income": 0.38,
        "applicant_id": "cust_77392"
    }
)
is_compliant = engine.check_compliance(decision, policy_id="loan_approval")
print(f"Compliance result: {is_compliant}")  # → True

# 4️⃣ Record policy application for audit trail

engine.record_policy_application(
    decision_id=decision.decision_id,
    policy_id="loan_approval",
    version="1.0.0"
)

# 5️⃣ Handle exception for high-value client requiring override

exception_id = engine.record_exception(
    decision_id="dec_2024_001848",
    policy_id="loan_approval",
    reason="Credit score below minimum threshold",
    approver="regional_director_215",
    justification="Existing customer with $2M deposits; executive approval obtained"
)

# 6️⃣ Analyze impact before tightening credit standards

proposed_rules = {
    "min_credit_score": 700,      # Increased from 650

    "max_loan_amount": 500000,    # Unchanged

    "max_debt_to_income": 0.40    # Tightened from 0.43

}
impact_report = engine.analyze_policy_impact("loan_approval", proposed_rules)
print(f"Decisions affected: {impact_report['affected_count']}")
print(f"Risk increase: {impact_report['risk_increase']:.2%}")

```

## Integration with Decision Models

The Policy Engine relies on dataclasses defined in [`semantica/context/decision_models.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/context/decision_models.py):

- **`Policy`** – Encapsulates policy rules, versioning metadata, and categorization
- **`PolicyException`** – Records justification, approver identity, and timestamp for every override
- **`Decision`** – The evaluation target containing outcome, confidence, and structured metadata

These models enforce type safety across the compliance pipeline and enable serialization to graph nodes.

## Testing and Validation

The comprehensive test suite in [`tests/test_030_realworld_comprehensive.py`](https://github.com/semantica-agi/semantica/blob/main/tests/test_030_realworld_comprehensive.py) validates:

- Policy creation with auto-generated versus explicit IDs
- Version chain integrity across multiple updates
- Compliance evaluation for edge cases (boundary values, missing fields)
- Exception workflow with full relationship verification

For realistic enforcement scenarios, [`tests/test_030_context_graph_realworld_extended.py`](https://github.com/semantica-agi/semantica/blob/main/tests/test_030_context_graph_realworld_extended.py) demonstrates `enforce_decision_policy` integration with production-like decision pipelines.

## Summary

- The **Policy Engine** provides backend-agnostic compliance enforcement through capability detection at initialization
- **Versioned policies** with `VERSION_OF` edges enable historical analysis and rollback
- **Rule naming conventions** (`min_`, `max_`, `required_`, `min_confidence`, `allowed_outcomes`) simplify policy authoring
- **Complete auditability** via `APPLIED_POLICY` relationships and `Exception` nodes with dual linking
- **Impact analysis** quantifies risk before policy changes affect production decisions

## Frequently Asked Questions

### What graph backends are compatible with the Policy Engine?

Any backend implementing either a Cypher `execute_query` method (tested with **Neo4j** and **FalkorDB**) or a generic node/edge API (`add_node`, `add_edge`, `find_nodes`). The engine auto-detects capabilities and optimizes queries accordingly.

### How does policy versioning handle concurrent updates?

The `update_policy` method creates a new version node and links it to the previous version via a `VERSION_OF` edge. Each version is immutable; concurrent updates result in branching version chains that can be reconciled through `get_policy_history`.

### Can compliance rules reference nested decision metadata?

Yes. The `_evaluate_compliance` method traverses nested dictionary structures in decision metadata. Use dot-notation keys in rules (e.g., `"min_applicant.income": 50000`) or flatten metadata before policy evaluation.

### What happens when a required rule field is missing from decision metadata?

Missing fields fail `required_` checks and cause numeric comparisons to return non-compliant. The `check_compliance` method returns `False` with detailed failure reasons accessible through extended logging at the `DEBUG` level.