tessl/pypi-langgraph-prebuilt

Library with high-level APIs for creating and executing LangGraph agents and tools.

Overview

Eval results

Files

Tool Validation

Name: tessl/pypi-langgraph-prebuilt
Author: tessl

Schema-based validation functionality for validating tool calls against Pydantic schemas, useful for structured output generation and data extraction workflows.

Capabilities

ValidationNode Class

A node that validates tool requests from the last AIMessage against predefined schemas. Useful for extraction and structured output use cases where you need to generate data that conforms to complex schemas without losing original messages and tool IDs.

class ValidationNode(RunnableCallable):
    def __init__(
        self,
        schemas: Sequence[Union[BaseTool, Type[BaseModel], Callable]],
        *,
        format_error: Optional[
            Callable[[BaseException, ToolCall, Type[BaseModel]], str]
        ] = None,
        name: str = "validation",
        tags: Optional[list[str]] = None,
    ) -> None

Parameters:

schemas: List of schemas to validate tool calls against (BaseModel classes, BaseTool instances, or functions)
format_error: Custom function for formatting validation error messages
name: Node name for graph identification
tags: Optional metadata tags

Returns: List of ToolMessages with validated content or error messages

Usage Examples:

from pydantic import BaseModel, field_validator
from langgraph.prebuilt import ValidationNode
from langchain_core.messages import AIMessage

# Define validation schemas
class PersonInfo(BaseModel):
    name: str
    age: int
    email: str

    @field_validator("age")
    def age_must_be_positive(cls, v):
        if v < 0:
            raise ValueError("Age must be positive")
        return v

    @field_validator("email")
    def email_must_contain_at(cls, v):
        if "@" not in v:
            raise ValueError("Invalid email format")
        return v

class ProductOrder(BaseModel):
    product_name: str
    quantity: int
    price: float

    @field_validator("quantity")
    def quantity_must_be_positive(cls, v):
        if v <= 0:
            raise ValueError("Quantity must be positive")
        return v

# Create validation node
validation_node = ValidationNode([PersonInfo, ProductOrder])

# Validate successful tool calls
tool_calls = [
    {
        "name": "PersonInfo",
        "args": {"name": "Alice", "age": 30, "email": "alice@example.com"},
        "id": "1",
        "type": "tool_call"
    }
]
ai_message = AIMessage(content="", tool_calls=tool_calls)
result = validation_node.invoke({"messages": [ai_message]})

# result will contain ToolMessage with validated JSON content

Custom Error Formatting

def custom_error_formatter(
    error: BaseException,
    call: ToolCall,
    schema: Type[BaseModel]
) -> str:
    """Custom error formatting for validation failures."""
    return f"Validation failed for {schema.__name__}: {str(error)}. Please provide valid data."

validation_node = ValidationNode(
    [PersonInfo],
    format_error=custom_error_formatter
)

Default Error Formatter

def _default_format_error(
    error: BaseException,
    call: ToolCall,
    schema: Union[Type[BaseModel], Type[BaseModelV1]],
) -> str:
    """Default error formatting function."""
    return f"{repr(error)}\n\nRespond after fixing all validation errors."

Schema Types

Pydantic BaseModel Schemas

from pydantic import BaseModel, Field

class UserProfile(BaseModel):
    username: str = Field(min_length=3, max_length=20)
    age: int = Field(ge=13, le=120)
    interests: list[str] = Field(max_items=10)

validation_node = ValidationNode([UserProfile])

BaseTool Schemas

from langchain_core.tools import BaseTool, create_schema_from_function

class CustomTool(BaseTool):
    name: str = "custom_tool"
    description: str = "A custom tool"
    args_schema: Type[BaseModel] = UserProfile

    def _run(self, **kwargs):
        return "Tool executed"

validation_node = ValidationNode([CustomTool()])

Function-Based Schemas

def process_order(product: str, quantity: int, customer_email: str) -> str:
    """Process a product order."""
    return f"Order processed: {quantity}x {product} for {customer_email}"

