Tool Calling Issue

[PranavScogo]

I am working on a task which converts the Natural Language into SQL query, for which I have created 4 tools QuerySQLDatabaseTool(), InfoSQLDatabaseTool(), ListSQLDatabaseTool(),QuerySQLCheckerTool() similar to those which create_sql_agent is using in langchain.
But how to call these tools sequentially as required considdering examples provided in the prompt.

[PranavScogo]

This is the BaseAgent

"""This file contains the base agent class."""

import re
from datetime import datetime
from typing import Any, Dict, Generator, List, Optional

from llm import get_models_manager
from tools.registry import tools_registry

from agent_core.context import Context, _inject_ctx_into_tools

class BaseAgent:
"""Base class for creating agents that can interact with various tools and processes."""

def __init__(
    self,
    ctx: Optional[Context] = None,
    config: Optional[Dict[str, Any]] = None,
    name: Optional[str] = None,
    model: Optional[str] = None,
    description: Optional[str] = None,
    system_prompt: Optional[str] = None,
    tools: Optional[List[str]] = None,
):
    """Initialize the agent."""
    try:
        if config:
            # Initialize from config
            self.name = config["name"]
            self.model = config["model"]
            self.description = config["description"]
            self.system_prompt = config["system_prompt"]
            tool_names = config.get("tools", [])
        else:
            # Initialize from explicitly passed arguments
            if not all([name, model, description, system_prompt]):
                raise ValueError(
                    "If config is not provided, all of 'name', 'model', 'description', and 'system_prompt' must be specified."
                )
            self.name = name
            self.model = model
            self.description = description
            self.system_prompt = system_prompt
            tool_names = tools or []

        if len(tool_names) > 0:
            self.tools = tools_registry.load_tools(tool_names)
        else:
            self.tools = []

        self.context = ctx
        # Replace with actual model manager
        models_manager = get_models_manager()
        self.llm = models_manager.get(self.model)
        self._initialize_tools()
    except Exception as e:
        raise e

def _initialize_tools(self):
    """Inject context into tools."""
    if self.tools:
        _inject_ctx_into_tools(self.context, self.tools)

def _get_tool_descriptions(self) -> Dict[str, str]:
    """Generate descriptions of the tools."""
    return {tool.name: tool.description for tool in self.tools}

def _get_prompt_variables(self) -> Dict[str, Any]:
    """Get prompt variables."""
    _prompt_variables = {
        "agent_description": self.description,
        "tools": self._get_tool_descriptions(),
        "datetime": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
    }
    if self.context.get("prompt_variables"):
        _prompt_variables.update(self.context.get("prompt_variables"))
    return _prompt_variables

@staticmethod
def _format_system_prompt(
    prompt_template: str, prompt_variables: Dict[str, Any]
) -> str:
    """Format the system prompt similar to Langfuse's compile method."""
    placeholders = re.findall(r"\{\{(\w+)\}\}", prompt_template)
    missing_keys = [key for key in placeholders if key not in prompt_variables]
    if missing_keys:
        raise KeyError(f"Missing variables in prompt_variables: {missing_keys}")

    for key in placeholders:
        placeholder = f"{{{{{key}}}}}"
        prompt_template = prompt_template.replace(
            placeholder, str(prompt_variables[key])
        )

    return prompt_template

def run(
    self, messages: List[Dict[str, Any]], stream: bool = False
) -> Dict[str, Any] | Generator[Dict[str, Any], None, None]:
    """Run the agent with the provided message and interaction history."""
    system_prompt = self._format_system_prompt(
        self.system_prompt, self._get_prompt_variables()
    )
    try:
        return self.llm.invoke(
            system_prompt=system_prompt,
            messages=messages,
            ctx=self.context,
            tools=self.tools if self.tools else None,
            stream=stream,
        )
    except Exception as e:
        raise e

def to_dict(self) -> Dict[str, Any]:
    """Convert the agent to a dictionary."""
    return {
        "name": self.name,
        "description": self.description,
        "tools": [tool.to_dict() for tool in self.tools],
    }

[manimovassagh]

The key insight is that you don't call the tools sequentially yourself -- you let the model decide the order by running a tool-calling loop. The model will call tools one (or several) at a time, you execute them and feed results back, and the model decides what to call next until it has enough info to generate the final SQL.

Here's the standard pattern using the OpenAI Python SDK:

from openai import OpenAI
import json

client = OpenAI()

# Define your tools as function schemas
tools = [
    {
        "type": "function",
        "function": {
            "name": "list_tables",
            "description": "List all tables in the database",
            "parameters": {"type": "object", "properties": {}, "required": []},
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_table_info",
            "description": "Get schema info (columns, types, sample rows) for specified tables",
            "parameters": {
                "type": "object",
                "properties": {
                    "tables": {"type": "array", "items": {"type": "string"}, "description": "Table names to inspect"}
                },
                "required": ["tables"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "check_sql",
            "description": "Validate a SQL query for correctness before executing",
            "parameters": {
                "type": "object",
                "properties": {"query": {"type": "string", "description": "SQL query to validate"}},
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "run_sql",
            "description": "Execute a SQL query and return results",
            "parameters": {
                "type": "object",
                "properties": {"query": {"type": "string", "description": "SQL query to execute"}},
                "required": ["query"],
            },
        },
    },
]

# Map tool names to your actual implementations
tool_functions = {
    "list_tables": lambda **kwargs: list_sql_database_tool(),
    "get_table_info": lambda tables, **kwargs: info_sql_database_tool(tables),
    "check_sql": lambda query, **kwargs: query_sql_checker_tool(query),
    "run_sql": lambda query, **kwargs: query_sql_database_tool(query),
}

def run_agent(user_question: str):
    messages = [
        {"role": "system", "content": "You are a SQL expert. Use the tools to explore the database schema, then write and validate a SQL query to answer the user's question."},
        {"role": "user", "content": user_question},
    ]

    # Tool-calling loop
    while True:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
        )

        msg = response.choices[0].message
        messages.append(msg)

        # If no tool calls, the model is done
        if not msg.tool_calls:
            return msg.content

        # Execute each tool call and append results
        for tool_call in msg.tool_calls:
            fn_name = tool_call.function.name
            fn_args = json.loads(tool_call.function.arguments)

            result = tool_functions[fn_name](**fn_args)

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result) if not isinstance(result, str) else result,
            })

answer = run_agent("How many orders were placed last month?")
print(answer)

The model will typically follow a sequence like: list_tables -> get_table_info -> write SQL -> check_sql -> run_sql -> return answer. But you don't hardcode this -- the loop lets the model figure out the right order based on what it learns at each step.