The code-review agent from the first lesson stops only when the model stops calling tools. tools_condition can’t tell a finished answer from a model that is stuck, and a run that runs out of steps ends in GraphRecursionError with no answer. Now you replace that edge with your own router: one conditional edge that sends each reply to the tools, to the answer, or to a stop node that ends the run with a named reason. Asked to keep checking CI until it passes, Gemini (gemini-3.8-flash) called the same tool twice, and the router stopped the run after 4 steps:
When you need this. Say your team wires the review assistant into a chat channel, and someone asks it to keep checking CI on #415 until it goes green. The model can keep requesting the same fixed CI result, adding model turns without new information. Set a stopping policy before unattended runs in a chat bot, a scheduled job or a queue worker.
1. Add the stop state to the graph
Download the complete lab, extract it and open a terminal in agent-state-machine/. The README lists its runtime and recorded examples.
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements-lock.txtFollow the lab setup and run instructions before executing this lesson.
Use the extracted lab and its virtual environment.
"""The code-review agent with its own router: it branches on each reply and stops inside a step limit."""
import json
import os
import sys
from typing import Annotated, Literal
from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, SystemMessage
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_ollama import ChatOllama
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.managed import RemainingSteps
from langgraph.prebuilt import ToolNode
from typing_extensions import TypedDict
from review_queue import UnknownPullRequest, get_ci_status, get_diff, list_open_prs
TOOLS = [list_open_prs, get_diff, get_ci_status]
RECURSION_LIMIT = 12 # graph steps per run; each node run is one step
SYSTEM = """You help a developer work through the team's code-review queue.
Use the tools to look up pull requests; never guess a diff or a CI result.
Answer in at most three sentences and name each pull request by its number."""
MODEL = os.environ.get("REVIEW_MODEL", "gemini-3.8-flash") # REVIEW_MODEL=gemma4:e2b runs it on Ollama
class State(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
remaining_steps: RemainingSteps # filled in by LangGraph: limit minus steps taken
stop_reason: str
def stop_reason(state: State) -> str | None:
"""Why the run must stop now, or None if it may continue."""
*earlier, last = state["messages"]
# Exact match on name and arguments; sort_keys so {"a": 1, "b": 2} and {"b": 2, "a": 1} are one call.
# A call that asks for the same thing in other words, such as "415" for 415, is not caught.
def same(c): return c["name"], json.dumps(c["args"], sort_keys=True)
seen = {same(c) for m in earlier if isinstance(m, AIMessage) for c in m.tool_calls}
if any(same(c) in seen for c in last.tool_calls):
return "repeated_call"
# Calling tools needs room for three more steps: tools, agent, and a possible stop.
if last.tool_calls and state["remaining_steps"] < 4:
return "step_limit"
return None
def chat_model():
if MODEL.startswith("gemini"):
return ChatGoogleGenerativeAI(model=MODEL) # reads GOOGLE_API_KEY from the environment
return ChatOllama(model=MODEL, temperature=0, reasoning=False)
def build_graph(model=None):
llm = (model or chat_model()).bind_tools(TOOLS)
def agent(state: State) -> dict:
return {"messages": [llm.invoke([SystemMessage(SYSTEM), *state["messages"]])]}
def route(state: State) -> Literal["tools", "stop", "__end__"]:
if stop_reason(state):
return "stop"
return "tools" if state["messages"][-1].tool_calls else END
def stop(state: State) -> dict:
reason = stop_reason(state)
return {"stop_reason": reason, "messages": [AIMessage(f"I stopped before answering ({reason}).")]}
graph = StateGraph(State)
graph.add_node("agent", agent)
graph.add_node("tools", ToolNode(TOOLS, handle_tool_errors=UnknownPullRequest))
graph.add_node("stop", stop)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", route)
graph.add_edge("tools", "agent")
graph.add_edge("stop", END)
return graph.compile()
def ask(graph, question: str) -> dict:
return graph.invoke({"messages": [HumanMessage(question)]}, {"recursion_limit": RECURSION_LIMIT})
if __name__ == "__main__":
result = ask(build_graph(), " ".join(sys.argv[1:]) or "Which open pull request has failing CI, and why?")
print(result.get("stop_reason", "answered"), "|", result["messages"][-1].text)The state has two fields beyond messages. stop_reason is how a caller tells an answer from a stop: the answer branch in this graph leaves it unset. remaining_steps is a managed value. The application does not write it; LangGraph fills it in before each node with the run’s recursion_limit minus the steps already taken, so a node or a router can see how much room is left.
The default limit. The default permits over ten thousand graph steps. This graph can make many model and tool calls before that limit trips. Pass recursion_limit in the config of every run. Check: test_the_default_limit_is_far_above_a_useful_run.
The repeated-call check compares a call with every earlier call in the thread, not only the previous one, so a model that alternates between two calls is caught on its third reply. The check is deliberately narrow. It matches exact arguments only: it compares each call’s name and its arguments as JSON with sorted keys, so key order does not matter, but {"number": "415"} and {"number": 415}, or the same question asked with different arguments, count as different calls and pass it. test_the_repeat_check_matches_exact_arguments_in_any_key_order pins both cases. Such a loop is still stopped by the step-limit check and by recursion_limit, which do not depend on the arguments. A tool whose result can change, such as a CI status you really want to poll, needs a wait between calls and a count, not a model that calls it in a loop.
2. Route each reply with a conditional edge
Now we replace tools_condition with a function of our own. Its return type, Literal["tools", "stop", "__end__"], lists every node it can go to, so LangGraph knows the graph’s edges when it compiles. The order of its checks is the design:
- A repeated call. If the reply asks for a tool call, with the same name and arguments, that an earlier reply already made, the result won’t change. The run stops with
repeated_callbefore the call runs again. - Not enough steps left. A tool round needs three more steps:
tools,agent, and possiblystop. With fewer than four left, the run stops withstep_limitwhile it can still end in a named state. Waiting forGraphRecursionErrorwould discard the run’s last state. - Otherwise the reply goes to
toolsif it has tool calls, and toENDif it doesn’t.
stop writes the reason and a short final message. A production caller would show the partial results or hand the task to a person; what matters is that the run ends where your code decided, not where an exception happened.
Note: stop_reason() runs twice, once in route and once in stop. A router can’t update the state; only nodes can. Recomputing keeps the decision in one pure function that the tests can call.
The model repeats a call. Asked to check CI until it passes, the model called get_ci_status for #415 twice. This local tool returns the same CI result; requesting it again adds no information. Stop on a repeated name and arguments, before the call runs. Check: test_the_same_call_twice_stops_before_it_runs_again.
The model keeps calling tools. With recursion_limit 5, the run needed one more tool round than it had steps for. Read remaining_steps and stop while a stop node can still run. Check: test_a_model_that_never_answers_stops_inside_the_limit.
No stop branch. With tools_condition and the scripted model that keeps requesting tools, invoke raises GraphRecursionError and returns no state. Route to a stop node; catch GraphRecursionError only as a last resort. Check: test_without_the_stop_branch_the_run_raises.
3. Run it on Gemini
record.py asks three questions with recursion_limit 12, then the first question again with a limit of 5, and saves every node path to runs.json. python graph.py "<question>" runs one question.
python record.py== Which open pull request has failing CI, and why? (recursion_limit=12)
path: agent -> tools -> agent -> tools -> agent -> tools -> agent
proposed: list_open_prs {}
proposed: get_ci_status {"number": 412}
proposed: get_ci_status {"number": 415}
proposed: get_ci_status {"number": 418}
proposed: get_diff {"number": 415}
outcome: answered after 7 steps
final: Pull request #415 has failing CI. It failed `tests/test_client.py::test_default_timeout` because the test expected a default timeout value of 10, but got 30 after `DEFAULT_TIMEOUT` was updated.
== Which open pull request has failing CI, and why? (recursion_limit=5)
path: agent -> tools -> agent -> stop
proposed: list_open_prs {}
proposed: get_ci_status {"number": 412}
proposed: get_ci_status {"number": 415}
proposed: get_ci_status {"number": 418}
outcome: step_limit after 4 steps
final: I stopped before answering (step_limit).The first question took 7 steps: the model listed the pull requests, checked CI on all three in one reply, then read the diff of the failing one. The second took 3. The step count is the model’s choice, not yours, which is why the limit is set per run and checked inside it.
Read a recorded step-limit stop
With recursion_limit=3, the recorded model proposed two tools, but the remaining-step guard routed directly from agent to stop. Neither tool ran. The stop node returned I stopped before answering (step_limit). That sentence comes from graph code, not the model.
Inspect the saved events
[
{
"event": "task",
"question": "For pull request #415, inspect its diff and CI status. In at most three sentences, explain why CI fails and which changed constant conflicts with the test. Use the supplied read-only tools; do not change files or CI.",
"recursion_limit": 3,
"policy": {
"lesson_id": "agent-state-machine",
"model": "gemini-3.8-flash",
"max_model_calls": 1,
"max_output_tokens": 4096,
"thinking_level": "low",
"timeout_seconds": 45,
"max_retries": 0,
"vertexai": false
}
},
{
"event": "model_request",
"ordinal": 1,
"policy": {
"lesson_id": "agent-state-machine",
"model": "gemini-3.8-flash",
"max_model_calls": 1,
"max_output_tokens": 4096,
"thinking_level": "low",
"timeout_seconds": 45,
"max_retries": 0,
"vertexai": false
}
},
{
"event": "model_response",
"ordinal": 1,
"message": {
"type": "ai",
"text": "",
"tool_calls": [
{
"name": "get_diff",
"args": {
"number": 415
},
"id": "call_79959"
},
{
"name": "get_ci_status",
"args": {
"number": 415
},
"id": "call_79960"
}
],
"usage": {
"input_tokens": 243,
"output_tokens": 34,
"total_tokens": 277,
"input_token_details": {
"cache_read": 0
}
},
"finish_reason": "STOP"
}
},
{
"event": "graph_step",
"node": "agent",
"messages": [
{
"type": "ai",
"text": "",
"tool_calls": [
{
"name": "get_diff",
"args": {
"number": 415
},
"id": "call_79959"
},
{
"name": "get_ci_status",
"args": {
"number": 415
},
"id": "call_79960"
}
],
"usage": {
"input_tokens": 243,
"output_tokens": 34,
"total_tokens": 277,
"input_token_details": {
"cache_read": 0
}
},
"finish_reason": "STOP"
}
]
},
{
"event": "graph_step",
"node": "stop",
"messages": [
{
"type": "ai",
"text": "I stopped before answering (step_limit).",
"tool_calls": [],
"usage": null,
"finish_reason": null
}
]
},
{
"event": "completion",
"status": "completed_native_step_limit",
"stop_reason": "step_limit",
"model_calls": 1,
"path": [
"agent",
"stop"
],
"tool_execution_count": 0,
"final": "I stopped before answering (step_limit).",
"boundary": "The real model proposed tools; RemainingSteps stopped before execution. No answer correctness claim."
}
]How it works
LangGraph runs a compiled graph in steps. In each step the scheduled nodes run, their updates go through the reducers, and the edges pick the next nodes. A conditional edge is a function of the state; it runs after agent and returns a node name. recursion_limit is the number of steps one run may take. The installed LangGraph uses a default of 10007 (DEFAULT_RECURSION_LIMIT, which the LANGGRAPH_DEFAULT_RECURSION_LIMIT environment variable overrides), while the graph API documentation gives 1000. Either way, the default is far above a useful review run, so set it on every call.
With a limit of 5, the second agent call ran at step 3 and asked for three get_ci_status calls. The steps left could not hold another tool round, so route sent the run to stop, which ran at step 4.
Test the router without a model
test_graph.py replaces Gemini with a scripted chat model that returns fixed replies, and runs the real graph. One test runs the scripted tool-request loop against each limit from 3 to 12. Each of those ten cases ends in stop without raising GraphRecursionError.
python -m pytest -v test_graph.pytest_graph.py::test_an_answer_ends_the_run PASSED [ 6%]
test_graph.py::test_the_same_call_twice_stops_before_it_runs_again PASSED [ 13%]
test_graph.py::test_the_repeat_check_matches_exact_arguments_in_any_key_order PASSED [ 20%]
test_graph.py::test_a_model_that_never_answers_stops_inside_the_limit[3] PASSED [ 26%]
test_graph.py::test_a_model_that_never_answers_stops_inside_the_limit[4] PASSED [ 33%]
test_graph.py::test_a_model_that_never_answers_stops_inside_the_limit[5] PASSED [ 40%]
test_graph.py::test_a_model_that_never_answers_stops_inside_the_limit[6] PASSED [ 46%]
test_graph.py::test_a_model_that_never_answers_stops_inside_the_limit[7] PASSED [ 53%]
test_graph.py::test_a_model_that_never_answers_stops_inside_the_limit[8] PASSED [ 60%]
test_graph.py::test_a_model_that_never_answers_stops_inside_the_limit[9] PASSED [ 66%]
test_graph.py::test_a_model_that_never_answers_stops_inside_the_limit[10] PASSED [ 73%]
test_graph.py::test_a_model_that_never_answers_stops_inside_the_limit[11] PASSED [ 80%]
test_graph.py::test_a_model_that_never_answers_stops_inside_the_limit[12] PASSED [ 86%]
test_graph.py::test_without_the_stop_branch_the_run_raises PASSED [ 93%]
test_graph.py::test_the_default_limit_is_far_above_a_useful_run PASSED [100%]Running teardown with pytest sessionfinish...
============================== 15 passed in 0.35s ==============================All 15 tests pass.
Read the test and recording scripts
"""Tests for the code-review agent graph. A scripted chat model replaces Gemini, so no network or API key is needed."""
from itertools import count
import pytest
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.errors import GraphRecursionError
from langgraph.graph import END, START, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition
import graph
class ScriptedChat(GenericFakeChatModel):
"""Returns the scripted messages in order; tool binding is a no-op."""
def bind_tools(self, tools, **kwargs):
return self
def call(name, **args):
return AIMessage("", tool_calls=[{"name": name, "args": args, "id": f"c{next(ids)}"}])
ids = count()
def endless_lookups():
for n in count(1):
yield call("get_diff", number=400 + n) # a new number each time, so no call repeats
def run(messages, limit=graph.RECURSION_LIMIT):
g = graph.build_graph(ScriptedChat(messages=iter(messages)))
path = []
state = None
for update in g.stream({"messages": [HumanMessage("q")]}, {"recursion_limit": limit},
stream_mode=["updates", "values"]):
kind, data = update
if kind == "updates":
path += list(data)
else:
state = data
return path, state
def test_an_answer_ends_the_run():
path, state = run([call("list_open_prs"), call("get_ci_status", number=415),
AIMessage("#415 fails test_default_timeout.")])
assert path == ["agent", "tools", "agent", "tools", "agent"]
assert "stop_reason" not in state
assert "expected 10, got 30" in state["messages"][-2].content # the tool result the model read
def test_the_same_call_twice_stops_before_it_runs_again():
path, state = run([call("get_ci_status", number=415), call("get_ci_status", number=415)])
assert path == ["agent", "tools", "agent", "stop"]
assert state["stop_reason"] == "repeated_call"
def test_the_repeat_check_matches_exact_arguments_in_any_key_order():
first = AIMessage("", tool_calls=[{"name": "get_diff", "args": {"number": 415, "path": "a.py"}, "id": "c-a"}])
reordered = AIMessage("", tool_calls=[{"name": "get_diff", "args": {"path": "a.py", "number": 415}, "id": "c-b"}])
as_text = AIMessage("", tool_calls=[{"name": "get_diff", "args": {"number": "415", "path": "a.py"}, "id": "c-c"}])
state = {"messages": [HumanMessage("q"), first], "remaining_steps": 20}
assert graph.stop_reason({**state, "messages": [*state["messages"], reordered]}) == "repeated_call"
assert graph.stop_reason({**state, "messages": [*state["messages"], as_text]}) is None # exact match only
@pytest.mark.parametrize("limit", range(3, 13))
def test_a_model_that_never_answers_stops_inside_the_limit(limit):
path, state = run(endless_lookups(), limit)
assert state["stop_reason"] == "step_limit" and path[-1] == "stop"
assert len(path) < limit
def test_without_the_stop_branch_the_run_raises():
plain = StateGraph(graph.State)
llm = ScriptedChat(messages=endless_lookups())
plain.add_node("agent", lambda s: {"messages": [llm.invoke(s["messages"])]})
plain.add_node("tools", ToolNode(graph.TOOLS, handle_tool_errors=True))
plain.add_edge(START, "agent")
plain.add_conditional_edges("agent", tools_condition)
plain.add_edge("tools", "agent")
with pytest.raises(GraphRecursionError):
plain.compile().invoke({"messages": [HumanMessage("q")]}, {"recursion_limit": 8})
def test_the_default_limit_is_far_above_a_useful_run():
seen = []
probe = StateGraph(graph.State)
probe.add_node("look", lambda s: seen.append(s["remaining_steps"]) or {})
probe.add_edge(START, "look")
probe.add_edge("look", END)
probe.compile().invoke({"messages": []})
assert seen[0] > 10_000 # so set recursion_limit on every run"""Run the code-review agent on Gemini and save the node path of every run."""
import json
from langchain_core.messages import HumanMessage
from langgraph.errors import GraphRecursionError
from graph import MODEL, RECURSION_LIMIT, build_graph
QUESTIONS = [
"Which open pull request has failing CI, and why?",
"What does pull request 418 change, and did its CI pass?",
"Has CI been re-run on #415 since the last push? Check until it passes.",
]
def run(graph, question: str, limit: int) -> dict:
path, calls, final, reason = [], [], None, "answered"
try:
for update in graph.stream({"messages": [HumanMessage(question)]},
{"recursion_limit": limit}, stream_mode="updates"):
for node, delta in update.items():
path.append(node)
message = delta["messages"][-1]
calls += [{"name": c["name"], "args": c["args"]} for c in getattr(message, "tool_calls", [])]
if node in ("agent", "stop") and not getattr(message, "tool_calls", None):
final = message.text
reason = delta.get("stop_reason", reason)
except GraphRecursionError as error:
reason, final = "GraphRecursionError", str(error).splitlines()[0]
return {"question": question, "recursion_limit": limit, "path": path, "steps": len(path),
"tool_calls": calls, "outcome": reason, "final": final}
def main() -> None:
graph = build_graph()
runs = [run(graph, q, RECURSION_LIMIT) for q in QUESTIONS]
# The first question again with a limit too small for two tool rounds: the step-limit branch.
runs.append(run(graph, QUESTIONS[0], 5))
with open("runs.json", "w") as f:
json.dump({"model": MODEL, "runs": runs}, f, indent=1)
for r in runs:
print(f"== {r['question']} (recursion_limit={r['recursion_limit']})")
print(" path: ", " -> ".join(r["path"]))
for c in r["tool_calls"]:
print(" proposed:", c["name"], json.dumps(c["args"]))
print(" outcome: ", r["outcome"], "after", r["steps"], "steps")
print(" final: ", r["final"])
if __name__ == "__main__":
main(){
"question": "Has CI been re-run on #415 since the last push? Check until it passes.",
"recursion_limit": 12,
"path": [
"agent",
"tools",
"agent",
"stop"
],
"steps": 4,
"tool_calls": [
{
"name": "get_ci_status",
"args": {
"number": 415
}
},
{
"name": "get_ci_status",
"args": {
"number": 415
}
}
],
"outcome": "repeated_call",
"final": "I stopped before answering (repeated_call)."
}Read the explanation
Let the graph end in a named state while it still can; treat GraphRecursionError as a last resort.
What’s next
- Trim conversation history to a token budget: a step limit bounds the number of calls; a token budget bounds what each call sends.
References
Official sources checked October 21, 2026, against langgraph 1.2.16 and langchain-core 1.6.12.
- LangGraph: Graph API: conditional edges,
recursion_limitand its documented default. - LangGraph: Use the graph API:
RemainingStepsand handling the recursion limit inside a graph. - LangGraph: GRAPH_RECURSION_LIMIT: the error and how to raise the limit.
- LangChain: ChatGoogleGenerativeAI: the Gemini chat model and tool binding.
- Gemini API: Models: the
gemini-3.8-flashmodel ID. - The default limit of 10007 is from the installed
langgraph/_internal/_config.py.