# Schema is automatically created from function signature
validation_node = ValidationNode([process_order])

Advanced Validation Patterns

Re-prompting on Validation Errors

from typing import Literal
from typing_extensions import TypedDict
from langchain_anthropic import ChatAnthropic
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages

class SelectNumber(BaseModel):
    number: int

    @field_validator("number")
    def number_must_be_meaningful(cls, v):
        if v != 37:
            raise ValueError("Only 37 is allowed")
        return v

# Create graph with validation and re-prompting
builder = StateGraph(Annotated[list, add_messages])
llm = ChatAnthropic(model="claude-3-7-haiku-latest").bind_tools([SelectNumber])
builder.add_node("model", llm)
builder.add_node("validation", ValidationNode([SelectNumber]))
builder.add_edge(START, "model")

def should_validate(state: list) -> Literal["validation", "__end__"]:
    """Route to validation if there are tool calls."""
    if state[-1].tool_calls:
        return "validation"
    return END

builder.add_conditional_edges("model", should_validate)

def should_reprompt(state: list) -> Literal["model", "__end__"]:
    """Check if validation errors occurred and reprompt if needed."""
    for msg in state[::-1]:
        if msg.type == "ai":
            return END
        if msg.additional_kwargs.get("is_error"):
            return "model"  # Re-prompt the model
    return END

builder.add_conditional_edges("validation", should_reprompt)

graph = builder.compile()
result = graph.invoke(("user", "Select a number, any number"))

Multi-Schema Validation

class ContactInfo(BaseModel):
    name: str
    phone: str
    email: str

class AddressInfo(BaseModel):
    street: str
    city: str
    zip_code: str

class PersonalInfo(BaseModel):
    age: int
    occupation: str

# Validate different types of information
validation_node = ValidationNode([ContactInfo, AddressInfo, PersonalInfo])

# Tool calls with different schemas
tool_calls = [
    {"name": "ContactInfo", "args": {"name": "John", "phone": "123-456-7890", "email": "john@example.com"}, "id": "1", "type": "tool_call"},
    {"name": "AddressInfo", "args": {"street": "123 Main St", "city": "Anytown", "zip_code": "12345"}, "id": "2", "type": "tool_call"}
]

Nested Schema Validation

from typing import Optional

class Address(BaseModel):
    street: str
    city: str
    state: str
    zip_code: str

class Person(BaseModel):
    name: str
    age: int
    address: Optional[Address] = None
    emergency_contact: Optional['Person'] = None  # Self-referential

class Company(BaseModel):
    name: str
    employees: list[Person]
    headquarters: Address

validation_node = ValidationNode([Person, Company])

Error Handling and Recovery

Validation Error Messages

When validation fails, ValidationNode returns ToolMessage objects with error details:

# Failed validation example
tool_call = {
    "name": "PersonInfo",
    "args": {"name": "Alice", "age": -5, "email": "invalid-email"},
    "id": "1",
    "type": "tool_call"
}

result = validation_node.invoke({"messages": [AIMessage(content="", tool_calls=[tool_call])]})

# Result contains ToolMessage with:
# - content: Error description
# - name: "PersonInfo"
# - tool_call_id: "1"
# - additional_kwargs: {"is_error": True}

Custom Error Recovery

def smart_error_handler(
    error: BaseException,
    call: ToolCall,
    schema: Type[BaseModel]
) -> str:
    """Provide helpful error messages with suggestions."""
    error_msg = str(error)
    suggestions = []

    if "age" in error_msg and "positive" in error_msg:
        suggestions.append("Age must be a positive number")
    if "email" in error_msg:
        suggestions.append("Email must contain @ symbol")
    if "required" in error_msg:
        suggestions.append("All required fields must be provided")

    base_msg = f"Validation failed for {schema.__name__}: {error_msg}"
    if suggestions:
        base_msg += f"\n\nSuggestions:\n" + "\n".join(f"- {s}" for s in suggestions)

    return base_msg

