Skip to main content

Guard Policies

Guard policies control what DriftGuard does when before_step() detects a warning that meets the confidence threshold.

Policies

PolicyBehavior
warnReturn warnings — agent decides what to do
blockRaise GuardrailTriggered exception
acknowledgeRaise GuardrailAcknowledgementRequired unless acknowledged=True
record_onlySkip review entirely — store memory only

Setting the default policy

from driftguard import DriftGuard, DriftGuardSettings

guard = DriftGuard(
settings=DriftGuardSettings(
guard_policy="block",
guard_min_confidence=0.8,
)
)

Overriding per call

review = guard.before_step(
"increase salt",
policy="acknowledge",
min_confidence=0.75,
)

Warn (default)

The default policy. Warnings are returned in the RetrievalResponse but no exception is raised. The agent can inspect and decide.

review = guard.before_step("increase salt")

for warning in review.warnings:
print(f"{warning.trigger}{warning.risk} ({warning.confidence:.2f})")

Block

Raises GuardrailTriggered when warnings exist above min_confidence.

from driftguard import GuardrailTriggered

try:
guard.before_step("increase salt", policy="block", min_confidence=0.8)
except GuardrailTriggered as e:
print(f"Blocked: {e}")
# agent takes alternative action

Acknowledge

Raises GuardrailAcknowledgementRequired unless the caller passes acknowledged=True. Useful for human-in-the-loop flows.

from driftguard import GuardrailAcknowledgementRequired

try:
guard.before_step("increase salt", policy="acknowledge")
except GuardrailAcknowledgementRequired:
# show warning to human, get confirmation
guard.before_step("increase salt", policy="acknowledge", acknowledged=True)

Record only

Skips the retrieval review entirely. Memory is still stored when you call record(). Useful for observation-only phases.

review = guard.before_step("increase salt", policy="record_only")
# review.warnings == []
# review.confidence == 0.0

guard_step decorator

Apply guardrails to any agent step function:

from driftguard import guard_step

@guard_step(guard, policy="block", min_confidence=0.8)
def execute_action(action: str):
...

Custom context extraction:

@guard_step(
guard,
input_getter=lambda payload: payload["next_action"],
policy="acknowledge",
acknowledged_getter=lambda payload: payload["confirmed"],
)
def execute(payload: dict):
...