Tool Methods in Python Class Missing self Argument When Used in Custom Agents

[quartermaine]

I am working on a custom implementation of AI agents using Python, where the agents rely on langchain tools (methods) for interacting with a PostgreSQL database. The ToolWithConnection class initializes a database connection and defines tools like list_tables, tables_schema, check_sql, and execute_sql_to_df. These tools are later used by CrewAI agents to perform various tasks.

Here’s the relevant part of the code:

• Connection Method:

def connection(x):
    postgres_schm = x["database_schm"]
    postgres_host = x["database_host"]
    postgres_port = x["database_port"]
    postgres_user = x["database_user"]
    postgres_pass = x["database_pass"]
    postgres_dryr = x["database_dryr"]

    if postgres_dryr == "true":
        postgres_pass_dec = postgres_pass        
    else:
        postgres_pass_dec = decrypt_string(postgres_pass)
        
    CONNECTION_STRING = (
        f"postgresql+psycopg2://{postgres_user}:{postgres_pass_dec}@{postgres_host}:{postgres_port}/{postgres_schm}"
    )
    
    db = SQLDatabase.from_uri(
        CONNECTION_STRING,
        schema=postgres_schm,
        view_support=True
    )
    return db

• Tool Class:

class ToolWithConnection:
    def __init__(self, x):
        """Initialize with a connection based on the input 'x'."""
        self.db = connection(x)
        print('>> Initialize db connection ✔\n')

    @tool("list_tables")
    def list_tables(self) -> str:
        """List the available tables in the database."""
        return ListSQLDatabaseTool(db=self.db).invoke("")
        
    @tool("tables_schema")
    def tables_schema(self, tables: str) -> str:
        """Get the schema of specified tables."""
        tool = InfoSQLDatabaseTool(db=self.db)
        return tool.invoke(tables)

• CrewAI Wrapper :

class CrewAIWrapper:
    def __init__(self, x):
        self.x = x
        self.llm_crew = LLM(model='azure/GPT4o')
        tool_instance = ToolWithConnection(self.x)

        self.sql_dev = Agent(
            role="Senior Database Developer",
            goal="Construct and execute PostgreSQL queries based on a query {query}",
            tools=[
                tool_instance.list_tables,  
                tool_instance.tables_schema,
            ],
            allow_delegation=False,
            memory=False,
            verbose=True,
        )

    def kickoff_crew(self):
        extract_data_task = Task(
            description="Extract data that is required for the query {query}.",
            agent=self.sql_dev,
        )

        my_crew = Crew(
            agents=[self.sql_dev],
            tasks=[extract_data_task],
            verbose=True,
        )

        try:
            crew_result = my_crew.kickoff(
                inputs={"query": self.x['question']}
            )
            return crew_result.raw
        except Exception as e:
            print(f"Error during task execution: {str(e)}")

• Chain setup:

class CustomCrewAILLM(LLM):
    model: str
        
    def __init__(self, model='azure/GPT4o'):
        super().__init__(model=model)
        
    @property
    def _llm_type(self):
        return 'custom_crew_ai'
    
    def _call(self, obj: dict,) -> str:
        crew_ai_wrapper = CrewAIWrapper(obj, )
        result = crew_ai_wrapper.kickoff_crew()
        print(type(result))
        # print(result)
        return result
class InputType(BaseModel):
    crewai_input: Optional[str]
    db: Optional[SQLDatabase] = Field(default=None, exclude=True)  # Exclude from schema validation
    database_schm: str
    database_host: str
    database_port: str
    database_user: str
    database_pass: str
    database_name: str
    database_dryr: str
    class Config:
        arbitrary_types_allowed = True  # Allow non-Pydantic types like SQLDatabase

    # @field_validator("crewai_input")
    # def validate_input(cls, v):
    #     if not isinstance(v, str):
    #         raise ValidationError(f"Expected crewai_input to be a str, got {type(v)} instead.")
    #     return v

class OutputType(BaseModel):
    crewai_output: Optional[str]

    # @field_validator("crewai_output")
    # def validate_output(cls, v):
    #     if not isinstance(v, str):
    #         raise ValidationError(f"Expected crewai_output to be a str, got {type(v)} instead.")
    #     return v

chain = (
    RunnablePassthrough.assign(
        crewai_output=RunnableLambda(lambda x: CustomCrewAILLM()._call(x))
        )
        ).with_types(input_type=InputType,
                     output_type=OutputType
                     )

• The Problem:
  When I run the CrewAIWrapper with the agents and tools, I get the following error for the list_tables tool:

