DOC: Example of parsing a Pydantic object str returned by ToolNode #195

Closed
opened 2026-02-20 17:31:02 -05:00 by yindo · 3 comments
Owner

Originally created by @austinmw on GitHub (Aug 16, 2024).

Issue with current documentation:

I saw in the examples that ToolNode is meant to be used to handle errors and retrying for function calling. However I don't want to actually call a tool, I just want to return the pydantic object after successful validation. It seems that the object is sent back as a message string, and it's unclear to me how to parse it back into a Pydantic object before returning it.

Here's an example below. The parse_obj function is the one that's unclear to me:

image

from typing import Literal

from IPython.display import Image, display
from langchain.pydantic_v1 import BaseModel, conlist, Field
from langchain_aws import ChatBedrockConverse
from langgraph.graph import StateGraph, MessagesState
from langgraph.prebuilt import ToolNode

num_weeks = 4

class Day(BaseModel):
    day_of_week: str = Field(..., description="Day of the week")
    activities: list[str] = Field(..., description="List of 1-3 high-level workout descriptions or rest day activities")

class Week(BaseModel):
    monday: Day
    tuesday: Day
    wednesday: Day
    thursday: Day
    friday: Day
    saturday: Day
    sunday: Day

class WorkoutProgram(BaseModel):
    """Generate a workout program for an individual"""
    weeks: conlist(Week, min_items=num_weeks, max_items=num_weeks)


tool_node = ToolNode([WorkoutProgram], handle_tool_errors=True)

model = ChatBedrockConverse(
    model="anthropic.claude-3-haiku-20240307-v1:0",
    temperature=0,
    max_tokens=4096,
)
model_with_tools = model.bind_tools([WorkoutProgram], tool_choice="any")

class ProgramState(MessagesState):
    workout_program: WorkoutProgram | None = None

def should_retry(state: ProgramState) -> Literal["model", "parse_obj"]:
    """Decide whether to send error to agent or tool response to parser"""
    messages = state["messages"]
    last_message = messages[-1]
    # Side question: Any better way to check for error?
    if "Error: ValidationError" in last_message.content:
        return "model"
    return "parse_obj"

def parse_obj(state: ProgramState) -> Literal['__end__']:
    """Parse the object from the tool response"""
    messages = state["messages"]
    last_message = messages[-1]
    # ISSUE: This doesn't work — how to parse the string back to a BaseModel?
    workout_program = WorkoutProgram.parse_obj(dict(eval(last_message.content)))
    state["workout_program"] = workout_program
    return state

def call_model(state: ProgramState) -> ProgramState:
    """Call the model with the messages"""
    messages = state["messages"]
    response = model_with_tools.invoke(messages)
    state["messages"] = [response]
    return state

workflow = StateGraph(ProgramState)

# Define the two nodes
workflow.add_node("model", call_model)
workflow.add_node("tools", tool_node)
workflow.add_node("parse_obj", parse_obj)

workflow.add_edge("__start__", "model")
workflow.add_edge("model", "tools")
workflow.add_conditional_edges(
    "tools",
    should_retry # back to agent or to parse_obj
)
workflow.add_edge("parse_obj", "__end__")

app = workflow.compile()

display(Image(app.get_graph().draw_mermaid_png()))

response = app.invoke(
    {"messages": [("human", "Create a workout program for a beginner with the following characteristics: height: 5 feet 10 inches, weight: 180 lbs, age: 30, goals: Lose weight and improve overall fitness.")],
    },
    {"recursion_limit": 10},
)

workout_program = response["workout_program"]

