add fingerprint benchmark

This commit is contained in:
aaxiyao 2026-08-25 02:06:23 +00:00
parent 897f8fb1b9
commit 8b7dff96e7
6 changed files with 640 additions and 1 deletions

View File

@ -54,6 +54,10 @@ BENCHMARK_DOMAIN = {
'general_fc': '智能体与工具', 'general_fc': '智能体与工具',
'bfcl_v3': '智能体与工具', 'bfcl_v3': '智能体与工具',
'terminal_bench_v2_1': '智能体与工具', 'terminal_bench_v2_1': '智能体与工具',
# 指纹/安全类 benchmarkbash/fingerprint/ 下的独立执行器产出)
'llmmap': '模型安全与指纹',
'llm_verify': '模型安全与指纹',
'llm_fingerprint_detector': '模型安全与指纹',
} }
# Column order matching the reference CSV # Column order matching the reference CSV

114
bash/fingerprint/common.py Normal file
View File

@ -0,0 +1,114 @@
#!/usr/bin/env python3
"""Shared helpers for the three fingerprint benchmarks (LLMmap / LLM Verify /
llm-fingerprint-detector).
These benchmarks do not go through EvalScope's dataset pipeline. Each runner
script probes the target OpenAI-compatible endpoint with its own tool logic and
writes a report JSON shaped like EvalScope reports:
output/<folder>/<benchmark>/seed_<seed>/reports/<benchmark>.json
with at least ``score`` (float 0~1) and ``num`` so that
``bash/collect_results.py`` can aggregate them like any other benchmark.
"""
import argparse
import json
import urllib.error
import urllib.request
# 各工具报告文件名与其 benchmark 名一致
BENCHMARK_LLMMAP = 'llmmap'
BENCHMARK_LLM_VERIFY = 'llm_verify'
BENCHMARK_DETECTOR = 'llm_fingerprint_detector'
ALL_FINGERPRINT_BENCHMARKS = [BENCHMARK_LLMMAP, BENCHMARK_LLM_VERIFY, BENCHMARK_DETECTOR]
# 默认对被测端点关闭 thinking指纹探测需要稳定的可见回答
# 思考链会烧掉 max_tokens 且改变输出分布。sglang/vLLM 均支持该字段。
DEFAULT_EXTRA_BODY = {'chat_template_kwargs': {'thinking': False}}
def add_common_args(parser: argparse.ArgumentParser) -> argparse.ArgumentParser:
"""CLI arguments shared by all three fingerprint runners."""
parser.add_argument('--api-url', required=True,
help='Target OpenAI-compatible API base URL, e.g. http://localhost:30000/v1')
parser.add_argument('--model', required=True, help='Served model name to probe')
parser.add_argument('--report-path', required=True,
help='Where to write the EvalScope-style report JSON')
parser.add_argument('--timeout', type=int, default=120,
help='Per-request timeout in seconds (default: %(default)s)')
parser.add_argument('--thinking', action='store_true', default=False,
help='Do NOT disable thinking on the target (default: disabled)')
return parser
def chat_completion(api_url: str, model: str, user_prompt: str,
system_prompt: str = '', temperature: float = 1.0,
max_tokens: int = 512, timeout: int = 120,
extra_body: dict = None):
"""Minimal OpenAI chat-completions call (stdlib only).
Returns:
(content, error) exactly one of them is None.
"""
messages = []
if system_prompt:
messages.append({'role': 'system', 'content': system_prompt})
messages.append({'role': 'user', 'content': user_prompt})
payload = {
'model': model,
'messages': messages,
'temperature': temperature,
'max_tokens': max_tokens,
'stream': False,
}
payload.update(extra_body or {})
req = urllib.request.Request(
f"{api_url.rstrip('/')}/chat/completions",
data=json.dumps(payload).encode('utf-8'),
headers={'Content-Type': 'application/json'},
method='POST',
)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
data = json.loads(resp.read().decode('utf-8'))
except urllib.error.HTTPError as e:
detail = ''
try:
detail = e.read().decode('utf-8')[:200]
except Exception:
pass
return None, f'HTTP {e.code}: {detail}'
except Exception as e:
return None, f'request failed: {e}'
choices = data.get('choices') or []
if not choices:
return None, 'empty choices in response'
message = choices[0].get('message') or {}
content = message.get('content')
# 部分推理模型把可见内容放在 reasoning_content仅当 content 为空时兜底。
if not content:
content = message.get('reasoning_content') or ''
return str(content), None
def write_report(report_path: str, benchmark: str, score: float, num: int,
**details) -> None:
"""Write an EvalScope-style report JSON consumable by collect_results.py."""
from pathlib import Path
report_path = Path(report_path)
report_path.parent.mkdir(parents=True, exist_ok=True)
payload = {
'benchmark': benchmark,
'score': float(score),
'num': int(num),
}
payload.update(details)
report_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2),
encoding='utf-8')
print(f'[fingerprint] report written: {report_path}')

