el): tool_name: str parameters: Dict[str, Any]

class SchemaValidator: def init(self, allowed_tools: List[str]): self.allowed = set(allowed_tools) self.rescue_pattern = re.compile(r'{.*}', re.DOTALL)

def validate(self, raw_output: str) -> ToolContract:
    try:
        parsed = json.loads(raw_output)
        return ToolContract(**parsed)
    except (json.JSONDecodeError, ValidationError):
        return self._rescue_parse(raw_output)

def _rescue_parse(self, raw: str) -> ToolContract:
    match = self.rescue_pattern.search(raw)
    if not match:
        raise ValueError("No valid JSON structure detected")
    cleaned = match.group(0)
    parsed = json.loads(cleaned)
    if parsed.get("tool_name") not in self.allowed:
        raise ValueError(f"Unknown tool: {parsed.get('tool_name')}")
    return ToolContract(**parsed)

**Why this design:** Rescue parsing extracts valid JSON from noisy model outputs without consuming a full retry budget. The regex fallback handles truncated responses or markdown-wrapped payloads. Strict schema validation prevents execution of hallucinated tools, which is the primary cause of silent workflow failures.

### Step 2: Implement Step-Gated Execution
Multi-step workflows require ordered execution. Models frequently attempt to skip prerequisites or repeat terminal steps. A state machine enforces dependency chains.

