How to Set Up a Policy Engine with Configurable Rule Sets for Compliance Gates

You can configure Semantica’s built-in policy engine by creating Policy nodes in the context graph, defining rule sets as dictionaries or Python callables, and invoking enforce_decision_policy to validate decisions against compliance gates.

The semantica-agi/semantica repository provides a graph-native policy engine that lets you enforce compliance rules on any decision or reasoning result. By attaching Policy nodes to the context graph and supplying configurable rule sets, you can create automated compliance gates without modifying core reasoning code. The entire workflow is driven by a single enforcement entry point implemented in semantica/context/context_graph.py.

Core Architecture

Policy Nodes

A Policy node represents a compliance policy—such as privacy, insurance, or clinical guidelines—as a first-class entity in the Semantica context graph. Each node stores a unique policy_id, a human-readable name, a description, and optional metadata like confidence thresholds. You register a policy by calling graph.add_node(node_id=..., node_type="Policy", ...) so it can be referenced during enforcement.

Configurable Rule Sets

A rule set is a collection of constraints that define what makes a decision compliant. Rules can be expressed as simple key-value dictionaries, logical expressions, or custom Python callables. When you invoke the engine, you pass the rule set to the policy_rules parameter, allowing the same infrastructure to support diverse domains without code changes.

The enforce_decision_policy Entry Point

The function semantica.context.context_graph.enforce_decision_policy(decision_data, policy_rules=None) is the central evaluation mechanism. It extracts the decision payload, looks up relevant Policy nodes, applies the supplied rules, and returns a structured dictionary with the keys compliant, violations, and warnings.

Compliance Gates

A compliance gate is a higher-level wrapper that automatically halts execution when a decision fails policy checks. The gate calls enforce_decision_policy and raises an exception or returns a retry signal whenever the compliant flag is False.

Step-by-Step Setup

1. Create Policy Nodes

Add one or more Policy nodes to the context graph so the engine knows which domains to enforce.

graph.add_node(
    node_id="clinical_policy_htn_2024",
    node_type="Policy",
    content="Clinical policy: high-confidence AI decisions only",
)

2. Define Your Rule Set

Write a rule set that captures your compliance logic. The following YAML-style dictionary requires a confidence score of at least 0.9 and prohibits experimental treatments:

policy_id: clinical_policy_htn_2024
required_confidence: 0.9
prohibited_entities:
  - "experimental_treatment"

You can also express the same logic as a Python callable:

def clinical_rules(decision):
    return (
        decision["confidence"] >= 0.9
        and not any(e in decision["entities"] for e in ["experimental_treatment"])
    )

3. Enforce the Policy

Hand the decision payload to the engine alongside the rules you defined:

result = graph.enforce_decision_policy(
    decision_data=decision,
    policy_rules=clinical_rules,   # or the dictionary shown above

)
if not result["compliant"]:
    raise RuntimeError(f"Policy violation: {result['violations']}")

4. Integrate the Compliance Gate

Embed the enforcement call inside an MCP pipeline or CrewAI tool. The engine locates the relevant Policy node by the policy_id stored in the decision payload, applies the rules, and returns a compliance report.

Practical Code Examples

Enforce a Simple JSON Rule Set

The following example loads a declarative policy from YAML and evaluates a decision against it:

# policies.yaml

- policy_id: privacy_policy_v1
  required_confidence: 0.95
  allowed_entities:
    - "personal_data"
    - "transaction"

# enforcement.py

import yaml
from semantica.context import context_graph as cg

with open("policies.yaml") as f:
    rules = yaml.safe_load(f)

def load_rules(policy_id):
    for r in rules:
        if r["policy_id"] == policy_id:
            return r
    return {}

decision = {
    "policy_id": "privacy_policy_v1",
    "confidence": 0.96,
    "entities": ["personal_data"],
}
policy_rules = load_rules(decision["policy_id"])
result = cg.enforce_decision_policy(decision, policy_rules=policy_rules)
print(result)          # → {'compliant': True, 'violations': [], 'warnings': []}

