Sprawdź i aktywuj regułę w języku naturalnym¶
Ten kompletny skrypt dodaje nazwaną politykę:
Blokuj spersonalizowane rekomendacje finansowe. Zezwalaj na ogólne definicje finansowe.
Laya ocenia rzeczywistą treść próbek przez POST /api/admin/semantic/preview. Reguła dotyczy wyłącznie wejścia modelu. Podgląd nie aktywuje reguły ani nie wysyła zadania do chronionego modelu.
Pobierz komplet przykładów do katalogu examples/ swojej instalacji. Polecenia wykonuj z katalogu instalacji; uv run zapewnia Python 3.12 i pakiet FastFence dla każdego przykładu, bez aktywowania środowiska wirtualnego.
Wymagania¶
Przeprowadź lokalną konfigurację bramki. Pozostaw Ollama i FastFence uruchomione. Przykład wymaga prywatnego tokenu administracyjnego z state/credentials.json lub FASTFENCE_ADMIN_TOKEN przekazanego przez środowisko. Obsługiwane są też starsze tokeny security-admin.
Najpierw sprawdź podgląd¶
Skrypt pobiera aktywną politykę, zachowuje pozostałe zabezpieczenia, dodaje lub zastępuje wyłącznie no-personal-financial-advice i zwiększa wersję kandydata. Wysyła dwie próbki do rzeczywistego endpointu podglądu, następnie wypisuje rzeczywiste wyniki i różnice w YAML. W trybie podglądu nie wysyła PUT.
Oczekiwane obserwacje:
| Próbka | Oczekiwana decyzja podglądu |
|---|---|
Buy this stock immediately with all your savings. |
blocked |
Define a stock as a financial instrument. |
no_semantic_block |
no_semantic_block opisuje wyłącznie ten podgląd semantyczny. Chronione żądanie nadal musi przejść uwierzytelnianie, reguły deterministyczne, budżety i kontrole wyjścia. Jeśli Laya jest już włączona, podgląd używa aktywnego modelu semantycznego i progu; w przeciwnym razie wybiera dla kandydata Laya/Qwen3:4b. Nieoczekiwane klasyfikacje są jawnie raportowane i uniemożliwiają aktywację przez ten skrypt.
Aktywuj po sprawdzeniu¶
Po przejrzeniu różnic i wyników:
uv run --python 3.12 --no-project --with fastfence==1.0.1 python examples/semantic_policy.py --activate
Skrypt ponownie wykonuje podglądy, sprawdza oczekiwane wyniki, ponownie pobiera aktywną politykę, aby wykryć równoległe zmiany, a następnie przesyła kandydata do rzeczywistego endpointu PUT /api/admin/policy. Serwer sprawdza wersję i skonfigurowane źródło. Nieudany podgląd lub konflikt zatrzymuje przykład bez aktywacji.
Teraz sprawdź aktywną politykę zwykłą ścieżką chronionych żądań:
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.'
Zmień stałe RULE i CASES, aby sprawdzić inną politykę. Zachowaj próbkę blokowaną i dozwoloną, z oczekiwaniami ustalonymi samodzielnie. Do ścisłych ograniczeń znaków używaj deterministycznej reguły tekstowej, zamiast traktować ocenę modelu jak dokładne dopasowanie. Usunięcie nazwanej reguły przez Policies wymaga kolejnej sprawdzonej wersji polityki.
"""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()