Validation Patterns for E-Commerce Logistics RAG

Tuhin Paul
3h
98
0
0

Article

The Fragility of LLM Tool Use in Production

LangChain’s tool calling interface is elegant in notebooks and brittle in warehouses. When an LLM hallucinates a tracking number format, passes a string where an enum is expected, or omits a required field during peak shipping season, the consequences cascade: failed API calls trigger retries that slam carrier APIs, malformed warehouse queries return wrong pick lists, and customers receive conflicting delivery estimates. The root problem is architectural: LangChain treats tool validation as a schema concern, but in logistics it’s a business continuity concern. Pydantic models catch type errors; they don’t catch that carrier="fedex" with service_level="ground_freight" is a physically impossible combination, or that a ZIP code 00000 passes regex validation but represents no real destination. This article demonstrates building a defense-in-depth tool validation layer for e-commerce logistics RAG, integrated into a stateful LangGraph multi-agent system. We validate at four levels: schema, semantic, contextual, and temporal—each catching failure modes the previous level cannot.

Real-Time Use Case: Multi-Carrier Fulfillment Intelligence

The Scenario

A DTC e-commerce company processing 50K orders/day across 3PL warehouses uses an AI-powered logistics operations platform. Warehouse supervisors, customer service reps, and carrier liaisons query the system:

"Where is order ORD-2024-884721? Customer says it's been stuck for 5 days."
"Reroute all FedEx Ground packages destined for Miami to UPS after the hurricane alert."
"Why did we overspend $12K on expedited shipping last week in the Southwest region?"

The system integrates six live APIs: WMS (warehouse management), OMS (order management), three carrier APIs (FedEx, UPS, USPS), and a freight rate engine. Every tool call touches production systems where malformed inputs cause real operational damage.

Failure Modes We Must Prevent

Input Problem	Naive Tool Behavior	Business Impact
Tracking number with transposed digits	API returns "not found"; agent retries 3x	Carrier API rate limit hit; legitimate queries blocked
Invalid carrier + service level combo	API error; agent tries alternative carriers randomly	Packages routed to wrong service tier; SLA breach
Ambiguous warehouse code ("WH-3" vs "WH-03")	Wrong facility queried; inventory mismatch	Pick list sent to wrong building; shipment delayed
Date range spanning future dates	Rate engine returns projected rates as actual	Cost analysis wrong by 40%; budget decisions corrupted
Missing optional field that's conditionally required	Partial API response treated as complete	Customer given incomplete tracking info; support ticket escalates
Stale session context (old order state)	Tool executes against outdated assumptions	Reroute applied to already-delivered package; carrier chargeback

Architecture

Implementation

Step 1: State with Validation-Aware Fields

from typing import Annotated, List, Dict, Any, Optional, Literal
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from datetime import datetime
import operator

class ToolAuditEntry(TypedDict):
    """Immutable record of every tool call attempt."""
    timestamp: datetime
    tool_name: str
    raw_args: Dict[str, Any]
    validation_results: Dict[str, Any]  # Per-layer results
    repaired_args: Optional[Dict[str, Any]]
    execution_status: Literal["success", "failed", "rejected", "repaired"]
    latency_ms: float
    error_detail: Optional[str]

class LogisticsState(TypedDict):
    messages: Annotated[list, add_messages]
    
    # Session context for contextual validation
    session_context: Dict[str, Any]  # {"active_order_id": ..., "warehouse": ..., "user_role": ...}
    
    # Immutable audit trail
    tool_audit_log: Annotated[List[ToolAuditEntry], operator.add]
    
    # Cached validation results (avoid re-validating identical calls)
    validation_cache: Dict[str, Dict[str, Any]]
    
    # Current query
    current_query: str
    
    # Final response
    final_response: Optional[str]
    session_id: str

Step 2: Define Tools with Rich Schemas (Layer 1 Foundation)

Start with strict Pydantic schemas that encode as much business logic as possible at the type level.

from pydantic import BaseModel, Field, field_validator, model_validator
from enum import Enum
import re

class Carrier(str, Enum):
    FEDEX = "fedex"
    UPS = "ups"
    USPS = "usps"
    DHL = "dhl"

class ServiceLevel(str, Enum):
    GROUND = "ground"
    EXPRESS = "express"
    OVERNIGHT = "overnight"
    FREIGHT_LTL = "freight_ltl"
    FREIGHT_FTL = "freight_ftl"
    INTERNATIONAL_ECONOMY = "international_economy"

