Dual-Layer Rate Limiting Middleware for Enterprise LangGraph RAG

Large Language Models (LLMs) are already essential infrastructure in the corporate AI world of 2026, not only a curiosity. Nevertheless, LLM inference is computationally demanding and costly by nature. The "noisy neighbor" issue, budget-draining prompt injection loops, and misuse are all possible when a multi-agent Retrieval-Augmented Generation (RAG) system is exposed through an API.

 

A single client spamming your RAG endpoint has the potential to deplete your LLM token quotas, increase your cloud compute expenses, and worsen latency for all other users. An end-to-end tutorial for setting up a Dual-Layer Rate Limiting Middleware in FastAPI is given in this article. In order to enforce restrictions, we will return standard 429 Too Many Requests replies based on both Client IP (for anonymous/internal traffic) and API Key (for authorized partners). Lastly, a production-grade Multi-Agent LangGraph RAG system with permanent memory and state will be integrated with this middleware. 

Part 1: The Enterprise Use Case

Scenario: "Acme Corp" has deployed an internal/external AI Knowledge Assistant. The system uses a multi-agent LangGraph architecture to retrieve data from Confluence, Jira, and internal wikis, and synthesize answers.

The Access Tiers:

1. Authenticated Partners (API Key): Have a strict limit (e.g., 60 requests per minute) to prevent automated script abuse.

2. Internal Employees (Corporate IP): Have a higher limit (e.g., 120 requests per minute) as they are trusted internal users.

3. Anonymous Web Chat (Public IP): Have the lowest limit (e.g., 20 requests per minute) to prevent bot attacks.

The Architecture:

• FastAPI Gateway: Handles HTTP routing, authentication extraction, and rate limiting.

• Redis: Acts as the distributed, high-performance counter for rate limits.

• LangGraph Backend: Executes the multi-agent RAG workflow, maintaining conversation state via a Checkpointer.

Part 2: Designing the Dual-Layer Rate Limiter

To build an enterprise-grade rate limiter, we must address three challenges:

1. Distributed State: If your FastAPI app runs on multiple replicas (Kubernetes pods), in-memory counters will fail. We must use Redis.

2. Race Conditions: A simple GET then SET in Redis is not atomic. We will use Redis INCR with an expiration to ensure thread safety.

3. Header Standards: When rejecting a request with a 429, we must return standard headers (X-RateLimit-Limit, X-RateLimit-Remaining, Retry-After) so client applications can handle the backoff gracefully.

Part 3: The Code Implementation

Below is the complete, end-to-end implementation.

1. Dependencies

pip install fastapi uvicorn redis langchain-openai langgraph langchain-core pydantic

2. The Redis Rate Limiter & FastAPI Middleware

import time
import logging
from typing import Optional, Tuple
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
import redis.asyncio as redis

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Initialize Redis connection (Enterprise: Use a connection pool)
redis_client = redis.from_url("redis://localhost:6379", decode_responses=True)

class DualLayerRateLimiter:
    """Handles the logic for checking and incrementing rate limits in Redis."""

    def __init__(self, redis_conn: redis.Redis):
        self.redis = redis_conn

    async def check_limit(self, identifier: str, limit: int, window_seconds: int) -> Tuple[bool, int, int]:
        """
        Checks the rate limit. Uses a sliding/fixed window approach.
        Returns: (is_allowed, remaining_requests, limit)
        """
        # Create a time-bounded key (Fixed window per minute)
        current_window = int(time.time() // window_seconds)
        redis_key = f"ratelimit:{identifier}:{current_window}"

        # Atomic increment. If key doesn't exist, it's created with value 1.
        current_count = await self.redis.incr(redis_key)

        # Set expiration only on the first request of the window to prevent race conditions
        if current_count == 1:
            await self.redis.expire(redis_key, window_seconds + 1) # +1s buffer

        is_allowed = current_count <= limit
        remaining = max(0, limit - current_count)

        return is_allowed, remaining, limit

class RateLimitMiddleware(BaseHTTPMiddleware):
    """
    FastAPI Middleware that intercepts requests, extracts IP/API Key,
    and enforces dual-layer rate limiting.
    """
    def __init__(self, app, limiter: DualLayerRateLimiter):
        super().__init__(app)
        self.limiter = limiter
        # Define tiers: (limit per minute)
        self.api_key_limit = 60
        self.internal_ip_limit = 120
        self.public_ip_limit = 20
        self.window = 60 # 60 seconds

    async def dispatch(self, request: Request, call_next):
        # Skip rate limiting for health checks and docs
        if request.url.path in ["/health", "/docs", "/openapi.json"]:
            return await call_next(request)

        # 1. Extract Client IP (Handling reverse proxies like AWS ALB / Nginx)
        forwarded_for = request.headers.get("X-Forwarded-For")
        if forwarded_for:
            client_ip = forwarded_for.split(",")[0].strip()
        else:
            client_ip = request.client.host if request.client else "unknown"

        # 2. Extract API Key
        api_key = request.headers.get("X-API-Key") or request.headers.get("Authorization", "").replace("Bearer ", "")

        # 3. Determine Identifier and Limit Tier
        if api_key and api_key != "":
            identifier = f"key:{api_key}"
            limit = self.api_key_limit
            tier = "API_KEY"
        elif client_ip.startswith("10.") or client_ip.startswith("192.168."): # Mock internal IP check
            identifier = f"ip:{client_ip}"
            limit = self.internal_ip_limit
            tier = "INTERNAL_IP"
        else:
            identifier = f"ip:{client_ip}"
            limit = self.public_ip_limit
            tier = "PUBLIC_IP"

        # 4. Check Limit
        is_allowed, remaining, max_limit = await self.limiter.check_limit(identifier, limit, self.window)

        if not is_allowed:
            logger.warning(f"Rate limit exceeded for {tier} {identifier}. Limit: {max_limit}")
            return JSONResponse(
                status_code=429,
                content={"error": "Too Many Requests", "message": "Rate limit exceeded. Please slow down."},
                headers={
                    "X-RateLimit-Limit": str(max_limit),
                    "X-RateLimit-Remaining": "0",
                    "Retry-After": str(self.window),
                    "X-RateLimit-Tier": tier
                }
            )

        # 5. Proceed and inject rate limit headers into the successful response
        response = await call_next(request)
        response.headers["X-RateLimit-Limit"] = str(max_limit)
        response.headers["X-RateLimit-Remaining"] = str(remaining)
        response.headers["X-RateLimit-Tier"] = tier
        return response

3. The Multi-Agent LangGraph RAG Backend

Now, we build the AI backend. We will use a multi-agent approach: a Router Agent directs the query to a RAG Retriever Agent, which passes context to a Synthesizer Agent. We use LangGraph's Checkpointer for memory.

from typing import TypedDict, Annotated, Sequence, Literal
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.tools import tool

# --- 1. Define the State ---
class GraphState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], "The conversation history"]
    user_id: str
    retrieved_context: str
    next_agent: str

