LangGraph Agents · Lesson 7 of 17
Cycles and Loops: When Agents Retry
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 ENDApproach 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.8Approach 3: Recursion limit
Python
# Set at compile time
app = graph.compile(recursion_limit=25) # Default is 25Always 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.