class TrackShipmentInput(BaseModel):
    """Schema-level validation catches format errors before any API call."""
    
    tracking_number: str = Field(
        description="Carrier tracking number",
        min_length=8,
        max_length=34
    )
    carrier: Carrier = Field(description="Shipping carrier")
    
    @field_validator("tracking_number")
    @classmethod
    def validate_tracking_format(cls, v: str, info) -> str:
        """Format validation per carrier. Catches transposed/invalid numbers."""
        v = v.strip().upper()
        
        patterns = {
            Carrier.FEDEX: r'^\d{12}$',           # 12-digit numeric
            Carrier.UPS: r'^1Z[A-HJ-NP-TV-Z0-9]{16}$',  # 1Z prefix + 16 alphanum
            Carrier.USPS: r'^(9[0-9]|1[0-9]|[2-9]\d)\d{18}$',  # 20-digit
            Carrier.DHL: r'^\d{10}$',             # 10-digit numeric
        }
        
        # Extract carrier from validation context if available
        carrier = info.data.get("carrier")
        if carrier and carrier in patterns:
            if not re.match(patterns[carrier], v):
                raise ValueError(
                    f"Tracking number '{v}' does not match {carrier.value} format. "
                    f"Expected pattern: {patterns[carrier]}"
                )
        return v


class RerouteShipmentInput(BaseModel):
    """Semantic constraints encoded at schema level where possible."""
    
    order_id: str = Field(pattern=r'^ORD-\d{4}-\d{6}$')
    original_carrier: Carrier
    new_carrier: Carrier
    new_service_level: ServiceLevel
    reason_code: str = Field(min_length=2, max_length=10)
    
    @model_validator(mode="after")
    def validate_carrier_service_compatibility(self):
        """
        Layer 2 preview: Some business rules CAN be enforced at schema level.
        More complex rules are deferred to the semantic validator.
        """
        incompatible = {
            (Carrier.USPS, ServiceLevel.FREIGHT_LTL),
            (Carrier.USPS, ServiceLevel.FREIGHT_FTL),
            (Carrier.UPS, ServiceLevel.INTERNATIONAL_ECONOMY),  # UPS uses different naming
        }
        
        if (self.new_carrier, self.new_service_level) in incompatible:
            raise ValueError(
                f"{self.new_carrier.value} does not support {self.new_service_level.value}. "
                f"Valid combinations documented in carrier matrix."
            )
        
        if self.original_carrier == self.new_carrier and self.reason_code != "SERVICE_UPGRADE":
            raise ValueError(
                "Same-carrier reroute requires reason_code='SERVICE_UPGRADE'"
            )
        
        return self

Step 3: Semantic Validator (Layer 2) — Business Rules Beyond Schema

