More detailed ValidationError's with ToolNode

[austinmw]

Hi, I'm using ToolNode, which wraps a tool that takes as input a Pydantic BaseModel.

Currently when my LLM fails to generate the correct schema, the ToolNode fails with an error like this:

--> [625](https://file+.vscode-resource.vscode-cdn.net/Users/username/Documents/workout-planner/notebooks/~/Documents/POCs/Echelon/workout-planner/.venv/lib/python3.11/site-packages/pydantic/main.py:625) return cls.__pydantic_validator__.validate_json(json_data, strict=strict, context=context)

ValidationError: 1 validation error for PopulatedPlan
  Invalid JSON: expected value at line 1 column 1 [type=json_invalid, input_value="Error: 3 validation err...ease fix your mistakes.", input_type=str]
    For further information visit https://errors.pydantic.dev/2.9/v/json_invalid

This error is sent back to the LLM to help provide information for a better retry attempt, but the error doesn't really provide enough information to help the LLM. How can I replace this with a concatenation of the actual underlying validation errors?

Also my second question is, since my tool is simply this:

workout_tool(generated_plan: BaseModel):
    return generated_plan.model_dump()

tools = [workout_tool]
tool_node = ToolNode(tools)

Should I even be using ToolNode, or is something else more fitting like ValidationNode?

[vbarda]

@austinmw could you try creating the tool manually and setting the handle_validation_error for a tool? https://python.langchain.com/api_reference/core/tools/langchain_core.tools.base.BaseTool.html#langchain_core.tools.base.BaseTool.handle_validation_error. this parameter allows you to control what is put into the ToolMessage.content if a ValidationError is thrown

we're also thinking of adding a more general handle_validation_error parameter for the ToolNode as a whole

[austinmw]

@vbarda I'm actually not using @tool or BaseTool in this case, are these necessary?

One thing that complicates my workflow a bit is that I want my Pydantic schema to be dynamically created based on information gathered from a user.

I'm currently doing it like this:

# Function to dynamically create a Pydantic BaseModel schema
def create_workout_plan_template(num_weeks: int = 2, max_workouts_per_day: int = 1) -> Type[BaseModel]:
    
    """Dynamically create a generic workout plan template based on num_weeks and max_workouts_per_day input by user."""

    # Nested pydantic classes omitted for brevity (e.g. `Week`)

    class WorkoutPlan(BaseModel):
        """Multi-week workout program."""

        explanation: str = Field(
            min_length=30,
            description="Highly intelligent and knowledgeable (but very polite) reasoning/rationale for the workout program design.",
        )
        weeks: list[Week] = Field(description=f"List of exactly {num_weeks} weeks to structure the workout program.")

        @model_validator(mode='after')
        def validate_weeks_count(self) -> 'WorkoutPlan':
            actual_weeks = len(self.weeks)
            if actual_weeks != num_weeks:
                raise ValueError(
                    f"Invalid number of workout weeks. Expected {num_weeks} weeks, but got {actual_weeks} weeks. "
                    "Try the tool call again!"
                )
            return self


    return WorkoutPlan
    
# Function to return a tools list from a Pydantic BaseModel definition
def create_workout_plan_tools(workout_plan_template):
    """Create a list of tools (just one in this case)."""
    
    def create_workout_plan(request: workout_plan_template) -> dict:
        """Return a generated workout program"""
        return request.model_dump()
    
    tools = (create_workout_plan,)
    return tools


# Dynamically create Pydantic model by passing num_weeks obtained from user
workout_plan_template = create_workout_plan_template(num_weeks)

# Create list of tools
tools = create_workout_plan_tools(workout_plan_template)

# Create a tool node for use in my graph
tool_node = ToolNode(tools)

# Conditional edge based on tool node output:
def should_retry_plan(state: PlannerState) -> Literal["Success", "Failure"]:
    """Decide whether to send error to agent or tool response to parser"""
    messages = state["messages"]
    last_message = messages[-1]
    if "error" in last_message.content.lower():
        return "Failure"
    return "Success"