last_message.content in parse_obj is a str that may look like, for example:
"weeks=[Week(monday=Day(day_of_week='Monday', activities=['30 mins cardio (e.g. brisk walking, jogging)', '20 mins strength training (e.g. bodyweight exercises)']), tuesday=Day(day_of_week='Tuesday', activities=['Rest day']), wednesday=Day(day_of_week='Wednesday', activities=['30 mins cardio (e.g. cycling, swimming)', '20 mins strength training (e.g. resistance bands)']), thursday=Day(day_of_week='Thursday', activities=['Rest day']), friday=Day(day_of_week='Friday', activities=['45 mins cardio (e.g. elliptical, rowing)', '15 mins core exercises']), saturday=Day(day_of_week='Saturday', activities=['Rest day']), sunday=Day(day_of_week='Sunday', activities=['60 mins outdoor activity (e.g. hiking, bike ride)'])), Week(monday=Day(day_of_week='Monday', activities=['40 mins cardio (e.g. jogging, swimming)', '25 mins strength training (e.g. bodyweight exercises, light weights)']), tuesday=Day(day_of_week='Tuesday', activities=['Rest day']), wednesday=Day(day_of_week='Wednesday', activities=['40 mins cardio (e.g. cycling, rowing)', '25 mins strength training (e.g. resistance bands, dumbbells)']), thursday=Day(day_of_week='Thursday', activities=['Rest day']), friday=Day(day_of_week='Friday', activities=['50 mins cardio (e.g. elliptical, stair climber)', '20 mins core exercises']), saturday=Day(day_of_week='Saturday', activities=['Rest day']), sunday=Day(day_of_week='Sunday', activities=['75 mins outdoor activity (e.g. hiking, bike ride)']))]"

Idea or request for content:

Can a simple example be added to the examples section for combining retry logic (including error messages) with Pydantic object outputs?

Originally created by @austinmw on GitHub (Aug 16, 2024). ### Issue with current documentation: I saw in the examples that `ToolNode` is meant to be used to handle errors and retrying for function calling. However I don't want to actually call a tool, I just want to return the pydantic object after successful validation. It seems that the object is sent back as a message string, and it's unclear to me how to parse it back into a Pydantic object before returning it. Here's an example below. The `parse_obj` function is the one that's unclear to me: ![image](https://github.com/user-attachments/assets/e5943d87-eafb-4eba-8842-bbf156c62fae) ```python from typing import Literal from IPython.display import Image, display from langchain.pydantic_v1 import BaseModel, conlist, Field from langchain_aws import ChatBedrockConverse from langgraph.graph import StateGraph, MessagesState from langgraph.prebuilt import ToolNode num_weeks = 4 class Day(BaseModel): day_of_week: str = Field(..., description="Day of the week") activities: list[str] = Field(..., description="List of 1-3 high-level workout descriptions or rest day activities") class Week(BaseModel): monday: Day tuesday: Day wednesday: Day thursday: Day friday: Day saturday: Day sunday: Day class WorkoutProgram(BaseModel): """Generate a workout program for an individual""" weeks: conlist(Week, min_items=num_weeks, max_items=num_weeks) tool_node = ToolNode([WorkoutProgram], handle_tool_errors=True) model = ChatBedrockConverse( model="anthropic.claude-3-haiku-20240307-v1:0", temperature=0, max_tokens=4096, ) model_with_tools = model.bind_tools([WorkoutProgram], tool_choice="any") class ProgramState(MessagesState): workout_program: WorkoutProgram | None = None def should_retry(state: ProgramState) -> Literal["model", "parse_obj"]: """Decide whether to send error to agent or tool response to parser""" messages = state["messages"] last_message = messages[-1] # Side question: Any better way to check for error? if "Error: ValidationError" in last_message.content: return "model" return "parse_obj" def parse_obj(state: ProgramState) -> Literal['__end__']: """Parse the object from the tool response""" messages = state["messages"] last_message = messages[-1] # ISSUE: This doesn't work — how to parse the string back to a BaseModel? workout_program = WorkoutProgram.parse_obj(dict(eval(last_message.content))) state["workout_program"] = workout_program return state def call_model(state: ProgramState) -> ProgramState: """Call the model with the messages""" messages = state["messages"] response = model_with_tools.invoke(messages) state["messages"] = [response] return state workflow = StateGraph(ProgramState) # Define the two nodes workflow.add_node("model", call_model) workflow.add_node("tools", tool_node) workflow.add_node("parse_obj", parse_obj) workflow.add_edge("__start__", "model") workflow.add_edge("model", "tools") workflow.add_conditional_edges( "tools", should_retry # back to agent or to parse_obj ) workflow.add_edge("parse_obj", "__end__") app = workflow.compile() display(Image(app.get_graph().draw_mermaid_png())) response = app.invoke( {"messages": [("human", "Create a workout program for a beginner with the following characteristics: height: 5 feet 10 inches, weight: 180 lbs, age: 30, goals: Lose weight and improve overall fitness.")], }, {"recursion_limit": 10}, ) workout_program = response["workout_program"] ``` `last_message.content` in `parse_obj` is a `str` that may look like, for example: `"weeks=[Week(monday=Day(day_of_week='Monday', activities=['30 mins cardio (e.g. brisk walking, jogging)', '20 mins strength training (e.g. bodyweight exercises)']), tuesday=Day(day_of_week='Tuesday', activities=['Rest day']), wednesday=Day(day_of_week='Wednesday', activities=['30 mins cardio (e.g. cycling, swimming)', '20 mins strength training (e.g. resistance bands)']), thursday=Day(day_of_week='Thursday', activities=['Rest day']), friday=Day(day_of_week='Friday', activities=['45 mins cardio (e.g. elliptical, rowing)', '15 mins core exercises']), saturday=Day(day_of_week='Saturday', activities=['Rest day']), sunday=Day(day_of_week='Sunday', activities=['60 mins outdoor activity (e.g. hiking, bike ride)'])), Week(monday=Day(day_of_week='Monday', activities=['40 mins cardio (e.g. jogging, swimming)', '25 mins strength training (e.g. bodyweight exercises, light weights)']), tuesday=Day(day_of_week='Tuesday', activities=['Rest day']), wednesday=Day(day_of_week='Wednesday', activities=['40 mins cardio (e.g. cycling, rowing)', '25 mins strength training (e.g. resistance bands, dumbbells)']), thursday=Day(day_of_week='Thursday', activities=['Rest day']), friday=Day(day_of_week='Friday', activities=['50 mins cardio (e.g. elliptical, stair climber)', '20 mins core exercises']), saturday=Day(day_of_week='Saturday', activities=['Rest day']), sunday=Day(day_of_week='Sunday', activities=['75 mins outdoor activity (e.g. hiking, bike ride)']))]"` ### Idea or request for content: Can a simple example be added to the examples section for combining retry logic (including error messages) with Pydantic object outputs?
yindo closed this issue 2026-02-20 17:31:02 -05:00
Author
Owner