# --- 2. Define the Tools (RAG) ---
@tool
def search_internal_knowledge_base(query: str) -> str:
    """Searches the enterprise vector database for internal policies and documentation."""
    # In production, this queries Pinecone/Milvus/Weaviate
    return "Acme Corp Policy: Employees are entitled to 25 days of PTO. Rollover is permitted up to 5 days."

# --- 3. Define the Agent Nodes ---
llm = ChatOpenAI(model="gpt-4o", temperature=0)

async def router_node(state: GraphState):
    """Determines if the query requires RAG or can be answered directly."""
    # Simplified routing logic for demonstration
    messages = state["messages"]
    last_msg = messages[-1].content.lower()

    if "policy" in last_msg or "pto" in last_msg or "hr" in last_msg:
        return {"next_agent": "rag_retriever"}
    return {"next_agent": "synthesizer"}

async def rag_retriever_node(state: GraphState):
    """Executes the RAG tool and stores context in state."""
    last_msg = state["messages"][-1].content
    # In a real agent, the LLM would decide the tool arguments
    context = search_internal_knowledge_base.invoke({"query": last_msg})
    return {"retrieved_context": context, "next_agent": "synthesizer"}

async def synthesizer_node(state: GraphState):
    """Synthesizes the final answer using the LLM and retrieved context."""
    context = state.get("retrieved_context", "No specific context retrieved.")
    messages = state["messages"]

    prompt = [
        SystemMessage(content=f"You are an enterprise AI assistant. Use the following context to answer the user:\n\nContext: {context}\n\nIf the context doesn't contain the answer, state that you don't know."),
        *messages
    ]

    response = await llm.ainvoke(prompt)
    return {"messages": [response], "next_agent": "end"}

# --- 4. Build and Compile the Graph ---
def route_next(state: GraphState) -> Literal["rag_retriever", "synthesizer", "end"]:
    return state["next_agent"]

workflow = StateGraph(GraphState)
workflow.add_node("router", router_node)
workflow.add_node("rag_retriever", rag_retriever_node)
workflow.add_node("synthesizer", synthesizer_node)

workflow.set_entry_point("router")
workflow.add_conditional_edges("router", route_next)
workflow.add_edge("rag_retriever", "synthesizer")
workflow.add_edge("synthesizer", END)

# Initialize Memory (Enterprise: Use AsyncPostgresSaver)
memory = MemorySaver()
rag_graph = workflow.compile(checkpointer=memory)

4. Tying it Together: The FastAPI App

from pydantic import BaseModel

app = FastAPI(title="Enterprise RAG Gateway")

# Initialize Limiter and Middleware
limiter = DualLayerRateLimiter(redis_client)
app.add_middleware(RateLimitMiddleware, limiter=limiter)

class ChatRequest(BaseModel):
    message: str
    user_id: str
    thread_id: str # Used for LangGraph memory checkpointing

