funccall source

Exported source 
import inspect
from fastcore.utils import *
from fastcore.docments import docments

Function calling

Many LLMs do function calling (aka tool use) by taking advantage of JSON schema.

We’ll use docments to make getting JSON schema from Python functions as ergonomic as possible. Each parameter (and the return value) should have a type, and a docments comment with the description of what it is. Here’s an example:

def silly_sum(
    a:int, # First thing to sum
    b:int=1, # Second thing to sum
    c:list[int]=None, # A pointless argument
) -> int: # The sum of the inputs
    "Adds a + b."
    return a + b

This is what docments makes of that:

d = docments(silly_sum, full=True)
d
{ 'a': { 'anno': <class 'int'>,
         'default': <class 'inspect._empty'>,
         'docment': 'First thing to sum'},
  'b': {'anno': <class 'int'>, 'default': 1, 'docment': 'Second thing to sum'},
  'c': {'anno': list[int], 'default': None, 'docment': 'A pointless argument'},
  'return': { 'anno': <class 'int'>,
              'default': <class 'inspect._empty'>,
              'docment': 'The sum of the inputs'}}

Note that this is an AttrDict so we can treat it like an object, or a dict:

d.a.docment, d['a']['anno']
('First thing to sum', int)
Exported source 
def _types(t:type)->tuple[str,Optional[str]]:
    "Tuple of json schema type name and (if appropriate) array item name."
    if t is empty: raise TypeError('Missing type')
    tmap = {int:"integer", float:"number", str:"string", bool:"boolean", list:"array", dict:"object"}
    tmap.update({k.__name__: v for k, v in tmap.items()})
    if getattr(t, '__origin__', None) in  (list,tuple): return "array", tmap.get(t.__args__[0], "object")
    else: return tmap[t], None

This internal function is needed to convert Python types into JSON schema types.

_types(list[int]), _types(int), _types('int')
(('array', 'integer'), ('integer', None), ('integer', None))
Exported source 
def _param(name, info):
    "json schema parameter given `name` and `info` from docments full dict."
    paramt,itemt = _types(info.anno)
    pschema = dict(type=paramt, description=info.docment or "")
    if itemt: pschema["items"] = {"type": itemt}
    if info.default is not empty: pschema["default"] = info.default
    return pschema

This private function converts a key/value pair from the docments structure into the dict that will be needed for the schema.

n,o = first(d.items())
print(n,'//', o)
_param(n, o)
a // {'docment': 'First thing to sum', 'anno': <class 'int'>, 'default': <class 'inspect._empty'>}
{'type': 'integer', 'description': 'First thing to sum'}

source

get_schema

 get_schema (f:<built-infunctioncallable>, pname='input_schema')

Convert function f into a JSON schema dict for tool use.

Exported source 
def get_schema(f:callable, pname='input_schema')->dict:
    "Convert function `f` into a JSON schema `dict` for tool use."
    d = docments(f, full=True)
    ret = d.pop('return')
    d.pop('self', None) # Ignore `self` for methods
    paramd = {
        'type': "object",
        'properties': {n:_param(n,o) for n,o in d.items() if n[0]!='_'},
        'required': [n for n,o in d.items() if o.default is empty and n[0]!='_']
    }
    desc = f.__doc__
    assert desc, "Docstring missing!"
    if ret.anno is not empty: desc += f'\n\nReturns:\n- type: {_types(ret.anno)[0]}'
    if ret.docment: desc += f'\n- description: {ret.docment}'
    return {'name':f.__name__, 'description':desc, pname:paramd}

Putting this all together, we can now test getting a schema from silly_sum. The tool use spec doesn’t support return annotations directly, so we put that in the description instead.

s = get_schema(silly_sum)
desc = s.pop('description')
print(desc)
s
Adds a + b.