Validate with a Custom Callable

For complex logic, pass a callable that performs arbitrary checks:

def complex_rules(decision):
    prohibited = {"unverified_source", "experimental_treatment"}
    if decision["confidence"] < 0.92:
        return False
    if prohibited.intersection(set(decision["entities"])):
        return False
    return True

decision = {
    "policy_id": "clinical_policy_htn_2024",
    "confidence": 0.93,
    "entities": ["diagnosis", "treatment_standard"],
}
result = cg.enforce_decision_policy(decision, policy_rules=complex_rules)
print(result)          # → {'compliant': True, ...}

Wire Into an MCP Pipeline

Use the compliance gate inside a Semantica pipeline to block non-compliant decisions automatically:

from semantica.mcp.pipeline import PipelineEngine

def compliance_gate(decision):
    res = graph.enforce_decision_policy(decision, policy_rules=clinical_rules)
    if not res["compliant"]:
        raise RuntimeError("Compliance gate failed")
    return decision

engine = PipelineEngine()
engine.add_step("reasoning", some_reasoning_tool)
engine.add_step("compliance", compliance_gate)
engine.run(initial_context)

Key Source Files

• [semantica/context/context_graph.py](https://github.com/semantica-agi/semantica/blob/main/semantica/context/context_graph.py) — Implements enforce_decision_policy and the graph API for adding Policy nodes.
• [semantica/mcp/pipeline.py](https://github.com/semantica-agi/semantica/blob/main/semantica/mcp/pipeline.py) — Demonstrates how to plug the policy engine into an end-to-end pipeline.
• [tests/integrations/crewai/test_knowledge_source.py](https://github.com/semantica-agi/semantica/blob/main/tests/integrations/crewai/test_knowledge_source.py) — Shows how to add a policy node and assert its content.
• [tests/test_030_context_graph_realworld_extended.py](https://github.com/semantica-agi/semantica/blob/main/tests/test_030_context_graph_realworld_extended.py) — Contains concrete examples of enforce_decision_policy usage with custom rule sets.
• [ARCHITECTURE.md](https://github.com/semantica-agi/semantica/blob/main/ARCHITECTURE.md) — Provides a high-level overview of Semantica’s modular design, including the policy subsystem.

Summary

• Policy nodes are first-class graph entities that store domain-specific compliance requirements.
• Rule sets can be dictionaries, YAML, or Python callables, making the engine highly configurable.
• enforce_decision_policy in semantica/context/context_graph.py is the single entry point for evaluating decisions.
• Compliance gates wrap the enforcement call to halt pipelines when rules are violated.
• The policy engine integrates directly with Semantica’s MCP pipelines and CrewAI tools.

Frequently Asked Questions

What data structure does enforce_decision_policy return?

The function returns a Python dictionary with three keys: compliant (a boolean), violations (a list of failed rules), and warnings (a list of optional advisories). You can inspect these fields to decide whether to raise an exception, log a warning, or retry the decision.

Can I use YAML or JSON files instead of Python code for rule sets?

Yes. The engine accepts any serializable dictionary as the policy_rules argument. You can load rule definitions from YAML or JSON files, map them by policy_id, and pass the resulting dictionary directly to enforce_decision_policy without writing custom callables.

Where are the Policy nodes stored in Semantica?

Policy nodes live in the context graph alongside other domain entities. You create them via the graph API in semantica/context/context_graph.py, which means they are version-controlled, traceable, and visible in graph visualizations just like any other node.

How do I attach a compliance gate to an existing MCP pipeline?

Define a lightweight wrapper function that calls graph.enforce_decision_policy on the step’s output. Then register that function as a pipeline step using PipelineEngine.add_step(), as shown in semantica/mcp/pipeline.py. If the decision is non-compliant, the wrapper raises an exception and stops the pipeline.

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