LangChainLangChain 1.4 · Python 3.10+
0%
1
Curious builder0 XP earned · 300 to level 2
0 daysFinish a lesson to begin
Badge collection0 of 6 unlocked
43 small wins to finish your pathNext lesson

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.

Exampleticket.py
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

Exampleagent.py
from langchain.agents import create_agent
from ticket import Ticket
from ticket_model import TicketModel
from tools import lookup_order
Example
from 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.

Exampleticket_model.py, a new method of TicketModel
    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"}])
Exampleticket_model.py, the first lines of decide
        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.

Example
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.

Try it yourself
  • Ask about A17 and check which of the three words its status becomes.
  • Print result["messages"][-3].text in full and find the three allowed words in it.
  • Remove "waiting" from the Literal and see what fix chooses.

Little by little, you're building something great.