Validation errors, and the retry
A schema can reject what the model sends. By default the agent returns the validation error to the model and lets it try again, and the model needs to use it.
A free-text status is still hard to count. The ticket system accepts three words, so the schema says so with Literal.
from typing import Literal
from pydantic import BaseModel
class Ticket(BaseModel):
"""A support ticket for one order."""
order_id: str
status: Literal["shipped", "waiting", "unknown"]The lookup for C40 says "waiting for stock", and lesson 10's model copies that phrase into the status.
The error, turned off
from langchain.agents import create_agent
from ticket import Ticket
from ticket_model import TicketModel
from tools import lookup_orderfrom langchain.agents.structured_output import ToolStrategy
strict = ToolStrategy(Ticket, handle_errors=False)
agent = create_agent(TicketModel(), tools=[lookup_order], response_format=strict)
agent.invoke({"messages": [{"role": "user", "content": "Where is C40?"}]})Wrapping the class in ToolStrategy gives access to its options. With handle_errors=False, the Pydantic error ends the run. It names the field, the value the model sent, and the three that were allowed.
Reading the error and trying again
handle_errors defaults to True: the error goes back to the model as a tool message, and the loop continues. A hosted model reads it and corrects its call. Yours needs one more method.
def fix(self, messages):
error = messages[-1].text
allowed = re.findall(r"'(\w+)'", error.split("Input should be")[1].split("[")[0])
found = next(m.text for m in messages if m.type == "tool" and m.name != "Ticket")
status = next((word for word in allowed if word in found), "unknown")
args = {"order_id": found.split(" ")[0], "status": status}
return AIMessage("", tool_calls=[{"name": "Ticket", "args": args, "id": "call_retry"}]) if messages[-1].type == "tool" and messages[-1].text.startswith("Error"):
return self.fix(messages)fix pulls the allowed words out of the error, picks the one that appears in the lookup result, and calls Ticket again. Add import re at the top of the file.
agent = create_agent(TicketModel(), tools=[lookup_order], response_format=Ticket)
result = agent.invoke({"messages": [{"role": "user", "content": "Where is C40?"}]})
print(repr(result["structured_response"]))
print(result["messages"][-3].text[:70])The ticket arrived with waiting. Three messages from the end is the error the agent sent back, which ends with "Please fix your mistakes." in full. The first attempt is still in the conversation, so you can see a retry happened.
- Ask about A17 and check which of the three words its status becomes.
- Print
result["messages"][-3].textin full and find the three allowed words in it. - Remove
"waiting"from theLiteraland see whatfixchooses.
Little by little, you're building something great.