Hooks around the model

POST_MODEL_CALL hooks see every reply before the agent does and can replace it. PRE_MODEL_CALL hooks see every call before it is made and can stop it.

Lesson 12's hook guarded a tool. The same @on works on each model call. Here the writer's model has a scripted reply that contains a card number, which must never reach a customer.

Examplewriter.py

from crewai import Agent, Crew, Task
from shop_llm import ShopLLM

writer = Agent(
    role="Reply writer",
    goal="Write replies to customers",
    backstory="You write short, clear emails.",
    llm=ShopLLM(model="shop", script=["We refunded card 4111 1111 1111 1234."]),
)
task = Task(description="Tell the customer where the refund went.",
            expected_output="One sentence.", agent=writer)
crew = Crew(agents=[writer], tasks=[task])

script from lesson 9 makes the model return that exact sentence.

Changing the reply

Exampleredact.py

import re

from crewai.hooks import InterceptionPoint, on


@on(InterceptionPoint.POST_MODEL_CALL)
def hide_cards(ctx):
    if isinstance(ctx.response, str):
        return re.sub(r"\b(\d{4}) ?\d{4} ?\d{4} ?(\d{4})\b", r"\1 **** **** \2", ctx.response)

A POST_MODEL_CALL hook runs after the model answers, with the reply in ctx.response. Returning a string replaces the reply; returning nothing keeps it. A tool call is a list, not a string, so the hook leaves it alone.

Example

print(crew.kickoff().raw)

The agent, the task and the crew only ever saw the masked number. The guardrail in lesson 16 does a similar job on a task's finished answer, with a retry instead of an edit.

Stopping a call

Examplelimit.py

from crewai.hooks import HookAborted, InterceptionPoint, clear_all_hooks, on


@on(InterceptionPoint.PRE_MODEL_CALL)
def at_most_two(ctx):
    print("model call, iteration", ctx.iterations)
    if ctx.iterations >= 2:
        raise HookAborted(reason="too many model calls")

PRE_MODEL_CALL runs before each call. ctx.iterations counts the rounds of the agent loop so far, starting at 0.

Exampleloop.py

from crewai import Agent, Crew, Task
from shop_llm import ShopLLM
from tools import lookup_order

ask = [{"id": "c1", "type": "function",
        "function": {"name": "lookup_order", "arguments": '{"order_id": "A17"}'}}]
clerk = Agent(role="Order clerk", goal="Find orders", backstory="You look orders up.",
              llm=ShopLLM(model="shop", script=[ask] * 5), tools=[lookup_order])
task = Task(description="Where is my order A17?", expected_output="One sentence.", agent=clerk)
crew = Crew(agents=[clerk], tasks=[task])

This clerk's model is scripted to ask for the same lookup five times in a row.

Example

try:
    crew.kickoff()
except Exception as error:
    print(type(error).__name__, "|", error)
finally:
    clear_all_hooks()

The hook allowed two calls and aborted the third, and HookAborted ended the whole run. clear_all_hooks() in finally removes both hooks, so nothing registered here affects a crew that runs later in the same program. Where a tool hook's abort became a message to the model, a model hook's abort stops the crew. max_iter from lesson 11 ends the loop with an answer; this ends it with an error your code can catch.

Try it yourself

Change the scripted reply to contain two card numbers.
Make at_most_two allow five calls and count the lookups.
Print ctx.agent.role in hide_cards.

PreviousRefunds need approval: tool hooks

Slow is fine. Stopping is the only problem.