@gbaian10 commented on GitHub (Aug 17, 2024):

Under normal circumstances, ToolNode expects to receive a BaseTool object and will package the execution result of the tool into a ToolMessage.
If the return value of the tool function is not a string or list, it will be forcibly converted and inserted into the content of the ToolMessage.
If you pass in a BaseModel, it will convert the BaseModel object into a string and insert it into the content of the ToolMessage, making it difficult to convert it back to a Python object (the issue you are currently encountering).

If you always bind only one tool and ensure that only one result will be called, I believe using with_structured_output is the simplest approach.
If you will bind multiple tools or may have multiple tool_calls, you can use bind_tool() and then connect to PydanticToolsParser, storing the execution results in the state instead of having ToolNode execute it for you.

Here is an example, and I replaced your model with ChatOpenAI(model="gpt-4o").

from typing import Annotated, TypedDict

from dotenv import load_dotenv
from langchain.pydantic_v1 import BaseModel, Field, conlist
from langchain_core.messages import AnyMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, add_messages
from rich import print  # can remove, It's just a pretty output.

load_dotenv()


class Day(BaseModel):
    day_of_week: str = Field(..., description="Day of the week")
    activities: list[str] = Field(
        ...,
        description="List of 1-3 high-level workout descriptions or rest day activities",
    )


class Week(BaseModel):
    monday: Day
    tuesday: Day
    wednesday: Day
    thursday: Day
    friday: Day
    saturday: Day
    sunday: Day


