# How to Use Forge Guardrails Middleware in Custom Orchestration Loops

> Integrate Forge guardrails middleware into custom agent loops. This guide shows how to use check() and record() for validation, sequence enforcement, and error tracking without the WorkflowRunner.

- Repository: [Antoine/forge](https://github.com/antoinezambelli/forge)
- Tags: how-to-guide
- Published: 2026-05-22

---

**Forge's guardrails package exposes a lightweight two-method API (`check()` and `record()`) that validates LLM responses, enforces required step sequences, and tracks error budgets, allowing seamless integration of safety controls into any custom agent loop without adopting the full WorkflowRunner framework.**

The `antoinezambelli/forge` repository provides a modular guardrails system designed for interoperability with external orchestrators. Whether you are building a BFCL evaluation harness, an OpenClaw pipeline, or a bespoke agent controller, you can embed this middleware to enforce structured output validation and step-compliance without refactoring your existing control flow.

## Core Components of the Guardrails System

The guardrails façade bundles three independent, pure-Python middleware components defined in `src/forge/guardrails/`:

- **ResponseValidator** – Parses LLM responses to ensure they contain valid tool calls. If rescue mode is enabled, it attempts to extract tool calls from plain text. When validation fails, it generates a *retry* nudge.
- **StepEnforcer** – Tracks required step sequences (e.g., "search" must precede "lookup") and blocks terminal tools (e.g., "answer") until prerequisites are satisfied. It emits *step-blocked* nudges and aborts after a configurable number of premature attempts.
- **ErrorTracker** – Maintains counters for consecutive bad responses and tool-execution failures. When limits are exceeded, it forces a *fatal* outcome to prevent infinite loops.

## The Two-Method Facade API

The `Guardrails` class in [`src/forge/guardrails/guardrails.py`](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/guardrails.py) hides coordination logic behind a minimal interface designed for custom loops.

### Validating Responses with `check()`

Call `guardrails.check(response)` immediately after receiving an LLM generation but before executing any tools. This method returns a `CheckResult` object containing:

- `action` – One of four string values:
  - `execute` – The response contains valid `tool_calls` ready for execution.
  - `retry` – The response was unstructured text; inject `nudge` into the conversation and re-prompt.
  - `step_blocked` – A terminal tool was called too early; `nudge` cites the missing prerequisite steps.
  - `fatal` – The error budget is exhausted; abort the workflow.
- `tool_calls` – A validated list of `ToolCall` objects (populated when `action` is `execute`).
- `nudge` – Instructional text to inject when `action` is `retry` or `step_blocked`.

### Recording Execution with `record()`

After executing tools, call `guardrails.record(executed)` where `executed` is a list of tool names that ran successfully. This method updates the internal step tracker and error counters. It returns a boolean `done` that is `True` only when a terminal tool was called **and** all required steps are satisfied, signaling workflow completion.

## Implementing Guardrails in Custom Orchestration Loops

### Simple Integration with the Guardrails Facade

For most use cases, import the façade and initialize it once per session with your tool schema and step requirements:

```python
from forge.core.workflow import TextResponse, ToolCall
from forge.guardrails import Guardrails

# Initialise once per session

guardrails = Guardrails(
    tool_names=["search", "lookup", "answer"],
    required_steps=["search", "lookup"],
    terminal_tool="answer",
)

def loop_step(response):
    # 1️⃣ Validate the LLM response

    result = guardrails.check(response)

    if result.action == "fatal":
        raise RuntimeError(f"Guardrails fatal: {result.reason}")

    if result.action in ("retry", "step_blocked"):
        # Insert the nudge into the chat history and ask the model again

        inject_nudge(result.nudge)
        return  # early‑exit; the loop will resume with a new model turn

    # 2️⃣ Execute the validated tool calls

    executed = []
    for call in result.tool_calls:
        output = run_tool(call)          # ← your own tool‑dispatch

        executed.append(call.tool)

    # 3️⃣ Sync guardrails state and check for workflow completion

    done = guardrails.record(executed)
    if done:
        print("Workflow finished")
    else:
        print("Continue looping")

```

This pattern mirrors the reference implementation in [`examples/foreign_loop.py`](https://github.com/antoinezambelli/forge/blob/main/examples/foreign_loop.py).

### Granular Control Using Individual Components

For advanced scenarios requiring fine-grained control over state transitions, instantiate the components separately:

```python
from forge.guardrails import ResponseValidator, StepEnforcer, ErrorTracker
from forge.core.workflow import TextResponse, ToolCall

validator = ResponseValidator(
    tool_names=["search", "lookup", "answer"],
    rescue_enabled=True,
)
enforcer = StepEnforcer(
    required_steps=["search", "lookup"],
    terminal_tools=frozenset(["answer"]),
)
errors = ErrorTracker(max_retries=3, max_tool_errors=2)

def loop_step(response):
    # ---- Validation -------------------------------------------------

    val = validator.validate(response)
    if val.needs_retry:
        errors.record_retry()
        if errors.retries_exhausted:
            raise RuntimeError("Too many bad responses")
        inject_nudge(val.nudge)
        return

    errors.reset_retries()

    # ---- Step enforcement -------------------------------------------

    step = enforcer.check(val.tool_calls)
    if step.needs_nudge:
        if enforcer.premature_exhausted:
            raise RuntimeError("Model repeatedly skipped required steps")
        inject_nudge(step.nudge)
        return

    # ---- Execute -----------------------------------------------------

    for tc in val.tool_calls:
        run_tool(tc)                     # ← your execution code

        enforcer.record(tc.tool, tc.args)

    # ---- Bookkeeping ------------------------------------------------

    errors.reset_errors()
    enforcer.reset_premature()

    if enforcer.terminal_reached(val.tool_calls):
        print("All steps satisfied – workflow complete")

```

Both approaches assume you implement `run_tool()` for actual tool invocation and `inject_nudge()` to append nudge text to the LLM's conversation history.

## Key Source Files and Components

- [`src/forge/guardrails/guardrails.py`](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/guardrails.py) – Implements the `Guardrails` façade and `CheckResult` dataclass, coordinating the three sub-components.
- [`src/forge/guardrails/response_validator.py`](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/response_validator.py) – Contains the `ResponseValidator` class responsible for parsing and rescue logic.
- [`src/forge/guardrails/step_enforcer.py`](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/step_enforcer.py) – Houses `StepEnforcer`, which manages prerequisite step tracking and premature terminal attempt limits.
- [`src/forge/guardrails/error_tracker.py`](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/error_tracker.py) – Defines `ErrorTracker` for maintaining retry and tool-error budgets.
- [`src/forge/guardrails/nudge.py`](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/nudge.py) – Simple dataclass representing nudge messages passed back to the model.
- [`examples/foreign_loop.py`](https://github.com/antoinezambelli/forge/blob/main/examples/foreign_loop.py) – End-to-end demonstration of both simple and granular APIs in a standalone script.
- [`tests/unit/test_guardrails.py`](https://github.com/antoinezambelli/forge/blob/main/tests/unit/test_guardrails.py) – Comprehensive test suite validating the guardrails contract and state transitions.

## Summary

- **Forge guardrails** provide response validation, step enforcement, and error tracking through a façade exposing only `check()` and `record()` methods.
- The façade bundles three components: **ResponseValidator**, **StepEnforcer**, and **ErrorTracker**, each independently usable for granular control.
- **Action types** (`execute`, `retry`, `step_blocked`, `fatal`) dictate whether to run tools, re-prompt with a nudge, or abort the workflow.
- **Terminal tools** are blocked until required steps are satisfied, preventing premature completion of multi-step tasks.
- All components reside in `src/forge/guardrails/` and are side-effect-free, making them safe to embed in any external orchestration framework.

## Frequently Asked Questions

### What is the difference between `retry` and `step_blocked` actions?

The `retry` action indicates the LLM returned unstructured text rather than valid tool calls, triggering a format-correction nudge. The `step_blocked` action means the model attempted to call a terminal tool (such as "answer") before completing required prerequisite steps (such as "search" followed by "lookup"), prompting a nudge that cites the missing steps.

### How do I configure error budgets for consecutive failures?

Pass `max_retries` and `max_tool_errors` parameters to the `ErrorTracker` constructor (or the `Guardrails` façade). When consecutive invalid responses exceed `max_retries`, or when tool execution failures exceed `max_tool_errors`, the system returns a `fatal` action, forcing the orchestration loop to abort.

### Can guardrails be used independently of Forge's WorkflowRunner?

Yes. According to the source code in `src/forge/guardrails/`, all components are pure-Python with no hidden side-effects or framework dependencies. You can embed them into BFCL harnesses, OpenClaw pipelines, or custom agent loops without importing the full Forge runtime.

### What constitutes a terminal tool in step enforcement?

A terminal tool is the final operation that should only execute after all required steps are satisfied. Configure it via the `terminal_tool` string parameter in the façade or the `terminal_tools` frozenset in `StepEnforcer`. Attempting to invoke it prematurely triggers `step_blocked` nudges until prerequisites are met or the premature attempt limit is exhausted.