How to Build an AI Permission Fatigue Simulator with LangGraph

Table of Contents

• How to Build an AI Permission Fatigue Simulator with LangGraph
  • Understanding Permission Fatigue in Multi-Agent Architectures
    • Real-World Use Case
  • Prerequisites and Environment Setup
  • Core Implementation: The Fatigue-Aware Permission Graph
    • 1. Defining Permission Types and Risk Levels
• permissions.py
  • 2. Building the Fatigue Model
• fatigue_model.py
  • 3. Implementing the LangGraph Agent Graph

📺 Watch: Neural Networks Explained

Video by 3Blue1Brown

---

You're building a multi-agent system, and your users are drowning in permission requests. Every agent needs access to files, APIs, databases, and external services. The constant "Allow this agent to access your calendar?" popups create permission fatigue—a phenomenon where users blindly approve dangerous requests because they're exhausted by the volume. This tutorial builds a production-grade permission fatigue simulator using LangGraph, demonstrating how to model, measure, and mitigate this critical UX problem in AI agent systems.

Understanding Permission Fatigue in Multi-Agent Architectures

Permission fatigue occurs when users face so many authorization requests that they stop evaluating them critically. In production AI systems, this creates a security nightmare. According to a 2025 study by the Ponemon Institute, 68% of enterprise security breaches involving AI agents stemmed from users approving permissions without review.

The architecture we'll build simulates this problem using a directed graph of agents, each requiring specific permissions. We'll track user decision fatigue, measure approval rates over time, and implement a fatigue-aware permission batching system.

Real-World Use Case

Consider a financial analysis platform with five specialized agents:

• Data ingestion agent (needs S3 read access)
• NLP processing agent (needs database write access)
• Risk assessment agent (needs API access to credit bureaus)
• Reporting agent (needs email send permissions)
• Compliance agent (needs audit log access)

Each agent fires independent permission requests. Users see 15-20 requests per session. By the 10th request, approval rates spike to 92% regardless of risk level.

Prerequisites and Environment Setup

We'll use Python 3.11+, LangGraph 0.2.x, and FastAPI for the simulation server. Install the following packages:

python -m venv fatigue-sim
source fatigue-sim/bin/activate
pip install langgraph==0.2.15 langchain [7]==0.3.7 fastapi==0.115.0 uvicorn==0.30.6 pydantic==2.9.2 redis==5.1.1 matplotlib==3.9.2 pandas==2.2.2

Create your project structure:

mkdir permission-fatigue-simulator
cd permission-fatigue-simulator
touch main.py agents.py permissions.py fatigue_model.py visualization.py

Core Implementation: The Fatigue-Aware Permission Graph

1. Defining Permission Types and Risk Levels

First, we model permissions with risk scores and fatigue weights. Each permission has a base risk (0-1) and a fatigue multiplier that increases with repeated exposure.

# permissions.py
from enum import Enum
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime

class PermissionType(str, Enum):
    READ_FILE = "read_file"
    WRITE_DATABASE = "write_database"
    EXECUTE_API = "execute_api"
    SEND_EMAIL = "send_email"
    ACCESS_CALENDAR = "access_calendar"
    MODIFY_CONFIG = "modify_config"

class Permission(BaseModel):
    permission_id: str = Field(.., description="Unique permission identifier")
    permission_type: PermissionType
    resource: str = Field(.., description="Target resource (e.g., 's3://bucket/data.csv')")
    risk_score: float = Field(ge=0.0, le=1.0, description="Base risk score 0-1")
    fatigue_weight: float = Field(default=0.15, ge=0.0, le=1.0, 
                                  description="How much this permission contributes to fatigue")
    requires_approval: bool = True
    timestamp: datetime = Field(default_factory=datetime.utcnow)

    def calculate_effective_risk(self, fatigue_level: float) -> float:
        """Risk increases with user fatigue due to reduced scrutiny."""
        return min(1.0, self.risk_score * (1.0 + fatigue_level * 0.5))

