Preventing Infinite Loops in Enterprise Multi-Agent LangGraph Systems

Tuhin Paul
7h
99
0
0

Article

In the enterprise, AI agents are no longer isolated scripts; they are collaborative teams. When you build a Multi-Agent System using LangGraph, you are orchestrating a complex dance where a "Router" might send work to a "Researcher," who then passes it to a "Validator."

However, this collaboration introduces a critical risk: The Infinite Loop. Imagine a scenario where the Validator tells the Researcher, "This answer isn't good enough, try again." The Researcher tries again but retrieves the same data. The Validator rejects it again. Without safeguards, this cycle continues until your API budget is exhausted or your server times out. In this end-to-end guide, we will build a resilient Enterprise RAG Multi-Agent System that uses State Management, Memory, and Hard Limits to prevent infinite loops while maintaining high-quality responses.

The Real-World Use Case: TechCorp "Compliance & Policy" Assistant

You are building an internal assistant for TechCorp that helps employees navigate complex HR and Legal policies. The system uses three specialized agents:

The Router: Analyzes the user's intent (e.g., "Leave Policy" vs. "Code of Conduct").
The Researcher (RAG Agent): Queries the vector database to find relevant policy documents.
The Critic (Validator): Reviews the retrieved information. If the information is vague or incomplete, it instructs the Researcher to refine the search.

The Loop Trap

The loop occurs between the Researcher and the Critic. If the Critic keeps rejecting the Researcher's findings without a clear exit strategy, the graph will spin indefinitely.

The Three Pillars of Loop Prevention

To build a production-ready system, we will implement three layers of defense:

The Hard Stop (Recursion Limit): A global safety net provided by LangGraph that kills the process if it exceeds a certain number of steps.
The State Counter (Step Tracking): A manual counter in our state that allows us to make intelligent decisions (e.g., "If we've tried 3 times, stop trying and give the best available answer").
Semantic Memory (Action History): Tracking what we have already done. If the agent attempts the exact same search query twice, we detect the repetition and force a fallback.

End-to-End Implementation

We will use langgraph, pydantic, and standard Python libraries.

Step 1: Define the Robust State

Our state needs to track more than just the current message. It needs to remember the history of actions to detect loops.

from typing import TypedDict, List, Annotated, Optional
import operator
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    # The user's original question
    query: str
    
    # Memory: A list of all messages exchanged between agents
    messages: Annotated[List[str], operator.add]
    
    # The data retrieved from the vector DB
    context: List[str]
    
    # The final answer to be returned to the user
    response: str
    
    # Loop Prevention: How many times have we gone through the research/critic cycle?
    retry_count: int
    
    # Loop Prevention: What was the last search query used?
    last_search_query: str
    
    # Flag to indicate if the task is finished
    is_complete: bool

Step 2: Build the Agent Nodes

Node 1: The Router

This node initializes the process and sets the initial state.

def router_node(state: AgentState) -> AgentState:
    print("  [Router] Starting workflow...")
    return {
        "messages": ["Router: Intent identified as Policy Inquiry."],
        "retry_count": 0,
        "last_search_query": "",
        "is_complete": False
    }

Node 2: The Researcher (RAG)

This node simulates retrieving data. It includes Semantic Loop Detection. If it sees that it's about to run the same search as the last time, it changes its strategy or stops.

def researcher_node(state: AgentState) -> AgentState:
    print(f"  [Researcher] Attempt {state['retry_count'] + 1}")
    
    # SEMANTIC LOOP DETECTION
    # In a real app, this would be the actual embedding/search query
    current_query = f"Search for: {state['query']}" 
    
    if current_query == state["last_search_query"]:
        print("  [Researcher] Detected duplicate search query. Refining strategy...")
        # Force a different outcome to break the loop
        mock_docs = ["General Policy Summary (Fallback)"]
    else:
        # Normal retrieval
        mock_docs = ["TechCorp Leave Policy v4.2", "Employee Handbook Section 9"]
        
    return {
        "context": mock_docs,
        "last_search_query": current_query,
        "retry_count": state["retry_count"] + 1,
        "messages": [f"Researcher: Retrieved {len(mock_docs)} documents."]
    }

Node 3: The Critic (Validator)

