Move function format specification to function_tool.py

examples/function-calling/function_tool.py

@@ -3,6 +3,8 @@
 import inspect
 import re
 
+import json
+
 # Extract OpenAI function calling style definitions from functions
 #
 # Generated with: Create a python function to to generate the OpenAI function calling definition from a given function, getting the description, parameter type and parameter description from the function documentation, assuming the function documentation contains sphynx style parameter descriptions, marked with :param.
@@ -36,7 +38,7 @@ def get_function_tool_json(func):
 # Generate function definition schema from function definitions
 #
 # This is from llama-cpp-python, llama_chat_format.py
-def generate_schema_from_functions(functions, namespace="functions") -> str:
+def generate_functionary_schema_from_functions(functions, namespace="functions") -> str:
     schema = (
         "// Supported function definitions that should be called when necessary.\n"
     )
@@ -61,3 +63,31 @@ def generate_schema_from_functions(functions, namespace="functions") -> str:
 
     schema += "}} // namespace {}".format(namespace)
     return schema
+
+functionary_prompt_start = """<|start_header_id|>system<|end_header_id|>
+
+You are capable of executing available function(s) if required.
+Execute function(s) as needed.
+The function calls are not shown in the conversation and should be called covertly to answer questions.
+Ask for the required input to:recipient==all
+Use JSON for function arguments.
+Respond in this format:
+>>>${recipient}
+${content}
+Available functions:
+"""
+functionary_prompt_end = """<|eot_id|><|start_header_id|>system<|end_header_id|>
+
+When you send a message containing Python code to python, it will be executed in a stateful Jupyter notebook environment. python will respond with the output of the execution or time out after 60.0 seconds. The drive at '/mnt/data' can be used to save and persist user files.<|eot_id|><|start_header_id|>user<|end_header_id|>
+"""
+
+def get_chat_tool_format(args, tools):
+    return {
+        'prompt': functionary_prompt_start + generate_functionary_schema_from_functions(tools) + functionary_prompt_end,
+        'function_marker': '>>>',
+        'function_re': r'>>>([^\n]*)\n(.*)<\|eot_id\|>',
+        'user_start': '<|start_header_id|>user<|end_header_id|>\n',
+        'user_end': '<|eot_id|><|start_header_id|>assistant<|end_header_id|>' + '\n',
+        'tool_start': '',
+        'tool_end': '<|eot_id|><|start_header_id|>assistant<|end_header_id|>'
+    }

examples/function-calling/llama-cli-function-runner.py

@@ -10,28 +10,11 @@ import re
 import json
 
 import functions
-from function_tool import get_function_tool_json, generate_schema_from_functions
+from function_tool import get_function_tool_json, get_chat_tool_format
 
 function_name_list = [ name for name in dir(functions) if not name.startswith('_') ]
 function_lookup = { name: getattr(functions, name) for name in function_name_list }
 tools = [ get_function_tool_json(f) for (n, f) in function_lookup.items() ]
-function_schema = generate_schema_from_functions(tools)
-
-prompt = """<|start_header_id|>system<|end_header_id|>
-
-You are capable of executing available function(s) if required.
-Execute function(s) as needed.
-The function calls are not shown in the conversation and should be called covertly to answer questions.
-Ask for the required input to:recipient==all
-Use JSON for function arguments.
-Respond in this format:
->>>${recipient}
-${content}
-Available functions:
-""" + function_schema + """<|eot_id|><|start_header_id|>system<|end_header_id|>
-
-When you send a message containing Python code to python, it will be executed in a stateful Jupyter notebook environment. python will respond with the output of the execution or time out after 60.0 seconds. The drive at '/mnt/data' can be used to save and persist user files.<|eot_id|><|start_header_id|>user<|end_header_id|>
-"""
 
 def main():
     import argparse
