Source files / scripts · raw
tool-test.py
#!/usr/bin/env python3 """Tool-calling and agent-turn checks against an OpenAI-compatible server. tool-test.py [--port 8080] [--label NAME] [--repeat 3] Each case is a short conversation with tool definitions; the check looks at the tool calls (name, JSON arguments) or the final text. Thinking stays on, as in agent use. Cases run --repeat times to show consistency. Appends one JSON line per case and run to --out and prints a summary. """ import argparse, json, re, time, urllib.request ap = argparse.ArgumentParser() ap.add_argument("--port", type=int, default=8080) ap.add_argument("--label", default="") ap.add_argument("--repeat", type=int, default=3) ap.add_argument("--max-tokens", type=int, default=8192) ap.add_argument("--out", default="/opt/tersler/bench/tools.jsonl") a = ap.parse_args() URL = f"http://127.0.0.1:{a.port}/v1/chat/completions" def fn(name, desc, props, req): return {"type": "function", "function": {"name": name, "description": desc, "parameters": {"type": "object", "properties": props, "required": req}}} WEATHER = fn("get_weather", "Current weather for a city.", {"city": {"type": "string"}}, ["city"]) CALC = fn("calculator", "Evaluate an arithmetic expression exactly.", {"expression": {"type": "string"}}, ["expression"]) READ = fn("read_file", "Read a text file from the project.", {"path": {"type": "string"}}, ["path"]) EVENT = fn("create_event", "Create a calendar event.", {"title": {"type": "string"}, "start": {"type": "string", "description": "ISO 8601 local time"}, "duration_min": {"type": "integer"}}, ["title", "start", "duration_min"]) ALL = [WEATHER, CALC, READ, EVENT] def call(messages, tools): body = {"messages": messages, "tools": tools, "max_tokens": a.max_tokens, "temperature": 0.6, "top_p": 0.95} req = urllib.request.Request(URL, json.dumps(body).encode(), {"Content-Type": "application/json"}) t0 = time.time() with urllib.request.urlopen(req, timeout=1800) as r: out = json.load(r) m = out["choices"][0]["message"] calls = [] for c in m.get("tool_calls") or []: try: args = json.loads(c["function"]["arguments"]) except (json.JSONDecodeError, TypeError): args = {"_invalid_json": c["function"].get("arguments")} calls.append((c["function"]["name"], args, c.get("id", ""))) return m, calls, time.time() - t0 def user(t): return {"role": "user", "content": t} def tool_turn(m, calls, results): """The assistant turn with its tool calls, then one tool message per call.""" msgs = [{"role": "assistant", "content": m.get("content") or "", "tool_calls": m["tool_calls"]}] for (name, args, cid), res in zip(calls, results): msgs.append({"role": "tool", "tool_call_id": cid, "content": json.dumps(res)}) return msgs def case_weather(): m, c, _ = call([user("What's the weather in Hamburg right now?")], ALL) return len(c) == 1 and c[0][0] == "get_weather" and "hamburg" in c[0][1].get("city", "").lower(), c def case_no_tool(): m, c, _ = call([user("What is the capital of France? One word.")], ALL) return not c and "paris" in (m.get("content") or "").lower(), m.get("content") def case_pick_tool(): m, c, _ = call([user("Show me what's in docs/changelog.md.")], ALL) return len(c) == 1 and c[0][0] == "read_file" and c[0][1].get("path", "").endswith("docs/changelog.md"), c def case_args(): m, c, _ = call([user("Today is Tuesday 2026-11-10. Book 'Dentist' tomorrow at 9:30 for 45 minutes.")], ALL) ok = (len(c) == 1 and c[0][0] == "create_event" and c[0][1].get("duration_min") == 45 and str(c[0][1].get("start", "")).startswith("2026-11-11T09:30") and "dentist" in c[0][1].get("title", "").lower()) return ok, c def case_parallel(): m, c, _ = call([user("Compare the current weather in Berlin and in Paris.")], ALL) cities = sorted(x[1].get("city", "").lower() for x in c if x[0] == "get_weather") return any("berlin" in x for x in cities) and any("paris" in x for x in cities), c def case_use_result(): msgs = [user("What's the weather in Hamburg right now?")] m, c, _ = call(msgs, ALL) if not c: return False, "no tool call" msgs += tool_turn(m, c, [{"city": "Hamburg", "temp_c": 7.4, "condition": "drizzle", "wind_kmh": 31}] * len(c)) m2, c2, _ = call(msgs, ALL) t = m2.get("content") or "" return "7.4" in t and "drizzle" in t.lower(), t[-300:] def case_error_recovery(): msgs = [user("What's the weather in Hamburgg?")] m, c, _ = call(msgs, ALL) if not c: return False, "no tool call" msgs += tool_turn(m, c, [{"error": "unknown city 'Hamburgg'. Did you mean 'Hamburg'?"}] * len(c)) m2, c2, _ = call(msgs, ALL) # a retry with the corrected name counts, also as "Hamburg, Germany"; repeating the typo does not retried = any(x[0] == "get_weather" and "hamburg" in x[1].get("city", "").lower() and "hamburgg" not in x[1].get("city", "").lower() for x in c2) asked = not c2 and "hamburg" in (m2.get("content") or "").lower() return retried or asked, c2 or (m2.get("content") or "")[-200:] # A money task like Junk GPU Lab's payment reminder (Strata got it wrong 4/4 there). No tools needed; # the calculator is offered. 3 x 249.90 + 5 % twice + 4.50 x (2 + 4) started weeks = 801.69. INVOICE = ("Our rent invoices: January 249.90 EUR (paid on time), February 249.90 EUR (paid 9 days late), " "March 249.90 EUR (paid 23 days late). Late payment costs a fee of 5% of the invoice amount, plus " "4.50 EUR for every started 7-day period of delay. What did we pay in total for the three months, " "including all fees? Give the final total in EUR with two decimals on the last line as TOTAL: <amount>.") TOTAL = 3 * 249.90 + 2 * 0.05 * 249.90 + 4.50 * (2 + 4) # Feb: 2 started weeks, Mar: 4 started weeks def case_money(): msgs = [user(INVOICE)] for _ in range(6): # allow calculator round trips m, c, _ = call(msgs, [CALC]) if not c: break res = [] for name, args, cid in c: try: res.append({"result": round(eval(args.get("expression", ""), {"__builtins__": {}}), 6)}) except Exception as e: res.append({"error": str(e)}) msgs += tool_turn(m, c, res) t = m.get("content") or "" got = re.findall(r"TOTAL:\s*([0-9.,]+)", t) val = float(got[-1].replace(",", "")) if got else None return val is not None and abs(val - TOTAL) < 0.005, val CASES = [case_weather, case_no_tool, case_pick_tool, case_args, case_parallel, case_use_result, case_error_recovery, case_money] score = {} for f in CASES: for i in range(a.repeat): t0 = time.time() try: ok, detail = f() except Exception as e: ok, detail = False, f"exception: {e}" rec = {"case": f.__name__[5:], "run": i, "ok": bool(ok), "detail": str(detail)[:300], "seconds": round(time.time() - t0, 1), "label": a.label, "date": time.strftime("%Y-%m-%dT%H:%M:%S")} print(json.dumps(rec), flush=True) with open(a.out, "a") as fo: fo.write(json.dumps(rec) + "\n") score.setdefault(rec["case"], []).append(rec["ok"]) print("\nSUMMARY", a.label, f"(expected money total {TOTAL:.2f})") for k, v in score.items(): print(f" {k:16s} {sum(v)}/{len(v)}")