Cycles and Loops in LangGraph

Why Cycles Matter

Linear graphs (DAGs) model pipelines. Cyclic graphs model agents. An agent that retries failed tool calls, refines an answer until it meets quality criteria, or iterates over a list of items needs cycles.

LangGraph supports cycles natively — nodes can have edges that point back to earlier nodes.

Basic Retry Loop

An agent that retries until it produces valid output:

Python

from langgraph.graph import StateGraph, END
from typing import TypedDict

class AgentState(TypedDict):
    query: str
    response: str
    is_valid: bool
    attempts: int
    max_attempts: int

def generate(state: AgentState) -> AgentState:
    """Generate a response."""
    from openai import OpenAI
    client = OpenAI()

    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": state["query"]}],
        temperature=0.3,
    )
    return {
        **state,
        "response": resp.choices[0].message.content,
        "attempts": state["attempts"] + 1,
    }

def validate(state: AgentState) -> AgentState:
    """Check if response meets quality criteria."""
    response = state["response"]

    # Simple validation: response must be at least 50 words
    word_count = len(response.split())
    is_valid = word_count >= 50

    return {**state, "is_valid": is_valid}

def route_after_validate(state: AgentState) -> str:
    """Route: retry if invalid and under max attempts, else end."""
    if state["is_valid"]:
        return END
    if state["attempts"] >= state["max_attempts"]:
        return END  # Give up after max attempts
    return "generate"  # Loop back

# Build the graph
graph = StateGraph(AgentState)
graph.add_node("generate", generate)
graph.add_node("validate", validate)

graph.set_entry_point("generate")
graph.add_edge("generate", "validate")

# Conditional edge from validate: either loop or end
graph.add_conditional_edges(
    "validate",
    route_after_validate,
    {
        "generate": "generate",  # Loop back
        END: END,
    },
)

app = graph.compile()

result = app.invoke({
    "query": "Explain warfarin's mechanism in detail",
    "response": "",
    "is_valid": False,
    "attempts": 0,
    "max_attempts": 3,
})

print(f"Valid: {result['is_valid']}, Attempts: {result['attempts']}")
print(result["response"][:200])

ReAct Agent Loop

The classic Reason → Act → Observe loop:

Python

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

class ReActState(TypedDict):
    question: str
    thoughts: Annotated[list[str], operator.add]    # Accumulate
    actions: Annotated[list[str], operator.add]
    observations: Annotated[list[str], operator.add]
    final_answer: str
    max_steps: int
    step: int

def reason(state: ReActState) -> ReActState:
    """Think about what to do next."""
    from openai import OpenAI
    client = OpenAI()

    history = "\n".join(
        f"Thought: {t}\nAction: {a}\nObservation: {o}"
        for t, a, o in zip(state["thoughts"], state["actions"], state["observations"])
    )

    prompt = f"""Question: {state['question']}

{history}

Think step by step about what to do next. If you have enough information, provide the final answer.
Otherwise, describe what action to take (e.g., search for a drug interaction).

Respond as:
Thought: <your reasoning>
Action: <action to take> OR Final Answer: <answer>"""

    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1,
    )

    content = resp.choices[0].message.content
    thought = ""
    action = ""
    final_answer = ""

    for line in content.splitlines():
        if line.startswith("Thought:"):
            thought = line[8:].strip()
        elif line.startswith("Action:"):
            action = line[7:].strip()
        elif line.startswith("Final Answer:"):
            final_answer = line[13:].strip()

    if final_answer:
        return {**state, "thoughts": [thought], "final_answer": final_answer, "step": state["step"] + 1}

    return {**state, "thoughts": [thought], "actions": [action], "step": state["step"] + 1}

def act(state: ReActState) -> ReActState:
    """Execute the most recent action."""
    if not state["actions"] or state["final_answer"]:
        return {**state, "observations": [""]}

    action = state["actions"][-1]

    # Simulated tool execution
    if "interaction" in action.lower() and "warfarin" in action.lower():
        observation = "Warfarin and ibuprofen have a major drug interaction. NSAIDs increase bleeding risk and may displace warfarin from protein binding sites."
    else:
        observation = f"Executed: {action}. No specific result found."

    return {**state, "observations": [observation]}

def should_continue(state: ReActState) -> str:
    """Continue looping or end."""
    if state["final_answer"]:
        return END
    if state["step"] >= state["max_steps"]:
        return END
    return "reason"

# Build graph
builder = StateGraph(ReActState)
builder.add_node("reason", reason)
builder.add_node("act", act)
builder.set_entry_point("reason")

builder.add_conditional_edges("reason", should_continue, {"reason": "act", END: END})
builder.add_edge("act", "reason")  # Always loop back after acting

react_app = builder.compile()

result = react_app.invoke({
    "question": "What interaction should I worry about with warfarin and ibuprofen?",
    "thoughts": [],
    "actions": [],
    "observations": [],
    "final_answer": "",
    "max_steps": 5,
    "step": 0,
})

print(f"Final answer: {result['final_answer']}")
print(f"Steps taken: {result['step']}")

Preventing Infinite Loops

LangGraph doesn't automatically prevent infinite loops. You must build termination conditions:

Approach 1: Step counter

Python

# In routing function
if state["step"] >= state["max_steps"]:
    return END

Approach 2: Convergence check

Python

def has_converged(state) -> bool:
    """Stop if last two outputs are very similar."""
    if len(state["outputs"]) < 2:
        return False
    last = state["outputs"][-1]
    prev = state["outputs"][-2]
    # Simple string similarity
    return len(set(last.split()) & set(prev.split())) / len(set(last.split())) > 0.8

Approach 3: Recursion limit

Python

# Set at compile time
app = graph.compile(recursion_limit=25)  # Default is 25

Always pair cycles with explicit termination conditions. A missing return clause in a routing function that should return END is the most common source of infinite loops in LangGraph.

Cycles and Loops in LangGraph

Why Cycles Matter

Basic Retry Loop

ReAct Agent Loop

Preventing Infinite Loops

Enjoyed this article?

Leave a comment