validation_node = ValidationNode([PersonInfo], format_error=smart_error_handler)

Integration with Workflows

Data Extraction Pipeline

from langgraph.graph import StateGraph

class ExtractionState(TypedDict):
    messages: list
    extracted_data: dict

def extract_model_call(state):
    """Call LLM to extract structured data."""
    # Implementation to call model with extraction prompt
    pass

def process_extracted_data(state):
    """Process validated extraction results."""
    messages = state["messages"]
    for msg in reversed(messages):
        if msg.type == "tool" and not msg.additional_kwargs.get("is_error"):
            # Parse validated JSON content
            extracted_data = json.loads(msg.content)
            return {"extracted_data": extracted_data}
    return {"extracted_data": {}}

# Build extraction workflow
graph = StateGraph(ExtractionState)
graph.add_node("extract", extract_model_call)
graph.add_node("validate", ValidationNode([PersonInfo, ContactInfo]))
graph.add_node("process", process_extracted_data)

graph.add_edge("extract", "validate")
graph.add_edge("validate", "process")

Form Validation Workflow

class FormData(BaseModel):
    first_name: str = Field(min_length=2)
    last_name: str = Field(min_length=2)
    email: str = Field(pattern=r'^[^@]+@[^@]+\.[^@]+$')
    age: int = Field(ge=18, le=100)
    terms_accepted: bool = Field(...)

    @field_validator("terms_accepted")
    def terms_must_be_accepted(cls, v):
        if not v:
            raise ValueError("Terms and conditions must be accepted")
        return v

def form_validation_workflow():
    """Create a form validation workflow."""
    validation_node = ValidationNode([FormData])

    def validate_form_submission(state):
        """Validate form data and provide feedback."""
        result = validation_node.invoke(state)

        # Check for validation errors
        for msg in result.get("messages", []):
            if msg.additional_kwargs.get("is_error"):
                return {"validation_errors": msg.content, "form_valid": False}

        return {"form_valid": True, "validated_data": json.loads(result["messages"][0].content)}

    return validate_form_submission

Best Practices

Schema Design

# Good: Clear field validation with helpful error messages
class WellDesignedSchema(BaseModel):
    username: str = Field(min_length=3, max_length=20, description="Username between 3-20 characters")
    email: str = Field(pattern=r'^[^@]+@[^@]+\.[^@]+$', description="Valid email address")
    age: int = Field(ge=0, le=150, description="Age between 0-150")

    @field_validator("username")
    def username_alphanumeric(cls, v):
        if not v.replace('_', '').isalnum():
            raise ValueError("Username must contain only letters, numbers, and underscores")
        return v

Error Message Design

# Good: Specific, actionable error messages
@field_validator("password")
def validate_password(cls, v):
    if len(v) < 8:
        raise ValueError("Password must be at least 8 characters long")
    if not any(c.isupper() for c in v):
        raise ValueError("Password must contain at least one uppercase letter")
    if not any(c.islower() for c in v):
        raise ValueError("Password must contain at least one lowercase letter")
    if not any(c.isdigit() for c in v):
        raise ValueError("Password must contain at least one digit")
    return v

Validation Node Usage

# Good: Group related schemas together
user_schemas = [UserProfile, ContactInfo, AddressInfo]
product_schemas = [ProductInfo, OrderInfo, PaymentInfo]

user_validation = ValidationNode(user_schemas, name="user_validation")
product_validation = ValidationNode(product_schemas, name="product_validation")

# Good: Use meaningful names and tags
validation_node = ValidationNode(
    [PersonInfo],
    name="person_info_validation",
    tags=["validation", "user_data", "pii"]
)

Install with Tessl CLI

npx tessl i tessl/pypi-langgraph-prebuilt

docs

agent-creation.md

human-in-the-loop.md

index.md

state-store-injection.md

tool-execution.md

tool-validation.md

tile.json