@@ -39,13 +22,17 @@ def main():
     parser = argparse.ArgumentParser(epilog='For more options: llama-cli --help')
     parser.add_argument('--display-prompt', action=argparse.BooleanOptionalAction, default=False)
     parser.add_argument('--special', action=argparse.BooleanOptionalAction, default=False)
-    parser.add_argument('--reverse-prompt', type=str, default='<|start_header_id|>user<|end_header_id|>\n')
+    parser.add_argument('--reverse-prompt', type=str)
     parser.add_argument('--ctx-size', type=int, default=1024)
     args, other_args = parser.parse_known_args()
 
-    if args.display_prompt: print(prompt)
+    tool_format = get_chat_tool_format(args, tools)
+    if args.reverse_prompt is None: args.reverse_prompt = tool_format['user_start']
 
-    command = [ './llama-cli', '-i', '-p', prompt, '--reverse-prompt', args.reverse_prompt, '--escape', '--special', '--no-display-prompt', '--log-disable', '--simple-io', '--ctx-size',  str(args.ctx_size), *other_args]
+    if args.display_prompt: print(tool_format['prompt'])
+
+    command = [ './llama-cli', '-i', '-p', tool_format['prompt'], '--reverse-prompt', args.reverse_prompt, '--escape', '--special', '--no-display-prompt', '--log-disable', '--simple-io', '--ctx-size',  str(args.ctx_size), *other_args]
+    print("'" + "' '".join(command) + "'")
 
     process = subprocess.Popen(
         command,
@@ -57,14 +44,14 @@ def main():
     if process.stdout is not None: os.set_blocking(process.stdout.fileno(), False)
 
     try:
-        run_loop(process, args)
+        run_loop(process, args, tool_format)
     except KeyboardInterrupt:
         print("\nInterrupted by user.")
     finally:
         process.terminate()
         process.wait()
 
-def run_loop(process, args):
+def run_loop(process, args, tool_format):
     pbuffer = ''
     skip_output_until_result = False
     while True:
@@ -76,29 +63,32 @@ def run_loop(process, args):
                 if not pdata: continue
                 pbuffer += pdata
 
-                if(match := re.search(r'>>>([^\n]*)\n(.*)<\|eot_id\|>', pbuffer, re.S)):
+                if(match := re.search(tool_format['function_re'], pbuffer, re.S)):
                     if not args.special:
                         pdata = pdata[:match.pos]
                     pbuffer = ''
                     skip_output_until_result = False
-
+                    try:
+                        if 1 < len(match.groups()):
                             tool_name = match.group(1)
-                    tool_args = match.group(2)
+                            tool_args = json.loads(match.group(2))
+                        else:
+                            tool = json.loads(match.group(1))
+                            tool_name = tool['name']
+                            tool_args = tool['arguments']
 
                         if tool_name == 'python':
                             result = functions._run_python(tool_args);
                         else:
-                        try:
-                            tool_args = json.loads(tool_args)
                             result = function_lookup[tool_name](**tool_args)
                     except ValueError as e:
                         result = {'error': 'unknown'}
 
-                    result = json.dumps(result) + '<|eot_id|><|start_header_id|>assistant<|end_header_id|>'
+                    result = tool_format['tool_start'] + json.dumps(result) + tool_format['tool_end']
                     process.stdin.write(result + '\n')
                     process.stdin.flush()
                     if(args.special): pdata += '\n' + result
-                elif (n := pdata.find('>>>')) >= 0:
+                elif (n := pdata.find(tool_format['function_marker'])) >= 0:
                     if not args.special:
                         pdata = pdata[:n]
                         skip_output_until_result = True
@@ -114,7 +104,7 @@ def run_loop(process, args):
                 user_input = sys.stdin.readline()
                 if user_input:
                     user_input = user_input.rstrip()
-                    process.stdin.write(user_input + '<|eot_id|><|start_header_id|>assistant<|end_header_id|>' + '\n')
+                    process.stdin.write(user_input + tool_format['user_end'] + '\n')
                     process.stdin.flush()
 
 if __name__ == '__main__':