I encountered an error while trying to use the tool. This was the error: ToolWithConnection.list_tables() missing 1 required positional argument: 'self'.

• Question:

• Why is the self argument missing when calling list_tables?

• How can I properly pass instance methods (like list_tables) from a class to agents without encountering this error?

[dosubot[bot]]

The error you're encountering is due to the way the @tool decorator is handling the instance methods in your ToolWithConnection class. When you use the @tool decorator, it transforms the method into a tool, but it doesn't automatically bind the self argument, which is necessary for instance methods.

To resolve this issue, you need to ensure that the self argument is correctly bound when passing the instance methods to the agents. One way to achieve this is by using functools.partial to bind the self argument to the method before passing it to the agent. Here's how you can modify your code:

1. Import functools.partial at the beginning of your script:

   from functools import partial

2. Modify the CrewAIWrapper class to bind the self argument:

   class CrewAIWrapper:
       def __init__(self, x):
           self.x = x
           self.llm_crew = LLM(model='azure/GPT4o')
           tool_instance = ToolWithConnection(self.x)
   
           self.sql_dev = Agent(
               role="Senior Database Developer",
               goal="Construct and execute PostgreSQL queries based on a query {query}",
               tools=[
                   partial(tool_instance.list_tables),  
                   partial(tool_instance.tables_schema),
               ],
               allow_delegation=False,
               memory=False,
               verbose=True,
           )

By using partial, you bind the self argument to the instance methods, ensuring they are called correctly when used as tools in the agent. This should resolve the error you're encountering with the missing self argument.

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Other}

[dosubot[bot]]

To resolve the Pydantic validation error, you need to ensure that the tools you are using in the Agent class are either instances of BaseTool or objects with the required attributes ('name', 'func', and 'description').

Here's how you can define a tool as a subclass of BaseTool:

from langchain_core.tools.base import BaseTool

class MyTool(BaseTool):
    name = "my_tool"
    description = "This is my custom tool."

    def _run(self, *args, **kwargs):
        # Implement the tool's functionality here
        return "Tool executed successfully"

# Create an instance of the tool
my_tool_instance = MyTool()

Alternatively, you can use the @tool decorator to create a tool from a function, which automatically ensures the function is wrapped as a BaseTool instance:

from langchain_core.tools import tool

@tool
def list_tables_tool() -> str:
    """List the available tables in the database."""
    return "List of tables"

# This function is now a BaseTool instance
assert isinstance(list_tables_tool, BaseTool)

Ensure that when you add tools to the Agent class, they are either instances of BaseTool like my_tool_instance or objects with the required attributes. If you are using instance methods as tools, you might need to wrap them in a class that provides these attributes or use the @tool decorator to convert them into BaseTool instances ^[1].

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Other}

[quartermaine]

@dosubot
How to do that in my implementation?

[dosubot[bot]]

To ensure that your instance methods are valid tools for the Agent class, you can use the @tool decorator to transform them into BaseTool instances. Here's how you can adjust your implementation:

1. Ensure the @tool decorator is correctly applied: Make sure that the @tool decorator is applied to your methods in the ToolWithConnection class. This will automatically convert them into BaseTool instances.

2. Use the decorated methods directly: Once decorated, these methods should be directly usable as tools in your Agent class.

Here's how you can modify your ToolWithConnection class and CrewAIWrapper class:

from langchain_core.tools import tool

class ToolWithConnection:
    def __init__(self, x):
        """Initialize with a connection based on the input 'x'."""
        self.db = connection(x)
        print('>> Initialize db connection ✔\n')

    @tool
    def list_tables(self) -> str:
        """List the available tables in the database."""
        return ListSQLDatabaseTool(db=self.db).invoke("")
        
    @tool
    def tables_schema(self, tables: str) -> str:
        """Get the schema of specified tables."""
        tool = InfoSQLDatabaseTool(db=self.db)
        return tool.invoke(tables)

class CrewAIWrapper:
    def __init__(self, x):
        self.x = x
        self.llm_crew = LLM(model='azure/GPT4o')
        tool_instance = ToolWithConnection(self.x)

        # Use the decorated methods directly
        self.sql_dev = Agent(
            role="Senior Database Developer",
            goal="Construct and execute PostgreSQL queries based on a query {query}",
            tools=[
                tool_instance.list_tables,  
                tool_instance.tables_schema,
            ],
            allow_delegation=False,
            memory=False,
            verbose=True,
        )

By using the @tool decorator, your methods are transformed into BaseTool instances, which should satisfy the requirements for the Agent class ^[1][2].

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Other}

[quartermaine]

@dosubot

