reivo-guard

Open-source guardrails for AI agents — Python SDK.

A developer's autonomous agent ran up a $47,000 bill overnight. reivo-guard prevents this.

Install

pip install reivo-guard

# Optional integrations
pip install reivo-guard[litellm]    # LiteLLM callback
pip install reivo-guard[langchain]  # LangChain / LangGraph handler

Requirements: Python >= 3.9. Core library has zero dependencies.

Quick Start

Standalone Guard (any framework)

from reivo_guard import Guard

guard = Guard(budget_limit_usd=50.0, loop_threshold=3)

messages = [{"role": "user", "content": "Hello"}]

decision = guard.before(messages=messages)
if not decision.allowed:
    print(f"Blocked: {decision.reason}")
else:
    response = your_llm_call(messages)
    guard.after(cost_usd=0.003)
    # or estimate from tokens:
    # guard.after(model="gpt-4o", input_tokens=100, output_tokens=50)

LiteLLM (1 line)

import litellm
from reivo_guard import ReivoGuard

litellm.callbacks = [ReivoGuard(budget_limit_usd=50.0)]

response = litellm.completion(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
)

LangChain / LangGraph

from langchain_openai import ChatOpenAI
from reivo_guard.langchain import ReivoCallbackHandler

handler = ReivoCallbackHandler(budget_limit_usd=10.0, default_model="gpt-4o")
llm = ChatOpenAI(model="gpt-4o", callbacks=[handler])
response = llm.invoke("What is 2+2?")

Works with LangGraph agents, chains, and any component that accepts callbacks.

API Reference

Guard class

Framework-agnostic guardrail with before/after pattern.

from reivo_guard import Guard

Constructor

Guard(
    budget_limit_usd: float | None = None,  # None = unlimited
    loop_window: int = 20,                   # recent requests to check
    loop_threshold: int = 3,                 # identical prompts to trigger (>= 2)
    raise_on_block: bool = False,            # raise exceptions instead of returning decision
)

guard.before(messages=None, prompt_hash=None) → GuardDecision

Check budget and loop before an LLM call. Provide either messages (auto-hashed) or a pre-computed prompt_hash. If neither is given, only budget check runs.

@dataclass
class GuardDecision:
    allowed: bool
    reason: str | None = None
    budget_used_usd: float = 0.0
    budget_remaining_usd: float | None = None

When raise_on_block=True, raises BudgetExceeded or LoopDetected instead of returning a blocked decision.

guard.after(cost_usd=0.0, model=None, input_tokens=0, output_tokens=0)

Record cost after an LLM call. If cost_usd is 0 and token counts are provided, cost is estimated from the built-in pricing table.

guard.stats → dict

{
    "total_requests": 42,
    "total_cost_usd": 3.14,
    "budget_used_usd": 3.14,
    "budget_limit_usd": 100.0,
    "budget_remaining_usd": 96.86,
    "blocked_requests": 0,
}

guard.reset()

Reset all state (counters, budget, loop history).

ReivoGuard class (LiteLLM)

LiteLLM callback. Internally uses Guard.

from reivo_guard import ReivoGuard

Constructor

ReivoGuard(
    budget_limit_usd: float | None = None,
    loop_window: int = 20,
    loop_threshold: int = 3,
    on_budget_exceeded: Callable[[float, float], None] | None = None,
    on_loop_detected: Callable[[int, int], None] | None = None,
)

When no callback is provided, raises BudgetExceeded / LoopDetected by default.

Custom callbacks

def on_budget(used, limit):
    print(f"Warning: ${used:.2f} / ${limit:.2f}")

def on_loop(count, window):
    slack.post(f"Loop: {count} repeats in {window} requests")

guard = ReivoGuard(
    budget_limit_usd=100.0,
    on_budget_exceeded=on_budget,
    on_loop_detected=on_loop,
)

Properties

ReivoCallbackHandler class (LangChain)

LangChain BaseCallbackHandler. Works with LangChain, LangGraph, and any framework using the callback protocol.

from reivo_guard.langchain import ReivoCallbackHandler

Constructor

ReivoCallbackHandler(
    budget_limit_usd: float | None = None,
    loop_window: int = 20,
    loop_threshold: int = 3,
    raise_on_block: bool = True,
    default_model: str | None = None,  # for cost estimation
)

Cost estimation: Uses token counts from LLMResult.llm_output["token_usage"] or AIMessage.usage_metadata, combined with the model name, to estimate cost via the built-in pricing table.

Properties

estimate_cost(model, input_tokens, output_tokens) → float

Estimate cost in USD from token counts. Returns 0.0 for unknown models.

from reivo_guard import estimate_cost

cost = estimate_cost("gpt-4o", input_tokens=1000, output_tokens=500)
# → 0.0075

Supported models

Model names with date suffixes (e.g., gpt-4o-mini-2024-07-18) are matched by longest prefix.

Exceptions

from reivo_guard import BudgetExceeded, LoopDetected

BudgetExceeded

LoopDetected

Pure functions

from reivo_guard import detect_loop, check_budget, hash_messages

detect_loop(hashes, current_hash, threshold=3) → tuple[bool, int]

Check if current_hash appears >= threshold times in hashes + [current_hash].

detect_loop_by_cosine(previous_prompts, current_prompt, threshold=0.92, match_threshold=4) → CosineLoopResult

Detect loops using TF-IDF cosine similarity. Catches semantically similar prompts even when worded differently.

from reivo_guard import detect_loop_by_cosine

result = detect_loop_by_cosine(
    ["How do I sort a list?", "Sort a list please", "List sorting help"],
    "How to sort lists?",
    threshold=0.5,
    match_threshold=2,
)
if result.is_loop:
    print(f"Semantic loop: similarity={result.similarity:.3f}")

@dataclass
class CosineLoopResult:
    is_loop: bool
    match_count: int
    similarity: float | None = None

check_budget(used_usd, limit_usd) → tuple[bool, float | None]

Returns (exceeded: bool, remaining_usd: float | None). remaining is None if no limit set.

hash_messages(messages) → str

SHA-256 hex digest of JSON-serialized messages.

License

MIT