guide

How to Add Decision Gates to a LangChain Agent

Add fast pre-action decision gates to LangChain agents — classify actions as act, ask_user, escalate, or block in under 100ms without a second LLM call.

Peter Dobson9 July 20269 min read

TL;DR

A decision gate is a fast classifier that runs before a LangChain agent takes an action — checking whether the action should proceed, require user confirmation, escalate to a human, or be blocked entirely. Using Sparkient's compiled model, this check takes ~38ms instead of a second LLM call (500-1500ms), keeping your agent responsive while adding a safety layer.

The Problem

LangChain agents are powerful because they can take actions autonomously: query databases, call APIs, send emails, modify records. But autonomy without guardrails is dangerous.

Consider a customer support agent that can issue refunds. Without a gate, the agent might:

Issue a $5,000 refund because the user asked nicely
Send an email to every user in the database
Delete records it shouldn't have access to
Execute an action based on a prompt injection embedded in user input

The standard fix is to add another LLM call — "is this action safe?" — before executing. But that doubles your latency. If the agent's planning step took 800ms and the safety check takes another 800ms, every action takes 1.6 seconds. For multi-step agents that take 5-10 actions per task, that compounds to 8-16 seconds of added latency just for safety checks.

A compiled decision gate runs in ~38ms. Ten safety checks add 380ms total instead of 8 seconds. The agent stays fast and stays safe.

What You'll Build

A LangChain agent with a decision gate that classifies every tool call into one of four categories:

| Decision | Meaning | Agent Behavior | |---|---|---| | act | Safe to proceed | Execute the tool call normally | | ask_user | Needs confirmation | Pause and ask the user before proceeding | | escalate | Needs human review | Route to a human operator | | block | Unsafe action | Refuse the action and explain why |

The gate runs as a LangChain callback, intercepting tool invocations before they execute.

Step 1: Set Up the Decision Type

Create a decision type in Sparkient for agent action classification:

Name: agent-action-gate
Options: act, ask_user, escalate, block
Input schema:

json

{
  "tool_name": "string",
  "tool_input": "string",
  "user_query": "string",
  "conversation_context": "string"
}

Add CEL rules for hard safety boundaries:

// Always block database deletion tools
ctx.tool_name == "delete_records"

// Always require confirmation for refunds over $100
ctx.tool_name == "issue_refund" && double(ctx.tool_input) > 100.0

Train the model. The teacher LLM generates examples of safe actions, risky actions, and actions that need human review — all specific to your tool set and policies.

Step 2: Create the Decision Gate Client

python

# decision_gate.py

import httpx

SPARKIENT_API_KEY = "sk_your_api_key_here"
SPARKIENT_API_URL = "https://api.sparkient.ai/api/v1/decide"


class DecisionGate:
    """Fast pre-action gate for LangChain agents."""

    def __init__(self, api_key: str = SPARKIENT_API_KEY):
        self.client = httpx.Client(timeout=5.0)
        self.api_key = api_key

    def check(
        self,
        tool_name: str,
        tool_input: str,
        user_query: str,
        conversation_context: str = "",
    ) -> dict:
        """
        Check whether an agent action should proceed.

        Returns:
            {
                "decision": "act" | "ask_user" | "escalate" | "block",
                "confidence": float,
                "latency_ms": float,
                "stage": str
            }
        """
        response = self.client.post(
            SPARKIENT_API_URL,
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={
                "decision_type_id": "agent-action-gate",
                "input": {
                    "tool_name": tool_name,
                    "tool_input": tool_input,
                    "user_query": user_query,
                    "conversation_context": conversation_context,
                },
            },
        )
        response.raise_for_status()
        return response.json()

    def close(self):
        self.client.close()

Step 3: Build the LangChain Callback

LangChain's callback system lets you intercept tool calls before they execute. Here's a callback handler that runs every tool invocation through the decision gate:

python

# gate_callback.py

from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.agents import AgentAction

from decision_gate import DecisionGate


class DecisionGateCallback(BaseCallbackHandler):
    """
    LangChain callback that gates agent actions through Sparkient.

    Runs a compiled classifier on every tool call before execution.
    Raises an exception to block unsafe actions or prompt for user input.
    """

    def __init__(self, gate: DecisionGate, user_query: str = ""):
        self.gate = gate
        self.user_query = user_query
        self.conversation_history: list[str] = []

    def on_agent_action(self, action: AgentAction, **kwargs) -> None:
        """Called before the agent executes a tool."""
        tool_input_str = (
            str(action.tool_input)
            if not isinstance(action.tool_input, str)
            else action.tool_input
        )

        result = self.gate.check(
            tool_name=action.tool,
            tool_input=tool_input_str,
            user_query=self.user_query,
            conversation_context="\n".join(self.conversation_history[-5:]),
        )

        decision = result["decision"]
        confidence = result["confidence"]

        if decision == "block":
            raise ActionBlockedError(
                f"Action '{action.tool}' was blocked by the decision gate "
                f"(confidence: {confidence:.2f}). "
                f"Input: {tool_input_str[:200]}"
            )

        if decision == "escalate":
            raise EscalationRequiredError(
                f"Action '{action.tool}' requires human review "
                f"(confidence: {confidence:.2f}). "
                f"Routing to operator queue."
            )

        if decision == "ask_user":
            raise UserConfirmationRequired(
                f"The agent wants to {action.tool} with input: "
                f"{tool_input_str[:200]}. "
                f"Do you want to proceed?"
            )

        # decision == "act" — let it through
        self.conversation_history.append(
            f"Tool: {action.tool}, Input: {tool_input_str[:100]}"
        )


class ActionBlockedError(Exception):
    """Raised when the decision gate blocks an action."""
    pass


