评估集HELM/AlpacaEval在线回归测试

import asyncioimport hashlibimport jsonimport timefrom dataclasses import dataclass, fieldfrom typing import Dict, List, Optionalfrom datetime import datetime @dataclassclass EvalConfig:    model_name: str    model_version: str    eval_suite: str  # "helm" or "alpaca_eval"    dataset_path: str    concurrency: int = 5    rate_limit_rpm: int = 60    cache_ttl_seconds: int = 86400 @dataclassclass EvalResult:    model_name: str    model_version: str    eval_suite: str    timestamp: str    metrics: Dict[str, float]    num_samples: int    duration_seconds: float class CacheManager:    def __init__(self):        self._cache: Dict[str, dict] = {}     def _make_key(self, model_version: str, dataset_path: str) -> str:        raw = f"{model_version}:{dataset_path}"        return hashlib.md5(raw.encode()).hexdigest()     def get(self, model_version: str, dataset_path: str) -> Optional[dict]:        key = self._make_key(model_version, dataset_path)        entry = self._cache.get(key)        if entry and time.time() - entry["ts"] < 86400:            return entry["data"]        return None     def put(self, model_version: str, dataset_path: str, data: dict):        key = self._make_key(model_version, dataset_path)        self._cache[key] = {"data": data, "ts": time.time()} class RateLimiter:    def __init__(self, rpm: int):        self._interval = 60.0 / rpm        self._last_call = 0.0     async def acquire(self):        now = time.time()        wait = self._interval - (now - self._last_call)        if wait > 0:            await asyncio.sleep(wait)        self._last_call = time.time() class EvalPipeline:    def __init__(self, config: EvalConfig):        self.config = config        self.cache = CacheManager()        self.limiter = RateLimiter(config.rate_limit_rpm)        self.semaphore = asyncio.Semaphore(config.concurrency)     async def _run_single(self, sample: dict) -> dict:        async with self.semaphore:            await self.limiter.acquire()            # 调用模型推理接口            response = await self._call_model(sample["input"])            # 调用评判逻辑            score = await self._judge(sample, response)            return {"input": sample["input"], "response": response, "score": score}     async def _call_model(self, prompt: str) -> str:        await asyncio.sleep(0.01)  # 模拟API调用        return f"model_response_for: {prompt[:50]}"     async def _judge(self, sample: dict, response: str) -> float:        await asyncio.sleep(0.01)  # 模拟评判调用        return 0.85     async def run(self) -> EvalResult:        cached = self.cache.get(self.config.model_version, self.config.dataset_path)        if cached:            print(f"[Cache Hit] {self.config.model_version}")            return EvalResult(**cached)         with open(self.config.dataset_path, "r") as f:            dataset = json.load(f)         start = time.time()        tasks = [self._run_single(s) for s in dataset]        results = await asyncio.gather(*tasks)        duration = time.time() - start         metrics = self._aggregate(results)        eval_result = EvalResult(            model_name=self.config.model_name,            model_version=self.config.model_version,            eval_suite=self.config.eval_suite,            timestamp=datetime.now().isoformat(),            metrics=metrics,            num_samples=len(results),            duration_seconds=duration        )        self.cache.put(self.config.model_version, self.config.dataset_path,                       eval_result.__dict__)        return eval_result     def _aggregate(self, results: List[dict]) -> Dict[str, float]:        scores = [r["score"] for r in results]        return {            "mean_score": sum(scores) / len(scores),            "min_score": min(scores),            "max_score": max(scores),            "num_samples": len(scores)        }

from enum import Enumfrom typing import List, Tuple class SeverityLevel(Enum):    P0_BLOCK = "p0_block"       # 核心指标回归，阻塞部署    P1_WARN = "p1_warn"         # 非核心指标回归，告警    P2_INFO = "p2_info"         # 趋势异常，通知 class RegressionGate:    def __init__(self, thresholds: dict):        self.thresholds = thresholds  # {"accuracy": -0.02, "win_rate": -0.03}     def check(self, new_metrics: dict, baseline_metrics: dict) -> List[Tuple[str, SeverityLevel, float]]:        alerts = []        for metric, threshold in self.thresholds.items():            new_val = new_metrics.get(metric, 0)            old_val = baseline_metrics.get(metric, 0)            if old_val == 0:                continue            delta = (new_val - old_val) / old_val            if delta < threshold:                severity = SeverityLevel.P0_BLOCK if metric in ["accuracy", "win_rate"] else SeverityLevel.P1_WARN                alerts.append((metric, severity, delta))        return alerts     def should_block_deployment(self, alerts: List[Tuple[str, SeverityLevel, float]]) -> bool:        return any(sev == SeverityLevel.P0_BLOCK for _, sev, _ in alerts) # CI/CD集成示例def ci_cd_pipeline(model_version: str):    gate = RegressionGate({"accuracy": -0.02, "win_rate": -0.03, "robustness": -0.05})    # 运行评测...    # new_metrics = pipeline.run()    # alerts = gate.check(new_metrics, baseline)    # if gate.should_block_deployment(alerts):    #     notify_and_block()    # else:    #     proceed_deployment()    pass

