Implement policy-based guardrails

Introduction

When you build an agent that serves multiple user tiers, nothing stops a free-tier client from hammering the API at the same rate as an enterprise subscriber — or from calling premium features their plan doesn't include. Content safety filters catch harmful language, but they have no concept of quotas or subscription scope. By the end of this lesson, you'll be able to implement multi-window rate-limiting guardrails and user-tier policy checks that enforce organizational constraints before a request ever reaches the model.

Key Terminology

• Policy-based guardrail — a pre-model check that enforces organizational constraints such as rate limits and subscription-tier access rules, distinct from content-safety filters that evaluate the text of a request for harmful language.
• Rolling window — a sliding time interval (per-minute, per-hour, or per-day) used to count recent requests by comparing each timestamp in request_history against a computed cutoff such as now - timedelta(minutes=1), with no fixed reset point.
• Burst allowance — an additive tolerance applied exclusively to the per-minute window that permits short traffic spikes without relaxing hourly or daily caps; set via the burst_allowance constructor parameter of RateLimitGuardrail, making the effective per-minute ceiling requests_per_minute + burst_allowance.
• Request history — the per-user list of datetime timestamps stored in the request_history defaultdict; new timestamps are appended only after all three windows clear, so blocked requests never inflate the count and a periodic cleanup pass prunes entries older than 24 hours.
• User tier — the subscription level (free, pro, or enterprise) read from context["user_tier"] that selects which rate-limit configuration is passed to RateLimitGuardrail before delegating to check.

Concepts

Policy Guards Are Not Content Filters

Content-safety filters evaluate the text of a request — flagging harmful language, prompt-injection attempts, or policy-violating topics. Policy-based guardrails operate on an entirely different dimension: they enforce who can ask and how often, regardless of what the request says. A free-tier user submitting a perfectly benign question can still be blocked if they've exhausted their per-minute quota; an enterprise subscriber asking a borderline question might pass a content filter but still be rejected if their daily cap is blown.

This distinction matters architecturally. Content filters are stateless — each request is evaluated in isolation. Policy guardrails are inherently stateful: they must remember what a given user has done recently to enforce rolling limits. Both belong in the input-guardrail layer, but they serve orthogonal goals and should be implemented as separate, composable components rather than merged into a single check.

Why Three Windows Instead of One

A single rate limit catches only one type of misuse. A per-minute cap prevents instantaneous bursts but can't stop a user sending 59 requests per minute for hours. A per-day cap protects budget but doesn't defend against someone dumping their entire daily allowance in the first ten seconds.

Three cascading windows — per-minute, per-hour, and per-day — each catch a distinct abuse pattern while remaining transparent to normal usage. A user making steady, moderate use clears all three windows on every request. A script probing rapidly hits the per-minute wall. Slower-but-relentless hammering hits the per-hour wall. A heavy user who exhausts their daily credit gets a retry_after of 86,400 seconds.

Using rolling windows rather than fixed-interval resets is also fairer to users: there is no "reset at midnight" gaming opportunity, and a user who was rate-limited 59 minutes ago gradually earns back capacity as those timestamps slide outside the one-hour window.

Burst Allowance and Tier-Aware Dispatch

Legitimate traffic is bursty. A user navigating quickly through a multi-step flow or submitting a batch job will naturally spike above their steady-state per-minute rate. The burst_allowance parameter addresses this by adding a tolerance only to the per-minute window — the effective_minute_limit in the code is requests_per_minute + burst_allowance. Hourly and daily windows carry no burst tolerance because sustained elevated usage is precisely the pattern those windows are designed to catch (see Code Walkthrough).

User-tier enforcement sits above RateLimitGuardrail as a thin dispatch layer that reads context["user_tier"] and selects the appropriate constructor arguments before delegating to check. This keeps the rolling-window logic reusable across all tiers: free, pro, and enterprise all share the same RateLimitGuardrail implementation but receive different per-minute, per-hour, and per-day limits. Separating tier routing from window tracking makes it straightforward to add new tiers or adjust limits without touching the core rate-limiting code.

Code Walkthrough

Now that you understand how policy-based guardrails differ from content-safety filters — enforcing usage limits and tier-specific rules rather than flagging harmful language — you can see how those concepts translate directly into code.

The RateLimitGuardrail class below tracks per-user request timestamps in an in-memory dictionary and enforces limits across three rolling windows: per-minute, per-hour, and per-day. A burst allowance applies only to the per-minute window to accommodate legitimate traffic spikes without relaxing hourly or daily caps. A cleanup counter purges history older than 24 hours every 100 requests to prevent unbounded memory growth in long-running services.