```python
from enum import Enum, auto
from typing import Set

class WorkflowState(Enum):
    IDLE = auto()
    SEARCHING = auto()
    LOOKING_UP = auto()
    RESOLVING = auto()
    COMPLETE = auto()

class StepGatekeeper:
    def __init__(self, required_sequence: List[WorkflowState]):
        self.sequence = required_sequence
        self.completed: Set[WorkflowState] = set()
        self.current_index = 0

    def can_execute(self, requested_step: WorkflowState) -> bool:
        if requested_step not in self.sequence:
            return False
        if requested_step in self.completed:
            return False
        return requested_step == self.sequence[self.current_index]

    def advance(self, step: WorkflowState) -> str:
        if not self.can_execute(step):
            remaining = self.sequence[self.current_index:]
            return f"Prerequisites not met. Required: {[s.name for s in remaining]}"
        self.completed.add(step)
        self.current_index += 1
        if self.current_index >= len(self.sequence):
            return "Workflow complete."
        return f"Step {step.name} accepted. Next: {self.sequence[self.current_index].name}"

Why this design: Hardcoding step dependencies eliminates shortcutting and circular execution. The gatekeeper returns actionable feedback instead of silent failures, enabling the model to self-correct without external intervention.

Step 3: Hardware-Aware Context Management

Context windows fill rapidly during agentic loops. Unbounded growth causes VRAM exhaustion and instruction drift. A budget manager monitors token consumption and triggers compaction strategies.

import math

class ContextBudgetManager:
    def __init__(self, max_tokens: int, compaction_threshold: float = 0.75):
        self.max_tokens = max_tokens
        self.threshold = compaction_threshold
        self.current_usage = 0
        self.history: List[Dict[str, str]] = []

    def add_turn(self, role: str, content: str) -> None:
        estimated_tokens = math.ceil(len(content) / 4)
        self.current_usage += estimated_tokens
        self.history.append({"role": role, "content": content})

        if self.current_usage > self.max_tokens * self.threshold:
            self._compact()

    def _compact(self) -> None:
        keep_recent = self.history[-4:]
        older = self.history[:-4]
        summary = self._generate_summary(older)
        self.history = [{"role": "system", "content": summary}] + keep_recent
        self.current_usage = math.ceil(len(summary) / 4) + sum(
            math.ceil(len(t["content"]) / 4) for t in keep_recent
        )

    def _generate_summary(self, turns: List[Dict]) -> str:
        actions = [t["content"] for t in turns if t["role"] == "assistant"]
        return f"Previous actions: {' | '.join(actions)}. Maintain current workflow state."

Why this design: Token estimation prevents VRAM blowout by triggering compaction before the window saturates. Tiered compaction preserves recent turns verbatim while summarizing older context, maintaining instruction fidelity without exhausting memory. The threshold is configurable based on available GPU RAM.

Step 4: Targeted Retry Nudges

When validation fails or steps are blocked, blind retries reproduce the same error. A nudge system injects precise correction instructions into the conversation history.

class RetryNudgeEngine:
    def __init__(self, max_retries: int = 3):
        self.max_retries = max_retries
        self.attempts: Dict[str, int] = {}

    def generate_nudge(self, error_type: str, context: str) -> str:
        key = f"{error_type}_{context}"
        self.attempts[key] = self.attempts.get(key, 0) + 1
        if self.attempts[key] > self.max_retries:
            raise RuntimeError(f"Retry budget exhausted for {error_type}")

        templates = {
            "schema_violation": "Your last response did not match the required JSON schema. Use only registered tools and valid data types.",
            "step_blocked": "You cannot proceed yet. Complete the required prerequisites before calling this tool.",
            "ambiguous_output": "You must output a valid tool call. Do not generate plain text responses."
        }
        return templates.get(error_type, "Please correct your previous response.")

Why this design: Retry budgets prevent infinite loops. Template-based nudges exploit the model's training priors for error correction, transforming blind retries into guided state recovery. The engine tracks attempt counts per error type to avoid wasting tokens on unresolvable failures.

Pitfall Guide

1. Blind Retry Loops

Explanation: Resending the exact same prompt after a failure causes the model to reproduce the identical error. This consumes tokens, fills context windows, and degrades latency. Fix: Implement stateful retry tracking with targeted nudges. Inject specific error descriptions and schema reminders instead of repeating the original prompt.

2. Unbounded Context Growth

Explanation: Agentic loops accumulate request/response pairs exponentially. Without compaction, the context window saturates, pushing system instructions and tool schemas out of memory. Fix: Deploy a token budget manager with configurable compaction thresholds. Use tiered strategies that preserve recent turns while summarizing older history.

3. Text/Tool Choice Ambiguity

Explanation: Small models struggle to decide between generating conversational text and invoking a tool. Allowing free choice causes the orchestration loop to stall or proceed with missing data. Fix: Eliminate the choice entirely. Inject a synthetic respond tool into the schema so the model remains in tool-calling mode throughout the workflow.

4. Premature Step Execution

Explanation: Models frequently attempt to skip intermediate steps or execute terminal actions before prerequisites are satisfied. This breaks dependency chains and produces invalid outputs. Fix: Implement a step gatekeeper with explicit state transitions. Block unauthorized calls and return actionable prerequisite lists.

5. Static Context Compaction

Explanation: Fixed-window compaction discards critical early instructions or tool definitions, causing the model to hallucinate capabilities or forget constraints. Fix: Use adaptive compaction that preserves system prompts, tool schemas, and workflow state summaries. Only compress conversational turns.

6. Ignoring Hardware Memory Limits

Explanation: Context management that only tracks tokens ignores VRAM constraints. Large context windows on consumer GPUs cause OOM crashes or severe thrashing. Fix: Map token budgets to available VRAM at startup. Reduce compaction thresholds on hardware with limited memory. Monitor GPU utilization during runtime.

7. Over-Validating Tool Schemas

Explanation: Strict JSON validation that rejects minor formatting deviations (trailing commas, extra whitespace) causes unnecessary retries and degrades throughput. Fix: Implement rescue parsing with regex extraction. Normalize payloads before validation. Only reject structurally invalid or semantically incorrect outputs.

Production Bundle

Action Checklist

• Define explicit tool contracts with Pydantic schemas and register allowed tools at initialization
• Implement a step gatekeeper that enforces workflow dependencies and blocks premature execution
• Deploy a context budget manager with hardware-aware compaction thresholds
• Replace blind retries with targeted nudges that specify exact error types and corrections
• Inject a synthetic response tool to eliminate text/tool ambiguity in small models
• Add observability hooks to track retry counts, compaction triggers, and step transitions
• Configure circuit breakers that halt workflows after exhausting retry budgets
• Test compaction strategies under VRAM constraints before production deployment

Decision Matrix

Scenario	Recommended Approach	Why	Cost Impact
Local 8B Model Deployment	Tiered compaction + synthetic response tool	Prevents context overflow and eliminates text/tool ambiguity	Near-zero API cost, high GPU utilization
Cloud Frontier API	Sliding window compaction + strict schema validation	Reduces token spend while maintaining high completion rates	Moderate API cost, lower latency
Hybrid Workflow (Local + Cloud)	Proxy validation layer + step gatekeeper	Routes simple steps locally, escalates complex reasoning to cloud	Balanced cost, optimized throughput
High-Throughput Batch Processing	No compaction + aggressive retry nudges	Prioritizes speed over memory efficiency; relies on retry correction	Higher compute cost, faster completion

Configuration Template

reliability_engine:
  context:
    max_tokens: 8192
    compaction_threshold: 0.75
    strategy: tiered
    preserve_system_prompt: true
  validation:
    rescue_parsing: true
    strict_schema: true
    allowed_tools: ["search", "lookup", "resolve", "respond"]
  execution:
    max_retries: 3
    step_enforcement: true
    required_sequence: ["SEARCHING", "LOOKING_UP", "RESOLVING"]
  hardware:
    vram_budget_gb: 24
    compaction_on_oom: true
    token_to_vram_ratio: 0.003

Quick Start Guide

1. Initialize the reliability engine with your tool registry and workflow sequence. Configure context thresholds based on available GPU memory.
2. Wrap your model inference loop with the schema validator and step gatekeeper. Route all model outputs through validation before execution.
3. Attach the context budget manager to your conversation history. Enable compaction when token usage exceeds the configured threshold.
4. Deploy retry nudges for validation failures and step blocks. Monitor attempt counts to prevent infinite loops.
5. Run integration tests with synthetic workflows. Verify compaction triggers, step transitions, and retry behavior under VRAM constraints before production rollout.