The hardest failure for a workflow is a write whose response is lost. The ticket was created, but the client timed out, so as far as the workflow knows the step failed. Retry it and you get two tickets; don’t retry and the run fails although its work is done. In this lesson the review team’s workflow has Gemini (gemini-3.8-flash) draft a ticket for the failing CI job on #415 and files it in a tracker that loses one response. A LangGraph RetryPolicy retries the write, and an idempotency key turns the retry into a replay of the first request:
When you need this. The review assistant files a ticket for a failed CI job. If the tracker loses the response after creating it, retry that workflow’s write with the same key so it refers to the existing ticket. Use an idempotency contract for retried writes outside the graph, such as these tickets or review comments.
1. Model the tracker’s contract
Download the complete lab, extract it and open a terminal in build-idempotent-retries-for-workflows/. The README lists its runtime and recorded examples.
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements-lock.txtUse the extracted lab and its virtual environment.
"""A stub issue tracker. It honors an idempotency key, and it can lose the response after creating a ticket."""
class Tracker:
def __init__(self, lost_responses: int = 0):
self.tickets: list[dict] = []
self.keys: dict[str, dict] = {}
self.requests: list[str | None] = [] # the key sent with each create request
self.lost_responses = lost_responses
def create(self, title: str, labels: list[str], idempotency_key: str | None = None) -> dict:
self.requests.append(idempotency_key)
if idempotency_key in self.keys: # a repeat of a request it already handled
return {**self.keys[idempotency_key], "replayed": True}
ticket = {"id": f"REV-{101 + len(self.tickets)}", "title": title, "labels": labels}
self.tickets.append(ticket)
if idempotency_key:
self.keys[idempotency_key] = ticket
if self.lost_responses: # the ticket exists, but the caller never hears back
self.lost_responses -= 1
raise TimeoutError("tracker did not answer within 5s")
return ticketThe stub models idempotent creates: when a create request carries a key it has seen, it returns the ticket it made the first time and creates nothing. An HTTP API can expose such a key through an Idempotency-Key header; check the tracker’s API contract. lost_responses makes it create the ticket and then raise TimeoutError, which is what this client sees when the response is lost.
2. Retry the write step with RetryPolicy
"""The review team files a tracker ticket for a failed CI job. The write step retries, and an idempotency key stops duplicates."""
import os
from typing import Literal
from langchain_core.runnables import RunnableConfig
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_ollama import ChatOllama
from langgraph.graph import END, START, StateGraph
from langgraph.types import RetryPolicy
from pydantic import BaseModel, Field
from typing_extensions import TypedDict
# The default retry_on retries ConnectionError but no other OSError, including TimeoutError, so name it.
MODEL = os.environ.get("REVIEW_MODEL", "gemini-3.8-flash") # REVIEW_MODEL=gemma4:e2b runs it on Ollama
RETRY = RetryPolicy(max_attempts=3, initial_interval=0.5, retry_on=(ConnectionError, TimeoutError))
class Ticket(BaseModel):
title: str = Field(description="Under 70 characters: what failed and where.")
labels: list[Literal["bug", "flaky-test", "infra", "dependencies"]]
class State(TypedDict, total=False):
log: str
ticket: dict
filed: dict
def stable_key(config: RunnableConfig) -> str:
"""Same thread, same step: the same key on every retry and every resume."""
return f"{config['configurable']['thread_id']}:file_ticket"
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(tracker, drafter=None, retry=RETRY, key=stable_key, checkpointer=None):
drafter = drafter or chat_model().with_structured_output(Ticket)
def draft(state: State) -> dict:
ticket = drafter.invoke(f"Write a tracker ticket for this failed CI job:\n{state['log']}")
return {"ticket": ticket.model_dump()}
def file_ticket(state: State, config: RunnableConfig) -> dict:
return {"filed": tracker.create(**state["ticket"], idempotency_key=key(config) if key else None)}
graph = StateGraph(State)
graph.add_node("draft", draft) # the model call is its own step, so a retry below never repeats it
graph.add_node("file_ticket", file_ticket, retry_policy=retry)
graph.add_edge(START, "draft")
graph.add_edge("draft", "file_ticket")
graph.add_edge("file_ticket", END)
return graph.compile(checkpointer=checkpointer)add_node("file_ticket", file_ticket, retry_policy=RETRY) makes LangGraph retry that node for an exception that retry_on accepts, up to max_attempts total attempts including the first call, waiting initial_interval seconds and then longer by backoff_factor each time, with jitter. The policy belongs to one node, which is why the model call is a separate node: this retry of file_ticket reruns that node without re-entering the preceding draft node. A node that called the model and then wrote would pay for the model again on every retry, and could write a different ticket each time.
retry_on names the exceptions explicitly. The default, default_retry_on, retries ConnectionError and HTTP 5xx errors from httpx and requests, and does not retry ValueError, TypeError or any other OSError. ConnectionError is itself an OSError; the default checks for it first. TimeoutError is one of the other OSError subclasses, so the default policy treats a timeout as a bug and fails at once.
3. Send a key that stays the same
A retry that runs is only safe if the service can recognize it, so now we give every attempt the same name. file_ticket takes a second parameter, config: RunnableConfig. LangGraph sees it in the signature and passes the run’s config, which holds the thread_id. stable_key builds the key from the thread and the step, so every retry of this step, and every resume of this thread, sends the same key. The key identifies the operation, not the payload: if a resumed run drafted a different title, the tracker would still return the first ticket, which is the behavior you want for “file one ticket for this job”.
4. Run each choice on Gemini
record.py runs four policies against a tracker that loses the first response, then runs a thread with no retries left, lets it fail, and resumes it.
python record.py== default policy
model calls 1, create requests 1, tickets 1
error: TimeoutError: tracker did not answer within 5s
== retry, no key
model calls 1, create requests 2, tickets 2
filed: {"id": "REV-102", "title": "test_default_timeout failed in tests/test_client.py", "labels": ["bug"]}
== retry, key per attempt
model calls 1, create requests 2, tickets 2
filed: {"id": "REV-102", "title": "AssertionError in tests/test_client.py::test_default_timeout", "labels": ["bug"]}
== retry, stable key
model calls 1, create requests 2, tickets 1
filed: {"id": "REV-101", "title": "test_default_timeout failed in tests/test_client.py", "labels": ["bug"], "replayed": true}
== fail, then resume with a stable key
model calls 1, create requests 2, tickets 1
filed: {"id": "REV-101", "title": "test_default_timeout failed in tests/test_client.py", "labels": ["bug"], "replayed": true}
first run failed with TimeoutError: tracker did not answer within 5sEvery run made 1 model call. In these recorded runs, retries did not repeat draft, and the resume loaded the drafted ticket from the checkpoint.
How it works
A retry and a resume repeat the same thing: the node, from its start. A retry happens inside one run, in the same process, right after the exception. A resume happens in a later invocation from the last checkpoint. This lab uses InMemorySaver and an in-memory Tracker, so the recorded failure and resume happen in one process. Continuing after a process restart would need both durable graph checkpoints and a tracker that durably retains ticket IDs and idempotency keys; these runs do not test that combination. Either way the tracker receives a second create request for a ticket that already exists. Nothing in LangGraph can know whether the first request took effect, because only the tracker knows. The key moves that decision to the tracker: two requests with one key are one operation.
The key must come from something that is the same on every repeat and different for every real operation. The thread ID plus the node name works here because each thread files one ticket. A key made inside the node, such as uuid4(), is new on every attempt, which makes every repeat look like a new operation. A workflow that files several tickets in one thread would add the item, such as the job number, to the key.
Troubleshooting
| Failure | What happens | Fix | Test |
|---|---|---|---|
| The default policy and a timeout | RetryPolicy() did not retry TimeoutError: the run failed after 1 request, yet the tracker held 1 ticket. |
Name TimeoutError in retry_on for writes you can retry safely. |
test_the_default_policy_does_not_retry_a_timeout |
| A retry without a key | The retry succeeded and returned REV-102; the tracker held 2 tickets for one job. |
Send an idempotency key with every write. | test_a_retry_without_a_key_files_a_duplicate |
| A key made per attempt | With uuid4() as the key, the tracker again held 2 tickets. |
Derive the key from the thread and step, not the attempt. | test_a_key_made_per_attempt_does_not_stop_the_duplicate |
| Retries run out | With no retries left, the run fails; with a stable key, resuming the thread replayed REV-101 and created nothing. |
Resume the thread instead of starting a new one, and keep the key stable across runs. | test_a_resumed_thread_reuses_the_key_and_skips_the_model |
Retries also need an end. max_attempts bounds them per run, and test_retries_stop_at_max_attempts checks that a tracker stub that times out on every request receives exactly that many requests. Without a bound, a retry policy turns an outage of the tracker into a flood of requests to it.
Test the retries without a model
test_workflow.py replaces Gemini with a fixed drafter and uses a retry interval of 0.01 seconds, so every scenario runs in milliseconds.
python -m pytest -v test_workflow.pytest_workflow.py::test_the_default_policy_does_not_retry_a_timeout PASSED [ 11%]
test_workflow.py::test_a_retry_without_a_key_files_a_duplicate PASSED [ 22%]
test_workflow.py::test_a_key_made_per_attempt_does_not_stop_the_duplicate PASSED [ 33%]
test_workflow.py::test_a_stable_key_turns_the_retry_into_a_replay PASSED [ 44%]
test_workflow.py::test_retries_stop_at_max_attempts PASSED [ 55%]
test_workflow.py::test_a_resumed_thread_reuses_the_key_and_skips_the_model PASSED [ 66%]
test_workflow.py::test_the_default_policy_retries_connection_error_but_no_other_os_error PASSED [ 77%]
test_workflow.py::test_the_lesson_policy_names_timeout_error PASSED [ 88%]
test_workflow.py::test_recorded_runs_match_the_lesson PASSED [100%]Running teardown with pytest sessionfinish...
============================== 9 passed in 3.01s ===============================All 9 tests pass.
Read the test and recording scripts
"""Offline tests: a fixed drafter replaces Gemini; the stub tracker loses responses on demand."""
import json
import uuid
from pathlib import Path
import pytest
from langchain_core.runnables import RunnableLambda
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import RetryPolicy, default_retry_on
from tracker import Tracker
from workflow import RETRY, Ticket, build_graph, stable_key
RUN = json.loads(Path(__file__).with_name("runs.json").read_text())
FAST = RetryPolicy(max_attempts=3, initial_interval=0.01, retry_on=(ConnectionError, TimeoutError))
CONFIG = {"configurable": {"thread_id": "ci-812"}}
def drafter(calls):
return RunnableLambda(lambda prompt: calls.append(prompt) or Ticket(title="test_default_timeout fails on #415", labels=["bug"]))
def file(tracker, retry=FAST, key=stable_key, calls=None):
return build_graph(tracker, drafter([] if calls is None else calls), retry, key).invoke({"log": "FAILED"}, CONFIG)
def test_the_default_policy_does_not_retry_a_timeout():
tracker = Tracker(lost_responses=1)
with pytest.raises(TimeoutError):
file(tracker, retry=RetryPolicy(initial_interval=0.01))
assert len(tracker.requests) == 1 and len(tracker.tickets) == 1 # it failed, yet the ticket exists
def test_a_retry_without_a_key_files_a_duplicate():
tracker = Tracker(lost_responses=1)
file(tracker, key=None)
assert [t["id"] for t in tracker.tickets] == ["REV-101", "REV-102"]
def test_a_key_made_per_attempt_does_not_stop_the_duplicate():
tracker = Tracker(lost_responses=1)
file(tracker, key=lambda config: str(uuid.uuid4()))
assert len(tracker.tickets) == 2
def test_a_stable_key_turns_the_retry_into_a_replay():
tracker, calls = Tracker(lost_responses=1), []
filed = file(tracker, calls=calls)["filed"]
assert filed == {"id": "REV-101", "title": "test_default_timeout fails on #415", "labels": ["bug"], "replayed": True}
assert len(tracker.tickets) == 1 and len(calls) == 1 # the model step did not run again
def test_retries_stop_at_max_attempts():
tracker = Tracker(lost_responses=5)
with pytest.raises(TimeoutError):
file(tracker, key=None)
assert len(tracker.requests) == FAST.max_attempts
def test_a_resumed_thread_reuses_the_key_and_skips_the_model():
tracker, calls = Tracker(lost_responses=1), []
graph = build_graph(tracker, drafter(calls), RetryPolicy(max_attempts=1), stable_key, InMemorySaver())
with pytest.raises(TimeoutError):
graph.invoke({"log": "FAILED"}, CONFIG)
assert graph.invoke(None, CONFIG)["filed"]["replayed"] is True
assert len(tracker.tickets) == 1 and len(calls) == 1
def test_the_default_policy_retries_connection_error_but_no_other_os_error():
assert issubclass(ConnectionError, OSError) and issubclass(TimeoutError, OSError)
assert default_retry_on(ConnectionError()) and default_retry_on(ConnectionResetError())
assert not default_retry_on(TimeoutError()) and not default_retry_on(FileNotFoundError())
assert not default_retry_on(ValueError()) and not default_retry_on(TypeError())
def test_the_lesson_policy_names_timeout_error():
assert RETRY.retry_on == (ConnectionError, TimeoutError)
def test_recorded_runs_match_the_lesson():
by = {r["scenario"]: r for r in RUN["runs"]}
assert "error" in by["default policy"] and len(by["default policy"]["tickets"]) == 1
assert len(by["retry, no key"]["tickets"]) == 2 and len(by["retry, key per attempt"]["tickets"]) == 2
assert len(by["retry, stable key"]["tickets"]) == 1 and by["retry, stable key"]["filed"]["replayed"]"""Run each retry and key choice against a tracker that loses one response, with Gemini drafting the ticket."""
import json
import uuid
from langchain_core.runnables import RunnableLambda
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import RetryPolicy
from tracker import Tracker
from workflow import MODEL, RETRY, Ticket, build_graph, chat_model, stable_key
LOG = """job unit-tests #812 on pull request 415 (Raise the default HTTP timeout to 30s)
FAILED tests/test_client.py::test_default_timeout - AssertionError: expected 10, got 30
1 failed, 214 passed in 38.2s"""
SCENARIOS = { # name: (retry policy, key function)
"default policy": (RetryPolicy(), None),
"retry, no key": (RETRY, None),
"retry, key per attempt": (RETRY, lambda config: str(uuid.uuid4())),
"retry, stable key": (RETRY, stable_key),
}
def main() -> None:
model = chat_model().with_structured_output(Ticket)
runs = []
for name, (retry, key) in SCENARIOS.items():
tracker, calls = Tracker(lost_responses=1), []
drafter = RunnableLambda(lambda prompt: calls.append(1) or model.invoke(prompt))
graph = build_graph(tracker, drafter, retry, key)
try:
result = graph.invoke({"log": LOG}, {"configurable": {"thread_id": "ci-812"}})
outcome = {"filed": result["filed"]}
except TimeoutError as error:
outcome = {"error": f"TimeoutError: {error}"}
runs.append({"scenario": name, **outcome, "create_requests": len(tracker.requests),
"tickets": tracker.tickets, "model_calls": len(calls)})
# No retries left: the run fails after the ticket was created. Resume the thread from its checkpoint.
tracker, calls = Tracker(lost_responses=1), []
drafter = RunnableLambda(lambda prompt: calls.append(1) or model.invoke(prompt))
graph = build_graph(tracker, drafter, RetryPolicy(max_attempts=1), stable_key, InMemorySaver())
config = {"configurable": {"thread_id": "ci-812"}}
try:
graph.invoke({"log": LOG}, config)
except TimeoutError as error:
first = f"TimeoutError: {error}"
result = graph.invoke(None, config)
runs.append({"scenario": "fail, then resume with a stable key", "error": first, "filed": result["filed"],
"create_requests": len(tracker.requests), "tickets": tracker.tickets, "model_calls": len(calls)})
with open("runs.json", "w") as f:
json.dump({"model": MODEL, "log": LOG, "runs": runs}, f, indent=1)
for r in runs:
print(f"== {r['scenario']}")
print(f" model calls {r['model_calls']}, create requests {r['create_requests']}, tickets {len(r['tickets'])}")
print(" " + (f"error: {r['error']}" if "filed" not in r else f"filed: {json.dumps(r['filed'])}"))
if "filed" in r and "error" in r:
print(f" first run failed with {r['error']}")
if __name__ == "__main__":
main(){
"scenario": "retry, no key",
"filed": {
"id": "REV-102",
"title": "test_default_timeout failed in tests/test_client.py",
"labels": [
"bug"
]
},
"create_requests": 2,
"tickets": [
{
"id": "REV-101",
"title": "test_default_timeout failed in tests/test_client.py",
"labels": [
"bug"
]
},
{
"id": "REV-102",
"title": "test_default_timeout failed in tests/test_client.py",
"labels": [
"bug"
]
}
],
"model_calls": 1
}Read the explanation
A retry repeats the node. The key must identify the operation, so it can't depend on the attempt.
What’s next
- Plan, execute and verify a multi-step task in LangGraph: check what a run changed, not what the agent says it changed.
References
Official sources checked October 21, 2026, against langgraph 1.2.16 and langchain-core 1.6.12.
- LangGraph: Fault tolerance:
RetryPolicyand its parameters. - LangGraph: Use the graph API: adding a retry policy to a node, and accessing the config in a node.
- LangGraph: Persistence: threads and resuming from a checkpoint.
- LangChain: Structured output:
with_structured_outputwith a Pydantic model. - The exceptions
default_retry_onaccepts are from the installedlanggraph/types.py.