Pydantic not converting to tool for agent chain

Question

I have adapted this notebook from various tutorials:

https://colab.research.google.com/drive/1tAa5QUWplKRbYsUROZ1tmPiEZ4kY9sio?usp=sharing

I want to make this model into a tool and call the tool to parse human input as a node in langgraph:

class PlanItem(BaseModel):
    step: str
    tools: Optional[str] = []
    data_sources: Optional[str] = []
    sub_steps_needed: str

class Plan(BaseModel):
    plan: List[PlanItem] 

This function should create the agent for the node:

def create_agent(llm: ChatOpenAI, tools: list, system_prompt: str):
    # Each worker node will be given a name and some tools.
    prompt = ChatPromptTemplate.from_messages(
        [
            (
                "system",
                system_prompt,
            ),
            MessagesPlaceholder(variable_name="messages"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ]
    )
    agent = create_openai_tools_agent(llm, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools)
    return executor

but it doesn't recognise Plan as a BaseModel and I cannot convert it to a function I can pass as a tool.

Using pydantic.v1 I tried several approaches using PydanticOutputParser, PydanticOutputFunctionsParser and PydanticToolsParser. I also tried using convert_pydantic_to_openai_function:

output_parser = PydanticOutputParser(pydantic_object=Plan)

output_parser = PydanticOutputFunctionsParser(pydantic_schema=Plan)

output_parser = PydanticToolsParser(tools=[Plan])

output_parser = convert_pydantic_to_openai_function(Plan)

output_parser['description'] = "Parses data into Plan"

output_parser['title'] = "Plan Parser"

I seem to get errors which are variations on:

ValueError: Unsupported function

pydantic_schema=<class '__main__.Plan'>

Functions must be passed in as Dict, pydantic.BaseModel, or Callable. If they're a dict they must either be in OpenAI function format or valid JSON schema with top-level 'title' and 'description' keys.

There are tutorials on generally adding the parser as a tool in langchain, but I want the parser to only be invoked at a particular node.

Can anybody show me how to do this?

Answer 1

I personally had never used BaseTool, create_openai_tools_agent and AgentExecutor before, so this is some good learning for myself too.

Why your code not working:

AgentExecutor is the problem. If you look at the typing for tools, it is Sequence[BaseTool]. In Langchain's example, the tool used is

from langchain_community.tools.tavily_search import TavilySearchResults
tools = [TavilySearchResults(max_results=1)]

Try this and you'll see the class is derived from BaseTool, not BaseModel:

print(type(tools[0]).__bases__[0].__name__)

Your tool Plan is a child class of BaseModel, so it doesn't work.

Using BaseTool:

Now let's take a look at your tools:

from langchain_core.pydantic_v1 import BaseModel, Field
from typing import List, Optional, Type

class PlanItem(BaseModel):
    step: str
    tools: Optional[str] # Removed default value because [] is not a string 
    data_sources: Optional[str] # Removed default value because [] is not a string 
    sub_steps_needed: bool = False

class Plan(BaseModel):
    # """A tool to parse user input to detailed plans"""
    plan: List[PlanItem]

Please note that I have removed your default values of [], because they surely don't look like str to me. Also I have added a docstring in the your tool Plan, (which is commented out because we are not really going to use it as the tool). It is the description if you are using Plan directly as the tool.

But anyway, let's use BaseTool, and now your Plan is just the schema for the actual tool.

from langchain_community.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun

class PlanTool(BaseTool):
    name: str = 'PlanTool'
    # Description is put here instead of in the Plan
    description: str = ('A tool to parse user input to detailed plans')
    args_schema: Type[BaseModel] = Plan
    
    def _run(self, plan: List[PlanItem], run_manager: Optional[CallbackManagerForToolRun] = None) -> List:
        """Use the tool."""
        return plan

Now we try this tool. Note that I broke down your function to steps, so I got to check which step was wrong. You can put them back to a function again:

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.prompts.chat import MessagesPlaceholder

tools = [PlanTool()]
system_prompt = "You take the user input and use your tool `PlanTool` to convert it to a detailed plan."

prompt = ChatPromptTemplate.from_messages(
        [
            (
                "system",
                system_prompt,
            ),
            MessagesPlaceholder(variable_name="messages"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ]
    )
agent = create_openai_tools_agent(llm, tools, prompt) # llm is using ChatOpenAI, gpt-3.5-turbo
executor = AgentExecutor(agent=agent, tools=tools)

# An example input
input = """Pick a quality cut of steak that you enjoy, like ribeye, sirloin, or filet mignon. Make sure it's at room temperature before cooking.
Then, season the steak generously with salt and pepper. You can also add other spices or herbs based on your preference. Preheat the Pan or Grill: Heat a pan or grill over high heat. You want it to be very hot to get a good sear on the steak.
Place the steak on the hot pan or grill. For a medium-rare steak, cook it for about 3-4 minutes on each side, depending on the thickness of the steak.
Use a meat thermometer to check the internal temperature. For medium-rare, it should be around 130-135°F (54-57°C).
Once cooked to your liking, remove the steak from the heat and let it rest for a few minutes. This allows the juices to redistribute, keeping the steak juicy.
Slice the steak against the grain and serve it with your favorite side dishes."""
output = executor.invoke({'messages': [HumanMessage(input)]})
print(output['output'])

The result I got was:

I have created a detailed plan based on your instructions for cooking a steak:

1. Pick a quality cut of steak (e.g., ribeye, sirloin, filet mignon) and bring it to room temperature before cooking.
2. Season the steak generously with salt and pepper, and optionally add other spices or herbs based on preference.
3. Preheat a pan or grill over high heat to ensure it is very hot for searing the steak.
4. Place the seasoned steak on the hot pan or grill and cook for 3-4 minutes on each side for a medium-rare steak, adjusting based on thickness.
5. Use a meat thermometer to check the internal temperature, aiming for 130-135°F (54-57°C) for medium-rare.
6. Remove the steak from the heat when cooked to desired doneness and let it rest for a few minutes to allow the juices to redistribute.
7. Slice the rested steak against the grain and serve with preferred side dishes.

Does this answer your question?

Side notes:

When using LangGraph, my preference would be just use BaseModel in a chain like this: someChain = Prompt | llm.bind_tools(tools), where tools is just Plan with docstring uncommented here. I personally believe this makes it easier to track your state messages.