"""Modal H200 benchmark for HopCoder Mini 9B native tool-call LoRA adapter. Evaluates the adapter across all 8 targeted CLI tools plus general tool-call scenarios derived from xLAM-style queries. Reports per-tool accuracy, syntax validity, JSON parse rate, latency, and throughput. Run: python -m modal run --detach hopcoder_benchmark.py """ from __future__ import annotations import json import os import re import time from pathlib import Path from typing import Any, Dict, List, Optional import modal APP_NAME = "hopcoder-mini-9b-benchmark-h200" app = modal.App(APP_NAME) hf_cache_volume = modal.Volume.from_name( "hopcoder-hf-cache", create_if_missing=True, ) training_volume = modal.Volume.from_name( "hopcoder-training", create_if_missing=True, ) image = ( modal.Image.debian_slim(python_version="3.12") .apt_install("git", "curl") .uv_pip_install( "torch==2.10.0", "torchvision==0.25.0", "transformers==5.12.1", "datasets==5.0.0", "accelerate==1.14.0", "peft==0.19.1", "huggingface_hub==1.21.0", "hf-xet==1.5.1", "sentencepiece==0.2.1", "safetensors==0.7.0", "protobuf>=5,<7", "pillow>=11", "termcolor>=3", ) .env( { "HF_HOME": "/cache/huggingface", "HF_HUB_CACHE": "/cache/huggingface/hub", "HF_DATASETS_CACHE": "/cache/huggingface/datasets", "TORCH_HOME": "/cache/torch", "HF_XET_HIGH_PERFORMANCE": "1", "TOKENIZERS_PARALLELISM": "false", "PYTORCH_ALLOC_CONF": "expandable_segments:True,max_split_size_mb:256", } ) ) # --------------------------------------------------------------------------- # Benchmark cases # --------------------------------------------------------------------------- BENCHMARK_CASES: List[Dict[str, Any]] = [ # ask_user_question (5) {"query": "Before changing the database schema, ask me whether to add a nullable column, create a migration, or use a JSON field.", "expected_function": "ask_user_question", "expected_params": {"questions": True}, "category": "targeted"}, {"query": "I'm unsure about the deployment strategy. Show me interactive options for blue-green, canary, or rolling deployment.", "expected_function": "ask_user_question", "expected_params": {"questions": True}, "category": "targeted"}, {"query": "Ask me which test framework to use: pytest, unittest, or tox.", "expected_function": "ask_user_question", "expected_params": {"questions": True}, "category": "targeted"}, {"query": "I need to decide between REST and GraphQL for the new API. Prompt me with the choices.", "expected_function": "ask_user_question", "expected_params": {"questions": True}, "category": "targeted"}, {"query": "Use the question tool to clarify whether I want to cache in Redis, Memcached, or in-memory.", "expected_function": "ask_user_question", "expected_params": {"questions": True}, "category": "targeted"}, # todo_write (5) {"query": "Create a todo list: 1) audit the authentication module, 2) fix the token refresh bug, 3) write integration tests.", "expected_function": "todo_write", "expected_params": {"todos": True}, "category": "targeted"}, {"query": "Track these tasks with todo_write: review the pull request, run the CI pipeline, and deploy to staging.", "expected_function": "todo_write", "expected_params": {"todos": True}, "category": "targeted"}, {"query": "Add a structured task list for: investigate memory leak, profile the heap, and patch the allocator.", "expected_function": "todo_write", "expected_params": {"todos": True}, "category": "targeted"}, {"query": "Write these steps to the task tracker: update dependencies, rebuild the Docker image, and restart the service.", "expected_function": "todo_write", "expected_params": {"todos": True}, "category": "targeted"}, {"query": "Use todo_write to record: refactor the parser, add edge-case tests, and update the documentation.", "expected_function": "todo_write", "expected_params": {"todos": True}, "category": "targeted"}, # glob (5) {"query": "Find all TypeScript files in the src directory.", "expected_function": "glob", "expected_params": {"pattern": True}, "category": "targeted"}, {"query": "Use glob to locate all JSON config files in the project.", "expected_function": "glob", "expected_params": {"pattern": True}, "category": "targeted"}, {"query": "Search for all test files matching the pattern **/test_*.py", "expected_function": "glob", "expected_params": {"pattern": True}, "category": "targeted"}, {"query": "List all markdown files in the docs folder.", "expected_function": "glob", "expected_params": {"pattern": True}, "category": "targeted"}, {"query": "Find all files with the .yaml extension in the repository.", "expected_function": "glob", "expected_params": {"pattern": True}, "category": "targeted"}, # grep_search (5) {"query": "Search the codebase for all occurrences of 'def main(' in Python files.", "expected_function": "grep_search", "expected_params": {"pattern": True}, "category": "targeted"}, {"query": "Use grep_search to find all TODO comments in the source code.", "expected_function": "grep_search", "expected_params": {"pattern": True}, "category": "targeted"}, {"query": "Find all console.log statements in the JavaScript files.", "expected_function": "grep_search", "expected_params": {"pattern": True}, "category": "targeted"}, {"query": "Search for the regex pattern 'class.*Model' across the project.", "expected_function": "grep_search", "expected_params": {"pattern": True}, "category": "targeted"}, {"query": "Grep for 'import torch' in all Python files.", "expected_function": "grep_search", "expected_params": {"pattern": True}, "category": "targeted"}, # edit (5) {"query": "Edit /home/user/app.py to replace 'port = 8080' with 'port = 3000'.", "expected_function": "edit", "expected_params": {"file_path": True, "old_string": True, "new_string": True}, "category": "targeted"}, {"query": "Use edit to change 'DEBUG = False' to 'DEBUG = True' in /home/user/config.py.", "expected_function": "edit", "expected_params": {"file_path": True, "old_string": True, "new_string": True}, "category": "targeted"}, {"query": "Modify /home/user/utils.py: replace 'return x + y' with 'return x + y # sum helper'.", "expected_function": "edit", "expected_params": {"file_path": True, "old_string": True, "new_string": True}, "category": "targeted"}, {"query": "Edit the file /home/user/README.md to add a new section after '## Installation'.", "expected_function": "edit", "expected_params": {"file_path": True, "old_string": True, "new_string": True}, "category": "targeted"}, {"query": "Use edit to update the version string in /home/user/setup.py from 1.0.0 to 1.0.1.", "expected_function": "edit", "expected_params": {"file_path": True, "old_string": True, "new_string": True}, "category": "targeted"}, # run_shell_command (5) {"query": "Run 'npm run build' to build the project.", "expected_function": "run_shell_command", "expected_params": {"command": True, "description": True}, "category": "targeted"}, {"query": "Execute 'python -m pytest tests/' to run all tests.", "expected_function": "run_shell_command", "expected_params": {"command": True, "description": True}, "category": "targeted"}, {"query": "Use run_shell_command to check git status.", "expected_function": "run_shell_command", "expected_params": {"command": True, "description": True}, "category": "targeted"}, {"query": "Run 'docker build -t myapp .' to build the Docker image.", "expected_function": "run_shell_command", "expected_params": {"command": True, "description": True}, "category": "targeted"}, {"query": "Execute 'make clean' to remove build artifacts.", "expected_function": "run_shell_command", "expected_params": {"command": True, "description": True}, "category": "targeted"}, # general xLAM-style (10) {"query": "What's the weather like in San Francisco? Use the weather tool.", "expected_function": "get_weather", "expected_params": {"location": True}, "category": "general"}, {"query": "Search for flights from New York to London departing next Friday.", "expected_function": "search_flights", "expected_params": {"origin": True, "destination": True}, "category": "general"}, {"query": "Calculate the monthly mortgage payment for a $400,000 loan at 6.5% APR over 30 years.", "expected_function": "calculate_mortgage", "expected_params": {"loan_amount": True, "interest_rate": True}, "category": "general"}, {"query": "Send an email to john@example.com with the subject 'Meeting Tomorrow' and body 'See you at 10am'.", "expected_function": "send_email", "expected_params": {"to": True, "subject": True}, "category": "general"}, {"query": "Book a table at the Italian restaurant for 4 people at 7pm on Friday.", "expected_function": "book_restaurant", "expected_params": {"party_size": True, "time": True}, "category": "general"}, {"query": "Get the current stock price for AAPL.", "expected_function": "get_stock_price", "expected_params": {"symbol": True}, "category": "general"}, {"query": "Create a calendar event titled 'Team Standup' from 9am to 9:30am tomorrow.", "expected_function": "create_event", "expected_params": {"title": True, "start_time": True}, "category": "general"}, {"query": "Translate 'Hello, how are you?' from English to Spanish.", "expected_function": "translate_text", "expected_params": {"text": True, "source_lang": True, "target_lang": True}, "category": "general"}, {"query": "Get directions from 123 Main St to 456 Oak Ave avoiding highways.", "expected_function": "get_directions", "expected_params": {"origin": True, "destination": True}, "category": "general"}, {"query": "Set a reminder to call the dentist on Tuesday at 2pm.", "expected_function": "set_reminder", "expected_params": {"task": True, "time": True}, "category": "general"}, ] GENERAL_TOOLS: List[Dict[str, Any]] = [ {"type": "function", "function": {"name": "get_weather", "description": "Get the current weather for a location.", "parameters": {"type": "object", "properties": {"location": {"type": "string", "description": "City or coordinates"}, "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}}, "required": ["location"]}}}, {"type": "function", "function": {"name": "search_flights", "description": "Search for available flights between two cities.", "parameters": {"type": "object", "properties": {"origin": {"type": "string"}, "destination": {"type": "string"}, "date": {"type": "string"}, "passengers": {"type": "integer"}}, "required": ["origin", "destination"]}}}, {"type": "function", "function": {"name": "calculate_mortgage", "description": "Calculate monthly mortgage payment.", "parameters": {"type": "object", "properties": {"loan_amount": {"type": "number"}, "interest_rate": {"type": "number"}, "term_years": {"type": "integer"}}, "required": ["loan_amount", "interest_rate", "term_years"]}}}, {"type": "function", "function": {"name": "send_email", "description": "Send an email to a recipient.", "parameters": {"type": "object", "properties": {"to": {"type": "string"}, "subject": {"type": "string"}, "body": {"type": "string"}}, "required": ["to", "subject", "body"]}}}, {"type": "function", "function": {"name": "book_restaurant", "description": "Book a restaurant reservation.", "parameters": {"type": "object", "properties": {"restaurant_name": {"type": "string"}, "party_size": {"type": "integer"}, "time": {"type": "string"}, "date": {"type": "string"}}, "required": ["party_size", "time"]}}}, {"type": "function", "function": {"name": "get_stock_price", "description": "Get the current stock price for a ticker symbol.", "parameters": {"type": "object", "properties": {"symbol": {"type": "string"}}, "required": ["symbol"]}}}, {"type": "function", "function": {"name": "create_event", "description": "Create a calendar event.", "parameters": {"type": "object", "properties": {"title": {"type": "string"}, "start_time": {"type": "string"}, "end_time": {"type": "string"}}, "required": ["title", "start_time"]}}}, {"type": "function", "function": {"name": "translate_text", "description": "Translate text between languages.", "parameters": {"type": "object", "properties": {"text": {"type": "string"}, "source_lang": {"type": "string"}, "target_lang": {"type": "string"}}, "required": ["text", "source_lang", "target_lang"]}}}, {"type": "function", "function": {"name": "get_directions", "description": "Get driving directions between two addresses.", "parameters": {"type": "object", "properties": {"origin": {"type": "string"}, "destination": {"type": "string"}, "avoid_highways": {"type": "boolean"}}, "required": ["origin", "destination"]}}}, {"type": "function", "function": {"name": "set_reminder", "description": "Set a reminder for a specific time.", "parameters": {"type": "object", "properties": {"task": {"type": "string"}, "time": {"type": "string"}}, "required": ["task", "time"]}}}, ] TARGET_TOOLS: List[Dict[str, Any]] = [ {"type": "function", "function": {"name": "ask_user_question", "description": "Show one or more interactive questions in the CLI.", "parameters": {"type": "object", "properties": {"questions": {"type": "array", "items": {"type": "object", "properties": {"question": {"type": "string"}, "header": {"type": "string"}, "options": {"type": "array", "items": {"type": "object", "properties": {"label": {"type": "string"}, "description": {"type": "string"}}, "required": ["label", "description"]}}}, "required": ["question", "header", "options"]}}}, "required": ["questions"]}}}, {"type": "function", "function": {"name": "todo_write", "description": "Create or update the structured task list.", "parameters": {"type": "object", "properties": {"todos": {"type": "array", "items": {"type": "object", "properties": {"content": {"type": "string"}, "id": {"type": "string"}, "status": {"type": "string", "enum": ["pending", "in_progress", "completed"]}}, "required": ["content", "id", "status"]}}}, "required": ["todos"]}}}, {"type": "function", "function": {"name": "read_file", "description": "Read a UTF-8 text file.", "parameters": {"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]}}}, {"type": "function", "function": {"name": "search_code", "description": "Search source files for a text or regex pattern.", "parameters": {"type": "object", "properties": {"query": {"type": "string"}, "path": {"type": "string"}}, "required": ["query", "path"]}}}, {"type": "function", "function": {"name": "glob", "description": "Find files by glob pattern (e.g., **/*.py).", "parameters": {"type": "object", "properties": {"pattern": {"type": "string"}, "path": {"type": "string"}}, "required": ["pattern"]}}}, {"type": "function", "function": {"name": "grep_search", "description": "Search file contents for a regex pattern.", "parameters": {"type": "object", "properties": {"pattern": {"type": "string"}, "path": {"type": "string"}, "glob": {"type": "string"}}, "required": ["pattern"]}}}, {"type": "function", "function": {"name": "edit", "description": "Replace text in a file with new content.", "parameters": {"type": "object", "properties": {"file_path": {"type": "string"}, "old_string": {"type": "string"}, "new_string": {"type": "string"}, "replace_all": {"type": "boolean"}}, "required": ["file_path", "old_string", "new_string"]}}}, {"type": "function", "function": {"name": "run_shell_command", "description": "Execute a shell command and return output.", "parameters": {"type": "object", "properties": {"command": {"type": "string"}, "description": {"type": "string"}, "is_background": {"type": "boolean"}, "timeout": {"type": "integer"}}, "required": ["command", "description"]}}}, ] @app.function( image=image, gpu="H200", cpu=16.0, memory=65536, timeout=2 * 60 * 60, startup_timeout=2 * 60 * 60, retries=1, max_containers=1, volumes={ "/cache": hf_cache_volume, "/data": training_volume, }, secrets=[modal.Secret.from_name("huggingface")], ) def benchmark( adapter_dir: str = "Hopcoder-Mini-9B-Native-ToolCall-LoRA-H200", max_new_tokens: int = 384, ) -> dict[str, object]: """Benchmark the trained LoRA adapter on all 8 CLI tools plus general cases.""" import torch from peft import PeftModel from transformers import AutoModelForImageTextToText, AutoProcessor from huggingface_hub import login MODEL_ID = "TaimoorSiddiqui/Hopcoder-Mini-9B" ADAPTER_DIR = f"/data/{adapter_dir}" MAX_NEW_TOKENS = max_new_tokens HF_TOKEN = os.environ.get("HF_TOKEN") if HF_TOKEN: login(token=HF_TOKEN, add_to_git_credential=False) # Load processor and tokenizer processor = AutoProcessor.from_pretrained( MODEL_ID, trust_remote_code=True, token=HF_TOKEN, ) tokenizer = getattr(processor, "tokenizer", processor) if tokenizer.pad_token_id is None: tokenizer.pad_token = tokenizer.eos_token tokenizer.padding_side = "right" # Load base model in BF16 print("=== Loading base model (BF16) ===") compute_dtype = torch.bfloat16 model = AutoModelForImageTextToText.from_pretrained( MODEL_ID, trust_remote_code=True, token=HF_TOKEN, dtype=compute_dtype, device_map={"": 0}, low_cpu_mem_usage=True, attn_implementation="sdpa", ) model.config.use_cache = True model.eval() # Load LoRA adapter from Modal volume print(f"=== Loading LoRA adapter from {ADAPTER_DIR} ===") model = PeftModel.from_pretrained(model, ADAPTER_DIR) model.eval() model.config.use_cache = True # --- Helpers --- MAX_FUNCTION_DESCRIPTION_CHARS = 120 MAX_PARAMETER_DESCRIPTION_CHARS = 60 SYSTEM_PROMPT = ( "Use the provided tools whenever the request requires one.\n\n" "For a tool request, emit only complete native tool-call blocks. " "Never emit a function name as a top-level tag. Never leave unmatched " "parameter, function, or tool_call tags. Arrays and objects inside " "parameter blocks must be valid JSON. Do not use Markdown fences.\n\n" ) def truncate_description(value: Any, limit: int) -> Any: if not isinstance(value, str): return value value = " ".join(value.split()) return value if len(value) <= limit else value[: max(0, limit - 1)].rstrip() + "\u2026" def compact_schema_node(value: Any) -> Any: if isinstance(value, list): return [compact_schema_node(item) for item in value] if not isinstance(value, dict): return value compacted: Dict[str, Any] = {} for key, item in value.items(): if key == "description": compacted[key] = truncate_description(item, MAX_PARAMETER_DESCRIPTION_CHARS) elif key in {"examples", "example", "title", "$comment"}: continue else: compacted[key] = compact_schema_node(item) return compacted def compact_tool(tool: Dict[str, Any]) -> Dict[str, Any]: fn = tool["function"] return { "type": "function", "function": { "name": fn["name"], "description": truncate_description(fn.get("description", ""), MAX_FUNCTION_DESCRIPTION_CHARS), "parameters": compact_schema_node(fn.get("parameters", {"type": "object", "properties": {}})), }, } def schema_type_label(schema: Any) -> str: if not isinstance(schema, dict): return "any" t = schema.get("type", "any") if isinstance(t, list): return "|".join(str(i) for i in t) if t == "array": return f"array[{schema_type_label(schema.get('items', {}))}]" if t == "object": props = schema.get("properties", {}) if isinstance(props, dict) and props: keys = ",".join(list(props)[:5]) if len(props) > 5: keys += ",\u2026" return f"object{{{keys}}}" return "object" return str(t) def render_compact_tool_signatures(tools: List[Dict[str, Any]]) -> str: lines = [""] for tool in tools: fn = tool["function"] params = fn.get("parameters", {}) properties = params.get("properties", {}) if isinstance(params, dict) else {} required = set(params.get("required", []) if isinstance(params, dict) else []) parts: List[str] = [] if isinstance(properties, dict): for name, schema in properties.items(): suffix = "" if name in required else "?" parts.append(f"{name}{suffix}:{schema_type_label(schema)}") sig = ", ".join(parts) desc = truncate_description(fn.get("description", ""), MAX_FUNCTION_DESCRIPTION_CHARS) lines.append(f'{desc}') lines.append("") return "\n".join(lines) def build_prompt(query: str, tools: List[Dict[str, Any]]) -> str: system_content = SYSTEM_PROMPT.rstrip() + "\n\n" + render_compact_tool_signatures(tools) messages = [ {"role": "system", "content": system_content}, {"role": "user", "content": str(query).strip()}, ] template_kwargs = {"tokenize": False, "add_generation_prompt": True, "enable_thinking": False} try: return processor.apply_chat_template(messages, **template_kwargs) except (AttributeError, TypeError): try: return tokenizer.apply_chat_template(messages, **template_kwargs) except TypeError: template_kwargs.pop("enable_thinking", None) return tokenizer.apply_chat_template(messages, **template_kwargs) # --- Validation --- _LT = chr(60) _GT = chr(62) _TC_O = _LT + "tool_call" + _GT _TC_C = _LT + "/tool_call" + _GT _FN_O = _LT + "function=" _FN_C = _LT + "/function" + _GT _PM_O = _LT + "parameter=" _PM_C = _LT + "/parameter" + _GT FUNCTION_RE = re.compile( re.escape(_TC_O) + r"\s*" + re.escape(_FN_O) + r"([A-Za-z_][A-Za-z0-9_]*)" + re.escape(_GT) + r"\s*" r"(.*?)\s*" + re.escape(_FN_C) + r"\s*" + re.escape(_TC_C), flags=re.DOTALL, ) PARAMETER_RE = re.compile( re.escape(_PM_O) + r"([A-Za-z_][A-Za-z0-9_]*)" + re.escape(_GT) + r"\s*(.*?)\s*" + re.escape(_PM_C), flags=re.DOTALL, ) def validate_native_output(text, expected_function, expected_params): stripped = text.strip() match_objects = list(FUNCTION_RE.finditer(stripped)) matches = [(m.group(1), m.group(2)) for m in match_objects] errors = [] if not matches: errors.append("No complete native tool_call block found.") cursor = 0 outside_fragments = [] for m in match_objects: fragment = stripped[cursor:m.start()].strip() if fragment: outside_fragments.append(fragment) cursor = m.end() trailing = stripped[cursor:].strip() if trailing: outside_fragments.append(trailing) if outside_fragments: errors.append(f"Output contains text outside complete tool-call blocks: {outside_fragments!r}") if "```" in stripped: errors.append("Markdown fences are not allowed.") if _LT + expected_function + _GT in stripped: errors.append("Function name was emitted as an invalid top-level tag.") if stripped.count(_TC_O) != stripped.count(_TC_C): errors.append("Unbalanced tool_call tags.") if stripped.count(_PM_O) != stripped.count(_PM_C): errors.append("Unbalanced parameter tags.") if stripped.count(_FN_O) != stripped.count(_FN_C): errors.append("Unbalanced function tags.") functions = [name for name, _ in matches] if expected_function not in functions: errors.append(f"Expected function {expected_function!r}, received {functions!r}.") param_keys_present = set() for fn_name, body in matches: param_pairs = PARAMETER_RE.findall(body) residual = PARAMETER_RE.sub("", body).strip() if residual: errors.append(f"Unexpected content inside function {fn_name!r}: {residual!r}") for key, raw_value in param_pairs: value = raw_value.strip() if value.startswith("[") or value.startswith("{"): try: json.loads(value) except json.JSONDecodeError as exc: errors.append(f"Parameter {key!r} is not valid JSON: {exc}") param_keys_present.add(key) for param_name, required in expected_params.items(): if required and param_name not in param_keys_present: errors.append(f"Missing required parameter: {param_name!r}") return {"valid": len(errors) == 0, "errors": errors} # --- Generation --- @torch.inference_mode() def generate_tool_call(query, tools): prompt = build_prompt(query, tools) inputs = tokenizer(prompt, return_tensors="pt", add_special_tokens=False).to(model.device) start = time.time() outputs = model.generate( **inputs, max_new_tokens=MAX_NEW_TOKENS, do_sample=False, repetition_penalty=1.05, eos_token_id=tokenizer.eos_token_id, pad_token_id=tokenizer.pad_token_id, ) elapsed = time.time() - start generated = outputs[0, inputs["input_ids"].shape[1]:] text = tokenizer.decode(generated, skip_special_tokens=True).strip() return text, elapsed # --- Run benchmark --- print(f"\n=== Running benchmark: {len(BENCHMARK_CASES)} cases ===\n") results = [] per_tool_stats = {} for i, case in enumerate(BENCHMARK_CASES): query = case["query"] expected_function = case["expected_function"] expected_params = case["expected_params"] category = case["category"] tools = [compact_tool(t) for t in TARGET_TOOLS] if category == "targeted" else [compact_tool(t) for t in GENERAL_TOOLS] text, latency = generate_tool_call(query, tools) validation = validate_native_output(text, expected_function, expected_params) result = { "index": i, "category": category, "query": query[:120], "expected_function": expected_function, "generated_text": text[:500], "valid": validation["valid"], "errors": validation["errors"], "latency_s": round(latency, 3), } results.append(result) status = "\u2713" if validation["valid"] else "\u2717" print(f"[{i+1}/{len(BENCHMARK_CASES)}] {status} {expected_function} ({category}) - {latency:.2f}s") if not validation["valid"]: for err in validation["errors"]: print(f" ERROR: {err}") tool_name = expected_function if tool_name not in per_tool_stats: per_tool_stats[tool_name] = {"total": 0, "valid": 0, "latencies": []} per_tool_stats[tool_name]["total"] += 1 if validation["valid"]: per_tool_stats[tool_name]["valid"] += 1 per_tool_stats[tool_name]["latencies"].append(latency) # --- Summary --- total = len(results) valid_count = sum(1 for r in results if r["valid"]) overall_accuracy = valid_count / total if total else 0.0 total_latency = sum(r["latency_s"] for r in results) avg_latency = total_latency / total if total else 0.0 print("\n" + "=" * 70) print("BENCHMARK SUMMARY") print("=" * 70) print(f"Total cases: {total}") print(f"Valid outputs: {valid_count}/{total} ({overall_accuracy*100:.1f}%)") print(f"Total gen time: {total_latency:.2f}s") print(f"Average latency: {avg_latency:.2f}s") print() print("Per-tool breakdown:") print("-" * 60) for tool_name, stats in sorted(per_tool_stats.items()): acc = stats["valid"] / stats["total"] if stats["total"] else 0.0 avg_lat = sum(stats["latencies"]) / len(stats["latencies"]) if stats["latencies"] else 0.0 min_lat = min(stats["latencies"]) if stats["latencies"] else 0.0 max_lat = max(stats["latencies"]) if stats["latencies"] else 0.0 print(f" {tool_name:25s} {stats['valid']}/{stats['total']} ({acc*100:5.1f}%) avg={avg_lat:.2f}s min={min_lat:.2f}s max={max_lat:.2f}s") targeted_results = [r for r in results if r["category"] == "targeted"] general_results = [r for r in results if r["category"] == "general"] targeted_valid = sum(1 for r in targeted_results if r["valid"]) general_valid = sum(1 for r in general_results if r["valid"]) print() if targeted_results: print(f"Targeted CLI tools: {targeted_valid}/{len(targeted_results)} ({targeted_valid/len(targeted_results)*100:.1f}%)") if general_results: print(f"General tool-call: {general_valid}/{len(general_results)} ({general_valid/len(general_results)*100:.1f}%)") full_results = { "total_cases": total, "valid_outputs": valid_count, "overall_accuracy": round(overall_accuracy, 4), "total_latency_s": round(total_latency, 2), "avg_latency_s": round(avg_latency, 2), "per_tool": { tool_name: { "total": stats["total"], "valid": stats["valid"], "accuracy": round(stats["valid"] / stats["total"], 4) if stats["total"] else 0.0, "avg_latency_s": round(sum(stats["latencies"]) / len(stats["latencies"]), 2) if stats["latencies"] else 0.0, "min_latency_s": round(min(stats["latencies"]), 2) if stats["latencies"] else 0.0, "max_latency_s": round(max(stats["latencies"]), 2) if stats["latencies"] else 0.0, } for tool_name, stats in sorted(per_tool_stats.items()) }, "category_breakdown": { "targeted": { "total": len(targeted_results), "valid": targeted_valid, "accuracy": round(targeted_valid / len(targeted_results), 4) if targeted_results else 0.0, }, "general": { "total": len(general_results), "valid": general_valid, "accuracy": round(general_valid / len(general_results), 4) if general_results else 0.0, }, }, "results": results, } results_path = "/data/benchmark_results.json" with open(results_path, "w", encoding="utf-8") as f: json.dump(full_results, f, ensure_ascii=False, indent=2) print(f"\nFull results saved to: {results_path}") training_volume.commit() return { "status": "completed", "total_cases": total, "valid_outputs": valid_count, "overall_accuracy": round(overall_accuracy, 4), "per_tool": { tool_name: { "accuracy": round(stats["valid"] / stats["total"], 4) if stats["total"] else 0.0, "valid": stats["valid"], "total": stats["total"], } for tool_name, stats in sorted(per_tool_stats.items()) }, "category_breakdown": { "targeted": f"{targeted_valid}/{len(targeted_results)}", "general": f"{general_valid}/{len(general_results)}", }, } @app.local_entrypoint() def main( adapter_dir: str = "Hopcoder-Mini-9B-Native-ToolCall-LoRA-H200", max_new_tokens: int = 384, ): result = benchmark.remote( adapter_dir=adapter_dir, max_new_tokens=max_new_tokens, ) print(json.dumps(result, indent=2))