View File

@ -0,0 +1,123 @@
#!/usr/bin/env python3
"""llm-fingerprint-detector benchmark runner单 token 输出分布指纹)。
两种工作模式
1) --reference 提供同协议参考指纹 JSON 对被测端点采样一次并与参考比对
verify 模式硬比较
2) 未提供参考时自一致模式连续采样两次后互相比对衡量端点输出分布的
稳定性split-half 思路同时把 splitHalfJsd 记入报告
run.py 以子进程方式调用只需任意 Python + node需已 npm run build
<python> run_llm_detector.py --api-url ... --model ... --report-path ...
得分score [0,1]score = max(0, 1 - meanJSD)并记录 verdict
match 0.25 < uncertain 0.35 < mismatch论文基线标尺
"""
import argparse
import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path
from common import BENCHMARK_DETECTOR, add_common_args, write_report
def run_cli(cmd: list, timeout: int) -> dict:
"""Run the detector CLI with --json and return parsed stdout JSON."""
proc = subprocess.run(
cmd, capture_output=True, text=True, timeout=timeout,
env={**os.environ, 'LLM_FINGERPRINT_API_KEY': os.environ.get('LLM_FINGERPRINT_API_KEY', 'dummy')},
)
if proc.returncode not in (0, 2, 3): # 2=mismatch 3=uncertain 也是有效结论
raise RuntimeError(
f'detector CLI failed (rc={proc.returncode}):\n'
f'{proc.stdout[-500:]}\n{proc.stderr[-800:]}')
try:
return json.loads(proc.stdout)
except json.JSONDecodeError as e:
raise RuntimeError(f'cannot parse CLI --json output: {e}\n{proc.stdout[-300:]}')
def base_cmd(args) -> list:
root = Path(args.tools_root) / 'llm-fingerprint-detector'
cli = root / 'dist' / 'cli.js'
if not cli.exists():
raise FileNotFoundError(f'detector CLI not built: {cli} (run `npm run build` in the repo)')
return [args.node, str(cli)]
def endpoint_cmd(args) -> list:
return ['--base-url', args.api_url.rstrip('/'), '--model', args.model,
'--preset', args.preset, '--timeout', str(args.timeout * 1000),
'--concurrency', str(args.concurrency), '--json']
def main():
parser = argparse.ArgumentParser(description='llm-fingerprint-detector benchmark')
add_common_args(parser)
parser.add_argument('--tools-root', default='/data1/xii',
help='Directory containing the cloned llm-fingerprint-detector repo')
parser.add_argument('--node', default=os.environ.get('DETECTOR_NODE', 'node'),
help='Node executable (default: %(default)s)')
parser.add_argument('--reference', default=None,
help='Same-protocol reference fingerprint JSON; '
'omit for self-consistency mode')
parser.add_argument('--preset', default='standard',
choices=['quick', 'standard', 'strict'])
parser.add_argument('--concurrency', type=int, default=4)
args = parser.parse_args()
cmd = base_cmd(args) + endpoint_cmd(args)
if args.reference:
# ---- verify 模式:与参考指纹硬比较 ----
out = run_cli(cmd + ['verify', '--reference', args.reference], timeout=args.timeout * 40)
mean_jsd = float(out.get('meanJsd', out.get('comparison', {}).get('meanJsd', 1.0)))
verdict = out.get('verdict', 'insufficient')
mode = 'reference_verify'
reference = args.reference
split_half = None
cells = out.get('comparison', {}).get('cells') or out.get('cells') or []
else:
# ---- 自一致模式:采两次互相比较 ----
tmp_dir = Path(args.report_path).resolve().parent.parent / 'detector_tmp'
tmp_dir.mkdir(parents=True, exist_ok=True)
fa, fb = tmp_dir / 'fp_a.json', tmp_dir / 'fp_b.json'
run_a = run_cli(cmd + ['fingerprint', '--out', str(fa)], timeout=args.timeout * 40)
run_b = run_cli(cmd + ['fingerprint', '--out', str(fb)], timeout=args.timeout * 40)
cmp_out = run_cli(base_cmd(args) + ['compare', str(fa), str(fb), '--json'],
timeout=60)
mean_jsd = float(cmp_out.get('meanJsd', 1.0))
verdict = cmp_out.get('verdict', 'insufficient')
mode = 'self_consistency'
reference = None
split_half = (run_a.get('run') or {}).get('splitHalfJsd')
cells = cmp_out.get('cells') or []
score = max(0.0, min(1.0, 1.0 - mean_jsd))
write_report(
args.report_path, BENCHMARK_DETECTOR, score,
num=len(cells),
mode=mode,
verdict=verdict,
mean_jsd=mean_jsd,
split_half_jsd=split_half,
reference=reference,
preset=args.preset,
most_divergent=[
{'cell': c.get('cellId'), 'jsd': c.get('jsd')} for c in cells[:5]
],
)
print(f"[llm_fingerprint_detector] mode={mode} verdict={verdict} "
f"meanJSD={mean_jsd:.3f} -> score={score:.3f}")
if __name__ == '__main__':
main()