Still the same error

[dosubot[bot]]

The error you're encountering suggests that the methods decorated with @tool are not being correctly transformed into BaseTool instances. Let's ensure that the @tool decorator is applied correctly and that the methods are being used as intended.

1. Check the @tool Decorator: Ensure that the @tool decorator is correctly applied to the methods in your ToolWithConnection class. The decorator should automatically convert these methods into BaseTool instances.

2. Verify the Decorator's Functionality: The @tool decorator should handle the transformation of methods into callable tools. If this isn't happening, there might be an issue with how the decorator is being used or applied.

3. Directly Use the Decorated Methods: Once decorated, these methods should be directly usable as tools in your Agent class. Ensure that you are not wrapping them in any additional functions or lambdas that might interfere with their transformation.

Here's a refined approach to ensure everything is set up correctly:

from langchain_core.tools import tool

class ToolWithConnection:
    def __init__(self, x):
        """Initialize with a connection based on the input 'x'."""
        self.db = connection(x)
        print('>> Initialize db connection ✔\n')

    @tool
    def list_tables(self) -> str:
        """List the available tables in the database."""
        return ListSQLDatabaseTool(db=self.db).invoke("")
        
    @tool
    def tables_schema(self, tables: str) -> str:
        """Get the schema of specified tables."""
        tool = InfoSQLDatabaseTool(db=self.db)
        return tool.invoke(tables)

class CrewAIWrapper:
    def __init__(self, x):
        self.x = x
        self.llm_crew = LLM(model='azure/GPT4o')
        tool_instance = ToolWithConnection(self.x)

        # Use the decorated methods directly
        self.sql_dev = Agent(
            role="Senior Database Developer",
            goal="Construct and execute PostgreSQL queries based on a query {query}",
            tools=[
                tool_instance.list_tables,  
                tool_instance.tables_schema,
            ],
            allow_delegation=False,
            memory=False,
            verbose=True,
        )

Ensure that the @tool decorator is correctly applied and that the methods are being used directly as tools. If the error persists, double-check that the @tool decorator is functioning as expected in your environment and that there are no additional layers or wrappers affecting the tool instances ^[1][2].

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Other}

[quartermaine]

I managed to find the solution by making each tool a separate class. Below is an example of how I structured the tools:

Tool Classes

Each tool is implemented as a standalone class, with attributes for its name, functionality, and description. This encapsulation provides clarity and modularity.

class ListTablesTool:
    def __init__(self, db_connection):
        self.name = "list_tables"
        self.func = self.list_tables
        self.description = "Lists the available tables in the database"
        self.db_connection = db_connection

    def list_tables(self) -> str:
        """List the available tables in the database"""
        return ListSQLDatabaseTool(db=self.db_connection).invoke("")

class TablesSchemaTool:
    def __init__(self, db_connection):
        self.name = "tables_schema"
        self.func = self.tables_schema
        self.description = "Gets schema and sample rows for specified tables"
        self.db_connection = db_connection

    def tables_schema(self, tables: str) -> str:
        """Fetch schema for the provided tables"""
        tool = InfoSQLDatabaseTool(db=self.db_connection)
        return tool.invoke(tables)

Initializing the Tools

In the main workflow or kickoff process, the tools are initialized like this:

self.list_tables = ListTablesTool(self.db)
self.tables_schema = TablesSchemaTool(self.db)

I hope this helps!

[quartermaine]

I managed to find the solution by making each tool a separate class. Below is an example of how I structured the tools:

Tool Classes

Each tool is implemented as a standalone class, with attributes for its name, functionality, and description. This encapsulation provides clarity and modularity.

class ListTablesTool:
    def __init__(self, db_connection):
        self.name = "list_tables"
        self.func = self.list_tables
        self.description = "Lists the available tables in the database"
        self.db_connection = db_connection

    def list_tables(self) -> str:
        """List the available tables in the database"""
        return ListSQLDatabaseTool(db=self.db_connection).invoke("")

class TablesSchemaTool:
    def __init__(self, db_connection):
        self.name = "tables_schema"
        self.func = self.tables_schema
        self.description = "Gets schema and sample rows for specified tables"
        self.db_connection = db_connection

    def tables_schema(self, tables: str) -> str:
        """Fetch schema for the provided tables"""
        tool = InfoSQLDatabaseTool(db=self.db_connection)
        return tool.invoke(tables)

Initializing the Tools

In the main workflow or kickoff process, the tools are initialized like this:

self.list_tables = ListTablesTool(self.db)
self.tables_schema = TablesSchemaTool(self.db)

I hope this helps!