How to Use Forge Guardrails Middleware in Custom Orchestration Loops

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 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:

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.

Granular Control Using Individual Components

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

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 – Implements the Guardrails façade and CheckResult dataclass, coordinating the three sub-components.
• src/forge/guardrails/response_validator.py – Contains the ResponseValidator class responsible for parsing and rescue logic.
• src/forge/guardrails/step_enforcer.py – Houses StepEnforcer, which manages prerequisite step tracking and premature terminal attempt limits.
• src/forge/guardrails/error_tracker.py – Defines ErrorTracker for maintaining retry and tool-error budgets.
• src/forge/guardrails/nudge.py – Simple dataclass representing nudge messages passed back to the model.
• examples/foreign_loop.py – End-to-end demonstration of both simple and granular APIs in a standalone script.
• 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}