How to use Function calling in DSPy?

[BoxiYu]

I am trying to build an Agent using Function Calling and try to implement it with DSPy, but I haven't found the instructions for using DSPy with function calling.

[CShorten]

Hey @BoxiYu! It does not look like DSPy is working towards supporting black-box tool APIs. Please see this example of how the dspy.ReAct module interfaces tools in DSPy: https://github.com/stanfordnlp/dspy/blob/main/examples/agents/multi_agent.ipynb.

Please let me know if this solution works for you, happy to continue discussing! cc @CyrusOfEden @KCaverly @okhat

[thickett]

do you mean that you want the agent to do the function calling? if so this is quite simple;

Function calling can be achieved using tools that you can define within the DSPY framework.

you can read about it here:

it is rather simple;

1. define your python functions, which can quite literaly do anything, and can take input parameters that are generated as part of your agent, i.e you can have the LLM generate inputs, feed user-inputs through etc.

2. create a toolbox

3. supply this toolbox to the ReAct module, you can either use this with the default signature, or of course create your own specific for your task.

4. you can then simply use this as a building block in your program, add suggest/assert statements etc as you would do with any other dspy program.

here is an example of defining some basic tools to give a general idea of how you would create a module for function calling:

`

# define a generic class that our tool's will follow, they will need a name, description, 
# a python function and presumably some inputs from your llm.
@dataclass
class Tool:
    name: str
    desc: str
    input_variable: str
    function: Callable

    def __call__(self, *args, **kwargs) -> Tuple[str, str]:
        return self.function(*args, **kwargs)


def tool_one(*args, **kwargs):
    # simple tool that will just return '1' if called.
    return "1"


def tool_two(sql_code):

    # assuming we have some function that gets our dburi and creates a connection to our database.
    engine = create_engine()
    connection = engine.connect()

    data = pd.read_sql_query(sql_code, connection)
    # we can do what we want here, lets just save the sql code and the data. just as you may do in a 
    # typical python function.
    with open(r"./sql_code.sql", "w") as f:
        f.write(sql_code)
    data_directory = r"./data.csv"
    data.to_csv(data_directory, index=False)
    data_sample = data.sample(5)
    return {
    "columns": list(data.columns),
    "data": data_sample, # return what ever you want the llm to see for its next step in react.
}
toolbox: List[Tool] = [
    Tool(
        name="tool_one",
        input_variable="user query",
        desc="a simple tool that just returns '1'. ",
        function=tool_one,
    ),
    Tool(
        name="tool_2",
        input_variable="SQL code you have generated based on the user's query.",
        desc="A tool used to connect to a SQL database and run SQL you have generated.",
        function=tool_two,
    ),
]
# you can then go on to use this in a ReAct module for example, 
react_module = dspy.ReAct("question -> answer", tools=toolbox)
#(you will want to write your own signiture rather than using the pre-made question -> answer most likely)

`

[BoxiYu]