View File

@ -0,0 +1,114 @@
#!/usr/bin/env python3
"""LLM Verify fraud-detection benchmark runner.
对被测端点跑 LLM Verify 的一键深度分析identity/capability/fingerprint 三套件
32 条取证探测得到红旗与裁决并映射为 [0,1] 得分
run.py 以子进程方式调用解释器需带 fastapi/httpx/pydantic
默认 llmverify conda 环境
<verify-python> run_llm_verify.py --api-url ... --model ... --report-path ...
得分score [0,1]fail-closed证据不足绝不给高分
NO_FRAUD_SIGNALS -> 1.0 无欺诈信号且证据充分
INCONCLUSIVE -> 0.5 证据不足无法下结论
SUSPICIOUS -> 0.25 存在异常信号
FRAUD_DETECTED -> 0.0 多个独立强欺诈信号
"""
import argparse
import json
import os
import sys
from pathlib import Path
from common import BENCHMARK_LLM_VERIFY, add_common_args, write_report
VERDICT_SCORE = {
'NO_FRAUD_SIGNALS': 1.0,
'INCONCLUSIVE': 0.5,
'SUSPICIOUS': 0.25,
'FRAUD_DETECTED': 0.0,
}
def main():
parser = argparse.ArgumentParser(description='LLM Verify deep-analysis benchmark')
add_common_args(parser)
parser.add_argument('--tools-root', default='/data1/xii',
help='Directory containing the cloned llm-verify repo (default: %(default)s)')
parser.add_argument('--protocol', default='openai', choices=['openai', 'anthropic'],
help='API protocol spoken by the target (default: %(default)s)')
parser.add_argument('--suites', default='identity,capability,fingerprint',
help='Comma-separated prompt suites (default: %(default)s)')
# fail-closed 需要 >=8 条成功探测GLM 等思考模型较慢,放宽默认超时
parser.add_argument('--bench-timeout', type=int, default=90,
help='LLM Verify per-probe timeout seconds via BENCHMARK_TIMEOUT '
'(default: %(default)s)')
args = parser.parse_args()
verify_root = os.path.join(args.tools_root, 'llm-verify')
if not os.path.isdir(verify_root):
print(f'ERROR: llm-verify repo not found at {verify_root}')
sys.exit(1)
# 必须在导入 src.* 之前设置pydantic-settings 在模块导入时实例化
os.environ['BENCHMARK_TIMEOUT'] = str(args.bench_timeout)
os.environ.setdefault('MAX_CONCURRENT_CALLS', '5')
os.environ.pop('SUSPECT_API_BASE_URL', None) # 强制走命令行传入的 api_url
# 把 sqlite 工作库放到报告目录旁,避免污染仓库根目录
work_dir = Path(args.report_path).resolve().parent.parent
work_dir.mkdir(parents=True, exist_ok=True)
os.chdir(work_dir)
sys.path.insert(0, verify_root)
from fastapi.testclient import TestClient # 进程内调用 FastAPI无需起服务
from src.main import app
payload = {
'name': f'evalstone-fingerprint-{args.model}',
'model_configs': [{
'model_name': args.model,
'provider': 'suspect',
'protocol': args.protocol,
# 注意httpx 拒绝空 Bearer 头Illegal header value b'Bearer '
# 本地无鉴权端点也必须给非空占位 key
'api_key': os.environ.get('SUSPECT_API_KEY') or 'dummy',
'api_base_url': args.api_url,
}],
'suites': [s.strip() for s in args.suites.split(',') if s.strip()],
}
with TestClient(app) as client:
# 注意TestClient 不支持请求级 timeout单探测超时由 BENCHMARK_TIMEOUT 控制
resp = client.post('/api/v1/analysis/deep', json=payload)
if resp.status_code != 200:
print(f'ERROR: deep analysis failed: HTTP {resp.status_code}: {resp.text[:300]}')
sys.exit(1)
report = resp.json()
verdict = report.get('verdict', 'INCONCLUSIVE')
score = VERDICT_SCORE.get(verdict, 0.5)
total_probes, success_probes, avg_latency = 0, 0, None
for mr in report.get('model_reports', []):
total_probes += mr.get('total_probes', 0) or 0
success_probes += mr.get('successful_probes', 0) or 0
if mr.get('avg_latency_ms') is not None:
avg_latency = mr.get('avg_latency_ms')
write_report(
args.report_path, BENCHMARK_LLM_VERIFY, score,
num=total_probes,
verdict=verdict,
successful_probes=success_probes,
avg_latency_ms=avg_latency,
red_flags=report.get('red_flags', []),
summary=report.get('summary', ''),
)
print(f"[llm_verify] verdict={verdict} ({success_probes}/{total_probes} probes ok) "
f"-> score={score:.2f}")
if __name__ == '__main__':
main()