This node decides if the work is done. It implements the Step Counter Logic.

def critic_node(state: AgentState) -> AgentState:
    print("✅ [Critic] Evaluating quality...")
    
    # HARD STOP LOGIC: If we have retried too many times, force completion
    if state["retry_count"] >= 3:
        print("🛑 [Critic] Max retries reached. Forcing completion with best available data.")
        return {
            "response": f"Best available answer based on: {', '.join(state['context'])}",
            "is_complete": True,
            "messages": ["Critic: Max retries exceeded. Returning partial result."]
        }
    
    # Simulate a quality check
    # If context is empty or too short, ask for more
    if len(state["context"]) < 2:
        return {
            "is_complete": False,
            "messages": ["Critic: Information insufficient. Requesting deeper search."]
        }
        
    # If quality is good, finish
    return {
        "response": f"Final Answer: {', '.join(state['context'])}",
        "is_complete": True,
        "messages": ["Critic: Quality check passed."]
    }

Step 3: Define Conditional Edges (The Traffic Cop)

This is where we control the flow. We never let the Critic call the Researcher directly; we use a routing function that checks the state.

def route_after_critic(state: AgentState) -> str:
    if state["is_complete"]:
        return "end"
    else:
        return "researcher"

def build_enterprise_graph():
    workflow = StateGraph(AgentState)

    workflow.add_node("router", router_node)
    workflow.add_node("researcher", researcher_node)
    workflow.add_node("critic", critic_node)

    workflow.set_entry_point("router")
    
    # Router always goes to researcher first
    workflow.add_edge("router", "researcher")
    
    # Researcher always goes to critic for validation
    workflow.add_edge("researcher", "critic")
    
    # Critic decides: either end or go back to researcher
    workflow.add_conditional_edges(
        "critic",
        route_after_critic,
        {
            "researcher": "researcher",
            "end": END
        }
    )

    return workflow.compile()

app = build_enterprise_graph()

Running the System with Safety Nets

When invoking the graph, we must use the recursion_limit configuration. This is the ultimate "kill switch" provided by LangGraph.

initial_state = {
    "query": "What is the maternity leave policy?",
    "messages": [],
    "context": [],
    "response": "",
    "retry_count": 0,
    "last_search_query": "",
    "is_complete": False
}

# Invoke with a hard limit of 10 steps
try:
    result = app.invoke(initial_state, config={"recursion_limit": 10})
    print("\n--- Final Output ---")
    print(result["response"])
    print(f"Total Retries: {result['retry_count']}")
except Exception as e:
    print(f"System halted due to safety limit: {e}")

Expected Output Trace (Scenario: Loop Detected)

Router: Starts workflow. (retry_count: 0)
Researcher: Retrieves docs. (retry_count: 1)
Critic: Says "Not enough info." (is_complete: False)
Researcher: Tries again. (retry_count: 2)
Critic: Says "Still not enough." (is_complete: False)
Researcher: Tries again. (retry_count: 3)
Critic: Sees retry_count >= 3. Forces completion. (is_complete: True)
END: Returns the "Best available answer."

Enterprise Best Practices for Multi-Agent Stability

Idempotency in RAG: Always store the document_ids or search_queries in your state. If an agent generates a query that has already been executed, intercept it before it hits the vector database.
Time-Outs per Node: In a production environment, wrap your node logic in a timeout decorator. If a specific LLM call takes longer than 15 seconds, treat it as a failure and trigger a fallback.
Observability with LangSmith: Never deploy a multi-agent system without tracing. LangSmith allows you to see the "Tree" of your execution. If you see a branch that looks like a spiral, you know exactly which conditional edge is missing a termination condition.
Graceful Degradation: As shown in the code, when a loop is detected or limits are reached, don't just throw an error. Return a "Partial Result" or a "Human Handoff" message. In the enterprise, a partial answer is better than a crashed application.

Conclusion

Preventing infinite loops in LangGraph isn't about restricting the intelligence of your agents; it's about providing them with a clear set of boundaries. By combining LangGraph's built-in recursion limits with custom state-based counters and semantic memory, you can build a multi-agent RAG system that is both powerful and predictable. In the enterprise, reliability is the most important feature. A system that knows when to stop is far more valuable than one that never gives up.