class SemanticValidator:
    """
    Validates business logic that Pydantic cannot express.
    Stateless, deterministic, fast (<5ms per validation).
    """
    
    # Carrier-service compatibility matrix (loaded from config DB at init)
    CARRIER_SERVICE_MATRIX = {
        Carrier.FEDEX: {ServiceLevel.GROUND, ServiceLevel.EXPRESS, ServiceLevel.OVERNIGHT, ServiceLevel.FREIGHT_LTL},
        Carrier.UPS: {ServiceLevel.GROUND, ServiceLevel.EXPRESS, ServiceLevel.OVERNIGHT, ServiceLevel.FREIGHT_LTL, ServiceLevel.FREIGHT_FTL},
        Carrier.USPS: {ServiceLevel.GROUND, ServiceLevel.EXPRESS, ServiceLevel.INTERNATIONAL_ECONOMY},
        Carrier.DHL: {ServiceLevel.EXPRESS, ServiceLevel.INTERNATIONAL_ECONOMY},
    }
    
    # Warehouse code canonicalization map
    WAREHOUSE_ALIASES = {
        "WH-3": "WH-03", "WH3": "WH-03", "WAREHOUSE-3": "WH-03",
        "WH-12": "WH-12", "WH12": "WH-12",
        "DFW": "WH-07", "LAX": "WH-02", "JFK": "WH-11",
    }
    
    # Known-invalid ZIP codes that pass regex
    INVALID_ZIPS = {"00000", "99999", "12345"}  # Test/placeholder codes
    
    @classmethod
    def validate_track_shipment(cls, args: TrackShipmentInput) -> Dict[str, Any]:
        issues = []
        repairs = {}
        
        # Check for known test/tracking numbers that shouldn't hit prod APIs
        test_numbers = {"1Z999AA10123456784", "123456789012"}
        if args.tracking_number in test_numbers:
            issues.append({
                "severity": "block",
                "message": f"Test tracking number detected. Blocked to prevent API pollution.",
                "field": "tracking_number"
            })
        
        # Luhn-style checksum for FedEx (catches single-digit transpositions)
        if args.carrier == Carrier.FEDEX:
            digits = [int(d) for d in args.tracking_number]
            checksum = sum(digits[:-1]) % 10
            if checksum != digits[-1]:
                issues.append({
                    "severity": "warn",
                    "message": f"FedEx tracking checksum mismatch. Possible transposition.",
                    "field": "tracking_number",
                    "suggestion": "Verify with customer or check OMS for correct number"
                })
        
        return {
            "valid": not any(i["severity"] == "block" for i in issues),
            "issues": issues,
            "repairs": repairs
        }
    
    @classmethod
    def validate_reroute(cls, args: RerouteShipmentInput, session_context: Dict) -> Dict[str, Any]:
        issues = []
        repairs = {}
        
        # Validate carrier-service compatibility from live matrix
        valid_services = cls.CARRIER_SERVICE_MATRIX.get(args.new_carrier, set())
        if args.new_service_level not in valid_services:
            issues.append({
                "severity": "block",
                "message": f"{args.new_carrier.value} does not offer {args.new_service_level.value}",
                "field": "new_service_level",
                "valid_options": [s.value for s in valid_services]
            })
        
        # Warehouse-aware routing rules
        active_wh = session_context.get("warehouse")
        if active_wh:
            canonical = cls.WAREHOUSE_ALIASES.get(active_wh.upper(), active_wh)
            if canonical != active_wh:
                repairs["warehouse_canonical"] = canonical
                issues.append({
                    "severity": "info",
                    "message": f"Warehouse code '{active_wh}' normalized to '{canonical}'",
                    "field": "warehouse"
                })
        
        # Reason code validation against allowed set for user role
        user_role = session_context.get("user_role", "viewer")
        allowed_reasons = {
            "viewer": set(),  # Can't reroute
            "cs_rep": {"CUSTOMER_REQUEST", "ADDRESS_CORRECTION", "SERVICE_UPGRADE"},
            "supervisor": {"CUSTOMER_REQUEST", "ADDRESS_CORRECTION", "SERVICE_UPGRADE", 
                          "CARRIER_FAILURE", "WEATHER_DIVERSION", "COST_OPTIMIZATION"},
            "admin": {"CUSTOMER_REQUEST", "ADDRESS_CORRECTION", "SERVICE_UPGRADE",
                     "CARRIER_FAILURE", "WEATHER_DIVERSION", "COST_OPTIMIZATION", "SYSTEM_TEST"}
        }
        
        if args.reason_code not in allowed_reasons.get(user_role, set()):
            issues.append({
                "severity": "block",
                "message": f"Role '{user_role}' cannot use reason_code '{args.reason_code}'",
                "field": "reason_code",
                "valid_options": list(allowed_reasons.get(user_role, set()))
            })
        
        return {
            "valid": not any(i["severity"] == "block" for i in issues),
            "issues": issues,
            "repairs": repairs
        }

Step 4: Contextual Validator (Layer 3) — State-Aware Checks

This layer accesses LangGraph state to validate against session context and prior tool results. This is impossible with standalone LangChain tools.

