Preview and activate a natural-language rule¶
This complete script adds a named policy:
Block personalized financial recommendations. General financial definitions are allowed.
Laya evaluates actual sample text through POST /api/admin/semantic/preview.
The rule applies only to model input. Preview does not activate the rule or send
a completion request to your protected model.
Download the complete examples into your installation's examples/ directory. Run commands from the installation directory; uv run supplies Python 3.12 and the FastFence package for each example, without activating a virtual environment.
Prerequisites¶
Complete the local gateway setup. Keep
Ollama and FastFence running. This example requires the private management
credential from state/credentials.json, or FASTFENCE_ADMIN_TOKEN supplied
through your environment. Legacy security-admin credentials are also supported.
Preview first¶
The script fetches the active policy, preserves its other controls, adds or replaces
only no-personal-financial-advice, and increments the candidate version. It asks
the real preview endpoint to classify two samples, then prints their actual results
and a YAML diff. No PUT is sent in preview mode.
Expected observations:
| Sample | Expected preview decision |
|---|---|
Buy this stock immediately with all your savings. |
blocked |
Define a stock as a financial instrument. |
no_semantic_block |
no_semantic_block describes only this semantic preview. A protected request still
has to pass authentication, deterministic rules, budgets and any output checks.
Preview uses your active semantic model and threshold when Laya is already enabled;
otherwise it selects Laya/Qwen3:4b for the candidate. Unexpected classifications are
reported honestly and prevent this script's activation step.
Activate deliberately¶
After reviewing the diff and results:
uv run --python 3.12 --no-project --with fastfence==1.0.1 python examples/semantic_policy.py --activate
This reruns the previews, checks their expected outcomes, refetches the active
policy to reject concurrent changes, then submits the candidate through the actual
PUT /api/admin/policy endpoint. The server validates the version and configured
source. A failed preview or conflict stops the example without activation.
Now test the active policy through the normal protected request path:
uv run --python 3.12 --no-project --with fastfence==1.0.1 python examples/protected_request.py \
--prompt 'Buy this stock immediately with all your savings.'
uv run --python 3.12 --no-project --with fastfence==1.0.1 python examples/protected_request.py \
--prompt 'Define a stock as a financial instrument.'
Edit the RULE and CASES constants to explore another policy. Keep a blocked and
a permitted sample with expectations you chose independently. For exact character
restrictions, use a deterministic text rule instead of treating model judgment as
exact matching. Removing the named rule through Policies requires another
reviewed policy version.
Compound-rule limitation¶
The local Qwen3:4b assessment has a reproduced false negative for a rule requiring both a person's full name and an email address: the combined input was allowed even though the named rule reached Laya correctly. A natural-language conjunction is not a reliable substitute for deterministic privacy controls. Keep applicable PII controls enabled and include combined, partial and exception cases in your preview tests; a passing example does not establish general detection accuracy.
The same semantic configuration can also block content permitted by a literal rule because the layers enforce separate restrictions. Inspect the decision reason and input/output assessment results when a literal nonmatch is blocked.
"""Preview a named natural-language rule with real Laya; activate only explicitly."""
import argparse
import copy
import difflib
import json
import os
from pathlib import Path
import httpx
import yaml
from fastfence.modules.control.domain.models import Policy
RULE = {
"id": "no-personal-financial-advice",
"instruction": "Block personalized financial recommendations. General financial definitions are allowed.",
"direction": "input",
"target": "model",
}
CASES = [
("Buy this stock immediately with all your savings.", "blocked"),
("Define a stock as a financial instrument.", "no_semantic_block"),
]
def admin_token(path: Path) -> str:
if value := os.environ.get("FASTFENCE_ADMIN_TOKEN"):
return value
if path == Path("state/credentials.json") and not path.exists():
path = Path("state/demo-tokens.json")
values = json.loads(path.read_text())
return values.get("local-admin") or values["security-admin"]
def read_policy(client: httpx.Client, token: str) -> dict:
response = client.get(
"/api/admin/status", headers={"Authorization": "Bearer " + token}
)
response.raise_for_status()
return response.json()["policy"]
def prepare(client: httpx.Client, token: str) -> tuple[dict, dict, list[dict]]:
base = read_policy(client, token)
candidate = copy.deepcopy(base)
candidate["version"] += 1
semantic = candidate["semantic"]
if semantic["provider"] != "laya":
semantic.update(
provider="laya",
model="qwen3:4b",
timeout_ms=max(30000, semantic["timeout_ms"]),
)
semantic["rules"] = [
rule for rule in semantic.get("rules", []) if rule["id"] != RULE["id"]
] + [RULE]
candidate = Policy.model_validate(candidate).model_dump(mode="json")
results = []
for text, expected in CASES:
response = client.post(
"/api/admin/semantic/preview",
headers={"Authorization": "Bearer " + token},
json={
"rule": RULE,
"text": text,
"direction": "input",
"target": "model",
"base_version": base["version"],
},
)
response.raise_for_status()
result = response.json()
results.append({"expected": expected, **result})
return base, candidate, results
def activate(
client: httpx.Client,
token: str,
base: dict,
candidate: dict,
results: list[dict],
) -> dict:
if len(results) != len(CASES) or not all(
result["decision"] == result["expected"] and result["rule_applied"]
for result in results
):
raise ValueError(
"Preview did not match expected classifications; no activation."
)
if read_policy(client, token) != base:
raise ValueError(
"Active policy changed; preview again before activation."
)
response = client.put(
"/api/admin/policy",
headers={"Authorization": "Bearer " + token},
json=candidate,
)
response.raise_for_status() # Server also rejects version/source conflicts.
return response.json()
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--url", default="http://127.0.0.1:8000")
parser.add_argument(
"--credentials", type=Path, default=Path("state/credentials.json")
)
parser.add_argument("--activate", action="store_true")
args = parser.parse_args()
token = admin_token(args.credentials)
with httpx.Client(
base_url=args.url, timeout=120, trust_env=False
) as client:
base, candidate, results = prepare(client, token)
print(
"".join(
difflib.unified_diff(
yaml.safe_dump(base, sort_keys=False).splitlines(
keepends=True
),
yaml.safe_dump(candidate, sort_keys=False).splitlines(
keepends=True
),
fromfile="active-policy.yaml",
tofile="candidate-policy.yaml",
)
)
)
print(json.dumps(results, indent=2))
if args.activate:
print(json.dumps(activate(client, token, base, candidate, results)))
else:
print(
"Preview only. Review the diff; rerun with --activate to apply."
)
if __name__ == "__main__":
main()