Checking an answer: guardrails

A guardrail is a function that checks a task's answer. If it fails, CrewAI sends the reason back to the agent and runs the task again, up to a limit.

Lesson 13 masked card numbers by editing the reply. A guardrail rejects the reply instead and asks the agent to write it again, the way an editor sends a draft back.

Examplecrew.py, from lesson 14

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

clerk = Agent(role="Order clerk", goal="Find the status of customers' orders",
              backstory="You can look up any order in the shop's system.",
              llm=ShopLLM(model="shop"), tools=[lookup_order])
writer = Agent(role="Reply writer", goal="Write replies to customers",
               backstory="You write short, friendly emails.",
               llm=ShopLLM(model="shop"))

look = Task(description="Find the order in this message: {question}",
            expected_output="The order's status.", agent=clerk)
reply = Task(description="Write the customer a reply.",
             expected_output="A short, friendly email.", agent=writer)
crew = Crew(agents=[clerk, writer], tasks=[look, reply])

Exampleguard.py

import re


def no_card_numbers(output):
    if re.search(r"\d{4} ?\d{4} ?\d{4} ?\d{4}", output.raw):
        return (False, "Remove the card number. Never repeat one to a customer.")
    return (True, output.raw)

A guardrail takes the task's output and returns a pair: (True, value) to accept, or (False, reason) to reject with a reason the agent will read.

Example

reply.guardrail = no_card_numbers
writer.llm.script = ["Refunded to card 4111 1111 1111 1234.", "Refunded to your card."]

print(crew.kickoff(inputs={"question": "Where is my order A17?"}).raw)

The card number went straight through. Task sets up its guardrails when it is created, so assigning one to an existing task does nothing in 1.15.22, and no error says so. It has to be passed to Task.

Passing it when the task is made

Examplecrew.py, the reply task

reply = Task(description="Write the customer a reply.",
             expected_output="A short, friendly email.", agent=writer,
             guardrail=no_card_numbers)
crew = Crew(agents=[clerk, writer], tasks=[look, reply])

Example

writer.llm.script = ["Refunded to card 4111 1111 1111 1234.", "Refunded to your card."]

print(crew.kickoff(inputs={"question": "Where is my order A17?"}).raw)

The first reply failed, the task ran again, and the second passed. The writer's model is scripted with both replies so you can see the retry. On the second attempt the task's prompt carried the failed reply and your reason under Previous attempt failed validation, which is what a hosted model reads to fix its answer.

When it never passes

Example

reply.guardrail_max_retries = 1
writer.llm.script = ["Card 4111 1111 1111 1234."] * 2

try:
    crew.kickoff(inputs={"question": "Where is my order A17?"})
except Exception as error:
    print(error)

guardrail_max_retries defaults to 3; this task allowed one retry. Once the retries are spent the task fails with the last reason, and nothing with a card number leaves the crew. A task can also take a list, guardrails=[...], run in order, and a guardrail can be a sentence the agent's model judges; a function gives the same answer every time.

Try it yourself

Return (True, output.raw.upper()) on success and print the result.
Add a second guardrail that rejects replies longer than 60 characters.
Set guardrail_max_retries to 0 with the first script.

PreviousAnswers with a fixed shape

Little by little, you're building something great.