class ContextualValidator:
    """
    Validates tool calls against current graph state.
    Prevents actions based on stale or inconsistent context.
    """
    
    @staticmethod
    async def validate(state: LogisticsState, tool_name: str, args: Dict) -> Dict[str, Any]:
        issues = []
        
        session = state.get("session_context", {})
        audit_log = state.get("tool_audit_log", [])
        
        # CONTEXT CHECK: Is the referenced order still in a mutable state?
        if tool_name in ("reroute_shipment", "cancel_shipment", "update_address"):
            order_id = args.get("order_id")
            if order_id:
                # Check if a prior tool call already modified this order
                prior_modifications = [
                    entry for entry in audit_log
                    if entry.get("tool_name") in ("reroute_shipment", "cancel_shipment")
                    and entry.get("raw_args", {}).get("order_id") == order_id
                    and entry.get("execution_status") == "success"
                ]
                
                if prior_modifications:
                    last_mod = prior_modifications[-1]
                    age_s = (datetime.utcnow() - last_mod["timestamp"]).total_seconds()
                    
                    if age_s < 300:  # 5-minute cooldown
                        issues.append({
                            "severity": "block",
                            "message": (
                                f"Order {order_id} was modified {age_s:.0f}s ago via "
                                f"'{last_mod['tool_name']}'. Cooldown period: 300s. "
                                f"Wait or escalate to supervisor."
                            ),
                            "field": "order_id"
                        })
        
        # AMBIGUITY RESOLUTION: If warehouse code is ambiguous, check session context
        if tool_name == "check_inventory":
            wh_code = args.get("warehouse")
            if wh_code and wh_code.upper() in SemanticValidator.WAREHOUSE_ALIASES:
                canonical = SemanticValidator.WAREHOUSE_ALIASES[wh_code.upper()]
                session_wh = session.get("warehouse")
                
                if session_wh and session_wh != canonical:
                    issues.append({
                        "severity": "warn",
                        "message": (
                            f"Warehouse '{wh_code}' resolves to '{canonical}' but session "
                            f"context indicates '{session_wh}'. Confirm intended facility."
                        ),
                        "field": "warehouse",
                        "session_value": session_wh,
                        "resolved_value": canonical
                    })
        
        # RATE LIMIT PROTECTION: Count recent calls to same API endpoint
        if tool_name == "track_shipment":
            recent_tracks = [
                e for e in audit_log[-50:]
                if e["tool_name"] == "track_shipment"
                and (datetime.utcnow() - e["timestamp"]).total_seconds() < 60
            ]
            
            if len(recent_tracks) > 10:
                issues.append({
                    "severity": "warn",
                    "message": f"High tracking query volume ({len(recent_tracks)}/min). Consider batch lookup.",
                    "field": "_rate_limit"
                })
        
        return {
            "valid": not any(i["severity"] == "block" for i in issues),
            "issues": issues
        }

Step 5: Unified Validation Pipeline + Execution Wrapper

import time
import uuid
from langchain_core.tools import BaseTool