class WorkoutProgram(BaseModel):
    """Generate a workout program for an individual"""

    weeks: conlist(item_type=Week, min_items=4, max_items=4)


model = ChatOpenAI(model="gpt-4o-2024-08-06")
# model = ChatAnthropic(model="claude-3-haiku-20240307")
model_with_tools = model.with_structured_output(WorkoutProgram)


class ProgramState(TypedDict, total=False):
    messages: Annotated[list[AnyMessage], add_messages]
    workout_program: WorkoutProgram


def call_model(state: ProgramState) -> ProgramState:
    return {
        "workout_program": model_with_tools.invoke(state["messages"]),
    }


workflow = StateGraph(ProgramState)
workflow.set_entry_point("call_model")
workflow.add_node("call_model", call_model)
workflow.set_finish_point("call_model")
app = workflow.compile()

inputs: ProgramState = {
    "messages": [
        (
            "human",
            "Create a workout program for a beginner with the following characteristics: "
            "height: 5 feet 10 inches, weight: 180 lbs, age: 30, "
            "goals: Lose weight and improve overall fitness.",
        )
    ]
}
response: ProgramState = app.invoke(inputs, {"recursion_limit": 10})
print(response["workout_program"])

image


If you still insist on using ToolNode, I think converting it to BaseTool and using artifacts is also an option.
It will be able to store your Python object in ToolMessage, making it easier for subsequent nodes to get it.

By the way, your original class ProgramState(MessagesState), which is a TypedDict, setting its initial value = None has no effect.