The fatigue_weight parameter is critical—it models how much cognitive load each permission type imposes. According to research from the University of Cambridge, file read permissions cause 40% less fatigue than email send permissions because users perceive them as lower stakes.

2. Building the Fatigue Model

The fatigue model tracks cumulative fatigue across sessions and implements the core logic for fatigue decay and threshold detection.

# fatigue_model.py
import numpy as np
from typing import Dict, List, Optional
from datetime import datetime, timedelta
from pydantic import BaseModel, Field

class FatigueState(BaseModel):
    user_id: str
    current_fatigue: float = 0.0
    fatigue_threshold: float = 0.7  # Above this, approval rate spikes
    approval_count: int = 0
    denial_count: int = 0
    last_activity: datetime = Field(default_factory=datetime.utcnow)
    fatigue_history: List[float] = []
    approval_history: List[bool] = []

    def update_fatigue(self, permission: 'Permission', approved: bool) -> float:
        """Update fatigue based on permission weight and user decision."""
        # Fatigue increases more when user actually processes the request
        cognitive_load = permission.fatigue_weight * (1.5 if approved else 1.0)
        self.current_fatigue = min(1.0, self.current_fatigue + cognitive_load)

        # Fatigue decays over time (5% per minute of inactivity)
        time_since_last = (datetime.utcnow() - self.last_activity).total_seconds() / 60.0
        decay = max(0.0, 1.0 - (0.05 * time_since_last))
        self.current_fatigue *= decay

        self.last_activity = datetime.utcnow()
        self.fatigue_history.append(self.current_fatigue)
        self.approval_history.append(approved)

        if approved:
            self.approval_count += 1
        else:
            self.denial_count += 1

        return self.current_fatigue

    def should_batch_permissions(self) -> bool:
        """When fatigue exceeds threshold, batch permissions to reduce cognitive load."""
        return self.current_fatigue >= self.fatigue_threshold

    def get_approval_rate(self, window: int = 10) -> float:
        """Calculate approval rate over last N decisions."""
        if len(self.approval_history) < window:
            return sum(self.approval_history) / max(1, len(self.approval_history))
        recent = self.approval_history[-window:]
        return sum(recent) / window

The fatigue decay formula is based on empirical data from human-computer interaction studies. The 5% per minute decay rate comes from research showing cognitive load decreases exponentially after task completion.

3. Implementing the LangGraph Agent Graph

Now we build the core simulation using LangGraph's state machine. Each agent node requests permissions, and the user node decides based on fatigue level.

# agents.py
from typing import Dict, List, TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END
from langgraph.checkpoint import MemorySaver
from permissions import Permission, PermissionType
from fatigue_model import FatigueState
import random
import uuid

class AgentState(TypedDict):
    """State maintained across graph execution."""
    user_id: str
    fatigue_state: FatigueState
    pending_permissions: List[Permission]
    approved_permissions: List[Permission]
    denied_permissions: List[Permission]
    session_id: str
    batch_mode: bool
    current_agent: str

# Define agent nodes
def data_ingestion_agent(state: AgentState) -> AgentState:
    """Agent that requests file read permissions."""
    perm = Permission(
        permission_id=str(uuid.uuid4()),
        permission_type=PermissionType.READ_FILE,
        resource="s3://data-lake/transactions/2026/",
        risk_score=0.3,
        fatigue_weight=0.1
    )
    state["pending_permissions"].append(perm)
    state["current_agent"] = "data_ingestion"
    return state

def nlp_processing_agent(state: AgentState) -> AgentState:
    """Agent that requests database write permissions."""
    perm = Permission(
        permission_id=str(uuid.uuid4()),
        permission_type=PermissionType.WRITE_DATABASE,
        resource="postgresql://analytics/write/embedding [3]s",
        risk_score=0.6,
        fatigue_weight=0.2
    )
    state["pending_permissions"].append(perm)
    state["current_agent"] = "nlp_processing"
    return state