import asyncioimport hashlibimport jsonimport timefrom dataclasses import dataclass, fieldfrom typing import Dict, List, Optional, Tuplefrom datetime import datetimefrom enum import Enum @dataclassclass EvalConfig:    model_name: str    model_version: str    eval_suite: str    dataset_path: str    concurrency: int = 5    rate_limit_rpm: int = 60 @dataclassclass EvalResult:    model_name: str    model_version: str    eval_suite: str    timestamp: str    metrics: Dict[str, float]    num_samples: int class CacheManager:    def __init__(self):        self._cache: Dict[str, dict] = {}     def _make_key(self, model_version: str, dataset_path: str) -> str:        raw = f"{model_version}:{dataset_path}"        return hashlib.md5(raw.encode()).hexdigest()     def get(self, model_version: str, dataset_path: str) -> Optional[dict]:        key = self._make_key(model_version, dataset_path)        entry = self._cache.get(key)        if entry and time.time() - entry["ts"] < 86400:            return entry["data"]        return None     def put(self, model_version: str, dataset_path: str, data: dict):        key = self._make_key(model_version, dataset_path)        self._cache[key] = {"data": data, "ts": time.time()} class RateLimiter:    def __init__(self, rpm: int):        self._interval = 60.0 / rpm        self._last_call = 0.0     async def acquire(self):        now = time.time()        wait = self._interval - (now - self._last_call)        if wait > 0:            await asyncio.sleep(wait)        self._last_call = time.time() class RegressionGate:    def __init__(self, thresholds: Dict[str, float]):        self.thresholds = thresholds     def check(self, new_metrics: dict, baseline: dict) -> List[Tuple[str, str, float]]:        alerts = []        for metric, threshold in self.thresholds.items():            new_val = new_metrics.get(metric, 0)            old_val = baseline.get(metric, 0)            if old_val == 0:                continue            delta = (new_val - old_val) / old_val            if delta < threshold:                severity = "P0_BLOCK" if metric in ["accuracy", "win_rate"] else "P1_WARN"                alerts.append((metric, severity, delta))        return alerts     def should_block(self, alerts: List[Tuple[str, str, float]]) -> bool:        return any(sev == "P0_BLOCK" for _, sev, _ in alerts) class EvalPipeline:    def __init__(self, config: EvalConfig):        self.config = config        self.cache = CacheManager()        self.limiter = RateLimiter(config.rate_limit_rpm)        self.semaphore = asyncio.Semaphore(config.concurrency)     async def _run_single(self, sample: dict) -> dict:        async with self.semaphore:            await self.limiter.acquire()            response = await self._call_model(sample["input"])            score = await self._judge(sample, response)            return {"input": sample["input"], "score": score}     async def _call_model(self, prompt: str) -> str:        await asyncio.sleep(0.01)        return f"response: {prompt[:30]}"     async def _judge(self, sample: dict, response: str) -> float:        await asyncio.sleep(0.01)        return 0.85     async def run(self) -> EvalResult:        cached = self.cache.get(self.config.model_version, self.config.dataset_path)        if cached:            return EvalResult(**cached)        with open(self.config.dataset_path, "r") as f:            dataset = json.load(f)        tasks = [self._run_single(s) for s in dataset]        results = await asyncio.gather(*tasks)        scores = [r["score"] for r in results]        metrics = {"mean_score": sum(scores) / len(scores)}        return EvalResult(            model_name=self.config.model_name,            model_version=self.config.model_version,            eval_suite=self.config.eval_suite,            timestamp=datetime.now().isoformat(),            metrics=metrics,            num_samples=len(results)        ) def ci_cd_regression_test(model_version: str, baseline_metrics: dict):    config = EvalConfig(model_name="llm-v2", model_version=model_version,                        eval_suite="helm", dataset_path="eval_data.json")    pipeline = EvalPipeline(config)    result = asyncio.run(pipeline.run())    gate = RegressionGate({"accuracy": -0.02, "win_rate": -0.03})    alerts = gate.check(result.metrics, baseline_metrics)    if gate.should_block(alerts):        print(f"BLOCKED: Regression detected in {model_version}")        return False    print(f"PASSED: {model_version} deployed successfully")    return True