trustcall

You can use trustcall with any tool-calling LLM from the LangChain ecosystem. The create_extractor function takes an LLM and a list of tools (which can be Pydantic models, regular Python functions, or JSON schemas) to create an extractor that validates and cleans tool calls.

When a tool call fails validation (e.g., a Pydantic @validator raises an error), trustcall automatically re-prompts the LLM to fix the output, often by generating a JSON Patch to correct the invalid arguments.

from langchain_fireworks import ChatFireworks
from pydantic.v1 import BaseModel, Field, validator
from trustcall import create_extractor

class Preferences(BaseModel):
    foods: List[str] = Field(description="Favorite foods")

    @validator("foods")
    def at_least_three_foods(cls, v):
        if len(v) < 3:
            raise ValueError("Must have at least three favorite foods")
        return v

llm = ChatFireworks(model="accounts/fireworks/models/firefunction-v2")

# Create the extractor with the LLM and the tool schema
extractor = create_extractor(llm, tools=[Preferences], tool_choice="Preferences")

# Invoke the extractor with a message
res = extractor.invoke({"messages": [("user", "I like apple pie and ice cream.")]})

# Access the validated tool calls and the parsed Pydantic objects
print(res["messages"][-1].tool_calls)
print(res["responses"])

The extraction process is driven by state objects that track the conversation history and attempt counts.

• ExtractionState: The base state containing:

  • messages: A list of AnyMessage objects. It uses a custom reducer _reduce_messages to handle MessageOp updates (like delete, update_tool_call, or update_tool_name).
  • attempts: An integer tracking how many times extraction has been attempted.
  • msg_id: A unique identifier for the message being patched.
  • existing: An optional dictionary representing existing data if you are performing an update rather than a fresh extraction.

• ExtendedExtractState: Inherits from ExtractionState and adds:

  • tool_call_id: The specific ID of the tool call that needs to be patched.
  • bump_attempt: A boolean flag used to signal if the current attempt should be incremented.

from dataclasses import dataclass
from typing import Annotated, List, Optional, Dict, Any
from langchain_core.messages import AnyMessage

@dataclass(kw_only=True)
class ExtractionState:
    messages: Annotated[List[AnyMessage], _reduce_messages]
    attempts: Annotated[int, operator.add]
    msg_id: Annotated[str, _keep_first]
    existing: Optional[Dict[str, Any]] = None

@dataclass(kw_only=True)
class ExtendedExtractState(ExtractionState):
    tool_call_id: str = ""
    bump_attempt: bool = False

Instead of re-generating an entire tool call from scratch when a validation error occurs, trustcall uses JsonPatch operations to modify the existing invalid call. This is more token-efficient and reliable.

The JsonPatch Schema

A patch operation requires:

• op: One of "add", "remove", or "replace".
• path: A JSON Pointer string referencing the location in the document.
• value: The data to be used (required for add and replace).

Patching Tools

• PatchFunctionErrors: Used when a tool call fails validation. It requires a json_doc_id, planned_edits (reasoning), and a list of patches.
• PatchFunctionName: Used when the LLM calls an unrecognized tool name. It requires a json_doc_id, reasoning, and the fixed_name.
• PatchDoc: Used to update existing structured data. It follows a specific planning order: replace first, then remove (from highest index to lowest to avoid shifting), then add (using /- for array appends).

# Example of a JsonPatch structure
{
    "op": "replace",
    "path": "/path/to/my_array/1",
    "value": "the newer value to be patched"
}

Because the extractor returned by create_extractor follows the LangChain Runnable interface and returns chat messages containing validated tool calls, it can be used directly as a node in a LangGraph state machine. This allows for conversational agents that can automatically recover from invalid tool-calling attempts before the tools are actually executed.

from langgraph.graph import START, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition
from trustcall import create_extractor

# ... define tools and LLM ...

# The extractor acts as the agent node
agent = create_extractor(llm, tools=[save_user_information, lookup_time])

class State(TypedDict):
    messages: Annotated[list, operator.add]

builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("tools", ToolNode([save_user_information, lookup_time]))
builder.add_edge("tools", "agent")
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)

graph = builder.compile(checkpointer=MemorySaver())