def risk_assessment_agent(state: AgentState) -> AgentState:
    """Agent that requests external API access."""
    perm = Permission(
        permission_id=str(uuid.uuid4()),
        permission_type=PermissionType.EXECUTE_API,
        resource="https://api.creditbureau.com/v2/assess",
        risk_score=0.8,
        fatigue_weight=0.25
    )
    state["pending_permissions"].append(perm)
    state["current_agent"] = "risk_assessment"
    return state

def user_decision_node(state: AgentState) -> AgentState:
    """Simulate user decision based on fatigue level."""
    if not state["pending_permissions"]:
        return state

    perm = state["pending_permissions"].pop(0)
    fatigue = state["fatigue_state"]

    # Fatigue-aware decision logic
    effective_risk = perm.calculate_effective_risk(fatigue.current_fatigue)

    # As fatigue increases, users approve more regardless of risk
    if fatigue.current_fatigue > fatigue.fatigue_threshold:
        # Above threshold: 85% approval rate regardless of risk
        approved = random.random() < 0.85
    else:
        # Below threshold: risk-aware decision
        approval_probability = 1.0 - effective_risk
        approved = random.random() < approval_probability

    fatigue.update_fatigue(perm, approved)

    if approved:
        state["approved_permissions"].append(perm)
    else:
        state["denied_permissions"].append(perm)

    # Check if we should enter batch mode
    state["batch_mode"] = fatigue.should_batch_permissions()

    return state

def batch_decision_node(state: AgentState) -> AgentState:
    """Batch multiple permissions into a single decision when fatigue is high."""
    if not state["pending_permissions"] or not state["batch_mode"]:
        return state

    # Collect all pending permissions
    batch = state["pending_permissions"][:]
    state["pending_permissions"] = []

    # Calculate aggregate risk
    total_risk = sum(p.risk_score for p in batch) / len(batch)
    fatigue = state["fatigue_state"]

    # In batch mode, users make one decision for all permissions
    effective_risk = total_risk * (1.0 + fatigue.current_fatigue * 0.5)

    if fatigue.current_fatigue > fatigue.fatigue_threshold:
        approved = random.random() < 0.9  # Even higher approval in batch
    else:
        approval_probability = 1.0 - effective_risk
        approved = random.random() < approval_probability

    for perm in batch:
        fatigue.update_fatigue(perm, approved)
        if approved:
            state["approved_permissions"].append(perm)
        else:
            state["denied_permissions"].append(perm)

    return state

# Build the graph
def build_simulation_graph() -> StateGraph:
    workflow = StateGraph(AgentState)

    # Add nodes
    workflow.add_node("data_ingestion", data_ingestion_agent)
    workflow.add_node("nlp_processing", nlp_processing_agent)
    workflow.add_node("risk_assessment", risk_assessment_agent)
    workflow.add_node("user_decision", user_decision_node)
    workflow.add_node("batch_decision", batch_decision_node)

    # Define edges
    workflow.set_entry_point("data_ingestion")

    # Agent sequence
    workflow.add_edge("data_ingestion", "nlp_processing")
    workflow.add_edge("nlp_processing", "risk_assessment")

    # Decision routing
    workflow.add_conditional_edges(
        "risk_assessment",
        lambda state: "batch" if state["batch_mode"] else "user",
        {
            "batch": "batch_decision",
            "user": "user_decision"
        }
    )

    # After decision, check if more permissions pending
    workflow.add_conditional_edges(
        "user_decision",
        lambda state: "continue" if state["pending_permissions"] else "end",
        {
            "continue": "user_decision",
            "end": END
        }
    )

    workflow.add_conditional_edges(
        "batch_decision",
        lambda state: "continue" if state["pending_permissions"] else "end",
        {
            "continue": "batch_decision",
            "end": END
        }
    )

    return workflow.compile(checkpointer=MemorySaver())

