guide

How to Add Decision Gates to a LangChain Agent

Build and evaluate a pre-action decision gate for LangChain agents using rules and a compiled classifier.

Peter Dobson9 July 20268 min read

TL;DR

A decision gate runs before an agent action and chooses among outcomes such as proceed, ask the user, escalate, or block. Sparkient is a candidate when that choice is repeated and measurable: the compiled stage targets under 100ms, but you must validate false allows, false blocks, full-loop latency, and the handling of uncertain or high-impact actions.

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

One approach is to add another live-model call—"is this action safe?"—before execution. That adds another model request and its token, latency, retry, and provider profile to every gated action.

A separately trained compiled gate can have a smaller runtime profile for stable outcomes. Trace the complete workflow before and after; a faster gate is not evidence that the policy is 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

{
  "type": "object",
  "properties": {
    "tool_name": {"type": "string"},
    "tool_input": {"type": "string"},
    "user_query": {"type": "string"},
    "conversation_context": {"type": "string"}
  },
  "required": ["tool_name", "tool_input", "user_query"]
}

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 os

import httpx

SPARKIENT_API_KEY = os.environ["SPARKIENT_API_KEY"]
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": "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 after the gate returns "act"

    # 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 → Execute or block
                                       │
                                       ├── "act" → Execute tool
                                       ├── "ask_user" → Pause for confirmation
                                       ├── "escalate" → Route to human
                                       └── "block" → Refuse action

Measure p50, p95, and p99 gate latency, the number of gated actions per task, optional cloud escalation, and the full task duration. Four controlled synthetic Sparkient runs report 33–42ms batch-average time per item in other domains; they do not establish an agent-gating result or per-request p95.

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

Any component that consumes attacker-controlled text can fail under distribution shift or adversarial input. A separately trained classifier can provide a different signal, while CEL rules enforce deterministic constraints, but neither replaces least privilege, tool allowlists, user confirmation for consequential actions, audit logs, or a task-specific security evaluation.

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?

The overhead depends on the cloud network path, model, stage mix, and client. Measure it per tool and across a complete task. You can also skip low-risk read-only tools and reserve the gate for consequential actions, but encode that distinction in an explicit policy rather than assuming a tool is harmless.

Want to evaluate a decision gate in your LangChain agent? Start with Sparkient's free tier—5,000 credits, no credit card—and test it in shadow mode before enforcement.

Ready to get started?

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

Start Free