Code snippetpython
1from datetime import datetime, timedelta
2from typing import Dict, List
3from collections import defaultdict
4
5class RateLimitGuardrail:
6    def __init__(
7        self,
8        requests_per_minute: int = 60,
9        requests_per_hour: int = 1000,
10        requests_per_day: int = 10000,
11        burst_allowance: int = 10,
12        name: str = "rate_limiter",
13    ):
14        self.name = name
15        self.priority = 5
16        self.requests_per_minute = requests_per_minute
17        self.requests_per_hour = requests_per_hour
18        self.requests_per_day = requests_per_day
19        self.burst_allowance = burst_allowance
20        self.request_history: Dict[str, List[datetime]] = defaultdict(list)
21        self._cleanup_counter = 0
22
23    def _cleanup_old_entries(self, user_id: str, now: datetime) -> None:
24        cutoff = now - timedelta(days=1)
25        self.request_history[user_id] = [
26            t for t in self.request_history[user_id] if t > cutoff
27        ]
28
29    def check(self, input_text: str, context: dict) -> dict:
30        user_id = context.get("user_id", "anonymous")
31        now = datetime.now()
32
33        self._cleanup_counter += 1
34        if self._cleanup_counter >= 100:
35            self._cleanup_old_entries(user_id, now)
36            self._cleanup_counter = 0
37
38        history = self.request_history[user_id]
39        one_minute_ago = now - timedelta(minutes=1)
40        one_hour_ago = now - timedelta(hours=1)
41        one_day_ago = now - timedelta(days=1)
42
43        requests_last_minute = sum(1 for t in history if t > one_minute_ago)
44        requests_last_hour = sum(1 for t in history if t > one_hour_ago)
45        requests_last_day = sum(1 for t in history if t > one_day_ago)
46
47        effective_minute_limit = self.requests_per_minute + self.burst_allowance
48
49        if requests_last_minute >= effective_minute_limit:
50            return {"allowed": False, "reason": "per-minute limit exceeded", "retry_after": 60}
51        if requests_last_hour >= self.requests_per_hour:
52            oldest_in_window = min(t for t in history if t > one_hour_ago)
53            retry_after = int((oldest_in_window + timedelta(hours=1) - now).total_seconds())
54            return {"allowed": False, "reason": "per-hour limit exceeded", "retry_after": retry_after}
55        if requests_last_day >= self.requests_per_day:
56            return {"allowed": False, "reason": "per-day limit exceeded", "retry_after": 86400}
57
58        self.request_history[user_id].append(now)
59        return {"allowed": True, "reason": None, "retry_after": 0}

The check method records each passing request's timestamp only after all three windows clear, so the history never accumulates timestamps from blocked requests. The burst allowance is additive to the per-minute cap, meaning a guardrail configured with requests_per_minute=3 and burst_allowance=2 allows up to five calls per minute before blocking. Hourly and daily limits have no burst tolerance — when either is exceeded, the response includes a precise retry_after derived from the oldest timestamp still inside the rolling window.

For user-tier enforcement, wrap RateLimitGuardrail with a thin policy layer that reads context["user_tier"] and selects the appropriate limit configuration before delegating to check. This keeps rate-limiting logic reusable across free, pro, and enterprise tiers without duplicating window tracking.

Confirm that your guardrail is working: instantiate RateLimitGuardrail(requests_per_minute=3, burst_allowance=0) and call check() four times with the same user_id; you'll know it works when the fourth call returns {"allowed": False, "reason": "per-minute limit exceeded", "retry_after": 60} and the first three return {"allowed": True, ...}.

Do's and Don'ts

Having walked through implementing policy-based guardrails above, the following Do's and Don'ts distill it into practice.

Do's

1. ✓Do apply burst_allowance only to the per-minute window — the lesson's RateLimitGuardrail design deliberately exempts hourly and daily caps from burst tolerance, so traffic spikes get a short-term release valve without letting a single user exhaust their daily quota through sustained bursts.
2. ✓Do append the request timestamp to request_history only after all three window checks pass — recording blocked requests would inflate the history, causing legitimate follow-up calls to be counted against the user and making rolling windows permanently over-report usage.
3. ✓Do keep tier policy selection outside RateLimitGuardrail by reading context["user_tier"] in a wrapping layer — this lets the three-window tracking logic remain a single reusable class while free, pro, and enterprise limits can differ without duplicating the _cleanup_old_entries or window-counting logic.

Don'ts

1. ✗Don't skip the periodic _cleanup_old_entries call or raise its 100-request threshold arbitrarily — the in-memory request_history dictionary grows without bound in long-running services if timestamps older than 24 hours are never purged, eventually consuming enough memory to degrade or crash the process.
2. ✗Don't use a static retry_after value for the per-hour limit — the lesson computes retry_after dynamically from the oldest timestamp still inside the rolling hour window; a hardcoded 3600 would tell a user to wait far longer than necessary when their oldest request is about to age out of the window.
3. ✗Don't treat RateLimitGuardrail as a substitute for content-safety filters — it enforces quotas and tier scope only; it has no concept of harmful language, so deploying it alone leaves the pipeline unprotected against dangerous inputs that stay within the allowed request volume.