Detect Loops
Notice when an agent repeats the same actions without making progress, and interrupt it before the loop uses up the run’s request, token, or cost budget. Use it on agents that run autonomously for many steps.
While Pydantic AI Harness is on 0.x releases, the API may change between minor releases; when it does, deprecation warnings and release-note migration guidance tell you (or your agent) exactly how to upgrade. See the version policy.
pip install pydantic-ai-harness
uv add pydantic-ai-harness
from pydantic_ai import Agent
from pydantic_ai_harness import DetectLoops
agent = Agent('anthropic:claude-opus-5-5', capabilities=[DetectLoops()])
DetectLoops records each model response’s tool calls and text as it arrives, and each call’s result from the request that follows it. Before each model request, it checks what the run has done so far. When it finds a loop, it adds a system prompt to that request telling the model which loop it is in and asking it to change its approach. The capability adds no tools or instructions, so a run without a loop sends the same requests it would without it.
| Pattern | Reported when | Setting (default) |
|---|---|---|
repeated_call | The same tool call (same tool, same arguments) is made this many times within the last window tool calls, whatever it returned. | repeated_call_threshold (5) |
repeated_result | The same tool call returned the same result each of the last this-many times it was made within window. | repeated_result_threshold (3) |
alternating_calls | Two different tool calls are made in turn, A-B-A-B, for this many A-B cycles. | alternating_calls_threshold (3) |
repeated_text | This many consecutive responses have no tool calls and the same text, ignoring case and whitespace. This happens when output validation keeps asking the model to retry. | repeated_text_threshold (3) |
window (default 10) is how many of the most recent tool calls the three call-based checks look at. Set a threshold to None to turn that check off. A threshold must be at least 2, and a call-based threshold has to fit in window (alternating_calls_threshold needs twice its value), so a check that could never fire raises ValueError when the capability is created.
How calls and results are compared:
- Arguments are compared as canonical JSON with sorted keys, so
{"path": "a", "limit": 1}and{"limit": 1, "path": "a"}, or the same object sent as a JSON string, are the same call. - A result is the tool return’s outcome, its content as structured JSON with sorted keys, and the identifiers of any files it returned. A tool that raises
ModelRetrywith the same message each time returns the same result, so a call that keeps failing the same way is arepeated_resultloop. - Every call in a response counts, in order, including parallel calls in a response that already completes a loop.
- A result is matched to its call only within the request that follows the call’s response, so models that reuse tool call ids across responses are still compared correctly.
- Calls and results are recorded when they arrive.
DetectLoopsruns outermost among the agent’s capabilities, so it reads a call’s result before others, such as compaction, rewrite the request. Compaction that later drops old messages, or replaces old results and arguments with placeholders, doesn’t change what was recorded, so it can’t make different calls look alike. - Only function tool calls count. Calls the provider runs itself (native tools) are not compared.
At most one loop is reported per model request, checked in the order repeated_result, alternating_calls, repeated_call, repeated_text. The responses that formed it are then not counted again, so the same loop has to build up from scratch before it is reported a second time.
Only the current run’s responses are recorded. A run that continues an earlier conversation through message_history starts with nothing counted.
A tool polled until a job finishes, with arguments that don’t change, is a repeated_call loop after 5 polls within 10 calls, and a repeated_result loop after 3 polls in a row that return the same status. If your agent polls on purpose, raise those thresholds or turn the checks off, or have the tool return something that changes between polls, such as the elapsed time.
action decides what happens when a loop is detected.
The capability enqueues a SystemPromptPart with RunContext.enqueue, which joins the request being prepared. It keeps the default conversation scope, so like a course correction it keeps applying on later requests instead of being shown once, and later requests send it unchanged. For example:
Your last 3 calls to `read_file` used the same arguments and returned the same result, so calling it again will not change the outcome. Change your approach, or explain what is blocking you.
The text names tools using only letters, digits, and _; any other character becomes _. The real names, and the pattern, are in the part’s metadata:
{'origin': 'pydantic-ai-harness.detect_loops.nudge.v1', 'pattern': 'repeated_result', 'tool_names': ['read_file']}
Find nudges in history by that origin, not by their text (identifying system prompts). On a model without inline system prompts, Pydantic AI sends the part as <system> user text, as it does any mid-conversation system prompt.
A nudge doesn’t stop a model that ignores it. Combine it with UsageLimits for a hard stop.
The capability raises LoopDetectedError, a subclass of AgentRunError, which ends the run. The detected loop is on its loop attribute:
from pydantic_ai import Agent
from pydantic_ai.messages import ModelMessage, ModelResponse, ToolCallPart
from pydantic_ai.models.function import AgentInfo, FunctionModel
from pydantic_ai_harness import DetectLoops
from pydantic_ai_harness.detect_loops import LoopDetectedError
def stuck_model(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
return ModelResponse(parts=[ToolCallPart('read_file', {'path': 'missing.txt'})])
agent = Agent(FunctionModel(stuck_model), capabilities=[DetectLoops(action='error')])
@agent.tool_plain
def read_file(path: str) -> str:
return 'Error: file not found'
try:
agent.run_sync('Summarize missing.txt')
except LoopDetectedError as error:
print(error.loop.pattern, error.loop.tool_names, error.loop.count)
#> repeated_result ('read_file',) 3
Pass a sync or async function to decide yourself. It’s called with the run context and a DetectedLoop: raise from it to stop the run, or call ctx.enqueue(...) to steer the model your own way.
import logging
from pydantic_ai import Agent, RunContext
from pydantic_ai.messages import SystemPromptPart
from pydantic_ai_harness import DetectLoops
from pydantic_ai_harness.detect_loops import DetectedLoop
logger = logging.getLogger(__name__)
async def on_loop(ctx: RunContext[object], loop: DetectedLoop) -> None:
logger.warning('Agent loop: %s %s', loop.pattern, loop.tool_names)
ctx.enqueue(SystemPromptPart('You are repeating yourself. Summarize what you know and stop.'))
agent = Agent('anthropic:claude-opus-5-5', capabilities=[DetectLoops(action=on_loop)])
A DetectedLoop has:
pattern: which of the four patterns it is.tool_names: the tools involved, one or two (none forrepeated_text).count: how many calls, cycles, or responses repeated.window: how many recent tool calls (responses, forrepeated_text) the check looked at.fingerprints: short stable hashes of the repeated calls (tool name and canonical arguments) or text. The same call has the same fingerprint in every run. They don’t contain the arguments in the clear, but a short argument, such as a user id, can be recovered by hashing candidate values.message: the text the nudge would send.
What a run recorded lives in memory, in the instance for_run creates for that run. Under durable execution the recording and detection hooks (after_model_request and before_model_request) run in workflow code, so when Temporal, DBOS, or Prefect replays a run, they run again in the same order over the journaled responses and reach the same decisions. A handler runs there too, so keep it deterministic and free of I/O: log through the engine’s workflow logger, or move I/O into a durable capability operation. LoopDetectedError is an AgentRunError, so under Temporal it fails the workflow rather than the workflow task.
Each detection emits a zero-duration loop detected span through ctx.tracer, so it shows in the run’s trace next to the model request it changed. It’s a no-op unless the agent is instrumented.
| Attribute | Value |
|---|---|
detect_loops.pattern | The pattern, such as repeated_result |
detect_loops.action | nudge, error, or handler |
detect_loops.count | DetectedLoop.count |
detect_loops.window | DetectedLoop.window |
detect_loops.tool_names | The tools involved |
detect_loops.fingerprints | The hashes of the repeated calls or text, only when trace_include_content is enabled |
The span contains no arguments, results, or response text. The fingerprints are left out unless trace_include_content is enabled, because a short argument can be recovered from its hash. Checks that find nothing emit nothing.
- Several
DetectLoopson one agent run independently. Each reports the loops its own settings describe, so you can, for example, nudge at the defaults and add a second one with higher thresholds andaction='error'to stop a run that ignores the nudge. - System Reminders re-inject guidance on a cadence or condition, and Trajectory Judge asks a second model whether the run is on track.
DetectLoopsneeds no second model and only acts on exact repetition.
DetectLoops does not detect a run that keeps exceeding the model’s context window, and it doesn’t truncate, restart, or roll back the run: compaction and your own handler cover those.
See the source.
Bases: AbstractCapability[AgentDepsT]
Detect when an agent repeats the same actions without making progress, and interrupt it.
Each model response’s tool calls and text are recorded as it arrives, and each call’s result from the
request that follows it. Before each model request, the run so far is checked for four patterns: the same tool call
made repeated_call_threshold times within the last window tool calls, the same tool call
returning the same result repeated_result_threshold times in a row, two tool calls alternating
for alternating_calls_threshold cycles, and repeated_text_threshold consecutive responses
without tool calls whose text is the same. Set a threshold to None to turn that check off.
On detection, action runs: 'nudge' (the default) adds a system prompt to the request asking the
model to change its approach, 'error' raises LoopDetectedError, and a LoopHandler is called
with the DetectedLoop. The calls that formed the loop are then not counted again, so the same loop
has to build up again before it is reported a second time.
from pydantic_ai import Agent
from pydantic_ai_harness import DetectLoops
agent = Agent('anthropic:claude-opus-5-5', capabilities=[DetectLoops()])
Only the current run’s responses are recorded, so a run that continues an earlier conversation starts with nothing counted. Compaction that later drops or rewrites messages doesn’t change what was recorded.
How many of the most recent tool calls the call-based checks look at.
Type: int Default: 10
Report a tool call (same tool, same arguments) made this many times within window.
Report a tool call that returned the same result each of the last this-many times it was made
within window, with no different result in between.
Report two different tool calls made in turn (A-B-A-B) for this many A-B cycles.
Report this many consecutive responses without tool calls whose text is the same, ignoring case and whitespace.
What to do when a loop is detected.
'nudge': add a system prompt to the current request telling the model which loop it is in and asking it to change its approach.'error': raiseLoopDetectedError, ending the run.- a
LoopHandler: call it with the run context and theDetectedLoop.
Type: Literal[‘nudge’, ‘error’] | LoopHandler[AgentDepsT] Default: 'nudge'
def get_ordering() -> CapabilityOrdering
Run outermost, so the newest results are read before other capabilities, such as compaction, rewrite them.
CapabilityOrdering
@async
def for_run(ctx: RunContext[AgentDepsT]) -> DetectLoops[AgentDepsT]
Return a fresh instance, so runs don’t share what they recorded.
DetectLoops[AgentDepsT]
@async
def after_model_request(
ctx: RunContext[AgentDepsT],
*,
request_context: ModelRequestContext,
response: ModelResponse,
) -> ModelResponse
Record the response’s tool calls, or its text when it has none.
@async
def before_model_request(
ctx: RunContext[AgentDepsT],
request_context: ModelRequestContext,
) -> ModelRequestContext
Record the results of the previous response’s tool calls, then act on the first loop found.
A loop DetectLoops found in the run.
Passed to a LoopHandler, and carried by LoopDetectedError as loop.
Which kind of loop it is.
Type: LoopPattern
The tools involved: one for 'repeated_call' and 'repeated_result', two for
'alternating_calls', none for 'repeated_text'.
How many times the pattern repeated: calls for 'repeated_call' and 'repeated_result',
A-B cycles for 'alternating_calls', responses for 'repeated_text'.
Type: int
How many recent tool calls (responses, for 'repeated_text') the detection looked at.
Type: int
Stable hashes of what repeated.
For the call patterns, each covers a tool name and its canonical arguments: one for 'repeated_call'
and 'repeated_result', two for 'alternating_calls'. For 'repeated_text', it’s the hash of the
normalized text. They don’t contain argument values or text in the clear, but a short argument such
as a user id can be recovered by hashing candidate values.
The explanation sent to the model as the nudge, and used as the LoopDetectedError message.
Type: str
Bases: AgentRunError
Raised by DetectLoops(action='error') when it detects a loop.
Subclasses AgentRunError, so code that already stops on agent run errors stops on this one too.
The loop that was detected.
Type: DetectedLoop Default: loop
A sync or async function DetectLoops calls with each loop it detects.
Raise from it to stop the run, or call ctx.enqueue(...) to steer the model your own way.
Type: TypeAlias Default: Callable[[RunContext[AgentDepsT], DetectedLoop], Awaitable[None] | None]
Which kind of loop was detected.
'repeated_call': one tool call (same tool, same arguments) made many times among the recent calls.'repeated_result': one tool call that returned the same result each of the last several times it was made.'alternating_calls': two different tool calls made in turn, A-B-A-B.'repeated_text': consecutive responses without tool calls whose text is the same.
Type: TypeAlias Default: Literal['repeated_call', 'repeated_result', 'alternating_calls', 'repeated_text']