Skip to content

Handle unreliable models

Some models, often small or local ones, are less steady than others. A stream breaks halfway through a reply. The model stops without calling a tool, in the middle of the task. It calls the same tool with the same arguments again and again. Each recipe below handles one of these in your own code.

The model Recipe
Its stream breaks after text started arriving Retry a stream that dropped mid-reply
Replies with no tool call before the task is done Hand back a reply with no tool call
Repeats the same tool calls Stop when the model repeats itself

Retry a stream that dropped mid-reply

A Model that wraps another Model can retry the reply:

from alpineagents import Agent, Anthropic, AuthError, ContextTooLongError, Model, ModelEvent, ProviderError


class _Watch:
    """Passes text chunks on and remembers whether any arrived."""

    def __init__(self, on_text):
        self.on_text = on_text
        self.started = False

    def __call__(self, chunk):
        self.started = True
        if self.on_text is not None:
            self.on_text(chunk)


class RetryDroppedStream(Model):
    """Wraps a Model. Starts a reply over when its stream breaks after text arrived."""

    def __init__(self, model: Model, attempts: int = 3):
        self.model = model
        self.attempts = attempts
        self.name = model.name
        self.provider = model.provider
        self.supports = model.supports
        self.price = model.price
        self.max_tokens = model.max_tokens

    @property
    def context_window(self):
        return self.model.context_window

    def count_tokens(self, request):
        return self.model.count_tokens(request)

    def mark_cache(self, request):
        return self.model.mark_cache(request)

    def compact(self, request, instructions=None, on_event=None):
        return self.model.compact(request, instructions, on_event)

    async def acompact(self, request, instructions=None, on_event=None):
        return await self.model.acompact(request, instructions, on_event)

    def respond(self, request, on_text=None, on_event=None):
        for attempt in range(1, self.attempts + 1):
            watch = _Watch(on_text)
            try:
                return self.model.respond(request, watch, on_event)
            except ProviderError as error:
                if not self._retry(error, watch, attempt, on_event):
                    raise

    async def arespond(self, request, on_text=None, on_event=None):
        for attempt in range(1, self.attempts + 1):
            watch = _Watch(on_text)
            try:
                return await self.model.arespond(request, watch, on_event)
            except ProviderError as error:
                if not self._retry(error, watch, attempt, on_event):
                    raise

    def _retry(self, error, watch, attempt, on_event) -> bool:
        if isinstance(error, (AuthError, ContextTooLongError)):
            return False  # retrying cannot fix these
        if not watch.started or attempt == self.attempts:
            return False  # before the stream, the SDK already retried
        if on_event is not None:
            on_event(ModelEvent("retry", f"stream dropped ({error}), starting the reply over", {"attempt": attempt}))
        return True


agent = Agent(model=RetryDroppedStream(Anthropic("claude-sonnet-5")))
  1. RetryDroppedStream passes every request to the wrapped Model, and copies its name, prices and settings.
  2. _Watch wraps on_text, so it knows whether any text arrived before the failure.
  3. A ProviderError after the stream started is retried, up to attempts in total. The request is sent again as it was, so the model starts the reply over. The partial text is not kept.
  4. AuthError and ContextTooLongError are raised as they are. Sending the same request again cannot fix them.
  5. Before each retry, on_event(ModelEvent("retry", ...)) tells the Reporter, and the Agent records it in history as model_event.
  6. arespond does the same for athink and arun.

The Reporter has already shown the partial text. A Reporter that keeps the text, for a UI that redraws it, throws it away on the event:

from alpineagents import Agent, Reporter


class LiveText(Reporter):
    """Keeps the text of the reply being written, for a UI that redraws it."""

    def __init__(self):
        self.text = ""

    def on_think_start(self, state):
        self.text = ""

    def on_text(self, state, chunk):
        self.text += chunk

    def on_model_event(self, state, event):
        if event.kind == "retry":
            self.text = ""  # the reply starts over


agent = Agent(model="claude-sonnet-5", reporter=LiveText())

Terminal cannot take back printed text. It prints the event on its own line, model: stream dropped (...), starting the reply over, and the new reply below it.

The tokens of a dropped attempt are not in state.usage, because no Reply came back for it.

Why the framework does not retry

The provider SDK retries a failed request (retries=2 by default), but only before the stream starts: a connection error, a 429, a 500. Once text is streaming, the SDK raises, and the Model raises ProviderError. Retrying then has a cost: the text shown so far is thrown away, the tokens are paid again, and the new reply can differ from the first. Whether that is worth it, and what the person sees, is up to your app.

Hand back a reply with no tool call

Some models stop to say what they will do next ("I will now read the file") instead of doing it. That reply has no tool call, so waiting_for_user is true and the loop stops (Write your own loop). This loop hands it back:

from alpineagents import Agent, Message, State, loop, tool, waiting_for_user

NUDGES = 2
NUDGE = (
    "Your last reply called no tool. If the task is not done, continue it with a tool call. "
    "If it is done, give your final answer again."
)


@tool
def look() -> str:
    """Look around the project"""
    return "a README and a setup.py"


@loop(until=waiting_for_user, limit=40)
def nudging(agent: Agent, state: State):
    agent.think(state)
    if state.pending_calls:
        with state.edit_extra_data() as data:
            data["nudges"] = 0
        agent.use_tools(state)
        return
    nudges = state.extra_data.get("nudges", 0)
    if nudges < NUDGES:
        with state.edit_extra_data() as data:
            data["nudges"] = nudges + 1
        state.add_message(Message.notice(NUDGE))


agent = Agent(model="ollama/qwen3:8b", tools=[look], loop=nudging)
print(agent.run("Look around and tell me what this project is"))
  1. When the reply calls tools, the counter in state.extra_data goes back to 0.
  2. When it calls none, state.add_message(Message.notice(NUDGE)) asks the model to go on, up to NUDGES times in a row.
  3. The notice comes after the reply, so waiting_for_user is false and the loop runs another turn.
  4. After NUDGES notices, nothing is added. waiting_for_user is true and the loop stops.

The notice is in history as kind notice, and the model sees it as a user message that starts with [notice].

A finished task costs up to NUDGES more replies, because the model gives its answer again after each notice. Keep NUDGES low.

Stop when the model repeats itself

An until function can look at the last replies and stop the loop when the model is stuck:

from alpineagents import Agent, State, loop, tool, waiting_for_user

REPEATS = 3


@tool
def look(path: str) -> str:
    """Look at a path"""
    return "nothing here"


def repeating(state: State) -> bool:
    replies = [entry.content for entry in state.history if entry.kind == "model_reply"][-REPEATS:]
    if len(replies) < REPEATS:
        return False
    calls = [[(call.name, call.args) for call in reply.tool_calls] for reply in replies]
    return bool(calls[0]) and all(same == calls[0] for same in calls)


@loop(until=[waiting_for_user, repeating], limit=40)
def watched(agent: Agent, state: State):
    agent.think(state)
    if state.pending_calls:
        agent.use_tools(state)


agent = Agent(model="ollama/qwen3:8b", tools=[look], loop=watched)
print(agent.run("Find the config file"))
  1. repeating takes the tool calls of the last REPEATS model_reply entries of state.history.
  2. It compares names and arguments, not call ids, which are new in every reply.
  3. When all of them are the same, the loop stops before the next turn with state.stopped == StoppedByUntil("repeating").

To give the model a chance first, add a notice such as "You made the same calls 3 times. Try something else." from the loop, as in the recipe above, and stop only if it repeats again.