@app.post("/v1/chat")
async def chat_endpoint(payload: ChatRequest, request: Request):
    """
    The main endpoint. Protected by the RateLimitMiddleware.
    Invokes the LangGraph asynchronously.
    """
    # Extract user identity for LangGraph state
    user_id = payload.user_id
    thread_id = payload.thread_id

    # Prepare LangGraph config for memory persistence
    config = {"configurable": {"thread_id": thread_id}}

    # Initial state input
    inputs = {
        "messages": [HumanMessage(content=payload.message)],
        "user_id": user_id,
        "retrieved_context": "",
        "next_agent": ""
    }

    try:
        # Invoke the multi-agent graph asynchronously
        final_state = await rag_graph.ainvoke(inputs, config)

        # Extract the final AI response
        ai_message = final_state["messages"][-1].content

        return {
            "response": ai_message,
            "thread_id": thread_id,
            "metadata": {
                "agents_used": ["router", "rag_retriever", "synthesizer"],
                "user_id": user_id
            }
        }
    except Exception as e:
        logger.error(f"LangGraph execution failed: {str(e)}")
        raise HTTPException(status_code=500, detail="Internal AI processing error")

@app.get("/health")
async def health_check():
    return {"status": "healthy", "timestamp": time.time()}

Part 4: Testing the Implementation

To test this, run the FastAPI app (uvicorn main:app --reload) and use curl to simulate traffic.

Test 1: Normal Request (API Key)

curl -X POST "http://localhost:8000/v1/chat" \
-H "X-API-Key: partner_key_123" \
-H "Content-Type: application/json" \
-d '{"message": "What is the PTO policy?", "user_id": "user_1", "thread_id": "thread_99"}' \
-i

Expected Response: 200 OK with headers X-RateLimit-Limit: 60 and X-RateLimit-Remaining: 59. The AI will correctly retrieve and synthesize the PTO policy using the RAG agent.

Test 2: Triggering the 429 Rate Limit

Run a quick loop to exhaust the public IP limit (20 requests/minute).

for i in {1..25}; do
  curl -s -o /dev/null -w "%{http_code}\n" -X POST "http://localhost:8000/v1/chat" \
  -H "Content-Type: application/json" \
  -d '{"message": "Hello", "user_id": "user_2", "thread_id": "thread_100"}'
done

Expected Output: The first 20 requests will return 200. The remaining 5 will return 429.

If you inspect the headers of the 429 response:

HTTP/1.1 429 Too Many Requests
X-RateLimit-Limit: 20
X-RateLimit-Remaining: 0
Retry-After: 60
X-RateLimit-Tier: PUBLIC_IP
Content-Type: application/json

{"error": "Too Many Requests", "message": "Rate limit exceeded. Please slow down."}

Test 3: Verifying Memory and State

Send a follow-up message using the same thread_id to prove the LangGraph Checkpointer is maintaining state across requests.

curl -X POST "http://localhost:8000/v1/chat" \
-H "X-API-Key: partner_key_123" \
-H "Content-Type: application/json" \
-d '{"message": "Can I rollover more than 5 days?", "user_id": "user_1", "thread_id": "thread_99"}'

Expected Response: The AI will remember the previous context about the 5-day rollover limit without needing the RAG tool to fetch it again, demonstrating successful state persistence.

Part 5: Enterprise Best Practices & Next Steps

Implementing this architecture gives you a robust, production-ready AI gateway. However, to fully harden this for a Fortune 500 deployment, consider the following enhancements:

1. Atomic Lua Scripts in Redis: While INCR and EXPIRE work well, a strict enterprise environment should use a Redis Lua script to perform the check-and-set atomically, preventing edge-case race conditions during high-concurrency spikes.

2. Token-Based Rate Limiting: Instead of limiting by requests, limit by LLM tokens. A request that generates 10,000 tokens should cost more against the rate limit than a request generating 100 tokens. You can intercept the LangGraph output, calculate token usage, and decrement a Redis token bucket.

3. Distributed Checkpointing: In the code above, we used MemorySaver for brevity. In production, you must swap this for AsyncPostgresSaver or AsyncRedisSaver so that conversation memory survives FastAPI pod restarts and scales horizontally.

4. Circuit Breakers: Combine this rate limiter with a circuit breaker (like pybreaker). If the underlying vector database or LLM provider goes down, the circuit breaker should immediately return a 503 without wasting time executing the LangGraph nodes.

By combining FastAPI's dual-layer rate limiting with LangGraph's stateful multi-agent execution, you ensure that your enterprise AI is not only intelligent and context-aware but also financially predictable and highly resilient to abuse.

ASP.NET Core 10.0 Hosting Recommendation

One of the most important things when choosing a good ASP.NET Core 9.0 hosting is the feature and reliability. HostForLIFE is the leading provider of Windows hosting and affordable ASP.NET Core, their servers are optimized for PHP web applications. The performance and the uptime of the hosting service are excellent and the features of the web hosting plan are even greater than what many hosting providers ask you to pay for. 

At HostForLIFE.eu, customers can also experience fast ASP.NET Core hosting. The company invested a lot of money to ensure the best and fastest performance of the datacenters, servers, network and other facilities. Its datacenters are equipped with the top equipments like cooling system, fire detection, high speed Internet connection, and so on. That is why HostForLIFEASP.NET guarantees 99.9% uptime for ASP.NET Core. And the engineers do regular maintenance and monitoring works to assure its Orchard hosting are security and always up.