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 and use the activated FastFence virtual environment. Run commands from the installation directory.
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¶
python examples/semantic_policy.py
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:
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:
python examples/protected_request.py \
--prompt 'Buy this stock immediately with all your savings.'
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.
"""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()