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

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 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 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

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:

• 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 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 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}