@gbaian10 commented on GitHub (Aug 17, 2024): Under normal circumstances, `ToolNode` expects to receive a `BaseTool` object and will package the execution result of the tool into a `ToolMessage`. If the return value of the tool function is not a string or list, it will be forcibly converted and inserted into the content of the `ToolMessage`. If you pass in a `BaseModel`, it will convert the `BaseModel` object into a string and insert it into the `content` of the `ToolMessage`, making it difficult to convert it back to a Python object (the issue you are currently encountering). If you always bind `only one tool` and ensure that `only one result` will be called, I believe using `with_structured_output` is the simplest approach. If you will bind `multiple tools` or may have `multiple tool_calls`, you can use `bind_tool()` and then connect to `PydanticToolsParser`, storing the execution results in the state instead of having ToolNode execute it for you. Here is an example, and I replaced your model with `ChatOpenAI(model="gpt-4o")`. ```py from typing import Annotated, TypedDict from dotenv import load_dotenv from langchain.pydantic_v1 import BaseModel, Field, conlist from langchain_core.messages import AnyMessage from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, add_messages from rich import print # can remove, It's just a pretty output. load_dotenv() class Day(BaseModel): day_of_week: str = Field(..., description="Day of the week") activities: list[str] = Field( ..., description="List of 1-3 high-level workout descriptions or rest day activities", ) class Week(BaseModel): monday: Day tuesday: Day wednesday: Day thursday: Day friday: Day saturday: Day sunday: Day class WorkoutProgram(BaseModel): """Generate a workout program for an individual""" weeks: conlist(item_type=Week, min_items=4, max_items=4) model = ChatOpenAI(model="gpt-4o-2024-08-06") # model = ChatAnthropic(model="claude-3-haiku-20240307") model_with_tools = model.with_structured_output(WorkoutProgram) class ProgramState(TypedDict, total=False): messages: Annotated[list[AnyMessage], add_messages] workout_program: WorkoutProgram def call_model(state: ProgramState) -> ProgramState: return { "workout_program": model_with_tools.invoke(state["messages"]), } workflow = StateGraph(ProgramState) workflow.set_entry_point("call_model") workflow.add_node("call_model", call_model) workflow.set_finish_point("call_model") app = workflow.compile() inputs: ProgramState = { "messages": [ ( "human", "Create a workout program for a beginner with the following characteristics: " "height: 5 feet 10 inches, weight: 180 lbs, age: 30, " "goals: Lose weight and improve overall fitness.", ) ] } response: ProgramState = app.invoke(inputs, {"recursion_limit": 10}) print(response["workout_program"]) ``` ![image](https://github.com/user-attachments/assets/645a4239-99e8-4a78-a69d-60459862e787) --- If you still insist on using `ToolNode`, I think converting it to `BaseTool` and using [artifacts](https://python.langchain.com/v0.2/docs/how_to/tool_artifacts/) is also an option. It will be able to store your Python object in `ToolMessage`, making it easier for subsequent nodes to get it. By the way, your original `class ProgramState(MessagesState)`, which is a `TypedDict`, setting its initial value = None has no effect.
Author
Owner

@austinmw commented on GitHub (Aug 17, 2024):

@gbaian10 Thanks for your reply. The main functionality I wanted out of ToolNode was was Pydantic verification and error passing logic to add an intelligent retry loop in the case when an LLM returns a structured output with schema violations. I found that I can actually use ValidationNode instead of ToolNode:

import ast
from typing import Literal

from IPython.display import Image, display
from langchain.pydantic_v1 import BaseModel, Field, conlist
from langchain_aws import ChatBedrockConverse
from langgraph.graph import MessagesState, StateGraph
from langgraph.prebuilt import ValidationNode

num_weeks = 5

class Day(BaseModel):
    day_of_week: str = Field(..., description="Day of the week")
    activities: list[str] = Field(..., description="List of 1-3 high-level workout descriptions or rest day activities")

class Week(BaseModel):
    monday: Day
    tuesday: Day
    wednesday: Day
    thursday: Day
    friday: Day
    saturday: Day
    sunday: Day

class WorkoutProgram(BaseModel):
    """Generate a workout program for an individual"""
    weeks: conlist(Week, min_items=num_weeks, max_items=num_weeks)

validation_node = ValidationNode([WorkoutProgram])

model = ChatBedrockConverse(
    model="anthropic.claude-3-haiku-20240307-v1:0",
    temperature=0,
    max_tokens=4096,
)
model_with_tools = model.bind_tools([WorkoutProgram], tool_choice="any")

class ProgramState(MessagesState):
    workout_program: WorkoutProgram | None = None

def should_retry(state: ProgramState) -> Literal["model", "parse_obj"]:
    """Decide whether to send error to agent or tool response to parser"""
    messages = state["messages"]
    last_message = messages[-1]
    print(f"should_retry: {last_message.content}")

    # Side question: Any better way to check for error?
    if "ValidationError" in last_message.content:
        return "model"
    return "parse_obj"

def parse_obj(state: ProgramState) -> Literal['__end__']:
    """Parse the object from the tool response"""
    messages = state["messages"]
    last_message = messages[-1]
    workout_program = WorkoutProgram.parse_obj(ast.literal_eval(last_message.content))
    state["workout_program"] = workout_program
    return state

def call_model(state: ProgramState) -> ProgramState:
    """Call the model with the messages"""
    messages = state["messages"]
    last_message = messages[-1]
    print(f"call_model: {last_message.content}")
    response = model_with_tools.invoke(messages)
    state["messages"] = [response]
    return state

workflow = StateGraph(ProgramState)

# Define the two nodes
workflow.add_node("model", call_model)
workflow.add_node("validate", validation_node)
workflow.add_node("parse_obj", parse_obj)

workflow.add_edge("__start__", "model")
workflow.add_edge("model", "validate")
workflow.add_conditional_edges(
    "validate",
    should_retry # back to agent or to parse_obj
)
workflow.add_edge("parse_obj", "__end__")

app = workflow.compile()

display(Image(app.get_graph().draw_mermaid_png()))

response = app.invoke(
    {"messages": [("human", "Create a workout program for a beginner with the following characteristics: height: 5 feet 8 inches, weight: 250 lbs, age: 50, goals: Lose weight and improve overall fitness.")],
    },
    {"recursion_limit": 10},
)

workout_program = response["workout_program"]
@austinmw commented on GitHub (Aug 17, 2024): @gbaian10 Thanks for your reply. The main functionality I wanted out of ToolNode was was Pydantic verification and error passing logic to add an intelligent retry loop in the case when an LLM returns a structured output with schema violations. I found that I can actually use `ValidationNode` instead of ToolNode: ```python import ast from typing import Literal from IPython.display import Image, display from langchain.pydantic_v1 import BaseModel, Field, conlist from langchain_aws import ChatBedrockConverse from langgraph.graph import MessagesState, StateGraph from langgraph.prebuilt import ValidationNode num_weeks = 5 class Day(BaseModel): day_of_week: str = Field(..., description="Day of the week") activities: list[str] = Field(..., description="List of 1-3 high-level workout descriptions or rest day activities") class Week(BaseModel): monday: Day tuesday: Day wednesday: Day thursday: Day friday: Day saturday: Day sunday: Day class WorkoutProgram(BaseModel): """Generate a workout program for an individual""" weeks: conlist(Week, min_items=num_weeks, max_items=num_weeks) validation_node = ValidationNode([WorkoutProgram]) model = ChatBedrockConverse( model="anthropic.claude-3-haiku-20240307-v1:0", temperature=0, max_tokens=4096, ) model_with_tools = model.bind_tools([WorkoutProgram], tool_choice="any") class ProgramState(MessagesState): workout_program: WorkoutProgram | None = None def should_retry(state: ProgramState) -> Literal["model", "parse_obj"]: """Decide whether to send error to agent or tool response to parser""" messages = state["messages"] last_message = messages[-1] print(f"should_retry: {last_message.content}") # Side question: Any better way to check for error? if "ValidationError" in last_message.content: return "model" return "parse_obj" def parse_obj(state: ProgramState) -> Literal['__end__']: """Parse the object from the tool response""" messages = state["messages"] last_message = messages[-1] workout_program = WorkoutProgram.parse_obj(ast.literal_eval(last_message.content)) state["workout_program"] = workout_program return state def call_model(state: ProgramState) -> ProgramState: """Call the model with the messages""" messages = state["messages"] last_message = messages[-1] print(f"call_model: {last_message.content}") response = model_with_tools.invoke(messages) state["messages"] = [response] return state workflow = StateGraph(ProgramState) # Define the two nodes workflow.add_node("model", call_model) workflow.add_node("validate", validation_node) workflow.add_node("parse_obj", parse_obj) workflow.add_edge("__start__", "model") workflow.add_edge("model", "validate") workflow.add_conditional_edges( "validate", should_retry # back to agent or to parse_obj ) workflow.add_edge("parse_obj", "__end__") app = workflow.compile() display(Image(app.get_graph().draw_mermaid_png())) response = app.invoke( {"messages": [("human", "Create a workout program for a beginner with the following characteristics: height: 5 feet 8 inches, weight: 250 lbs, age: 50, goals: Lose weight and improve overall fitness.")], }, {"recursion_limit": 10}, ) workout_program = response["workout_program"] ```
Author
Owner

@gbaian10 commented on GitHub (Aug 17, 2024):

It seems you have found the answer that suits you.


Let me say something.

The part WorkoutProgram.parse_obj(ast.literal_eval(last_message.content)) can be changed to

json_str = last_message.content
WorkoutProgram.parse_raw(json_str)  # if in pydantic v2, use `.model_validate_json()`

ValidationNode convert a pydantic object into a JSON string and store it in content attr.
If you want to convert the JSON string back into a BaseModel object, doing so is simpler and more readable.

@gbaian10 commented on GitHub (Aug 17, 2024): It seems you have found the answer that suits you. --- ~~Let me say something.~~ The part `WorkoutProgram.parse_obj(ast.literal_eval(last_message.content))` can be changed to ```py json_str = last_message.content WorkoutProgram.parse_raw(json_str) # if in pydantic v2, use `.model_validate_json()` ``` `ValidationNode` convert a pydantic object into a JSON string and store it in `content` attr. If you want to convert the JSON string back into a `BaseModel` object, doing so is simpler and more readable.
Sign in to join this conversation.
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: langchain-ai/langgraph#195