Thank you for your comments @CShorten @thickett, it is useful for me to learn how to use tools in DSPy. I might not explain my question clearly. I want to figure out how I can use the function calling feature of Openai API (https://platform.openai.com/docs/guides/function-calling) in DSPy.

[mepwang]

Thank you for your comments @CShorten @thickett, it is useful for me to learn how to use tools in DSPy. I might not explain my question clearly. I want to figure out how I can use the function calling feature of Openai API (https://platform.openai.com/docs/guides/function-calling) in DSPy.

Yes it can be done by:

1. write a custom llm client to parse llm response, to strip out tool calling part, which can refer to: https://github.com/stanfordnlp/dspy/issues/192#issuecomment-2058281547

2. define tools: tools = [ { 'type': 'function', 'function': { 'name': 'get_current_time', 'description': '当你想知道现在的时间时非常有用。', 'parameters': {} } }, { 'type': 'function', 'function': { 'name': 'get_stock_info', 'description': '获取股票信息', 'parameters': { 'type': 'object', 'properties': { 'stock_code': { 'type': 'string', 'description': '股票代码', }, 'start_date': { 'type': 'string', 'description': '开始日期，格式为YYYYMMDD', }, 'end_date': { 'type': 'string', 'description': '结束日期，格式为YYYYMMDD', }, 'adjust': { 'type': 'string', 'description': '复权类型，可选参数为qfq、hfq', 'default': 'qfq' } } } } } ]

3. call dspy.Predict: gen = dspy.Predict('task -> answer', tools=tools)(task=task)

job done.

To implement llm tool calling using the dspy.ReAct module, a slight modification to the ReAct source code is required.

[whisper-bye]

@mepwang Could you give provide more details? I want to call my local function using dspy.OllamaLocal and dspy.Predict

[mepwang]

@mepwang Could you give provide more details? I want to call my local function using dspy.OllamaLocal and dspy.Predict

a custom llm client is required to implement tool calling.

1. define a cusom llm client which is openai compatible. the llm client parse tool calling as well as completions.

from openai import OpenAI
from dsp import LM
from datetime import datetime
import os
from typing import Any, Dict
import dspy
from pydantic import BaseModel, Field
from ast import literal_eval

from dotenv import load_dotenv
load_dotenv()


class OpenAICompatible(LM):
    def __init__(
            self,
            model: str,
            api_key: str,
            base_url: str,
            **kwargs
    ):
        super().__init__(model=model)
        self.base_url = base_url
        self.api_key = api_key
        self.model = model
        self.kwargs.update(kwargs)

    def basic_request(self, prompt, **kwargs):
        config = dict()
        config.update(self.kwargs)
        config['model'] = self.model
        config.update(kwargs)

        messages = [
            {
                "role": "system",
                "content": "You are a helpful assistant."
            },
            {
                'role': 'user',
                'content': prompt
            }
        ]

        client = OpenAI(
            api_key=self.api_key,
            base_url=self.base_url
        )

        response = client.chat.completions.create(
            messages=messages,
            **config,
        )

        self.history.append({
            'prompt': prompt,
            'response': response,
            'kwargs': config,
        })

        return response

    def get_choice_text(self, choice: dict[str, Any]) -> str:
        message = choice.message
        content = message.content
        if message.tool_calls != None:
            return str([
                {
                    'name': tool.function.name,
                    'arguments': tool.function.arguments
                } for tool in message.tool_calls])
        return content

    def __call__(self, prompt, only_completed=True, return_sorted=False, **kwargs):
        response = self.request(prompt, **kwargs)

        completions = [self.get_choice_text(choice)
                       for choice in response.choices]

        return completions

2. define tools.

class ToolBase(BaseModel):
    '''Base model for tool'''
    ...


class WeatherTool(ToolBase):
    '''useful when get weather data'''
    location: str = Field(..., description="Location to get weather for")
    date: datetime = Field(..., description="Date to get weather for")

    def __call__(self):
        print(f'Getting weather for {self.location} on {self.date}')

3. define util function to convert function to tool calling json string.

def convert_to_schema(model: ToolBase) -> Dict[str, Any]:
    json_schema = model.model_json_schema()
    function_calling_json = {
        'type': 'function',
        'function': {
            'name': model.__name__,
            'description': model.__doc__,
            'parameters': json_schema
        }
    }
    return function_calling_json

4. make the tool call

lm = OpenAICompatible(
    model=os.environ.get('MODEL_NAME'),
    api_key=os.environ.get('OPENAI_API_KEY'),
    base_url=os.environ.get('OPENAI_API_BASE'),
    max_tokens=2048,
    temperature=0.1
)

dspy.settings.configure(lm=lm)

question = f'Today is {datetime.now()}, what is the weather like tomorrow in London?'

gen = dspy.Predict('question -> answer', tools=[convert_to_schema(WeatherTool)])(question=question)

try:
    json_calling = literal_eval(gen.answer)[0]
    tool_name = json_calling['name']
    tool_args = literal_eval(json_calling['arguments'])

    if tool_name == 'WeatherTool':
        weather_data = WeatherTool(**tool_args)()
except Exception:
    print(f'got answer: {gen.answer}')

[whisper-bye]

@mepwang cool! thanks a lot.