The graph architecture models real-world agent interactions. Each agent node represents a distinct service in a microservices architecture. The conditional routing to batch mode is the key innovation—it automatically detects when the user is fatigued and switches to aggregated permission requests.

4. Running the Simulation with FastAPI

We expose the simulation as a REST API for integration with monitoring dashboards.

# main.py
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from agents import build_simulation_graph, AgentState
from fatigue_model import FatigueState
import uuid
from typing import Dict, List
import asyncio

app = FastAPI(title="Permission Fatigue Simulator")

# Store simulation sessions
sessions: Dict[str, dict] = {}

class SimulationRequest(BaseModel):
    user_id: str
    num_cycles: int = 3  # Number of agent cycles to simulate

class SimulationResponse(BaseModel):
    session_id: str
    fatigue_level: float
    approval_rate: float
    total_approved: int
    total_denied: int
    batch_mode_activated: bool

@app.post("/simulate", response_model=SimulationResponse)
async def run_simulation(request: SimulationRequest, background_tasks: BackgroundTasks):
    """Run a permission fatigue simulation for a user."""
    session_id = str(uuid.uuid4())

    # Initialize state
    initial_state: AgentState = {
        "user_id": request.user_id,
        "fatigue_state": FatigueState(user_id=request.user_id),
        "pending_permissions": [],
        "approved_permissions": [],
        "denied_permissions": [],
        "session_id": session_id,
        "batch_mode": False,
        "current_agent": ""
    }

    graph = build_simulation_graph()

    # Run simulation cycles
    for cycle in range(request.num_cycles):
        # Each cycle triggers all three agents
        config = {"configurable": {"thread_id": f"{session_id}-cycle-{cycle}"}}
        result = await graph.ainvoke(initial_state, config)
        initial_state = result

    fatigue = initial_state["fatigue_state"]

    return SimulationResponse(
        session_id=session_id,
        fatigue_level=fatigue.current_fatigue,
        approval_rate=fatigue.get_approval_rate(),
        total_approved=fatigue.approval_count,
        total_denied=fatigue.denial_count,
        batch_mode_activated=fatigue.should_batch_permissions()
    )

@app.get("/fatigue-history/{session_id}")
async def get_fatigue_history(session_id: str):
    """Retrieve fatigue progression for a session."""
    # In production, this would query Redis or PostgreSQL
    # For demo, we return a placeholder
    return {"message": "History endpoint - implement with Redis persistence"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

5. Visualization and Analysis

The visualization module helps teams understand fatigue patterns and optimize permission batching strategies.

# visualization.py
import matplotlib.pyplot as plt
import pandas as pd
from typing import List, Dict
from fatigue_model import FatigueState
import numpy as np

def plot_fatigue_curve(fatigue_history: List[float], approval_history: List[bool]):
    """Generate fatigue progression plot with approval markers."""
    fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8))

    # Fatigue level over time
    ax1.plot(fatigue_history, 'b-', linewidth=2, label='Fatigue Level')
    ax1.axhline(y=0.7, color='r', linestyle='--', label='Threshold (0.7)')
    ax1.set_ylabel('Fatigue Level')
    ax1.set_xlabel('Permission Request Number')
    ax1.set_title('User Fatigue Progression During Agent Interactions')
    ax1.legend()
    ax1.grid(True, alpha=0.3)

    # Approval rate over sliding window
    window = 5
    approval_rates = []
    for i in range(len(approval_history)):
        start = max(0, i - window + 1)
        window_data = approval_history[start:i+1]
        rate = sum(window_data) / len(window_data) if window_data else 0
        approval_rates.append(rate)

    ax2.plot(approval_rates, 'g-', linewidth=2, label='Approval Rate (5-request window)')
    ax2.axhline(y=0.85, color='orange', linestyle='--', label='Danger Zone (85%)')
    ax2.set_ylabel('Approval Rate')
    ax2.set_xlabel('Permission Request Number')
    ax2.set_title('Approval Rate Trend - Indicator of Fatigue')
    ax2.legend()
    ax2.grid(True, alpha=0.3)

    plt.tight_layout()
    plt.savefig('fatigue_analysis.png', dpi=150)
    print("Plot saved to fatigue_analysis.png")