class ValidatedLogisticsTool(BaseTool):
    """
    Wraps any logistics tool with the 4-layer validation pipeline.
    Drop-in replacement for standard LangChain tools in LangGraph agents.
    """
    
    name: str
    description: str
    args_schema: type[BaseModel]
    _actual_func: Any  # The real API call function
    _semantic_validator: Any
    _state_accessor: Any  # Injected at graph compile time
    
    async def _run(self, **kwargs) -> Dict[str, Any]:
        start = time.monotonic()
        audit_entry: ToolAuditEntry = {
            "timestamp": datetime.utcnow(),
            "tool_name": self.name,
            "raw_args": kwargs,
            "validation_results": {},
            "repaired_args": None,
            "execution_status": "rejected",
            "latency_ms": 0,
            "error_detail": None
        }
        
        try:
            # === LAYER 1: Schema Validation ===
            try:
                parsed_args = self.args_schema(**kwargs)
                audit_entry["validation_results"]["schema"] = {"valid": True}
            except Exception as e:
                audit_entry["validation_results"]["schema"] = {
                    "valid": False, "error": str(e)
                }
                audit_entry["error_detail"] = f"Schema: {e}"
                audit_entry["latency_ms"] = (time.monotonic() - start) * 1000
                
                # Return structured error to agent (NOT exception)
                return self._format_validation_error("schema", str(e), kwargs)
            
            # === LAYER 2: Semantic Validation ===
            semantic_result = self._semantic_validator(parsed_args)
            audit_entry["validation_results"]["semantic"] = semantic_result
            
            if not semantic_result["valid"]:
                blocking = [i for i in semantic_result["issues"] if i["severity"] == "block"]
                audit_entry["error_detail"] = "; ".join(i["message"] for i in blocking)
                audit_entry["latency_ms"] = (time.monotonic() - start) * 1000
                return self._format_validation_error("semantic", blocking, kwargs)
            
            # Apply repairs
            repaired_kwargs = {**kwargs, **semantic_result.get("repairs", {})}
            if semantic_result.get("repairs"):
                parsed_args = self.args_schema(**repaired_kwargs)
                audit_entry["repaired_args"] = repaired_kwargs
            
            # === LAYER 3: Contextual Validation ===
            state = self._state_accessor.get_state()
            contextual_result = await ContextualValidator.validate(state, self.name, repaired_kwargs)
            audit_entry["validation_results"]["contextual"] = contextual_result
            
            if not contextual_result["valid"]:
                blocking = [i for i in contextual_result["issues"] if i["severity"] == "block"]
                audit_entry["error_detail"] = "; ".join(i["message"] for i in blocking)
                audit_entry["latency_ms"] = (time.monotonic() - start) * 1000
                return self._format_validation_error("contextual", blocking, repaired_kwargs)
            
            # === LAYER 4: Execute with Protection ===
            idempotency_key = f"{self.name}:{hash(frozenset(repaired_kwargs.items()))}:{audit_entry['timestamp'].strftime('%Y%m%d%H%M')}"
            
            result = await self._actual_func(
                **repaired_kwargs,
                _idempotency_key=idempotency_key
            )
            
            audit_entry["execution_status"] = "success"
            audit_entry["latency_ms"] = (time.monotonic() - start) * 1000
            
            # Log warnings from non-blocking issues
            warnings = [
                i["message"] for layer in ["semantic", "contextual"]
                for i in audit_entry["validation_results"].get(layer, {}).get("issues", [])
                if i["severity"] == "warn"
            ]
            if warnings:
                result["_validation_warnings"] = warnings
            
            return result
            
        except Exception as e:
            audit_entry["execution_status"] = "failed"
            audit_entry["error_detail"] = str(e)
            audit_entry["latency_ms"] = (time.monotonic() - start) * 1000
            return {
                "error": True,
                "message": f"Tool execution failed: {e}",
                "tool": self.name,
                "suggestion": "Check parameters and retry, or escalate if persistent."
            }
        finally:
            # ALWAYS log, even on rejection
            self._state_accessor.append_audit(audit_entry)
    
    def _format_validation_error(self, layer: str, detail: Any, args: Dict) -> Dict:
        """
        Returns STRUCTURED error to agent, not exception.
        This prevents the agent from entering retry loops on invalid inputs.
        """
        return {
            "error": True,
            "validation_layer": layer,
            "detail": detail,
            "original_args": args,
            "suggestion": f"Fix {layer} validation issues before retrying. Do NOT retry with same parameters.",
            "do_not_retry": True  # Signal to agent framework
        }

Step 6: Integrate into LangGraph with State Access

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver

# Create validated tools
track_tool = ValidatedLogisticsTool(
    name="track_shipment",
    description="Track a shipment by carrier and tracking number. Validates format, checksum, and rate limits.",
    args_schema=TrackShipmentInput,
    _actual_func=fedex_track_api_call,  # Your actual API function
    _semantic_validator=SemanticValidator.validate_track_shipment,
    _state_accessor=None  # Injected below
)

reroute_tool = ValidatedLogisticsTool(
    name="reroute_shipment",
    description="Reroute a shipment to a different carrier/service. Validates compatibility, permissions, and cooldowns.",
    args_schema=RerouteShipmentInput,
    _actual_func=oms_reroute_api_call,
    _semantic_validator=SemanticValidator.validate_reroute,
    _state_accessor=None
)

# Graph node that injects state access before tool execution
async def logistics_agent_node(state: LogisticsState) -> dict:
    """Inject state accessor into tools, then run agent."""
    accessor = StateAccessor(state)
    track_tool._state_accessor = accessor
    reroute_tool._state_accessor = accessor
    
    agent = create_react_agent(
        model=ChatOpenAI(model="gpt-4o", temperature=0),
        tools=[track_tool, reroute_tool],
        prompt=LOGISTICS_SYSTEM_PROMPT
    )
    
    result = await agent.ainvoke({"messages": state["messages"]})
    return {"messages": result["messages"]}

# Build graph
workflow = StateGraph(LogisticsState)
workflow.add_node("agent", logistics_agent_node)
workflow.add_edge(START, "agent")
workflow.add_edge("agent", END)

app = workflow.compile(checkpointer=PostgresSaver.from_conn_string("postgresql://logistics-rag"))

Step 7: End-to-End Execution with Malformed Input

config = {"configurable": {"thread_id": "cs-rep-maria-session-4421"}}