View File

@ -0,0 +1,136 @@
#!/usr/bin/env python3
"""LLMmap fingerprint benchmark runner.
把目标端点当作"未知模型"向其发送 LLMmap 8 条指纹查询收集回答后用
LLMmap 预训练 open-set 模型与 52 个已知模板比对输出 Top-K 及得分
必须用装好 torch/transformers 的解释器运行默认 llmmap conda 环境
run.py 以子进程方式调用
<llmmap-python> run_llmmap.py --api-url ... --model ... --report-path ...
得分score [0,1]
- 提供 --expected-model Top-1 模板与期望模型名匹配 1.0否则 0.0
匹配为归一化后的包含关系 "GLM-5.2" 可匹配 "zai-org/GLM-5.2"
- 未提供时置信度 score = max(0, 1 - top1_distance / --distance-scale)
"""
import argparse
import os
import sys
# 嵌入模型已缓存到本地,禁止联网检查更新
os.environ.setdefault('HF_HUB_OFFLINE', '1')
os.environ.setdefault('TRANSFORMERS_OFFLINE', '1')
from common import BENCHMARK_LLMMAP, add_common_args, chat_completion, write_report
def normalize_name(name: str) -> str:
"""小写并去掉组织前缀/斜杠/冒号后的空白,便于宽松匹配。"""
n = str(name).strip().lower()
if '/' in n:
n = n.split('/')[-1]
return n.replace('-', '').replace('_', '').replace('.', '')
def main():
parser = argparse.ArgumentParser(description='LLMmap fingerprint benchmark')
add_common_args(parser)
parser.add_argument('--tools-root', default='/data1/xii',
help='Directory containing the cloned LLMmap repo (default: %(default)s)')
parser.add_argument('--llmmap-model-path', default=None,
help='Pretrained LLMmap open-set model directory '
'(default: <tools-root>/LLMmap/data/pretrained_models/default)')
parser.add_argument('--device', default='cpu', choices=['cpu', 'cuda'])
parser.add_argument('--temperature', type=float, default=0.7,
help='Sampling temperature when querying the target (default: %(default)s)')
parser.add_argument('--max-tokens', type=int, default=512,
help='Max tokens per target answer (default: %(default)s)')
parser.add_argument('--expected-model', default=None,
help='Ground-truth model identity; when set, score is a strict match flag')
parser.add_argument('--distance-scale', type=float, default=60.0,
help='Confidence normalizer when no expected model is given '
'(observed: same-family ~20, others ~40+)')
parser.add_argument('-k', type=int, default=5, help='Top-K templates to record')
args = parser.parse_args()
llmmap_root = os.path.join(args.tools_root, 'LLMmap')
if not os.path.isdir(llmmap_root):
print(f'ERROR: LLMmap repo not found at {llmmap_root}')
sys.exit(1)
model_path = args.llmmap_model_path or os.path.join(
llmmap_root, 'data', 'pretrained_models', 'default')
sys.path.insert(0, llmmap_root)
from LLMmap.inference import load_LLMmap
conf, llmmap = load_LLMmap(model_path, device=args.device)
# 逐条向被测端点发送指纹查询
extra_body = None if args.thinking else {'chat_template_kwargs': {'thinking': False}}
answers, errors = [], []
for i, query in enumerate(llmmap.queries, 1):
content, err = chat_completion(
args.api_url, args.model, query,
temperature=args.temperature, max_tokens=args.max_tokens,
timeout=args.timeout, extra_body=extra_body,
)
if err:
print(f' query {i}/{len(llmmap.queries)} failed: {err}')
errors.append({'query_index': i - 1, 'error': err})
content = ''
else:
print(f' query {i}/{len(llmmap.queries)} ok ({len(content)} chars)')
answers.append(content or '')
# 与已知模板比对open-set 距离检索)
# 端点大面积失败时回答为空,距离毫无意义 —— 直接判失败而不是给假分数
n_ok = len(answers) - len(errors)
if n_ok <= len(answers) // 2:
write_report(
args.report_path, BENCHMARK_LLMMAP, 0.0,
num=len(answers),
score_mode='error',
top1=None,
topk=[],
expected_model=args.expected_model,
n_query_errors=len(errors),
query_errors=errors[:5],
error=f'too many failed queries ({len(errors)}/{len(answers)}); '
f'is the endpoint up and serving --model?',
)
print(f'[llmmap] FAILED: {len(errors)}/{len(answers)} queries errored')
sys.exit(1)
distances = llmmap(answers)
order = sorted(range(len(distances)), key=lambda i: distances[i])
label_map = llmmap.label_map # {index: template_name}
topk = [{'name': label_map[i], 'distance': float(distances[i])}
for i in order[:max(1, args.k)]]
top1_name, top1_dist = topk[0]['name'], topk[0]['distance']
if args.expected_model:
matched = normalize_name(args.expected_model) in normalize_name(top1_name) or \
normalize_name(top1_name) in normalize_name(args.expected_model)
score = 1.0 if matched else 0.0
score_mode = 'identity_match'
else:
score = max(0.0, 1.0 - float(top1_dist) / args.distance_scale)
score_mode = 'confidence'
write_report(
args.report_path, BENCHMARK_LLMMAP, score,
num=len(answers),
score_mode=score_mode,
top1=topk[0],
topk=topk,
expected_model=args.expected_model,
n_query_errors=len(errors),
query_errors=errors[:5],
)
print(f"[llmmap] Top-1: {top1_name} (distance={top1_dist:.4f}) -> score={score:.4f}")
if __name__ == '__main__':
main()

