Function Calling

Simple guide to understanding and implementing function calling.

Installing the SDK

Automatic Tool Calling

Typescript

Here we define the tool function in json

import { RunnableToolFunction } from 'propulsionai/lib/RunnableFunction';

const tools: RunnableToolFunction<any>[] = [
  {
    type: 'function',
    function: {
      name: 'list',
      description: 'list queries books by genre, and returns a list of names of books',
      parameters: {
        type: 'object',
        properties: {
          genre: { type: 'string', enum: ['mystery', 'nonfiction', 'memoir', 'romance', 'historical'] },
        },
      },
      function: list,
      parse: JSON.parse,
    },
  } as RunnableToolFunction<{ genre: string }>,
  {
    type: 'function',
    function: {
      name: 'search',
      description: 'search queries books by their name and returns a list of book names and their ids',
      parameters: {
        type: 'object',
        properties: {
          name: { type: 'string' },
        },
      },
      function: search,
      parse: JSON.parse,
    },
  } as RunnableToolFunction<{ name: string }>,
  {
    type: 'function',
    function: {
      name: 'get',
      description:
        "get returns a book's detailed information based on the id of the book. Note that this does not accept names, and only IDs, which you can get by using search.",
      parameters: {
        type: 'object',
        properties: {
          id: { type: 'string' },
        },
      },
      function: get,
      parse: JSON.parse,
    },
  } as RunnableToolFunction<{ id: string }>,
];

async function list({ genre }: { genre: string }) {
  console.log('calling list: ', genre);
  return db.filter((item) => item.genre === genre).map((item) => ({ name: item.name, id: item.id }));
}

async function search({ name }: { name: string }) {
  console.log('calling search: ', name);
  return db.filter((item) => item.name.includes(name)).map((item) => ({ name: item.name, id: item.id }));
}

async function get({ id }: { id: string }) {
  console.log('calling get: ', id);
  return db.find((item) => item.id === id);
}

Using run_tools for automatic function calling with or without streaming.

async function main() {
  const runner = await client.chat.completions
    .runTools({
      deployment: '<deployment_id>',
      stream: true, // can be false
      tools, // as defined earlier
      messages: [
        {
          role: 'system',
          content:
            'Please use our book database, which you can access using functions to answer the following questions.',
        },
        {
          role: 'user',
          content:
            'I really enjoyed reading Where the Crawdads Sing, could you recommend me a book that is similar and tell me why?',
        },
      ],
    })
    .on('message', (msg) => console.log('msg', msg))
    .on('functionCallResult', (functionCallResult) => console.log('functionCallResult', functionCallResult))
    .on('content', (diff) => process.stdout.write(diff))
    .on('totalUsage', (totalUsage) => console.log('totalUsage', totalUsage))
    .on('task_id', (task_id) => console.log('task_id', task_id));

  const result = await runner.finalChatCompletion();
  console.log();
  console.log('messages');
  console.log(JSON.stringify(runner.messages));

  console.log();
  console.log('final chat completion');
  console.dir(result, { depth: null });
}

main()

Python

Here we define the tool function, its parameters and examples (optional)

def get_current_weather(location: str, format: str) -> str:
    """
    Retrieves the current weather for a specified location.

    This function fetches the current weather data for the given location
    and returns it in the specified format.

    Parameters:
    -----------
    location : str
        The name of the location (city, country, etc.) to get weather for.
    format : str
        The desired output format. Accepted values are 'celsius', 'fahrenheit',
        or 'kelvin' for temperature units.

    Examples:
    ---------
    >>> get_current_weather("New York", "celsius")
    'Current weather in New York: 22°C, Partly Cloudy'

    >>> get_current_weather("London", "fahrenheit")
    'Current weather in London: 59°F, Rainy'
    """
    # print(f"Function Log: Getting weather for {location} in {format} format...")
    return f'{{"temperature": 22, "description": "Partly Cloudy", "location": "{location}", "units": "{format}"}}'

Using run_tools for automatic function calling without streaming.

def sync_main() -> None:
    response = client.chat.completions.run_tools(
        deployment="<deployment_id>",
        messages=[
            {
                "role": "system",
                "content": "You are a helpful assistant.",
            },
            {
                "role": "user",
                "content": "What is the weather in SF and NY?",
            },
        ],
        tools=[get_current_weather],
        stream=False,
        tool_debug=False,
    )
    print(response.to_json())
sync_main()

Using run_tools for automatic function calling while streaming.

def sync_main() -> None:
    response = client.chat.completions.run_tools(
        deployment="<deployment_id>",
        messages=[
            {
                "role": "system",
                "content": "You are a helpful assistant.",
            },
            {
                "role": "user",
                "content": "What is the weather in SF and NY?",
            },
        ],
        tools=[get_current_weather],
        stream=True,
        tool_debug=False,
    )
    for data in response:
      if data:
          chunk = data.to_json()
          chunk_json = json.loads(chunk)
          
          if chunk_json["choices"] and chunk_json["choices"][0]["delta"] is not None:
              delta = chunk_json["choices"][0]["delta"]
              
              # Check if 'content' exists in delta and is not None
              if "content" in delta and delta["content"] is not None:
                  sys.stdout.write(delta["content"])
                  sys.stdout.flush()  # Ensure content is displayed immediately
              
              # Check for tool calls
              # if "tool_calls" in delta:
                  # print("Tool call detected:", delta["tool_calls"])
              
              # Check for finish reason
              # if "finish_reason" in chunk_json["choices"][0] and chunk_json["choices"][0]["finish_reason"] == "stop":
              #     print("\nGeneration complete.")
sync_main()

Manualy manage function calling

Typescript

// TODO

Python

// TODO