Courier vs Ollama: Tool-Calling, Measured

#!/usr/bin/env python
"""FC (function-calling) harness for BFCL against any OpenAI-compatible endpoint.

WHY THIS EXISTS
---------------
BFCL's built-in Gemma handler treats Gemma as a PROMPT model (tools embedded as
text, output parsed by a Gemma-specific parser). That is not what we want to
measure. Courier's value is that it exposes NATIVE `tool_calls` with FSM-verified
validity on /v1/chat/completions — turning a model that isn't natively
tool-trained into a reliable tool-caller. To measure THAT, we run the FC path:
send OpenAI `tools`, read back `tool_calls`, and score with BFCL's OWN AST
checker. Same model, same endpoint contract on both runtimes; the runtime's
tool-calling layer is the only variable.

Scoring is BFCL's official `ast_checker` — NOT a reimplementation — so nobody
can say we graded ourselves. We only supply the transport (call each endpoint)
and the format conversion (OpenAI tool_calls -> BFCL's decoded shape).

Run once per endpoint (Courier :9100, Ollama :11434), same category, compare.
"""
import argparse
import json
import os
import time

import requests

from bfcl_eval.constants.enums import Language
from bfcl_eval.eval_checker.ast_eval.ast_checker import ast_checker

_DATA = "bfcl-env/lib/python3.10/site-packages/bfcl_eval/data"

# BFCL function schemas use Gorilla's type names; the OpenAI tools API wants
# JSON-schema names. Map the ones that differ; leave the rest.
_TYPE_MAP = {"dict": "object", "float": "number", "tuple": "array", "any": "string"}


def _convert_schema(node):
    """Recursively rewrite Gorilla type names to JSON-schema names in place."""
    if isinstance(node, dict):
        if "type" in node and isinstance(node["type"], str):
            node["type"] = _TYPE_MAP.get(node["type"], node["type"])
        for v in node.values():
            _convert_schema(v)
    elif isinstance(node, list):
        for v in node:
            _convert_schema(v)
    return node


def _to_openai_tools(functions):
    tools = []
    for fn in functions:
        params = _convert_schema(json.loads(json.dumps(fn.get("parameters", {}))))
        if params.get("type") != "object":
            params["type"] = "object"
        tools.append({"type": "function", "function": {
            "name": fn["name"], "description": fn.get("description", ""), "parameters": params,
        }})
    return tools


def _decode_tool_calls(message):
    """OpenAI tool_calls -> BFCL decoded shape: [{func_name: {arg: value}}]."""
    out = []
    for tc in (message.get("tool_calls") or []):
        fn = tc.get("function", {})
        name = fn.get("name", "")
        try:
            args = json.loads(fn.get("arguments") or "{}")
        except (json.JSONDecodeError, TypeError):
            args = {"__unparseable__": fn.get("arguments")}
        out.append({name: args})
    return out


def _call(base_url, model, api_key, messages, tools, timeout):
    headers = {"Content-Type": "application/json"}
    if api_key:
        headers["Authorization"] = f"Bearer {api_key}"
    body = {"model": model, "messages": messages, "tools": tools,
            "tool_choice": "auto", "temperature": 0.0, "stream": False}
    r = requests.post(f"{base_url}/chat/completions", headers=headers,
                      json=body, timeout=timeout)
    r.raise_for_status()
    return r.json()["choices"][0]["message"]


def run(category, base_url, model, api_key, limit, timeout, out_path):
    with open(f"{_DATA}/BFCL_v4_{category}.json") as f:
        cases = [json.loads(l) for l in f if l.strip()]
    with open(f"{_DATA}/possible_answer/BFCL_v4_{category}.json") as f:
        answers = {json.loads(l)["id"]: json.loads(l) for l in f if l.strip()}
    if limit:
        cases = cases[:limit]

    # A registered model name so ast_checker's convert_func_name lookup resolves;
    # gemma-3-27b is underscore_to_dot=False so it never mangles names.
    score_name = "google/gemma-3-27b-it"
    lang = Language.JAVA if "java" in category else (
        Language.JAVASCRIPT if "javascript" in category else Language.PYTHON)

    results, correct, errors, latencies = [], 0, 0, []
    for i, case in enumerate(cases):
        cid = case["id"]
        tools = _to_openai_tools(case["function"])
        messages = case["question"][0]  # single-turn: first turn's message list
        rec = {"id": cid}
        try:
            t0 = time.time()
            msg = _call(base_url, model, api_key, messages, tools, timeout)
            latencies.append(time.time() - t0)
            decoded = _decode_tool_calls(msg)
            verdict = ast_checker(case["function"], decoded, answers[cid]["ground_truth"],
                                  lang, category, score_name)
            rec["valid"] = bool(verdict.get("valid"))
            rec["n_calls"] = len(decoded)
            if not rec["valid"]:
                rec["error"] = verdict.get("error")
            correct += rec["valid"]
        except Exception as e:  # transport/parse failure counts as a miss, recorded
            errors += 1
            rec["valid"] = False
            rec["exception"] = str(e)[:200]
        results.append(rec)
        print(f"  [{i+1}/{len(cases)}] {cid}: {'OK' if rec['valid'] else 'x'}", flush=True)

    n = len(cases)
    summary = {
        "category": category, "endpoint": base_url, "model": model,
        "n": n, "correct": correct, "accuracy": round(correct / n, 4) if n else 0.0,
        "transport_errors": errors,
        "median_latency_s": round(sorted(latencies)[len(latencies)//2], 2) if latencies else None,
    }
    with open(out_path, "w") as f:
        json.dump({"summary": summary, "results": results}, f, indent=2)
    print(json.dumps(summary, indent=2))
    return summary


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--category", default="simple_python")
    ap.add_argument("--base-url", required=True, help="e.g. http://127.0.0.1:9100/v1")
    ap.add_argument("--model", required=True)
    ap.add_argument("--api-key", default=os.getenv("BFCL_ENDPOINT_KEY", ""))
    ap.add_argument("--limit", type=int, default=0)
    ap.add_argument("--timeout", type=int, default=300)
    ap.add_argument("--out", required=True)
    a = ap.parse_args()
    run(a.category, a.base_url, a.model, a.api_key, a.limit, a.timeout, a.out)