import argparse import csv import datetime import hashlib import io import json import os from pathlib import Path import re import subprocess import sys import tempfile import time ROOT = Path(__file__).resolve().parents[1] DEFAULT_OUT = ROOT / 'public/evidence/model-tier-benchmark-20261009' API = 'https://api.quickrouter.ai' MAX_OUTPUT_TOKENS = 4096 COHORT = [ {'id': 'gpt-6.1-sol', 'protocol': 'responses', 'protocol_label': 'Responses API', 'endpoint': '/v1/responses', 'positioning': '均衡主力', 'effort': 'medium'}, {'id': 'claude-sonnet-5-5', 'protocol': 'messages', 'protocol_label': 'Messages API', 'endpoint': '/v1/messages', 'positioning': '均衡主力', 'effort': 'medium'}, {'id': 'gemini-3.8-flash', 'protocol': 'chat', 'protocol_label': 'Chat 兼容', 'endpoint': '/v1/chat/completions', 'positioning': '均衡主力', 'effort': 'medium'}, ] TOOL = { 'name': 'search_order', 'description': '查询虚构演示订单状态；这里只生成调用提案，不执行查询。', 'parameters': {'type': 'object', 'properties': {'order_id': {'type': 'string'}}, 'required': ['order_id'], 'additionalProperties': False}, } ORDERS = [ {'order_id': '00128', 'quantity': 3, 'unit_price': '19.90', 'shipping': '6.00', 'total': '65.70'}, {'order_id': '00047', 'quantity': 2, 'unit_price': '28.50', 'shipping': '0.00', 'total': '57.00'}, {'order_id': '00906', 'quantity': 4, 'unit_price': '12.35', 'shipping': '8.00', 'total': '57.40'}, ] GROUNDED = [ ('未拆封商品在签收后七天内可申请退货；普通订单付款后两个工作日内发货；没有提供国际配送资料。', '签收三天、未拆封，可以申请退货吗？支持国际配送吗？', {'return_allowed': True, 'international_shipping': None}), ('未拆封商品在签收后五天内可申请退货；超过五天不接受退货；没有提供国际配送资料。', '签收八天、未拆封，可以申请退货吗？支持国际配送吗？', {'return_allowed': False, 'international_shipping': None}), ('未拆封商品在签收后十天内可申请退货；拆封商品不接受退货；没有提供国际配送资料。', '签收两天、已拆封，可以申请退货吗？支持国际配送吗？', {'return_allowed': False, 'international_shipping': None}), ] TASK_IDS = ['structured-extraction', 'grounded-answer', 'tool-call'] METHOD = { 'selectedBeforeRun': True, 'rounds': 3, 'planned_requests': 27, 'max_output_tokens': MAX_OUTPUT_TOKENS, 'requested_generation_cap': MAX_OUTPUT_TOKENS, 'budget_mapping_independently_verified': False, 'effort': 'medium', 'automatic_retry': False, 'failure_only_recheck': False, 'tool_execution': False, 'scope': '同一产品定位的三个平台型号 ID；每轮三个任务使用固定新样本，所有模型接收相同样本。不是综合能力排名，也未独立验证上游型号身份。', 'budget_scope': '4096 是请求参数中的生成上限。各协议的推理 token 计入方式及 medium 含义可能不同，不代表相同计算量。网关的 cap/effort 透传与 Gem Chat 兼容预算映射未独立验证；HTTP 200 不能证明参数生效。输出截断单列。', 'latency_scope': '耗时为客户端完整请求墙钟时间，包含网络与队列，不是首 token 延迟。', 'billing_scope': '实际账单金额未经独立匹配，保持 null。公开报价估算与实际扣费必须分开。', } def now(): return datetime.datetime.now(datetime.timezone.utc).isoformat() def task_for(task_id, round_index): order = ORDERS[round_index] if task_id == 'structured-extraction': prompt = f"这是虚构测试资料。订单编号{order['order_id']}，买了{order['quantity']}件单价{order['unit_price']}元的商品，运费{order['shipping']}元。只输出 JSON 对象，不要 Markdown。字段：order_id 字符串保留前导零、quantity 整数、unit_price 字符串保留两位小数、shipping 字符串保留两位小数、total 字符串保留两位小数。" expected = order.copy() elif task_id == 'grounded-answer': fragment, question, expected = GROUNDED[round_index] prompt = f'只根据以下虚构商店资料回答：{fragment}只输出 JSON 对象，字段 return_allowed:boolean，international_shipping:null 或 boolean。问题：{question}不要补充资料以外的事实。' expected = expected.copy() elif task_id == 'tool-call': prompt = f"请使用 search_order 工具查询虚构订单 {order['order_id']} 的状态。不猜测状态，不直接回答订单结果。" expected = {'order_id': order['order_id']} else: raise ValueError('Unknown fixed task') return {'id': task_id, 'sample': round_index + 1, 'prompt': prompt, 'expected': expected} def request_body(model, task): body = {'model': model['id'], 'stream': False} protocol = model['protocol'] if protocol == 'responses': body.update({'input': [{'role': 'user', 'content': task['prompt']}], 'max_output_tokens': MAX_OUTPUT_TOKENS, 'reasoning': {'effort': 'medium'}, 'store': False}) if task['id'] == 'tool-call': body['tools'] = [{'type': 'function', **TOOL, 'strict': False}] body['tool_choice'] = 'auto' else: body['messages'] = [{'role': 'user', 'content': task['prompt']}] if protocol == 'messages': body.update({'max_tokens': MAX_OUTPUT_TOKENS, 'thinking': {'type': 'adaptive'}, 'output_config': {'effort': 'medium'}}) if task['id'] == 'tool-call': body['tools'] = [{'name': TOOL['name'], 'description': TOOL['description'], 'input_schema': TOOL['parameters']}] body['tool_choice'] = {'type': 'auto'} elif protocol == 'chat': body.update({'max_completion_tokens': MAX_OUTPUT_TOKENS, 'reasoning_effort': 'medium'}) if task['id'] == 'tool-call': body['tools'] = [{'type': 'function', 'function': TOOL}] body['tool_choice'] = 'auto' else: raise ValueError('Unknown native protocol') return body def plan(): trials = [] for round_index in range(3): order = COHORT[round_index:] + COHORT[:round_index] for task_id in TASK_IDS: task = task_for(task_id, round_index) for model in order: trials.append({'trial_id': f"r{round_index + 1}-{task_id}-{model['id']}", 'round': round_index + 1, 'model': model['id'], 'protocol': model['protocol'], 'endpoint': model['endpoint'], 'task': task, 'request': request_body(model, task)}) return {'schema_version': 1, 'cohort': COHORT, 'method': METHOD, 'trials': trials} def redact(value, key=''): if isinstance(value, str): text = value.replace(key, '[REDACTED]') if key else value return re.sub(r'sk-(?:proj-)?[A-Za-z0-9_-]{16,}', '[REDACTED]', text) if isinstance(value, list): return [redact(item, key) for item in value] if isinstance(value, dict): return {name: redact(item, key) for name, item in value.items()} return value def runtime_key(use_stdin): if use_stdin and sys.stdin.isatty(): import termios previous = termios.tcgetattr(sys.stdin) hidden = termios.tcgetattr(sys.stdin) hidden[3] &= ~termios.ECHO termios.tcsetattr(sys.stdin, termios.TCSANOW, hidden) print('Waiting for runtime key; terminal echo disabled.', flush=True) try: key = sys.stdin.readline().strip() finally: termios.tcsetattr(sys.stdin, termios.TCSANOW, previous) elif use_stdin: key = sys.stdin.readline().strip() else: key = os.environ.get('QUICKROUTER_API_KEY', '') if not key or any(char in key for char in ['\n', '\r', '"', '\\']) or any(ord(char) < 32 for char in key): raise ValueError('A valid runtime key is required; it is never saved.') return key def request(endpoint, key, body=None, protocol='chat'): started = time.monotonic() with tempfile.TemporaryDirectory() as temporary: directory = Path(temporary) response_file = directory / 'response.json' headers_file = directory / 'headers.txt' command = ['curl', '-sS', '--connect-timeout', '15', '--max-time', '120', '--max-redirs', '0', '--proto', '=https', '--config', '-', '-D', str(headers_file), '-o', str(response_file), '-w', '%{http_code}', API + endpoint] config = 'header = "Authorization: Bearer ' + key + '"\n' if protocol == 'messages': config += 'header = "x-api-key: ' + key + '"\nheader = "anthropic-version: 2023-06-01"\n' if body is not None: payload_file = directory / 'request.json' payload_file.write_text(json.dumps(body, ensure_ascii=False)) command += ['-H', 'Content-Type: application/json', '--data-binary', '@' + str(payload_file)] environment = os.environ.copy() environment.pop('QUICKROUTER_API_KEY', None) process = subprocess.run(command, input=config, text=True, capture_output=True, env=environment) try: response = json.loads(response_file.read_text()) except (OSError, ValueError): response = {} if not isinstance(response, dict): response = {} request_id = None if headers_file.exists(): for line in headers_file.read_text(errors='replace').splitlines(): if line.lower().startswith(('x-request-id:', 'request-id:')): request_id = redact(line.split(':', 1)[1].strip(), key) status = int(process.stdout) if process.stdout.isdigit() and process.stdout != '000' else None return response, {'http_status': status, 'elapsed_seconds': round(time.monotonic() - started, 3), 'request_id': request_id, 'curl_exit': process.returncode, 'transport_status': 'confirmed' if process.returncode == 0 else 'unconfirmed'} def count(value): return value if type(value) is int and value >= 0 else None def usage_metrics(value): if isinstance(value, dict): return {name: usage_metrics(item) for name, item in value.items() if isinstance(item, dict) or type(item) is int} return count(value) def usage_from(response, protocol): usage = response.get('usage') if not isinstance(usage, dict): return None allowed = {'input_tokens', 'output_tokens', 'total_tokens', 'prompt_tokens', 'completion_tokens', 'input_tokens_details', 'output_tokens_details', 'prompt_tokens_details', 'completion_tokens_details', 'cache_creation_input_tokens', 'cache_read_input_tokens', 'cache_creation'} raw = {name: usage_metrics(value) for name, value in usage.items() if name in allowed} detail = usage.get('input_tokens_details' if protocol == 'responses' else 'prompt_tokens_details') detail = detail if isinstance(detail, dict) else {} output_detail = usage.get('output_tokens_details' if protocol in {'responses', 'messages'} else 'completion_tokens_details') output_detail = output_detail if isinstance(output_detail, dict) else {} return { 'raw': raw, 'input_tokens': count(usage.get('prompt_tokens' if protocol == 'chat' else 'input_tokens')), 'output_tokens': count(usage.get('completion_tokens' if protocol == 'chat' else 'output_tokens')), 'total_tokens': count(usage.get('total_tokens')), 'cached_input_tokens': count(usage.get('cache_read_input_tokens') if protocol == 'messages' else detail.get('cached_tokens')), 'cache_creation_input_tokens': count(usage.get('cache_creation_input_tokens')) if protocol == 'messages' else None, 'reasoning_tokens': count(output_detail.get('thinking_tokens' if protocol == 'messages' else 'reasoning_tokens')), 'input_scope': 'uncached input; cache read/write reported separately' if protocol == 'messages' else 'total input; cached input is a subset when reported', } def answer_from(response, protocol): raw = [] texts = [] calls = [] finish = None incomplete_reason = None if protocol == 'responses': finish = response.get('status') details = response.get('incomplete_details') incomplete_reason = details.get('reason') if isinstance(details, dict) else None output = response.get('output') for item in output if isinstance(output, list) else []: if not isinstance(item, dict): continue if item.get('type') == 'message': content = item.get('content') parts = [] for part in content if isinstance(content, list) else []: if isinstance(part, dict) and part.get('type') in {'output_text', 'refusal'}: field = 'text' if part['type'] == 'output_text' else 'refusal' parts.append({'type': part['type'], field: part.get(field)}) if isinstance(part.get('text'), str): texts.append(part['text']) raw.append({'type': 'message', 'id': item.get('id'), 'role': item.get('role'), 'content': parts}) elif item.get('type') == 'function_call': raw.append({name: item.get(name) for name in ['type', 'id', 'call_id', 'name', 'arguments', 'status']}) calls.append({'id': item.get('call_id'), 'type': 'function', 'native_type': 'function_call', 'name': item.get('name'), 'arguments': item.get('arguments')}) truncated = finish == 'incomplete' and incomplete_reason == 'max_output_tokens' elif protocol == 'messages': finish = response.get('stop_reason') content = response.get('content') for item in content if isinstance(content, list) else []: if not isinstance(item, dict): continue if item.get('type') == 'text': raw.append({'type': 'text', 'text': item.get('text')}) if isinstance(item.get('text'), str): texts.append(item['text']) elif item.get('type') == 'tool_use': raw.append({name: item.get(name) for name in ['type', 'id', 'name', 'input']}) calls.append({'id': item.get('id'), 'type': 'function', 'native_type': 'tool_use', 'name': item.get('name'), 'arguments': item.get('input')}) truncated = finish == 'max_tokens' else: choices = response.get('choices') first = choices[0] if isinstance(choices, list) and choices and isinstance(choices[0], dict) else {} finish = first.get('finish_reason') message = first.get('message') message = message if isinstance(message, dict) else {} text = message.get('content') if isinstance(text, str): texts.append(text) raw.append({'role': message.get('role'), 'content': text if isinstance(text, str) else None}) tool_calls = message.get('tool_calls') for item in tool_calls if isinstance(tool_calls, list) else []: if not isinstance(item, dict): calls.append({'invalid': True}) continue function = item.get('function') function = function if isinstance(function, dict) else {} raw.append({'type': item.get('type'), 'id': item.get('id'), 'function': {'name': function.get('name'), 'arguments': function.get('arguments')}}) calls.append({'id': item.get('id'), 'type': item.get('type'), 'native_type': item.get('type'), 'name': function.get('name'), 'arguments': function.get('arguments')}) truncated = finish == 'length' complete_states = {'responses': {'completed'}, 'messages': {'end_turn', 'tool_use', 'stop_sequence'}, 'chat': {'stop', 'tool_calls', 'function_call'}} return {'content': ''.join(texts), 'tool_calls': calls, 'raw': raw, 'finish_reason': finish, 'incomplete_reason': incomplete_reason, 'truncated': truncated, 'completion_confirmed': finish in complete_states[protocol]} def exact(expected, actual): if type(expected) is not type(actual): return False if isinstance(expected, dict): return expected.keys() == actual.keys() and all(exact(value, actual[name]) for name, value in expected.items()) if isinstance(expected, list): return len(expected) == len(actual) and all(exact(left, right) for left, right in zip(expected, actual)) return expected == actual def check(task, answer): if task['id'] == 'tool-call': calls = answer.get('tool_calls') if not isinstance(calls, list) or len(calls) != 1 or not isinstance(calls[0], dict): return False, 'exactly-one-tool-proposal-required', None call = calls[0] if not isinstance(call.get('id'), str) or not call['id'].strip() or call.get('type') != 'function' or call.get('native_type') not in {'function', 'function_call', 'tool_use'}: return False, 'complete-tool-id-and-type-required', None arguments = call.get('arguments') try: parsed = json.loads(arguments) if isinstance(arguments, str) else arguments except (ValueError, TypeError): return False, 'parseable-tool-arguments-required', None passed = call.get('name') == TOOL['name'] and exact(task['expected'], parsed) return passed, 'tool-id-type-name-and-exact-arguments', parsed try: parsed = json.loads(answer.get('content', '')) except (ValueError, TypeError): return False, 'plain-json-required', None return exact(task['expected'], parsed), 'exact-object-fields-types-and-values', parsed def exclusive_json(path, data): with Path(path).open('x') as file: json.dump(data, file, ensure_ascii=False, indent=2) file.write('\n') def update_json(path, data): temporary = Path(path).with_name(Path(path).name + '.tmp') temporary.write_text(json.dumps(data, ensure_ascii=False, indent=2) + '\n') os.replace(temporary, path) def lock_plan(out): expected = plan() path = out / 'plan.json' if path.exists(): if json.loads(path.read_text()) != expected: raise ValueError('Existing plan differs; use a new output directory.') else: exclusive_json(path, expected) return expected def catalog(out, key): path = out / 'catalog.json' if path.exists(): return json.loads(path.read_text()) claim = out / 'catalog.state.json' exclusive_json(claim, {'started_at': now(), 'status': 'submission-unconfirmed'}) response, transport = request('/v1/models', key) data = response.get('data') ids = sorted({item['id'] for item in data if isinstance(item, dict) and isinstance(item.get('id'), str)}) if isinstance(data, list) else [] selected = [model['id'] for model in COHORT] result = {'checked_at': now(), 'transport': transport, 'models': ids, 'selected_models': selected, 'missing_selected': [model for model in selected if model not in ids], 'scope': 'This runtime token catalog only; not native endpoint or upstream identity verification.'} exclusive_json(path, redact(result, key)) update_json(claim, {'status': 'recorded' if transport['transport_status'] == 'confirmed' else 'unconfirmed'}) return result def make_record(trial, response, transport, key): protocol = trial['protocol'] answer = answer_from(response, protocol) confirmed = transport.get('transport_status') == 'confirmed' http_ok = transport.get('http_status') == 200 passed, rule, parsed = check(trial['task'], answer) if confirmed and http_ok else (None if not confirmed else False, 'transport-unconfirmed' if not confirmed else 'http-200-required', None) verdict = 'unknown' if not confirmed else 'http-error' if not http_ok else 'truncated' if answer['truncated'] else 'unknown' if not answer['completion_confirmed'] else 'pass' if passed else 'fail' task_check_rule = rule if confirmed and http_ok and not answer['truncated'] and not answer['completion_confirmed']: rule = 'completed-response-state-required' error = response.get('error') error = error if isinstance(error, dict) else {} record = { 'trial_id': trial['trial_id'], 'model': trial['model'], 'response_model': response.get('model'), 'response_id': response.get('id'), 'task': trial['task']['id'], 'sample': trial['task']['sample'], 'round': trial['round'], 'protocol': protocol, 'endpoint': trial['endpoint'], 'request': trial['request'], 'transport': transport, 'requested_generation_cap': MAX_OUTPUT_TOKENS, 'budget_mapping_independently_verified': False, 'passed': passed if verdict not in {'truncated', 'unknown'} else None, 'task_check_passed': passed, 'verdict': verdict, 'check': rule, 'task_check_rule': task_check_rule, 'parsed_result': parsed, 'truncated': answer['truncated'], 'completion_confirmed': answer['completion_confirmed'], 'finish_reason': answer['finish_reason'], 'incomplete_reason': answer['incomplete_reason'], 'raw_answer': answer['raw'], 'usage': usage_from(response, protocol), 'error': {name: error.get(name) for name in ['type', 'code', 'message']}, 'billed_usd': None, 'billing_status': 'unverified', 'reference_cost_usd': None, 'reference_cost_status': 'no-public-quote-matched', } return redact(record, key) def save_summary(out, fixed_plan, records): counts = [] for model in COHORT: selected = [record for record in records if record['model'] == model['id']] counts.append({**model, 'requests_recorded': len(selected), **{name: sum(record['verdict'] == name for record in selected) for name in ['pass', 'fail', 'truncated', 'unknown', 'http-error']}}) summary = {'checked_at': now(), 'cohort': COHORT, 'method': METHOD, 'planned_requests': len(fixed_plan['trials']), 'recorded_requests': len(records), 'model_counts': counts, 'billed_usd': None, 'reference_cost_usd': None, 'records': records} update_json(out / 'summary.json', summary) output = io.StringIO() fields = ['trial_id', 'model', 'response_model', 'round', 'task', 'sample', 'protocol', 'endpoint', 'started_at', 'completed_at', 'http_status', 'curl_exit', 'request_id', 'elapsed_seconds', 'verdict', 'passed', 'truncated', 'finish_reason', 'input_tokens', 'output_tokens', 'cached_input_tokens', 'cache_creation_input_tokens', 'reasoning_tokens', 'billed_usd', 'reference_cost_usd'] writer = csv.DictWriter(output, fieldnames=fields) writer.writeheader() for record in records: row = {field: record.get(field) for field in fields} row.update({field: record['transport'].get(field) for field in ['http_status', 'curl_exit', 'request_id', 'elapsed_seconds']}) usage = record.get('usage') or {} row.update({field: usage.get(field) for field in ['input_tokens', 'output_tokens', 'cached_input_tokens', 'cache_creation_input_tokens', 'reasoning_tokens']}) writer.writerow(row) temporary = out / 'records.csv.tmp' temporary.write_text(output.getvalue()) os.replace(temporary, out / 'records.csv') def run(out, key, catalog_only=False): out.mkdir(parents=True, exist_ok=True) fixed_plan = lock_plan(out) visible = catalog(out, key) if catalog_only: print(json.dumps({'catalog': True, 'transport': visible['transport'], 'selected_models': visible['selected_models'], 'missing_selected': visible['missing_selected']}, ensure_ascii=False), flush=True) return if visible['transport']['http_status'] != 200 or visible['transport']['transport_status'] != 'confirmed' or visible['missing_selected']: raise ValueError('Catalog does not confirm the whole fixed cohort; no trials submitted.') records = [] for trial in fixed_plan['trials']: digest = hashlib.sha256(trial['trial_id'].encode()).hexdigest()[:20] path = out / (digest + '.json') state = out / (digest + '.state.json') if path.exists(): recorded = json.loads(path.read_text()) records.append(recorded) if recorded.get('transport', {}).get('http_status') in {401, 402, 403}: print('Existing authorization or balance rejection; remaining trials stay stopped.', flush=True) break continue if state.exists(): print(json.dumps({'trial_id': trial['trial_id'], 'status': 'existing-submission-no-retry'}), flush=True) continue started_at = now() exclusive_json(state, {'trial_id': trial['trial_id'], 'started_at': started_at, 'status': 'submission-unconfirmed'}) response, transport = request(trial['endpoint'], key, trial['request'], trial['protocol']) record = make_record(trial, response, transport, key) record.update({'started_at': started_at, 'completed_at': now()}) exclusive_json(path, record) update_json(state, {'trial_id': trial['trial_id'], 'status': 'recorded' if transport['transport_status'] == 'confirmed' else 'unconfirmed', 'request_id': transport.get('request_id')}) records.append(record) save_summary(out, fixed_plan, records) print(json.dumps({'trial_id': record['trial_id'], 'transport': record['transport'], 'verdict': record['verdict'], 'billing_status': 'unverified'}, ensure_ascii=False), flush=True) if transport.get('http_status') in {401, 402, 403}: print('Authorization or balance rejected; remaining trials stopped.', flush=True) break save_summary(out, fixed_plan, records) def main(): parser = argparse.ArgumentParser(description='Fixed same-positioning benchmark; default preview makes no network request.') mode = parser.add_mutually_exclusive_group() mode.add_argument('--run', action='store_true', help='May incur charges: at most 27 generation requests; no automatic retries.') mode.add_argument('--catalog', action='store_true', help='Confirm runtime token model visibility without generation requests.') parser.add_argument('--out', type=Path, default=DEFAULT_OUT) parser.add_argument('--key-stdin', action='store_true', help='Read runtime key from stdin; disable echo on a terminal.') args = parser.parse_args() if not args.run and not args.catalog: print(json.dumps({'preview': True, **plan()}, ensure_ascii=False, indent=2)) return try: run(args.out.resolve(), runtime_key(args.key_stdin), args.catalog) except (ValueError, FileExistsError) as error: print(str(error), file=sys.stderr) raise SystemExit(1) if __name__ == '__main__': main()