Interrompre un agent qui se répète

turn 1: track_parcel → runs                      calls = 1
        track_parcel → runs                      calls = 2
        track_parcel → blocked, with a message   refusals = 1
        ... two more blocked calls ...           refusals = 3
        next model call → blocked: the turn ends with the hook's message
turn 2: track_parcel → runs: each turn starts a new count

import asyncio

from pydantic import Field

from mistralai import agents

# --- The carrier, faked -------------------------------------------------------------------------
# In production, track_parcel calls the carrier's API. Here the parcel stays in transit, and
# TRACKED records every call that reached the carrier.

TRACKED: list[str] = []

@agents.tool
def track_parcel(tracking_number: str) -> dict[str, str]:
    """The carrier's latest scan of a parcel."""
    print(f"carrier: tracked {tracking_number}")
    TRACKED.append(tracking_number)
    return {"tracking_number": tracking_number, "status": "in transit", "last_scan": "Lyon hub, 06:12"}

class StopRepeatedCalls(agents.Hook):
    """Refuse a tool once the turn has called it too often; end the turn if the model keeps trying."""

    max_calls: int = 2  # calls allowed to each tool in one turn
    max_refusals: int = 3  # refusals in one turn before the hook ends it
    calls: dict[str, int] = Field(default_factory=dict)  # this turn: tool id -> times called
    refusals: int = 0  # this turn: calls refused

    async def pre_agent_turn(self, ctx, request):
        self.calls = {}
        self.refusals = 0

    async def pre_tool_call(self, ctx, call):
        self.calls[call.tool_id] = self.calls.get(call.tool_id, 0) + 1
        if self.calls[call.tool_id] <= self.max_calls:
            return None
        self.refusals += 1
        print(f"loop breaker: refused {call.tool_id}, call {self.calls[call.tool_id]} of this turn")
        return agents.skip(
            f"You already called {call.tool_name} {self.max_calls} times in this turn, and its result will not "
            "change before the turn ends. Do not call it again: answer with what you know."
        )

    async def pre_llm_call(self, ctx, invocation):
        if self.refusals < self.max_refusals:
            return None
        print("loop breaker: ended the turn")
        return agents.skip("I stopped checking: the status did not change. Ask me again later.")

tracker = agents.Agent(
    model="mistral-large-4",
    name="parcel-tracker",
    instructions=(
        "You track parcels with track_parcel. Answer in at most two sentences. You cannot check again later: say so."
    ),
    tools=[track_parcel],
    hooks=[StopRepeatedCalls()],
)

def show(state: agents.TurnState[object]) -> None:
    if isinstance(state, agents.Completed):
        print(f"agent ({state.stop_reason or 'answered'}): {state.text}")
    else:
        print(f"agent: {type(state).__name__}")

async def main() -> None:
    async with agents.local_runtime():  # at exit: every local session of the event loop stops
        session = await tracker.start_session()
        print("--- turn 1")
        show(await session.run("Parcel FR-7731 is due today. Keep checking until it is delivered, then tell me."))
        print("--- turn 2")
        show(await session.run("Check FR-7731 once more."))
        await session.cancel()

    print(f"calls that reached the carrier: {len(TRACKED)}")

if __name__ == "__main__":
    asyncio.run(main())

--- turn 1
carrier: tracked FR-7731
carrier: tracked FR-7731
loop breaker: refused client.track_parcel, call 3 of this turn
agent (answered): I can't keep checking — the tracking result for this turn won't change and I have no way to check again later. As of now, parcel FR-7731 is **in transit**, with its last scan at the Lyon hub at 06:12; it has not yet been delivered.
--- turn 2
carrier: tracked FR-7731
agent (answered): Parcel FR-7731 is still **in transit** — last scan at the Lyon hub at 06:12, not yet delivered.
calls that reached the carrier: 3