Returns:
- type: integer
- description: The sum of the inputs
{'name': 'silly_sum',
 'input_schema': {'type': 'object',
  'properties': {'a': {'type': 'integer', 'description': 'First thing to sum'},
   'b': {'type': 'integer',
    'description': 'Second thing to sum',
    'default': 1},
   'c': {'type': 'array',
    'description': 'A pointless argument',
    'items': {'type': 'integer'},
    'default': None}},
  'required': ['a']}}

This also works with string annotations, e.g:

def silly_test(
    a: 'int',  # quoted type hint
):
    "Mandatory docstring"
    return a

get_schema(silly_test)

Python tool

In language model clients it’s often useful to have a ‘code interpreter’ – this is something that runs code, and generally outputs the result of the last expression (i.e like IPython or Jupyter).

In this section we’ll create the python function, which executes a string as Python code, with an optional timeout. If the last line is an expression, we’ll return that – just like in IPython or Jupyter, but without needing them installed.

Exported source 
import ast, time, signal, traceback
from fastcore.utils import *
Exported source 
def _copy_loc(new, orig):
    "Copy location information from original node to new node and all children."
    new = ast.copy_location(new, orig)
    for field, o in ast.iter_fields(new):
        if isinstance(o, ast.AST): setattr(new, field, _copy_loc(o, orig))
        elif isinstance(o, list): setattr(new, field, [_copy_loc(value, orig) for value in o])
    return new

This is an internal function that’s needed for _run to ensure that location information is available in the abstract syntax tree (AST), since otherwise python complains.

Exported source 
def _run(code:str ):
    "Run `code`, returning final expression (similar to IPython)"
    tree = ast.parse(code)
    last_node = tree.body[-1] if tree.body else None
    
    # If the last node is an expression, modify the AST to capture the result
    if isinstance(last_node, ast.Expr):
        tgt = [ast.Name(id='_result', ctx=ast.Store())]
        assign_node = ast.Assign(targets=tgt, value=last_node.value)
        tree.body[-1] = _copy_loc(assign_node, last_node)

    compiled_code = compile(tree, filename='<ast>', mode='exec')
    namespace = {}
    stdout_buffer = io.StringIO()
    saved_stdout = sys.stdout
    sys.stdout = stdout_buffer
    try: exec(compiled_code, namespace)
    finally: sys.stdout = saved_stdout
    _result = namespace.get('_result', None)
    if _result is not None: return _result
    return stdout_buffer.getvalue().strip()

This is the internal function used to actually run the code – we pull off the last AST to see if it’s an expression (i.e something that returns a value), and if so, we store it to a special _result variable so we can return it.

_run('import math;math.factorial(12)')
479001600
_run('print(1+1)')
'2'

We now have the machinery needed to create our python function.

source

python

 python (code, timeout=5)

Executes python code with timeout and returning final expression (similar to IPython). Raised exceptions are returned as a string, with a stack trace.

Type Default Details
code Code to execute
timeout int 5 Maximum run time in seconds before a TimeoutError is raised
Exported source 
def python(code, # Code to execute
           timeout=5 # Maximum run time in seconds before a `TimeoutError` is raised
          ): # Result of last node, if it's an expression, or `None` otherwise
    """Executes python `code` with `timeout` and returning final expression (similar to IPython).
    Raised exceptions are returned as a string, with a stack trace."""
    def handler(*args): raise TimeoutError()
    signal.signal(signal.SIGALRM, handler)
    signal.alarm(timeout)
    try: return _run(code)
    except Exception as e: return traceback.format_exc()
    finally: signal.alarm(0)

There’s no builtin security here – you should generally use this in a sandbox, or alternatively prompt before running code. It can handle multiline function definitions, and pretty much any other normal Python syntax.

python("""def factorial(n):
    if n == 0 or n == 1: return 1
    else: return n * factorial(n-1)
factorial(5)""")
120

If the code takes longer than timeout then it raises a TimeoutError.

try: python('import time; time.sleep(10)', timeout=1)
except TimeoutError: print('Timed out')