File size: 6,941 Bytes
6379283 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 | #!/usr/bin/env python3
"""
Evaluate Stack 2.9 on tool-use test suite.
Supports: local model (transformers) or vLLM API.
"""
import json
import argparse
from pathlib import Path
from typing import Dict, Any, List
import openai
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
def load_test_cases(path: str) -> List[Dict]:
with open(path, 'r') as f:
return json.load(f)
def predict_with_transformers(model_path: str, prompt: str) -> Dict:
"""Query local model for tool prediction."""
tokenizer = AutoTokenizer.from_pretrained(model_path)
model = AutoModelForCausalLM.from_pretrained(
model_path,
device_map="auto",
torch_dtype=torch.float16
)
inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
outputs = model.generate(
**inputs,
max_new_tokens=200,
temperature=0.2,
do_sample=False
)
response = tokenizer.decode(outputs[0], skip_special_tokens=True)
# Parse tool call from response (simple version - assumes structured format)
# This is a placeholder - real parser needs to extract JSON tool use
return {
"tool": "UnknownTool",
"params": {},
"raw_response": response
}
def predict_with_vllm(api_url: str, prompt: str, model_name: str = "stack-2.9") -> Dict:
"""Query vLLM server."""
client = openai.OpenAI(
base_url=api_url,
api_key="dummy"
)
response = client.chat.completions.create(
model=model_name,
messages=[{"role": "user", "content": prompt}],
max_tokens=200,
temperature=0.2
)
content = response.choices[0].message.content
# Parse tool call from content
return {
"tool": "UnknownTool",
"params": {},
"raw_response": content
}
def evaluate_predictions(
test_cases: List[Dict],
predictions: List[Dict],
tool_catalog: Dict[str, Any] = None
) -> Dict[str, Any]:
"""Compare predictions against ground truth."""
total = len(test_cases)
correct_tool = 0
correct_params = 0
exact_match = 0
per_tool = {}
for tc, pred in zip(test_cases, predictions):
expected_tool = tc["expected_tool"]
pred_tool = pred["tool"]
# Tool accuracy
tool_correct = pred_tool == expected_tool
if tool_correct:
correct_tool += 1
# Parameter accuracy (simple: exact match of params dict, or partial?)
expected_params = tc["expected_params"]
pred_params = pred["params"]
# For now, check if all expected params are present with same values
param_correct = all(
pred_params.get(k) == v for k, v in expected_params.items()
) if expected_params else True
if param_correct:
correct_params += 1
# Exact match (tool + all params)
if tool_correct and param_correct:
exact_match += 1
# Track per-tool stats
if expected_tool not in per_tool:
per_tool[expected_tool] = {"total": 0, "correct_tool": 0, "correct_params": 0}
per_tool[expected_tool]["total"] += 1
if tool_correct:
per_tool[expected_tool]["correct_tool"] += 1
if param_correct:
per_tool[expected_tool]["correct_params"] += 1
return {
"total_examples": total,
"tool_accuracy": correct_tool / total if total > 0 else 0,
"parameter_accuracy": correct_params / total if total > 0 else 0,
"exact_match_accuracy": exact_match / total if total > 0 else 0,
"per_tool_breakdown": per_tool
}
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--test-cases", type=str, default="stack-2.9-eval/tool_use/test_cases.json")
parser.add_argument("--catalog", type=str, default="training-data/tools/catalog.json")
parser.add_argument("--model-path", type=str, help="HuggingFace model path for transformers")
parser.add_argument("--vllm-api", type=str, help="vLLM API URL (e.g., http://localhost:8000)")
parser.add_argument("--output", type=str, default="stack-2.9-eval/tool_use/results.json")
parser.add_argument("--limit", type=int, help="Limit number of test cases to evaluate")
args = parser.parse_args()
test_cases_path = Path(args.test_cases)
catalog_path = Path(args.catalog)
if not test_cases_path.exists():
print(f"❌ Test cases not found: {test_cases_path}")
print(" Run generate_tool_use_tests.py first")
return
test_cases = load_test_cases(test_cases_path)
if args.limit:
test_cases = test_cases[:args.limit]
print(f"🧪 Evaluating {len(test_cases)} tool-use test cases")
# Load tool catalog
tool_catalog = None
if catalog_path.exists():
with open(catalog_path, 'r') as f:
tool_catalog = {t["tool"]: t for t in json.load(f)}
print(f"✅ Loaded tool catalog ({len(tool_catalog)} tools)")
# Generate predictions
predictions = []
for i, tc in enumerate(test_cases):
prompt = tc["prompt"]
if args.model_path:
pred = predict_with_transformers(args.model_path, prompt)
elif args.vllm_api:
pred = predict_with_vllm(args.vllm_api, prompt)
else:
# Mock predictor (random baseline)
import random
tools = list(tool_catalog.keys()) if tool_catalog else ["UnknownTool"]
pred = {
"tool": random.choice(tools),
"params": {},
"raw_response": "Mock prediction"
}
predictions.append(pred)
if (i+1) % 10 == 0:
print(f" Processed {i+1}/{len(test_cases)}...", end='\r')
print(f"\n📊 Evaluating predictions...")
# Evaluate
results = evaluate_predictions(test_cases, predictions, tool_catalog)
# Save results
output_path = Path(args.output)
output_path.parent.mkdir(parents=True, exist_ok=True)
with open(output_path, 'w') as f:
json.dump(results, f, indent=2)
print(f"\n✅ Evaluation results:")
print(f" Tool accuracy: {results['tool_accuracy']*100:.1f}%")
print(f" Param accuracy: {results['parameter_accuracy']*100:.1f}%")
print(f" Exact match: {results['exact_match_accuracy']*100:.1f}%")
print(f"\n Results saved to: {output_path}")
# Show per-tool breakdown (top 5 worst)
if results['per_tool_breakdown']:
print("\n📉 Worst performing tools:")
sorted_tools = sorted(
results['per_tool_breakdown'].items(),
key=lambda x: x[1]['correct_tool'] / x[1]['total'] if x[1]['total']>0 else 0
)[:5]
for tool, stats in sorted_tools:
acc = stats['correct_tool'] / stats['total'] * 100
print(f" {tool}: {acc:.1f}% ({stats['correct_tool']}/{stats['total']})")
if __name__ == "__main__":
main() |