# Test with intentionally malformed input
result = await app.ainvoke({
    "current_query": "Track FedEx package 123456789013 for order ORD-2024-884721",
    "messages": [{"role": "user", "content": "Track FedEx package 123456789013 for order ORD-2024-884721"}],
    "session_context": {"user_role": "cs_rep", "warehouse": "WH-3", "active_order_id": "ORD-2024-884721"},
    "tool_audit_log": [],
    "validation_cache": {},
    "final_response": None,
    "session_id": "cs-maria-20240805"
}, config=config)

# Inspect audit log
for entry in result["tool_audit_log"]:
    print(f"\n{'='*60}")
    print(f"Tool: {entry['tool_name']}")
    print(f"Status: {entry['execution_status']}")
    print(f"Latency: {entry['latency_ms']:.1f}ms")
    print(f"Schema: {entry['validation_results'].get('schema')}")
    print(f"Semantic: {entry['validation_results'].get('semantic')}")
    print(f"Contextual: {entry['validation_results'].get('contextual')}")
    if entry.get("repaired_args"):
        print(f"Repaired: {entry['repaired_args']}")
    if entry.get("error_detail"):
        print(f"Error: {entry['error_detail']}")

# Expected output for malformed tracking number:
# Tool: track_shipment
# Status: rejected
# Latency: 3.2ms  ← Rejected BEFORE API call
# Schema: {'valid': True}
# Semantic: {'valid': False, 'issues': [{'severity': 'warn', 'message': 'FedEx tracking checksum mismatch...'}]}
# Error: FedEx tracking checksum mismatch. Possible transposition.

Key Design Principles

1. Return Structured Errors, Never Raise Exceptions

When validation fails, return a dict with error: True and do_not_retry: True. Raising exceptions causes LangChain agents to enter generic retry loops, burning tokens on inputs that will never be valid. Structured errors let the agent understand why and adapt.

2. Separate Repairable from Blocking Issues

Not all validation failures are equal. A warehouse alias mismatch is repairable; a missing permission is blocking. The pipeline distinguishes these so agents can self-correct when possible and escalate only when necessary.

3. Audit Everything, Even Rejections

Every tool call attempt—successful, failed, or rejected—is logged immutably. This serves triple duty: debugging, compliance auditing, and training data for improving validation rules. In logistics, "why didn't the system do X?" is as important as "what did it do?"

4. Make Validation Fast and Deterministic

Layers 1-3 must complete in <10ms combined. No LLM calls, no external API lookups. Validation is pure computation against in-memory rules and state. If validation itself is slow, you've moved the bottleneck rather than solving it.

5. Cache Validation Results Judiciously

Identical tool calls within a session can skip re-validation. But cache keys must include relevant state (order status, user role) and have short TTLs. A cached "valid" result for a reroute becomes invalid the moment the order ships.

Production Monitoring

# Metrics to track per tool
validation_metrics = {
    "schema_rejection_rate": 0.02,    # Target: <5%
    "semantic_rejection_rate": 0.08,  # Target: <15%
    "contextual_rejection_rate": 0.03,# Target: <5%
    "repair_rate": 0.12,              # Healthy: 5-20%
    "avg_validation_latency_ms": 4.2, # Target: <10ms
    "false_positive_rate": 0.01,      # Target: <2% (validated via manual audit)
}

If schema rejection exceeds 5%, your Pydantic models are too strict or the LLM prompt needs better examples. If semantic rejection exceeds 15%, either business rules need updating or the LLM needs fine-tuning on valid parameter combinations. If repair rate drops below 5%, your repair logic may be broken.

Conclusion

Robust tool calling in enterprise logistics RAG is not about hoping the LLM generates correct parameters. It is about building a validation infrastructure that makes incorrect parameters harmless. The four-layer pipeline demonstrated here ensures that:

Malformed inputs are rejected in milliseconds, not after expensive API calls
Ambiguous inputs are repaired deterministically, not guessed at by the LLM
Contextually invalid inputs are caught against live state, not just static schemas
Every decision is auditable, reproducible, and improvable

In e-commerce logistics, a bad tool call isn't a hallucination—it's a misrouted pallet, a missed SLA, or a carrier penalty. Validation isn't overhead; it's the difference between an AI assistant and an operational liability. Build your tools assuming the LLM will be wrong, and make being wrong safe.