View File

@ -31,6 +31,7 @@ Examples:
import argparse import argparse
import json import json
import os import os
import subprocess
import sys import sys
import threading import threading
import time import time
@ -59,13 +60,19 @@ sys.path.insert(0, str(SCRIPT_DIR))
import collect_results as collect_results_module import collect_results as collect_results_module
import perf_backup as perf_backup_module import perf_backup as perf_backup_module
# Make fingerprint helpers importable失败报告写入用
sys.path.insert(0, str(SCRIPT_DIR / "fingerprint"))
import common as fingerprint_common
# ============================================================ # ============================================================
# Default configuration (override via CLI) # Default configuration (override via CLI)
# ============================================================ # ============================================================
DEFAULT_MODEL = 'DeepSeek-V4-Flash-Int8' DEFAULT_MODEL = 'DeepSeek-V4-Flash-Int8'
DEFAULT_API_URL = 'http://localhost:30000/v1' DEFAULT_API_URL = 'http://localhost:30000/v1'
DEFAULT_DATASET_DIR = str(PROJECT_ROOT) # 数据集缓存根evalscope 会在其下找 datasets/<名字>-<hash>。
# 镜像内通过 EVALSTONE_DATASET_DIR 指到挂载卷,宿主机目录直接命中已有缓存。
DEFAULT_DATASET_DIR = os.environ.get('EVALSTONE_DATASET_DIR', str(PROJECT_ROOT))
DEFAULT_OUTPUT_DIR = str(PROJECT_ROOT / 'output') DEFAULT_OUTPUT_DIR = str(PROJECT_ROOT / 'output')
DEFAULT_CONFIG = str(PROJECT_ROOT / 'config' / 'dpv4-int8_nothinking.yaml') DEFAULT_CONFIG = str(PROJECT_ROOT / 'config' / 'dpv4-int8_nothinking.yaml')
DEFAULT_TOKENIZER_PATH = '/data1/models/DeepSeek-V4-Flash-INT8' DEFAULT_TOKENIZER_PATH = '/data1/models/DeepSeek-V4-Flash-INT8'
@ -121,6 +128,23 @@ ALL_SINGLE_RUN = [
ALL_AGENT = ['tau2_bench', 'general_fc'] ALL_AGENT = ['tau2_bench', 'general_fc']
K3_SINGLE = ["gpqa_diamond", "hle", "terminal_bench_v2", "browsecomp", "mcp_atlas", "officeqa", "deepsearchqa", "jobbench", "automation_bench"] K3_SINGLE = ["gpqa_diamond", "hle", "terminal_bench_v2", "browsecomp", "mcp_atlas", "officeqa", "deepsearchqa", "jobbench", "automation_bench"]
# ============================================================
# Fingerprint / model-identity benchmarks
# ============================================================
# 这三个 benchmark 不经过 EvalScope 数据集管线:由 bash/fingerprint/ 下的
# 执行器直接探测 --api-url 端点,并产出与 EvalScope 同构的
# output/<folder>/<benchmark>/seed_<seed>/reports/<benchmark>.json含 score
# collect_results.py 可像普通 benchmark 一样汇总。
ALL_FINGERPRINT = ['llmmap', 'llm_verify', 'llm_fingerprint_detector']
FINGERPRINT_SCRIPTS = {
'llmmap': SCRIPT_DIR / 'fingerprint' / 'run_llmmap.py',
'llm_verify': SCRIPT_DIR / 'fingerprint' / 'run_llm_verify.py',
'llm_fingerprint_detector': SCRIPT_DIR / 'fingerprint' / 'run_llm_detector.py',
}
DEFAULT_TOOLS_ROOT = os.environ.get('FP_TOOLS_ROOT', '/data1/xii')
# 单个指纹 benchmark 的整体子进程超时。verify 的 32 条探测较慢,给足余量。
FP_OVERALL_TIMEOUT = 7200
# 分组基于 CSV 单次时间 + multi-run 后的 wall time 平衡: # 分组基于 CSV 单次时间 + multi-run 后的 wall time 平衡:
# Group1: ~61h | Group2: ~62h | Group3: ~55h # Group1: ~61h | Group2: ~62h | Group3: ~55h
SUITES = { SUITES = {
@ -175,6 +199,13 @@ SUITES = {
'tau2_bench' 'tau2_bench'
], ],
}, },
# 模型指纹/安全套件LLMmap 身份识别 + LLM Verify 欺诈检测 + 单 token 分布验证
'fingerprint': {
'multi': [],
'single': [],
'agent': [],
'fingerprint': ALL_FINGERPRINT,
},
} }
# ============================================================ # ============================================================
@ -291,6 +322,32 @@ def build_parser():
parser.add_argument('--truncation-tokens', type=int, default=DEFAULT_TRUNCATION_TOKENS, parser.add_argument('--truncation-tokens', type=int, default=DEFAULT_TRUNCATION_TOKENS,
help='Middle-truncation token budget for long-context benchmarks (default: %(default)s)') help='Middle-truncation token budget for long-context benchmarks (default: %(default)s)')
# Fingerprint benchmarks (llmmap / llm_verify / llm_fingerprint_detector)
parser.add_argument('--tools-root', default=DEFAULT_TOOLS_ROOT,
help='Root dir containing the three fingerprint tool repos '
'(LLMmap/, llm-verify/, llm-fingerprint-detector/) (default: %(default)s)')
parser.add_argument('--llmmap-python',
default=os.environ.get('LLMMAP_PYTHON',
'/root/miniconda3/envs/llmmap/bin/python'),
help='Python interpreter with torch/transformers for the LLMmap runner')
parser.add_argument('--verify-python',
default=os.environ.get('LLMVERIFY_PYTHON',
'/root/miniconda3/envs/llmverify/bin/python'),
help='Python interpreter with fastapi/httpx for the LLM Verify runner')
parser.add_argument('--detector-node', default=os.environ.get('DETECTOR_NODE', 'node'),
help='Node executable for the llm-fingerprint-detector runner')
parser.add_argument('--detector-reference', default=None,
help='Optional same-protocol reference fingerprint JSON for the detector; '
'omit to run in self-consistency mode')
parser.add_argument('--detector-preset', default='standard',
choices=['quick', 'standard', 'strict'],
help='Sampling preset for the detector benchmark (default: %(default)s)')
parser.add_argument('--expected-model', default=None,
help='Ground-truth model identity; when set, the llmmap score becomes a '
'strict Top-1 identity match flag instead of a distance confidence')
parser.add_argument('--fingerprint-timeout', type=int, default=120,
help='Per-request timeout (seconds) passed to fingerprint runners (default: %(default)s)')
# Result collection # Result collection
parser.add_argument('--no-summary', dest='write_summary', action='store_false', parser.add_argument('--no-summary', dest='write_summary', action='store_false',
help='Skip writing summary Excel/CSV after each benchmark') help='Skip writing summary Excel/CSV after each benchmark')
@ -668,11 +725,14 @@ def main():
single_run = [d for d in custom if d not in MULTI_RUN_CONFIG] single_run = [d for d in custom if d not in MULTI_RUN_CONFIG]
agent = [d for d in custom if d in ALL_AGENT] agent = [d for d in custom if d in ALL_AGENT]
single_run = [d for d in single_run if d not in ALL_AGENT] single_run = [d for d in single_run if d not in ALL_AGENT]
fingerprint = [d for d in custom if d in ALL_FINGERPRINT]
single_run = [d for d in single_run if d not in ALL_FINGERPRINT]
else: else:
suite = SUITES[args.suite] suite = SUITES[args.suite]
multi_run = list(suite['multi']) multi_run = list(suite['multi'])
single_run = list(suite['single']) single_run = list(suite['single'])
agent = list(suite['agent']) agent = list(suite['agent'])
fingerprint = list(suite.get('fingerprint', []))
# Apply --exclude # Apply --exclude
if args.exclude: if args.exclude:
@ -680,6 +740,7 @@ def main():
multi_run = [d for d in multi_run if d not in exclude] multi_run = [d for d in multi_run if d not in exclude]
single_run = [d for d in single_run if d not in exclude] single_run = [d for d in single_run if d not in exclude]
agent = [d for d in agent if d not in exclude] agent = [d for d in agent if d not in exclude]
fingerprint = [d for d in fingerprint if d not in exclude]
judge_model_args = { judge_model_args = {
'model_id': args.judge_model, 'model_id': args.judge_model,
@ -742,6 +803,7 @@ def main():
print(f'Multi-run datasets: {multi_run}') print(f'Multi-run datasets: {multi_run}')
print(f'Single-run datasets: {single_run}') print(f'Single-run datasets: {single_run}')
print(f'Agent datasets: {agent}') print(f'Agent datasets: {agent}')
print(f'Fingerprint datasets: {fingerprint}')
print(f'Write summary: {args.write_summary}') print(f'Write summary: {args.write_summary}')
print('=' * 60) print('=' * 60)
@ -755,8 +817,91 @@ def main():
f'max_tokens={DEFAULT_GENERATION_CONFIG["max_tokens"]})') f'max_tokens={DEFAULT_GENERATION_CONFIG["max_tokens"]})')
return {'generation_config': deepcopy(DEFAULT_GENERATION_CONFIG)} return {'generation_config': deepcopy(DEFAULT_GENERATION_CONFIG)}
def run_fingerprint_benchmark(dataset_name: str, benchmark_names: list,
write_summary_flag: bool = True,
summary_lock: threading.Lock = None):
"""Run one fingerprint benchmark via its standalone runner script.
产出与 EvalScope 一致的 output/<folder>/<bench>/seed_<seed>/reports/*.json
并复用 active_time 计时与 perf 备份保证 collect_results 可直接汇总
"""
work_dir = model_output_dir / dataset_name / f'seed_{args.seed}'
report_path = work_dir / 'reports' / f'{dataset_name}.json'
script = FINGERPRINT_SCRIPTS[dataset_name]
common_cmd = [
'--api-url', args.api_url,
'--model', args.model,
'--report-path', str(report_path),
'--timeout', str(args.fingerprint_timeout),
]
if dataset_name == 'llmmap':
cmd = [args.llmmap_python, str(script), *common_cmd,
'--tools-root', args.tools_root]
if args.expected_model:
cmd += ['--expected-model', args.expected_model]
elif dataset_name == 'llm_verify':
cmd = [args.verify_python, str(script), *common_cmd,
'--tools-root', args.tools_root]
else: # llm_fingerprint_detector
cmd = [sys.executable, str(script), *common_cmd,
'--tools-root', args.tools_root,
'--preset', args.detector_preset,
'--concurrency', '4']
if args.detector_node and args.detector_node != 'node':
cmd += ['--node', args.detector_node]
if args.detector_reference:
cmd += ['--reference', args.detector_reference]
print(f"\n{'='*60}")
print(f'Running: {dataset_name} (fingerprint benchmark, seed={args.seed})')
print(f"{'='*60}")
start_ts = time.monotonic()
try:
proc = subprocess.run(cmd, capture_output=True, text=True,
timeout=FP_OVERALL_TIMEOUT)
tail = '\n'.join((proc.stdout or '').strip().splitlines()[-20:])
if tail:
print(tail)
if proc.returncode != 0:
err_tail = '\n'.join((proc.stderr or '').strip().splitlines()[-10:])
print(f'ERROR in {dataset_name}: exit={proc.returncode}\n{err_tail}')
fingerprint_common.write_report(
str(report_path), dataset_name, 0.0, num=0,
error=f'runner exited with code {proc.returncode}',
stderr_tail=err_tail[-800:],
)
except Exception as e:
print(f'ERROR in {dataset_name}: {e}')
fingerprint_common.write_report(
str(report_path), dataset_name, 0.0, num=0,
error=str(e),
)
finally:
elapsed = time.monotonic() - start_ts
perf_backup_module.record_active_time(str(model_output_dir), dataset_name,
args.model, elapsed)
print(f'Active time for {dataset_name}: {elapsed:.1f}s')
backup_after_run(str(model_output_dir), dataset_name, args.model, work_dir)
if write_summary_flag:
if summary_lock is not None:
with summary_lock:
write_summary(str(model_output_dir), args.model, folder_name,
benchmark_names=benchmark_names)
else:
write_summary(str(model_output_dir), args.model, folder_name,
benchmark_names=benchmark_names)
def run_one(dataset_name, run_idx=0, benchmark_names=None, write_summary_flag=True, def run_one(dataset_name, run_idx=0, benchmark_names=None, write_summary_flag=True,
summary_lock=None): summary_lock=None):
if dataset_name in FINGERPRINT_SCRIPTS:
run_fingerprint_benchmark(dataset_name, benchmark_names or [dataset_name],
write_summary_flag=write_summary_flag,
summary_lock=summary_lock)
return
ds_cfg = get_dataset_config(dataset_name) ds_cfg = get_dataset_config(dataset_name)
task_cfg = build_task_config( task_cfg = build_task_config(
dataset_name, ds_cfg, args.batch_size, enable_thinking, args.seed, limit, dataset_name, ds_cfg, args.batch_size, enable_thinking, args.seed, limit,
@ -791,6 +936,9 @@ def main():
for dataset_name in agent: for dataset_name in agent:
benchmark_names.append(dataset_name) benchmark_names.append(dataset_name)
benchmark_units.append((dataset_name, 'agent')) benchmark_units.append((dataset_name, 'agent'))
for dataset_name in fingerprint:
benchmark_names.append(dataset_name)
benchmark_units.append((dataset_name, 'fingerprint'))
def run_benchmark_unit(dataset_name: str, kind: str, summary_lock=None): def run_benchmark_unit(dataset_name: str, kind: str, summary_lock=None):
"""Run one benchmark (all seeds/runs) and return its name.""" """Run one benchmark (all seeds/runs) and return its name."""