115 lines
4.6 KiB
Python
115 lines
4.6 KiB
Python
#!/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()
|