| |
| """ |
| Lightweight BFCL evaluator for local Ollama models. |
| Tests function calling accuracy on BFCL v3 simple/multiple/parallel categories. |
| |
| Usage: |
| python3 run_bfcl.py --model qwen3.5:9b --categories simple,multiple,parallel --limit 50 |
| """ |
|
|
| import argparse |
| import json |
| import os |
| import re |
| import sys |
| import time |
| import requests |
|
|
| SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) |
| REPO_DIR = os.path.dirname(SCRIPT_DIR) |
| RESULTS_DIR = os.environ.get('RESULTS_DIR', os.path.join(REPO_DIR, 'results', 'bfcl')) |
|
|
| |
| def _find_bfcl_data(): |
| """Find or download BFCL v3 dataset.""" |
| |
| hf_cache = os.path.expanduser('~/.cache/huggingface/hub') |
| ds_dir = os.path.join(hf_cache, 'datasets--gorilla-llm--Berkeley-Function-Calling-Leaderboard') |
| if os.path.isdir(ds_dir): |
| snaps = os.path.join(ds_dir, 'snapshots') |
| if os.path.isdir(snaps): |
| for d in os.listdir(snaps): |
| candidate = os.path.join(snaps, d) |
| if os.path.isfile(os.path.join(candidate, 'BFCL_v3_simple.json')): |
| return candidate |
| |
| try: |
| from datasets import load_dataset |
| ds = load_dataset('gorilla-llm/Berkeley-Function-Calling-Leaderboard', trust_remote_code=True) |
| |
| if os.path.isdir(snaps): |
| for d in os.listdir(snaps): |
| candidate = os.path.join(snaps, d) |
| if os.path.isfile(os.path.join(candidate, 'BFCL_v3_simple.json')): |
| return candidate |
| except ImportError: |
| pass |
| print('ERROR: BFCL dataset not found. Install datasets: pip install datasets', file=sys.stderr) |
| print('Then run: python -c "from datasets import load_dataset; load_dataset(\'gorilla-llm/Berkeley-Function-Calling-Leaderboard\', trust_remote_code=True)"', file=sys.stderr) |
| sys.exit(1) |
|
|
| DATA_DIR = os.environ.get('BFCL_DATA_DIR') or _find_bfcl_data() |
|
|
| BACKEND_CONFIG = { |
| 'ollama': {'url': 'http://localhost:11434/api/chat', 'type': 'ollama'}, |
| 'llama-cpp': {'url': 'http://localhost:8081/v1/chat/completions', 'type': 'openai'}, |
| 'mlx-vlm': {'url': 'http://localhost:8080/chat/completions', 'type': 'openai'}, |
| } |
|
|
| SYSTEM_PROMPT = """You are a helpful assistant that can call functions. |
| When the user asks you to do something, respond with ONLY a function call in this exact JSON format: |
| [{"name": "function_name", "arguments": {"param1": "value1", "param2": "value2"}}] |
| |
| If multiple function calls are needed, include all of them in the list. |
| Output ONLY the JSON list. No explanation, no markdown, no extra text.""" |
|
|
| CATEGORY_FILES = { |
| 'simple': ('BFCL_v3_simple.json', 'possible_answer/BFCL_v3_simple.json'), |
| 'multiple': ('BFCL_v3_multiple.json', 'possible_answer/BFCL_v3_multiple.json'), |
| 'parallel': ('BFCL_v3_parallel.json', 'possible_answer/BFCL_v3_parallel.json'), |
| 'parallel_multiple': ('BFCL_v3_parallel_multiple.json', 'possible_answer/BFCL_v3_parallel_multiple.json'), |
| 'irrelevance': ('BFCL_v3_irrelevance.json', None), |
| } |
|
|
|
|
| def load_jsonl(path): |
| with open(path) as f: |
| return [json.loads(line) for line in f if line.strip()] |
|
|
|
|
| def format_functions(functions): |
| """Format function definitions for the prompt.""" |
| lines = [] |
| for fn in functions: |
| name = fn.get('name', '') |
| desc = fn.get('description', '') |
| params = fn.get('parameters', {}) |
| props = params.get('properties', {}) |
| required = params.get('required', []) |
|
|
| param_strs = [] |
| for pname, pinfo in props.items(): |
| ptype = pinfo.get('type', 'any') |
| pdesc = pinfo.get('description', '') |
| req = '(required)' if pname in required else '(optional)' |
| param_strs.append(' - %s: %s %s %s' % (pname, ptype, req, pdesc[:60])) |
|
|
| lines.append('%s: %s' % (name, desc)) |
| if param_strs: |
| lines.append(' Parameters:') |
| lines.extend(param_strs) |
| lines.append('') |
| return '\n'.join(lines) |
|
|
|
|
| def call_llm(model, messages, backend='ollama'): |
| cfg = BACKEND_CONFIG[backend] |
| if cfg['type'] == 'ollama': |
| payload = { |
| 'model': model, |
| 'messages': messages, |
| 'stream': False, |
| 'think': False, |
| 'options': {'temperature': 0, 'num_predict': 1024}, |
| } |
| else: |
| payload = { |
| 'model': model, |
| 'messages': messages, |
| 'temperature': 0, |
| 'max_tokens': 1024, |
| } |
| try: |
| resp = requests.post(cfg['url'], json=payload, timeout=180) |
| resp.raise_for_status() |
| data = resp.json() |
| if cfg['type'] == 'ollama': |
| content = data.get('message', {}).get('content', '') |
| tokens = data.get('eval_count', 0) + data.get('prompt_eval_count', 0) |
| else: |
| content = data.get('choices', [{}])[0].get('message', {}).get('content', '') |
| usage = data.get('usage', {}) |
| tokens = usage.get('total_tokens', 0) |
| return content, tokens |
| except Exception as e: |
| return str(e), 0 |
|
|
|
|
| def parse_function_calls(text): |
| """Extract function calls from model output.""" |
| |
| text = re.sub(r'<think>.*?</think>', '', text, flags=re.DOTALL) |
| text = re.sub(r'<think>.*', '', text, flags=re.DOTALL) |
| text = re.sub(r'Thinking Process:.*?(?=\[|\{|$)', '', text, flags=re.DOTALL) |
| text = text.strip() |
|
|
| |
| try: |
| result = json.loads(text) |
| if isinstance(result, list): |
| return result |
| if isinstance(result, dict): |
| return [result] |
| except json.JSONDecodeError: |
| pass |
|
|
| |
| m = re.search(r'```(?:json)?\s*([\[\{].*?[\]\}])\s*```', text, re.DOTALL) |
| if m: |
| try: |
| result = json.loads(m.group(1)) |
| if isinstance(result, list): |
| return result |
| if isinstance(result, dict): |
| return [result] |
| except json.JSONDecodeError: |
| pass |
|
|
| |
| m = re.search(r'\[.*\]', text, re.DOTALL) |
| if m: |
| try: |
| result = json.loads(m.group()) |
| if isinstance(result, list): |
| return result |
| except json.JSONDecodeError: |
| pass |
|
|
| |
| m = re.search(r'\{.*\}', text, re.DOTALL) |
| if m: |
| try: |
| result = json.loads(m.group()) |
| if isinstance(result, dict): |
| return [result] |
| except json.JSONDecodeError: |
| pass |
|
|
| return [] |
|
|
|
|
| def check_function_call(predicted, ground_truth): |
| """Check if predicted function call matches ground truth. |
| |
| ground_truth format: {"func_name": {"param": [accepted_values]}} |
| predicted format: {"name": "func_name", "arguments": {"param": value}} |
| """ |
| if not predicted or not ground_truth: |
| return False |
|
|
| |
| return match_call_set(predicted, ground_truth) |
|
|
|
|
| def match_call_set(predicted_calls, gt_calls): |
| """Match a set of predicted calls against ground truth calls.""" |
| if not isinstance(gt_calls, list): |
| gt_calls = [gt_calls] |
|
|
| if len(predicted_calls) != len(gt_calls): |
| return False |
|
|
| |
| matched_gt = set() |
| for pred in predicted_calls: |
| pred_name = pred.get('name', '') |
| pred_args = pred.get('arguments', {}) |
|
|
| found = False |
| for i, gt in enumerate(gt_calls): |
| if i in matched_gt: |
| continue |
| |
| for gt_name, gt_params in gt.items(): |
| if pred_name == gt_name: |
| if match_params(pred_args, gt_params): |
| matched_gt.add(i) |
| found = True |
| break |
| if found: |
| break |
| if not found: |
| return False |
|
|
| return len(matched_gt) == len(gt_calls) |
|
|
|
|
| def match_params(pred_args, gt_params): |
| """Check if predicted params match ground truth accepted values.""" |
| for param_name, accepted_values in gt_params.items(): |
| if param_name not in pred_args: |
| |
| if accepted_values and accepted_values != ['']: |
| return False |
| continue |
|
|
| pred_val = pred_args[param_name] |
|
|
| |
| matched = False |
| for av in accepted_values: |
| if av == '': |
| matched = True |
| break |
| |
| if str(pred_val) == str(av): |
| matched = True |
| break |
| try: |
| if float(pred_val) == float(av): |
| matched = True |
| break |
| except (ValueError, TypeError): |
| pass |
| |
| if str(pred_val).lower() == str(av).lower(): |
| matched = True |
| break |
|
|
| if not matched: |
| return False |
|
|
| return True |
|
|
|
|
| def run_category(model, category, limit=50, backend='ollama'): |
| """Run BFCL evaluation for a category.""" |
| test_file, answer_file = CATEGORY_FILES[category] |
| tests = load_jsonl(os.path.join(DATA_DIR, test_file)) |
|
|
| if answer_file: |
| answers = load_jsonl(os.path.join(DATA_DIR, answer_file)) |
| answer_map = {a['id']: a['ground_truth'] for a in answers} |
| else: |
| answer_map = {} |
|
|
| if limit and limit < len(tests): |
| tests = tests[:limit] |
|
|
| print('\n=== %s: %s (%d tests) ===' % (model, category, len(tests))) |
|
|
| correct = 0 |
| total = 0 |
| parse_failures = 0 |
| total_tokens = 0 |
| total_time = 0 |
|
|
| for i, test in enumerate(tests): |
| test_id = test.get('id', str(i)) |
| question = test['question'][0] |
| functions = test.get('function', []) |
|
|
| |
| user_msg = question[-1]['content'] if isinstance(question, list) else question['content'] |
| func_desc = format_functions(functions) |
|
|
| messages = [ |
| {'role': 'system', 'content': SYSTEM_PROMPT + '\n\nAvailable functions:\n' + func_desc}, |
| {'role': 'user', 'content': user_msg}, |
| ] |
|
|
| t0 = time.time() |
| response, tokens = call_llm(model, messages, backend) |
| elapsed = time.time() - t0 |
| total_tokens += tokens |
| total_time += elapsed |
|
|
| predicted = parse_function_calls(response) |
| gt = answer_map.get(test_id) |
|
|
| if not predicted: |
| parse_failures += 1 |
| passed = False |
| elif gt: |
| passed = check_function_call(predicted, gt) |
| else: |
| |
| passed = len(predicted) == 0 |
|
|
| total += 1 |
| if passed: |
| correct += 1 |
|
|
| if i < 3 or (not passed and i < 10): |
| pred_short = json.dumps(predicted)[:80] if predicted else '(none)' |
| gt_short = json.dumps(gt)[:80] if gt else '(irrelevance)' |
| status = 'PASS' if passed else 'FAIL' |
| print(' [%d] %s | pred: %s' % (i, status, pred_short)) |
| if not passed: |
| print(' gt: %s' % gt_short) |
|
|
| if (i + 1) % 25 == 0: |
| print(' ... %d/%d done (%.0f%% correct)' % (i + 1, total, 100 * correct / total)) |
|
|
| accuracy = correct / total if total else 0 |
| avg_time = total_time / total if total else 0 |
| avg_tokens = total_tokens / total if total else 0 |
|
|
| print('\n Results: %d/%d correct (%.1f%%)' % (correct, total, 100 * accuracy)) |
| print(' Parse failures: %d' % parse_failures) |
| print(' Avg time: %.2fs, Avg tokens: %d' % (avg_time, avg_tokens)) |
|
|
| return { |
| 'model': model, |
| 'category': category, |
| 'correct': correct, |
| 'total': total, |
| 'accuracy': round(accuracy, 4), |
| 'parse_failures': parse_failures, |
| 'avg_time': round(avg_time, 2), |
| 'avg_tokens': round(avg_tokens), |
| } |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser() |
| parser.add_argument('--model', required=True) |
| parser.add_argument('--backend', default='ollama', choices=['ollama', 'llama-cpp', 'mlx-vlm']) |
| parser.add_argument('--categories', default='simple,multiple,parallel') |
| parser.add_argument('--limit', type=int, default=50, help='Max tests per category') |
| args = parser.parse_args() |
|
|
| categories = [c.strip() for c in args.categories.split(',')] |
|
|
| results = [] |
| for cat in categories: |
| if cat not in CATEGORY_FILES: |
| print('Unknown category: %s' % cat) |
| continue |
| r = run_category(args.model, cat, args.limit, args.backend) |
| results.append(r) |
|
|
| |
| print('\n' + '=' * 60) |
| print('BFCL SUMMARY: %s' % args.model) |
| print('=' * 60) |
| print('%-20s %8s %8s %8s' % ('Category', 'Score', 'Accuracy', 'AvgTime')) |
| print('-' * 50) |
| for r in results: |
| print('%-20s %3d/%-4d %7.1f%% %7.2fs' % ( |
| r['category'], r['correct'], r['total'], 100 * r['accuracy'], r['avg_time'])) |
|
|
| |
| model_tag = args.model.replace(':', '_').replace('/', '_') |
| os.makedirs(RESULTS_DIR, exist_ok=True) |
| backend_suffix = '_%s' % args.backend if args.backend != 'ollama' else '' |
| outfile = os.path.join(RESULTS_DIR, 'bfcl_%s%s.json' % (model_tag, backend_suffix)) |
| with open(outfile, 'w') as f: |
| json.dump(results, f, indent=2) |
| print('\nSaved to: %s' % outfile) |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|