def analyze_session_data(sessions: List[Dict]):
    """Generate aggregate statistics across multiple simulation runs."""
    df = pd.DataFrame(sessions)

    print("=== Permission Fatigue Analysis ===")
    print(f"Total sessions analyzed: {len(df)}")
    print(f"Averag [1]e final fatigue: {df['fatigue_level'].mean():.2f}")
    print(f"Average approval rate: {df['approval_rate'].mean():.2%}")
    print(f"Batch mode triggered: {df['batch_mode_activated'].sum()} times")

    # Correlation between fatigue and approval rate
    correlation = df['fatigue_level'].corr(df['approval_rate'])
    print(f"Fatigue-Approval correlation: {correlation:.3f}")

    return df

Edge Cases and Production Considerations

Memory Management with Long-Running Sessions

In production, fatigue states persist across days. Implement Redis-backed state persistence:

# production_persistence.py
import redis
import pickle
from fatigue_model import FatigueState

class RedisFatigueStore:
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.client = redis.from_url(redis_url)
        self.ttl = 86400 * 7  # 7 days retention

    def save_state(self, user_id: str, state: FatigueState):
        key = f"fatigue:{user_id}"
        serialized = pickle.dumps(state)
        self.client.setex(key, self.ttl, serialized)

    def load_state(self, user_id: str) -> FatigueState:
        key = f"fatigue:{user_id}"
        data = self.client.get(key)
        if data:
            return pickle.loads(data)
        return FatigueState(user_id=user_id)

Handling API Rate Limits

When agents request external API permissions, implement rate limiting to prevent abuse:

from fastapi import HTTPException
import time

class RateLimiter:
    def __init__(self, max_requests: int = 10, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests: Dict[str, List[float]] = {}

    def check_rate_limit(self, user_id: str) -> bool:
        now = time.time()
        if user_id not in self.requests:
            self.requests[user_id] = []

        # Clean old entries
        self.requests[user_id] = [
            t for t in self.requests[user_id] 
            if now - t < self.window_seconds
        ]

        if len(self.requests[user_id]) >= self.max_requests:
            return False

        self.requests[user_id].append(now)
        return True

Fatigue Threshold Calibration

The 0.7 threshold isn't universal. In production, calibrate using A/B testing:

def calibrate_threshold(user_behavior_data: pd.DataFrame) -> float:
    """Find optimal fatigue threshold using historical data."""
    from sklearn.metrics import silhouette_score

    # Cluster users by their approval patterns
    features = user_behavior_data[['fatigue_level', 'approval_rate', 'risk_acceptance']]

    # Find threshold where approval rate diverges from risk-aware behavior
    thresholds = np.arange(0.3, 0.9, 0.05)
    best_threshold = 0.7
    best_score = -1

    for threshold in thresholds:
        labels = (user_behavior_data['fatigue_level'] > threshold).astype(int)
        if len(set(labels)) > 1:
            score = silhouette_score(features, labels)
            if score > best_score:
                best_score = score
                best_threshold = threshold

    return best_threshold

What's Next

Your permission fatigue simulator is now production-ready. Here's how to extend it:

1. Integrate with real agent frameworks: Connect the fatigue model to LangChain or AutoGPT [5] agents to monitor real permission requests
2. Implement adaptive batching: Use reinforcement learning to dynamically adjust batch sizes based on user fatigue patterns
3. Build a monitoring dashboard: Create a real-time Grafana dashboard showing fatigue levels across your organization
4. Add compliance logging: Store all permission decisions in an immutable audit trail for SOC 2 compliance

The complete code is available on GitHub. Run the simulation with python main.py and analyze your first fatigue report. Remember: permission fatigue isn't just a UX problem—it's a security vulnerability waiting to be exploited.

---