When Agents Loop Forever: 4 Root Causes and How to Stop Them

def is_finished(response: str) -> bool:
    # Hard‑coded stop token – change per workflow
    return response.strip().endswith("END")

MAX_TOKENS_PER_REQUEST = 2048

def compute_step_budget(remaining_tokens: int) -> int:
    # Reserve at least 100 tokens for the final answer
    return max(50, (remaining_tokens - 100) // 3)

def call_tool(name: str, args: dict, depth: int = 0):
    if depth > 10:
        raise RecursionError("Tool recursion depth exceeded")
    # Tool dispatch table
    if name == "search_events":
        return search_events(args, depth + 1)
    # other tools …

import time

MAX_RECURSION_TIME = 2.0  # seconds

def call_tool(name: str, args: dict, start: float = None):
    if start is None:
        start = time.time()
    if time.time() - start > MAX_RECURSION_TIME:
        raise TimeoutError("Tool chain exceeded time budget")
    # dispatch …

llm = OpenAI(
    model="gpt-4",
    temperature=0.2,            # deterministic for orchestration
    max_tokens=512,
)

def orchestrate(prompt: str):
    response = llm.complete(prompt)
    if not schema.validate(response):
        # deterministic fallback
        response = template_fallback(prompt)
    return response

# Bad: creates new memory on each call
def handler(event, context):
    memory = AgentMemory()          # always new
    agent = MyAgent(memory=memory)
    return agent.run(event["prompt"])

import redis

r = redis.Redis(host="cache", port=6379)

def get_session_id(user_id: str) -> str:
    sid = r.get(f"session:{user_id}")
    if not sid:
        sid = uuid4().hex
        r.set(f"session:{user_id}", sid, ex=86400)  # 1‑day TTL
    return sid.decode()

import time
import threading
from functools import wraps
from prometheus_client import Histogram, Counter

# Prometheus metrics
LOOP_DURATION = Histogram(
    "agent_loop_seconds",
    "Time spent in an agent loop iteration",
    ["agent_name"]
)
LOOP_ITER = Counter(
    "agent_loop_iterations_total",
    "Number of loop iterations",
    ["agent_name", "outcome"]
)

def LoopGuard(
    max_steps: int = 5,
    token_budget: int = 2048,
    timeout_sec: float = 4.0,
    agent_name: str = "generic",
):
    """
    Decorator that adds safety guards around an `agent.run` method.
    """
    def decorator(run_fn):
        @wraps(run_fn)
        def wrapper(*args, **kwargs):
            start_time = time.time()
            steps = 0
            tokens_used = 0
            timed_out = False
            result = None

            # Watchdog thread – will set `timed_out` if over limit
            def watchdog():
                nonlocal timed_out
                time.sleep(timeout_sec)
                timed_out = True

            watch = threading.Thread(target=watchdog, daemon=True)
            watch.start()

            while steps < max_steps and not timed_out:
                # Assume the wrapped function returns a tuple (response, tokens)
                response, used = run_fn(*args, **kwargs)
                steps += 1
                tokens_used += used

                # Record per‑iteration metrics
                LOOP_DURATION.labels(agent_name).observe(time.time() - start_time)
                LOOP_ITER.labels(agent_name, "success").inc()

                # Stop‑token validation – configurable per workflow
                if isinstance(response, str) and response.strip().endswith("END"):
                    result = response
                    break

                # Token budget guard
                if tokens_used >= token_budget:
                    LOOP_ITER.labels(agent_name, "budget_exhausted").inc()
                    raise RuntimeError(
                        f"Token budget of {token_budget} exceeded after {steps} steps"
                    )

                # Prepare next iteration input (could be a refined prompt)
                kwargs["prompt"] = response  # simplistic example

            if timed_out:
                LOOP_ITER.labels(agent_name, "timeout").inc()
                raise TimeoutError(
                    f"Agent '{agent_name}' exceeded {timeout_sec}s timeout after {steps} steps"
                )

            if result is None:
                LOOP_ITER.labels(agent_name, "no_end_token").inc()
                raise RuntimeError(
                    f"Agent '{agent_name}' exited without stop token after {steps} steps"
                )

            return result

        return wrapper
    return decorator

# ----------------------------------------------------------------------
# Example usage with a LangChain‑style agent
# ----------------------------------------------------------------------
from langchain.llms import OpenAI
from langchain.agents import AgentExecutor, Tool

# Simple LLM with low temperature for deterministic orchestration
llm = OpenAI(model="gpt-4", temperature=0.2, max_tokens=512)

# Dummy tool just to illustrate recursion guarding
def dummy_tool(arg: str) -> str:
    return f"processed:{arg}"

tools = [Tool(name="dummy", func=dummy_tool, description="Echoes input")]

agent = AgentExecutor.from_agent_and_tools(
    agent=llm,
    tools=tools,
    verbose=False,
)

# Wrap the agent's `run` method
@LoopGuard(max_steps=5, token_budget=2048, timeout_sec=4.0, agent_name="order_fulfilment")
def guarded_run(prompt: str):
    # LangChain agents return a string; we approximate token usage
    response = agent.run(prompt)
    # Rough token count – replace with real tokeniser if available
    tokens = len(response.split())
    return response, tokens

# ----------------------------------------------------------------------
# Run the protected agent
# ----------------------------------------------------------------------
if __name__ == "__main__":
    try:
        answer = guarded_run("Process order #12345 and confirm shipping")
        print("✅ Finished:", answer)
    except Exception as exc:
        print("❌ Agent aborted:", exc)