Module 4: State Machines for Agents with LangGraph
4. Agent loops as cyclic graphs
Overview
In the previous capsule you designed your agent's typed state: AgentState with messages, plan, iteration_count, tool_results. In capsule 02 you learned to model an agent as a StateGraph with functional nodes and edges as transitions. Now you're going to combine both concepts to implement what really makes an agent an agent: the cycle.
An agent is not a linear chain that takes input and produces output. An agent is a loop: it reasons, acts, observes the result, and reasons again. That "reasons again" is the key point — it's an edge that goes back to an earlier node, creating a cycle in the graph. In Module 1 you implemented this loop by hand with a Python for. Now you're going to implement it as a cyclic graph in LangGraph, where the cycle is a structural property of the graph, not a control-flow trick.
The difference runs deep: when the cycle is part of the graph, you can visualize it (draw_mermaid_png), debug it (inspect each iteration in the state), control it (stop conditions as part of the graph), and extend it (add nodes to the cycle without rewriting the loop).
The agent as a cycle
Why cycles, not chains
Most LLM applications are linear chains: input → prompt → LLM → output. Even RAG is sequential: query → retrieval → augment → generate.
An agent is fundamentally different. You don't know ahead of time how many times it will run, what tools it will call, or when it will decide it's done. The agent iterates until it reaches its goal or runs out of attempts. That's a cycle:
┌─────────────────────────────────────────────────────────────┐
│ │
│ START │
│ │ │
│ ▼ │
│ ┌──────────┐ │
│ │ REASON │ The LLM analyzes the current state and │
│ │ │ decides: need more info? can I answer? │
│ └────┬─────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────┐ │
│ │ SHOULD_CONTINUE │ Any tool_calls? │
│ │ (conditional) │ Max iterations reached? │
│ └──┬────────────┬───┘ │
│ │ │ │
│ "tools" "end" │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────┐ ┌─────┐ │
│ │ TOOLS │ │ END │ │
│ │ Executes │ └─────┘ │
│ │ tools │ │
│ └────┬─────┘ │
│ │ │
│ │ ◀── THIS IS THE CYCLE │
│ │ The "tools" edge goes back to "reason" │
│ │ │
│ └──────────────► REASON (back to the top) │
│ │
└─────────────────────────────────────────────────────────────┘
The anatomy of the cycle
- Reasoning node (
reason): The LLM processes the state and decides what to do. - Conditional edge (
should_continue): Decides whether to continue the cycle or exit. It's the loop's "gate". - Action node (
tools): Runs the tools and adds the results to the state. - Return edge:
tools → reason. Without this edge, there is no loop.
The conditional edge is the most important component. It's the analog of Python's while condition: — but expressed as a property of the graph.
Cycles vs the manual loop from M1
In Module 1 you implemented the cycle as a Python for:
for iteration in range(max_iterations):
response = model.invoke(messages)
if not response.tool_calls:
break
In LangGraph, the cycle lives in the graph itself:
graph.add_edge("tools", "reason") # This edge CREATES the cycle
In the manual loop, the flow control is invisible — it's Python code. In the graph, the cycle is visible, serializable, and debuggable. You can draw it and see the cycle. You can inspect the state at any point. You can add nodes without touching the flow control.
Implementing a cycle in StateGraph
Setup: state, tools, and model
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import ToolMessage, SystemMessage, HumanMessage
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
iteration_count: int
max_iterations: int
@tool
def search_web(query: str) -> str:
"""Search the web for information about a topic."""
fake_results = {
"langchain": "LangChain is a framework for building applications with LLMs.",
"langgraph": "LangGraph lets you build agents as stateful graphs with cycles.",
"react agents": "ReAct combines reasoning and acting in an iterative loop.",
}
for key, value in fake_results.items():
if key in query.lower():
return value
return f"Results for '{query}': general information found."
@tool
def analyze_text(text: str) -> str:
"""Analyze a text and extract its key points."""
return f"Analysis ({len(text.split())} words): technical concepts identified."
tools = [search_web, analyze_text]
tools_by_name = {t.name: t for t in tools}
model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)
The nodes of the cycle
def reason_node(state: AgentState) -> dict:
"""The LLM analyzes the state and decides what to do."""
response = model_with_tools.invoke(state["messages"])
new_iteration = state.get("iteration_count", 0) + 1
return {
"messages": [response],
"iteration_count": new_iteration,
}
def tools_node(state: AgentState) -> dict:
"""Runs the tools the LLM requested."""
last_message = state["messages"][-1]
tool_messages = []
for tc in last_message.tool_calls:
if tc["name"] in tools_by_name:
result = tools_by_name[tc["name"]].invoke(tc["args"])
else:
result = f"Error: tool '{tc['name']}' does not exist."
tool_messages.append(
ToolMessage(content=str(result), tool_call_id=tc["id"])
)
return {"messages": tool_messages}
reason_node increments iteration_count every time it runs. Since it's the "start" of each iteration, counting here gives you precise tracking.
The conditional edge: the gate of the cycle
def should_continue(state: AgentState) -> str:
"""Decides whether to continue the cycle or finish."""
last_message = state["messages"][-1]
# Stop condition 1: The model didn't ask for tools (task complete)
if not last_message.tool_calls:
return "end"
# Stop condition 2: Iteration limit
if state.get("iteration_count", 0) >= state.get("max_iterations", 10):
return "end"
return "tools"
Building and compiling the graph
graph = StateGraph(AgentState)
graph.add_node("reason", reason_node)
graph.add_node("tools", tools_node)
graph.add_edge(START, "reason")
graph.add_conditional_edges(
"reason",
should_continue,
{"tools": "tools", "end": END}
)
# ═══════════════════════════════════════
# THIS EDGE IS THE CYCLE
# It goes from "tools" BACK to "reason"
# ═══════════════════════════════════════
graph.add_edge("tools", "reason")
agent = graph.compile()
graph.add_edge("tools", "reason") is the line that creates the cycle. Without it, the graph would be linear. With it, the graph forms a loop: reason → tools → reason → tools → ... until should_continue returns "end".
Running it
result = agent.invoke({
"messages": [
SystemMessage(content="You are a research agent. Use tools to investigate."),
HumanMessage(content="Research what LangGraph is.")
],
"iteration_count": 0,
"max_iterations": 5,
})
print(result["messages"][-1].content)
print(f"Iterations: {result['iteration_count']}")
# Iteration 1: reason → search_web("langgraph") → tools
# Iteration 2: reason → it has enough → END
# Iterations: 2
Stop conditions
Stop conditions decide when the agent stops iterating. Without them, the agent runs forever.
Condition 1: Task complete (no tool_calls)
The most natural one. When the LLM returns without tool_calls, it's saying "I already have enough information":
if not last_message.tool_calls:
return "end"
Condition 2: Max iterations
Protection against infinite loops. It should always be there as a safety net:
if state["iteration_count"] >= state["max_iterations"]:
return "end"
Condition 3: Quality score threshold
If your agent evaluates the quality of its own work (the reflection pattern in M5), stop when the quality is good enough: state.get("quality_score", 0.0) >= 0.9.
Condition 4: Budget exhausted
In production, every iteration consumes tokens. Stop when the cost exceeds a limit: state.get("budget_remaining", 1.0) <= 0.
Multiple simultaneous stop conditions
In practice you combine all of them. Evaluate the cheapest ones first:
def should_continue(state: AgentState) -> str:
"""Evaluates multiple stop conditions simultaneously."""
last_message = state["messages"][-1]
if not last_message.tool_calls:
return "end"
if state.get("iteration_count", 0) >= state.get("max_iterations", 10):
return "end"
if state.get("quality_score", 0.0) >= 0.9:
return "end"
if state.get("budget_remaining", 1.0) <= 0:
return "end"
return "tools"
Stop condition table
| Condition | When it stops | Type | Cost to evaluate |
|---|---|---|---|
not tool_calls | The model decides it's done | Natural | Free |
iteration_count >= max | The safety net is reached | Safety | Free |
quality_score >= threshold | Quality is good enough | Quality | Medium |
budget_remaining <= 0 | Out of budget | Cost | Low |
error_count >= max_errors | Too many failures | Resilience | Low |
time_elapsed >= timeout | Timeout | Time | Low |
Debugging cycles
When your agent has a cycle, debugging means understanding what happened in each iteration.
Technique 1: Logging in the reason node
def reason_node_debug(state: AgentState) -> dict:
iteration = state.get("iteration_count", 0) + 1
print(f"\n{'='*50}")
print(f"ITERATION {iteration} | Messages: {len(state['messages'])}")
response = model_with_tools.invoke(state["messages"])
if response.tool_calls:
for tc in response.tool_calls:
print(f" → {tc['name']}({tc['args']})")
else:
print(f" → Answer directly")
return {"messages": [response], "iteration_count": iteration}
Typical output:
==================================================
ITERATION 1 | Messages: 2
→ search_web({'query': 'langgraph'})
==================================================
ITERATION 2 | Messages: 4
→ analyze_text({'text': 'LangGraph lets you build...'})
==================================================
ITERATION 3 | Messages: 6
→ Answer directly
You can see the context grow (2 → 4 → 6 messages), what it decides on each round, and why it finished.
Technique 2: Visualize the graph with draw_mermaid_png
It's not a nice-to-have — it's the fastest tool for checking that your cycle is where it should be:
from IPython.display import Image, display
img = agent.get_graph().draw_mermaid_png()
display(Image(img))
If the return edge tools → reason doesn't appear in the diagram, your cycle doesn't exist. If there are unexpected edges, you have a routing bug.
Technique 3: Stream to see each step in real time
Use stream instead of invoke to see each node as it runs:
for event in agent.stream({
"messages": [HumanMessage(content="Research LangGraph")],
"iteration_count": 0,
"max_iterations": 5,
}):
for node_name, node_output in event.items():
print(f"--- Node: {node_name} ---")
if "messages" in node_output:
for msg in node_output["messages"]:
print(f" {type(msg).__name__}: {str(msg.content)[:80]}")
The danger of the infinite loop
What happens without stop conditions
If your should_continue always returns "tools", LangGraph has a safety net: recursion_limit.
# Bug: it NEVER returns "end"
def bad_should_continue(state: AgentState) -> str:
return "tools"
# LangGraph raises:
# GraphRecursionError: Recursion limit of 25 reached without hitting a stop condition.
recursion_limit
LangGraph has a default recursion_limit of 25 steps. Each node counts as one step. A reason → tools cycle uses 2 steps per iteration, allowing ~12 iterations.
result = agent.invoke(
{"messages": [HumanMessage(content="...")]},
config={"recursion_limit": 50}
)
Three layers of protection
| Layer | Mechanism | When it kicks in |
|---|---|---|
| 1. Logic | not tool_calls | The model decides it's done |
| 2. Application | iteration_count >= max | Your custom limit is reached |
| 3. Framework | recursion_limit | Last resort if layers 1 and 2 fail |
All three should be present. Layer 1 is the natural condition. Layer 2 is your explicit control. Layer 3 is the framework's safety net.
Error handling inside the cycle
What happens if a tool fails? The best option: tell the LLM and let it decide. Wrap the execution in try/except and turn the error into a ToolMessage:
for tc in last_message.tool_calls:
try:
result = tools_by_name[tc["name"]].invoke(tc["args"])
except Exception as e:
result = f"Error in {tc['name']}: {str(e)}. Try another tool or answer with what you have."
tool_messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
The LLM reads the error on the next iteration and decides: retry, use another tool, or answer with what it has. The cycle keeps going — the error is just another piece of data.
Comparison: manual loop (M1) vs StateGraph cycle
| Aspect | Manual loop (M1) | StateGraph cycle (M4) |
|---|---|---|
| Where the cycle lives | A for/while in Python | The tools → reason edge in the graph |
| Visibility | Invisible (code) | Visible (draw_mermaid_png) |
| State | A local variable (messages = []) | A typed, persistable AgentState |
| Stop conditions | if/break inside the loop | An explicit should_continue function |
| Debugging | Manual print() | stream(), state inspection, visualization |
| Checkpointing | No (process dies → it's lost) | Yes (M6: MemorySaver/PostgresSaver) |
| Extensibility | Rewrite the loop to add steps | Add nodes and edges without touching the cycle |
| Testing | Test the whole loop | Test individual nodes |
| Code | ~25 lines | ~40 lines (more setup, more structure) |
| When to use | Prototypes, learning | Production, complex flows |
The key difference: the manual loop mixes agent logic with flow control. The StateGraph separates them — the logic lives in the nodes, the control in the edges.
Connection with the project
In this module's project (capsule 08, Research Agent State Machine):
- The
reason → tools → reasoncycle is the heart of your Research Agent. Your graph will have planning, research (with a cycle), analysis, and synthesis nodes — but the cycle lives in the research phase. - The stop conditions are real:
max_iterationsas a safety net,task_completewhen the agent decides it has enough. draw_mermaid_pnglets you verify that the cycle is where it should be.
In later modules:
- M5 (Planning): The cycle includes self-evaluation. After
act, areflectnode evaluates the quality and decides whether to repeat. - M6 (Memory): The cycle's state is persisted with checkpointing. If the process dies on iteration 3, you resume from there.
- M8 (Multi-Agent): Each agent has its own cycle, and a supervisor decides when each one should stop.
Troubleshooting
Problem 1: The agent always stops after 1 iteration
Cause: The model answers directly without using tools.
Solution: Check that the tools are bound to the model and the system prompt tells it to use them:
model_with_tools = model.bind_tools(tools)
messages = [
SystemMessage(content="ALWAYS use tools before answering."),
HumanMessage(content=user_input)
]
Problem 2: Unexpected GraphRecursionError
Cause: The recursion_limit (default 25) was reached before your max_iterations.
Solution: With 2 nodes per iteration, recursion_limit=25 allows ~12 iterations. If you need more:
result = agent.invoke(input, config={"recursion_limit": 50})
Rule: recursion_limit >= max_iterations * 2 + 5.
Problem 3: The state doesn't update between iterations
Cause: The node doesn't return the keys it's supposed to update.
Solution: If reason_node is supposed to increment iteration_count, it has to return it:
return {
"messages": [response],
"iteration_count": state.get("iteration_count", 0) + 1,
}
If you leave it out, the state doesn't change and your stop condition never fires.
Problem 4: draw_mermaid_png doesn't show the cycle
Cause: The return edge wasn't added, or there's a typo in the names.
Solution: Check that the names match exactly:
graph.add_node("reason", reason_node) # "reason"
graph.add_node("tools", tools_node) # "tools"
graph.add_edge("tools", "reason") # tools → reason
graph.add_edge("tool", "reason") (singular) won't raise an error — the cycle simply won't exist.
Exercises
Exercise 1: Identify the cycle in a graph (Easy)
Given the following code, identify: (a) which edge creates the cycle, (b) how many maximum iterations it allows, and (c) what happens if the model never stops asking for tools.
graph = StateGraph(AgentState)
graph.add_node("planner", planner_node)
graph.add_node("executor", executor_node)
graph.add_edge(START, "planner")
graph.add_conditional_edges("planner", check_done, {"execute": "executor", "done": END})
graph.add_edge("executor", "planner")
View solution
(a) graph.add_edge("executor", "planner") — it goes from executor back to planner, closing the loop.
(b) We can't know without seeing check_done. If it doesn't check iteration_count, the iterations are potentially infinite.
(c) LangGraph raises GraphRecursionError when it hits recursion_limit (default 25). With 2 nodes per iteration, ~12 iterations.
Lesson: Always include a max_iterations check in your conditional edge.
Exercise 2: Add a stop condition for errors (Easy)
Modify should_continue so it also stops if the last ToolMessage contains "ERROR":
def should_continue(state: AgentState) -> str:
last_message = state["messages"][-1]
if not last_message.tool_calls:
return "end"
if state["iteration_count"] >= state["max_iterations"]:
return "end"
return "tools"
View solution
from langchain_core.messages import ToolMessage
def should_continue(state: AgentState) -> str:
last_message = state["messages"][-1]
if not last_message.tool_calls:
return "end"
if state["iteration_count"] >= state["max_iterations"]:
return "end"
last_tool = next(
(m for m in reversed(state["messages"]) if isinstance(m, ToolMessage)),
None
)
if last_tool and "ERROR" in last_tool.content.upper():
return "end"
return "tools"
We walk the messages in reverse to find the most recent ToolMessage. We only check the result of the current iteration.
Exercise 3: A cycle with full logging (Medium)
Build a StateGraph where reason_node prints the iteration and the tools it calls, and should_continue prints why it decided to continue or stop. Use search_web and analyze_text.
View solution
class DebugState(TypedDict):
messages: Annotated[list, add_messages]
iteration_count: int
max_iterations: int
def reason_debug(state: DebugState) -> dict:
iteration = state.get("iteration_count", 0) + 1
print(f"\n[REASON] Iteration {iteration} | Messages: {len(state['messages'])}")
response = model_with_tools.invoke(state["messages"])
if response.tool_calls:
for tc in response.tool_calls:
print(f"[REASON] → {tc['name']}({tc['args']})")
else:
print(f"[REASON] → Answer directly")
return {"messages": [response], "iteration_count": iteration}
def tools_debug(state: DebugState) -> dict:
last = state["messages"][-1]
results = []
for tc in last.tool_calls:
result = tools_by_name[tc["name"]].invoke(tc["args"])
results.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
print(f"[TOOLS] {tc['name']} → {str(result)[:50]}")
return {"messages": results}
def gate(state: DebugState) -> str:
last = state["messages"][-1]
if not last.tool_calls:
print(f"[GATE] STOP: task complete")
return "end"
if state.get("iteration_count", 0) >= state.get("max_iterations", 5):
print(f"[GATE] STOP: max iterations")
return "end"
print(f"[GATE] CONTINUE: {state['iteration_count']}/{state['max_iterations']}")
return "tools"
graph = StateGraph(DebugState)
graph.add_node("reason", reason_debug)
graph.add_node("tools", tools_debug)
graph.add_edge(START, "reason")
graph.add_conditional_edges("reason", gate, {"tools": "tools", "end": END})
graph.add_edge("tools", "reason")
debug_agent = graph.compile()
result = debug_agent.invoke({
"messages": [
SystemMessage(content="Research using tools."),
HumanMessage(content="What is LangChain?")
],
"iteration_count": 0,
"max_iterations": 5,
})
print(f"\nIterations: {result['iteration_count']}")
Exercise 4: Double cycle — research + refinement (Hard)
Design a graph with two cycles: research (search → reason → search...) and refinement (write → evaluate → write...). A conditional edge decides when to move from research to refinement. Design the structure (don't implement the full nodes).
View solution
class DualCycleState(TypedDict):
messages: Annotated[list, add_messages]
research_iterations: int
refinement_iterations: int
max_research: int
max_refinement: int
def after_research(state: DualCycleState) -> str:
last = state["messages"][-1]
if not last.tool_calls:
return "write"
if state["research_iterations"] >= state["max_research"]:
return "write"
return "research_tools"
def after_evaluate(state: DualCycleState) -> str:
if state["refinement_iterations"] >= state["max_refinement"]:
return "done"
return "write"
graph = StateGraph(DualCycleState)
graph.add_node("research_reason", research_reason)
graph.add_node("research_tools", research_tools)
graph.add_node("write", write_draft)
graph.add_node("evaluate", evaluate_draft)
graph.add_edge(START, "research_reason")
graph.add_conditional_edges("research_reason", after_research, {
"research_tools": "research_tools",
"write": "write",
})
graph.add_edge("research_tools", "research_reason") # CYCLE 1
graph.add_edge("write", "evaluate")
graph.add_conditional_edges("evaluate", after_evaluate, {
"write": "write", # CYCLE 2
"done": END,
})
Structure:
START → research_reason ─┬─→ research_tools → research_reason (CYCLE 1)
└─→ write → evaluate ─┬─→ write (CYCLE 2)
└─→ END
A graph can have multiple independent cycles. In M5 you'll implement the refinement cycle with real reflection.
Exercise 5: Budget tracking inside the cycle (Hard)
Extend AgentState with budget_remaining (float, starts at 1.0) and cost_per_iteration (float, 0.15). Modify reason_node to decrement the budget and should_continue to stop when the budget runs out. Test it with a budget of 0.50.
View solution
class BudgetState(TypedDict):
messages: Annotated[list, add_messages]
iteration_count: int
max_iterations: int
budget_remaining: float
cost_per_iteration: float
def reason_budget(state: BudgetState) -> dict:
iteration = state.get("iteration_count", 0) + 1
cost = state.get("cost_per_iteration", 0.15)
new_budget = state.get("budget_remaining", 1.0) - cost
print(f"[Iter {iteration}] Budget: ${new_budget:.2f}")
response = model_with_tools.invoke(state["messages"])
return {
"messages": [response],
"iteration_count": iteration,
"budget_remaining": new_budget,
}
def gate_budget(state: BudgetState) -> str:
last = state["messages"][-1]
if not last.tool_calls:
return "end"
if state.get("iteration_count", 0) >= state.get("max_iterations", 10):
return "end"
if state.get("budget_remaining", 1.0) <= state.get("cost_per_iteration", 0.15):
print(f"[STOP] Not enough budget for another iteration")
return "end"
return "tools"
graph = StateGraph(BudgetState)
graph.add_node("reason", reason_budget)
graph.add_node("tools", tools_node)
graph.add_edge(START, "reason")
graph.add_conditional_edges("reason", gate_budget, {"tools": "tools", "end": END})
graph.add_edge("tools", "reason")
budget_agent = graph.compile()
result = budget_agent.invoke({
"messages": [
SystemMessage(content="Research using tools."),
HumanMessage(content="What is LangGraph?")
],
"iteration_count": 0,
"max_iterations": 10,
"budget_remaining": 0.50,
"cost_per_iteration": 0.15,
})
# With a budget of 0.50 and a cost of 0.15, it allows ~3 iterations before stopping.
print(f"Final budget: ${result['budget_remaining']:.2f}")
print(f"Iterations: {result['iteration_count']}")
The condition <= cost_per_iteration (instead of <= 0) prevents the budget from going negative.
Summary
In this capsule you learned:
- An agent is a cycle, not a chain: reason → act → observe → reason. The
tools → reasonedge is what creates the loop. - In LangGraph, a cycle is an edge that goes back to an earlier node.
graph.add_edge("tools", "reason")turns a linear graph into a cyclic one. - The conditional edge (
should_continue) is the cycle's gate. On each iteration it decides whether to keep going ("tools") or finish ("end"). - Stop conditions are multiple and simultaneous: task complete, max iterations, quality threshold, budget exhausted.
- Three layers of protection against infinite loops: the model's logic, your code (
max_iterations), the framework (recursion_limit). - Debugging cycles: logging in the nodes,
draw_mermaid_pngto visualize,stream()for real-time output,iteration_countin the state. - vs. the manual loop from M1: the StateGraph cycle is visible, persistable, extensible, and testable node by node.
Next capsule: Conditional Routing for Agents — conditional edges not just as "continue/stop" but as decision points that route based on result type, model confidence, or remaining budget.
Additional resources
- LangGraph Cycles — official documentation — How LangGraph handles cycles in graphs
- LangGraph Recursion Limit — Configuring and handling the recursion limit
- ReAct Paper — The foundational paper on the reason-act loop
- LangGraph Visualization — draw_mermaid_png and visualization tools
- LangGraph Streaming — stream() for real-time debugging