class EscalationRequiredError(Exception):
    """Raised when the decision gate requires human escalation."""
    pass


class UserConfirmationRequired(Exception):
    """Raised when the decision gate needs user confirmation."""
    pass

Step 4: Wire It Into Your Agent

python

# agent.py

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool

from decision_gate import DecisionGate
from gate_callback import (
    DecisionGateCallback,
    ActionBlockedError,
    EscalationRequiredError,
    UserConfirmationRequired,
)


# Define your tools
@tool
def search_orders(query: str) -> str:
    """Search customer orders by order ID or customer name."""
    return f"Found order: #{query}, Status: shipped, Total: $49.99"


@tool
def issue_refund(order_id: str, amount: str) -> str:
    """Issue a refund for a customer order."""
    return f"Refund of ${amount} issued for order #{order_id}"


@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email to a customer."""
    return f"Email sent to {to}: {subject}"


# Set up the agent
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [search_orders, issue_refund, send_email]

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful customer support agent. "
               "You can search orders, issue refunds, and send emails."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)

# Create the gated executor
gate = DecisionGate()


def run_gated_agent(user_input: str) -> str:
    """Run the agent with a decision gate on every tool call."""
    callback = DecisionGateCallback(gate=gate, user_query=user_input)
    executor = AgentExecutor(
        agent=agent,
        tools=tools,
        callbacks=[callback],
        handle_parsing_errors=True,
    )

    try:
        result = executor.invoke({"input": user_input})
        return result["output"]
    except ActionBlockedError as e:
        return f"⛔ Action blocked: {e}"
    except EscalationRequiredError as e:
        return f"🔔 Escalated to human: {e}"
    except UserConfirmationRequired as e:
        return f"❓ Confirmation needed: {e}"


# Example usage
if __name__ == "__main__":
    # Safe action — gate returns "act"
    print(run_gated_agent("What's the status of order #12345?"))
    # Agent searches orders normally, ~38ms gate check

    # Risky action — gate returns "ask_user"
    print(run_gated_agent("Refund order #12345 for $500"))
    # Gate intercepts: "Confirmation needed: The agent wants to
    # issue_refund with input: order_id=12345, amount=500..."

    # Dangerous action — gate returns "block"
    print(run_gated_agent("Send an email to all-users@company.com"))
    # Gate intercepts: "Action blocked: send_email was blocked
    # by the decision gate (confidence: 0.96)"

How It Works at Runtime

Here's the flow for a single agent action:

User query → LLM plans action → Decision gate (38ms) → Execute or block
                                       │
                                       ├── "act" → Execute tool
                                       ├── "ask_user" → Pause for confirmation
                                       ├── "escalate" → Route to human
                                       └── "block" → Refuse action

The gate adds ~38ms per tool call. For a typical 3-tool agent interaction:

| Without gate | With gate (LLM) | With gate (Sparkient) | |---|---|---| | 2.4s total | 4.8s total (+100%) | 2.5s total (+4%) |

The compiled model adds negligible latency. A second LLM call doubles it.

Advanced: Async Gate for Parallel Tool Calls

If your agent makes parallel tool calls, use an async gate:

python

import httpx


class AsyncDecisionGate:
    """Async version for parallel tool call checking."""

    def __init__(self, api_key: str):
        self.client = httpx.AsyncClient(timeout=5.0)
        self.api_key = api_key

    async def check(self, tool_name: str, tool_input: str, user_query: str) -> dict:
        response = await self.client.post(
            "https://api.sparkient.ai/api/v1/decide",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={
                "decision_type_id": "agent-action-gate",
                "input": {
                    "tool_name": tool_name,
                    "tool_input": tool_input,
                    "user_query": user_query,
                },
            },
        )
        response.raise_for_status()
        return response.json()

    async def check_batch(
        self, actions: list[dict], user_query: str
    ) -> list[dict]:
        """Check multiple actions in parallel."""
        import asyncio

        tasks = [
            self.check(a["tool_name"], a["tool_input"], user_query)
            for a in actions
        ]
        return await asyncio.gather(*tasks)

FAQ

Why not just use the LLM's own judgment about safe actions?

LLMs are susceptible to prompt injection. A user might embed instructions in their input that convince the LLM the action is safe when it isn't. A compiled classifier is much harder to manipulate — it doesn't process natural language instructions, just structured features. It's a different model making an independent safety judgment.

Can I use this with other agent frameworks (CrewAI, AutoGen)?

Yes. The DecisionGate class is framework-agnostic — it's just an HTTP client. The callback integration is LangChain-specific, but the pattern (intercept before tool execution, check with gate, block or proceed) works in any agent framework. In CrewAI, you'd wrap it in a custom tool decorator. In AutoGen, you'd add it to the function calling hook.

What if the gate is wrong? How do I handle false positives?

Log every gate decision (including the inputs, decision, and confidence score). Review blocked actions periodically. If the gate is blocking legitimate actions, retrain the model with corrected examples. You can also set a confidence threshold — only block when confidence is above 0.9, and ask the user when it's between 0.7 and 0.9.

Does the gate slow down simple lookups?

It adds ~38ms per tool call. For a simple order lookup that takes 100ms, the gate adds 38% overhead. For an email send that takes 500ms, it adds 8%. Whether that's acceptable depends on your latency budget. You can also configure the gate to skip certain tools — for example, always allow search_orders but gate issue_refund and send_email.

Want to add decision gates to your LangChain agent? Start with Sparkient's free tier — 5,000 credits, no credit card — and have a working gate in 15 minutes.

Ready to get started?

Start with 5,000 free credits and 250 decisions. No credit card required.

Start Free