建LLM評(píng)估框架：實(shí)戰(zhàn)Harness實(shí)現(xiàn)與多模型對(duì)比測(cè)試)
1. 項(xiàng)目緣起從“黑盒”到“白盒”的探索最近在折騰大模型應(yīng)用開(kāi)發(fā)時(shí)我遇到了一個(gè)挺典型的問(wèn)題手頭有好幾個(gè)不同的LLM大語(yǔ)言模型服務(wù)比如OpenAI的GPT、Claude還有幾個(gè)開(kāi)源的模型。每次想測(cè)試一個(gè)新提示詞Prompt在不同模型上的效果或者對(duì)比不同模型對(duì)同一任務(wù)的響應(yīng)質(zhì)量都得手動(dòng)一個(gè)個(gè)去調(diào)用API然后把返回結(jié)果復(fù)制粘貼到文檔里再人工去比對(duì)。這個(gè)過(guò)程不僅繁瑣低效而且很難做到標(biāo)準(zhǔn)化評(píng)估比如響應(yīng)時(shí)間、輸出格式一致性、成本消耗這些關(guān)鍵指標(biāo)全靠感覺(jué)。這讓我想起了軟件測(cè)試?yán)锏摹皽y(cè)試工具鏈”。我們不會(huì)手動(dòng)去點(diǎn)每一個(gè)按鈕來(lái)測(cè)試功能而是會(huì)用像JUnit、Pytest這樣的框架來(lái)編寫自動(dòng)化測(cè)試用例。那么對(duì)于LLM應(yīng)用是不是也應(yīng)該有這樣一個(gè)“測(cè)試框架”呢這就是我接觸到“Harness”概念的起點(diǎn)。在AI工程領(lǐng)域Harness這個(gè)詞常被用來(lái)指代一套用于評(píng)估、測(cè)試和管理AI模型特別是LLM的框架或工具集。它的核心目標(biāo)是把模型評(píng)估這個(gè)事從隨意、手工的狀態(tài)變成系統(tǒng)化、自動(dòng)化、可重復(fù)的過(guò)程。網(wǎng)上確實(shí)有一些現(xiàn)成的工具比如著名的lm-evaluation-harness現(xiàn)在常被稱為EleutherAI LM Harness。它功能強(qiáng)大覆蓋了海量的評(píng)測(cè)任務(wù)。但我在實(shí)際想用的時(shí)候發(fā)現(xiàn)了一些痛點(diǎn)一是它通常更偏向于學(xué)術(shù)界的基準(zhǔn)測(cè)試對(duì)于我這種想快速驗(yàn)證業(yè)務(wù)場(chǎng)景提示詞效果的開(kāi)發(fā)者來(lái)說(shuō)有點(diǎn)“重”二是它的配置和擴(kuò)展對(duì)于不熟悉其代碼結(jié)構(gòu)的人來(lái)說(shuō)學(xué)習(xí)成本不低三也是最重要的我想徹底搞明白這背后的機(jī)制而不是當(dāng)一個(gè)“調(diào)包俠”。知其然更要知其所以然自己動(dòng)手實(shí)現(xiàn)一個(gè)輕量化的、貼合自身需求的Harness就成了一個(gè)很有吸引力的挑戰(zhàn)。所以這個(gè)項(xiàng)目的標(biāo)題“實(shí)戰(zhàn)復(fù)盤我是如何用代碼實(shí)現(xiàn) Harness 的”就是記錄我從零開(kāi)始用Python設(shè)計(jì)和編碼構(gòu)建一個(gè)屬于我自己的、輕量級(jí)LLM評(píng)估工具鏈的全過(guò)程。它不追求大而全而是聚焦于解決我實(shí)際開(kāi)發(fā)中的痛點(diǎn)自動(dòng)化、標(biāo)準(zhǔn)化地對(duì)比不同LLM API對(duì)同一組提示詞的響應(yīng)。2. 核心設(shè)計(jì)思路一個(gè)輕量評(píng)估框架的藍(lán)圖在開(kāi)始敲代碼之前我得先想清楚這個(gè)自研Harness到底要干什么以及怎么干。大方向是自動(dòng)化評(píng)估但具體落到設(shè)計(jì)上需要拆解出幾個(gè)核心模塊。2.1 需求定義與架構(gòu)選型我的核心需求很明確多模型支持能夠方便地接入不同的LLM服務(wù)提供商如OpenAI、Anthropic、Google等以及本地模型。任務(wù)定義能夠以結(jié)構(gòu)化的方式定義一組“評(píng)測(cè)任務(wù)”每個(gè)任務(wù)包含輸入提示詞Prompt和預(yù)期的評(píng)估標(biāo)準(zhǔn)不一定是標(biāo)準(zhǔn)答案可能是評(píng)分規(guī)則。自動(dòng)化執(zhí)行框架能自動(dòng)遍歷所有任務(wù)調(diào)用配置的模型執(zhí)行推理并收集結(jié)果。結(jié)果收集與比對(duì)將不同模型對(duì)同一任務(wù)的結(jié)果并排收集方便人工或自動(dòng)比對(duì)。需要記錄關(guān)鍵指標(biāo)如響應(yīng)內(nèi)容、耗時(shí)、Token使用量、成本如果API計(jì)費(fèi)等。可擴(kuò)展性易于添加新的模型接口、新的評(píng)估指標(biāo)或新的任務(wù)類型?；谶@些需求一個(gè)典型的分層架構(gòu)浮現(xiàn)在腦海中任務(wù)層 (Task Layer)負(fù)責(zé)定義和管理具體的評(píng)測(cè)任務(wù)。一個(gè)任務(wù)就是一個(gè)(prompt, evaluation_criteria)對(duì)。執(zhí)行層 (Execution Layer)核心引擎。負(fù)責(zé)加載任務(wù)調(diào)用配置好的模型執(zhí)行推理并處理可能的錯(cuò)誤如網(wǎng)絡(luò)超時(shí)、API限流。模型適配層 (Model Adapter Layer)這是實(shí)現(xiàn)多模型支持的關(guān)鍵。為每個(gè)支持的LLM服務(wù)編寫一個(gè)統(tǒng)一的“適配器”Adapter將框架內(nèi)部的通用請(qǐng)求格式轉(zhuǎn)換為特定API所需的格式并解析其返回結(jié)果。結(jié)果層 (Result Layer)定義標(biāo)準(zhǔn)化的結(jié)果數(shù)據(jù)結(jié)構(gòu)并負(fù)責(zé)將不同模型對(duì)同一任務(wù)的結(jié)果聚合、持久化如保存為JSON或CSV文件并生成易于閱讀的報(bào)告如Markdown表格。為什么不直接用langchain這類框架langchain的核心是構(gòu)建復(fù)雜的工作流鏈Chain其LLM類雖然也統(tǒng)一了接口但它的設(shè)計(jì)重心在于鏈?zhǔn)秸{(diào)用和記憶等高級(jí)功能對(duì)于純粹的、批量的、帶對(duì)比的模型評(píng)測(cè)場(chǎng)景封裝得不夠直接。自己實(shí)現(xiàn)可以更聚焦控制每一個(gè)細(xì)節(jié)比如精確計(jì)時(shí)、自定義重試邏輯、更靈活的結(jié)果收集。2.2 關(guān)鍵技術(shù)點(diǎn)與工具選型編程語(yǔ)言毫無(wú)疑問(wèn)是Python。它在AI和數(shù)據(jù)科學(xué)領(lǐng)域的生態(tài)是無(wú)與倫比的有豐富的HTTP客戶端、數(shù)據(jù)處理和序列化庫(kù)。異步 vs 同步考慮到可能會(huì)批量測(cè)試數(shù)十上百個(gè)提示詞使用異步IO可以大幅提升效率尤其是在網(wǎng)絡(luò)請(qǐng)求成為瓶頸時(shí)。Python的asyncio庫(kù)和aiohttp是首選。配置管理模型的API Key、Base URL等敏感信息不能硬編碼。使用pydantic配合.env文件或config.yaml來(lái)管理配置既安全又靈活。數(shù)據(jù)序列化任務(wù)定義和結(jié)果存儲(chǔ)使用JSON格式因?yàn)樗强缯Z(yǔ)言、人類可讀的并且Python的json模塊支持得很好。對(duì)于最終報(bào)告可以結(jié)合pandas將結(jié)果轉(zhuǎn)為DataFrame再輸出為CSV或Markdown。錯(cuò)誤處理與重試網(wǎng)絡(luò)請(qǐng)求不穩(wěn)定API有速率限制。必須實(shí)現(xiàn)健壯的錯(cuò)誤處理和指數(shù)退避的重試機(jī)制。tenacity庫(kù)是一個(gè)很好的選擇。這個(gè)設(shè)計(jì)藍(lán)圖看起來(lái)清晰了接下來(lái)就是把這些模塊用代碼搭建起來(lái)。3. 分步實(shí)現(xiàn)從零搭建代碼Harness我將按照自底向上的順序先實(shí)現(xiàn)最基礎(chǔ)的模型適配器再構(gòu)建執(zhí)行引擎最后定義任務(wù)和結(jié)果處理。3.1 第一步構(gòu)建統(tǒng)一的模型適配器接口適配器的目標(biāo)是對(duì)外提供統(tǒng)一的generate(prompt: str) - str方法對(duì)內(nèi)處理各自API的差異。首先定義一個(gè)抽象基類規(guī)定所有適配器必須實(shí)現(xiàn)的方法。# model_adapters/base.py import abc from typing import Optional, Dict, Any from pydantic import BaseModel class ModelResponse(BaseModel): 標(biāo)準(zhǔn)化的模型響應(yīng)結(jié)構(gòu) content: str # 模型返回的文本內(nèi)容 model_name: str # 模型標(biāo)識(shí)如 gpt-4 prompt_tokens: Optional[int] None completion_tokens: Optional[int] None total_tokens: Optional[int] None latency: Optional[float] None # 請(qǐng)求耗時(shí)單位秒 cost: Optional[float] None # 估算成本單位美元或其他貨幣 class BaseModelAdapter(abc.ABC): 模型適配器抽象基類 def __init__(self, model_name: str, **kwargs): self.model_name model_name # 可以在這里初始化客戶端如OpenAI客戶端 self.client self._initialize_client(**kwargs) abc.abstractmethod def _initialize_client(self, **kwargs): 初始化特定模型的客戶端 pass abc.abstractmethod async def generate_async(self, prompt: str, **generation_params) - ModelResponse: 異步生成文本的核心方法。 :param prompt: 輸入的提示詞 :param generation_params: 模型特定的生成參數(shù)溫度、top_p等 :return: 標(biāo)準(zhǔn)化的ModelResponse對(duì)象 pass # 也可以提供一個(gè)同步版本作為備選 def generate_sync(self, prompt: str, **kwargs) - ModelResponse: import asyncio return asyncio.run(self.generate_async(prompt, **kwargs))接下來(lái)實(shí)現(xiàn)一個(gè)具體的適配器比如OpenAI的。# model_adapters/openai_adapter.py import time from typing import Optional import openai from .base import BaseModelAdapter, ModelResponse class OpenAIModelAdapter(BaseModelAdapter): def _initialize_client(self, api_key: str, base_url: Optional[str] None, **kwargs): # 使用OpenAI官方庫(kù)或直接HTTP請(qǐng)求 client openai.AsyncOpenAI(api_keyapi_key, base_urlbase_url) return client async def generate_async(self, prompt: str, **generation_params) - ModelResponse: start_time time.time() try: # 設(shè)置默認(rèn)參數(shù)并允許通過(guò)kwargs覆蓋 params { model: self.model_name, messages: [{role: user, content: prompt}], temperature: 0.7, max_tokens: 1024, **generation_params # 用戶自定義參數(shù)優(yōu)先級(jí)最高 } response await self.client.chat.completions.create(**params) end_time time.time() latency end_time - start_time # 提取響應(yīng)內(nèi)容 content response.choices[0].message.content # 提取Usage信息 usage response.usage prompt_tokens usage.prompt_tokens if usage else None completion_tokens usage.completion_tokens if usage else None total_tokens usage.total_tokens if usage else None # 簡(jiǎn)單成本估算示例實(shí)際需根據(jù)官方定價(jià)計(jì)算 cost None if total_tokens: # 這里需要根據(jù)具體模型定價(jià)設(shè)置例如 gpt-4o 的輸入輸出價(jià)格 # 僅為示例非真實(shí)計(jì)算 cost total_tokens * 0.00001 return ModelResponse( contentcontent, model_nameself.model_name, prompt_tokensprompt_tokens, completion_tokenscompletion_tokens, total_tokenstotal_tokens, latencylatency, costcost ) except Exception as e: end_time time.time() # 返回一個(gè)包含錯(cuò)誤信息的響應(yīng)而不是直接拋出異常便于框架統(tǒng)一處理 return ModelResponse( contentf[ERROR] {str(e)}, model_nameself.model_name, latencyend_time - start_time )注意成本估算是非常粗略的實(shí)際應(yīng)用中需要根據(jù)每個(gè)模型的官方定價(jià)表如每1000個(gè)輸入/輸出Token的價(jià)格來(lái)實(shí)現(xiàn)精確計(jì)算邏輯并考慮是否區(qū)分輸入和輸出Token。這里僅作演示。同理可以創(chuàng)建AnthropicModelAdapter、GoogleGeminiAdapter等。對(duì)于通過(guò)ollama運(yùn)行的本地模型可以創(chuàng)建一個(gè)OllamaModelAdapter其_initialize_client可能只是一個(gè)配置了基礎(chǔ)URL的httpx.AsyncClient而generate_async方法則向http://localhost:11434/api/generate發(fā)送特定的POST請(qǐng)求。3.2 第二步定義任務(wù)與構(gòu)建執(zhí)行引擎任務(wù)定義很簡(jiǎn)單就是一個(gè)包含提示詞和元數(shù)據(jù)的對(duì)象。# tasks/task.py from pydantic import BaseModel from typing import Optional, Dict, Any class EvaluationTask(BaseModel): id: str # 任務(wù)唯一標(biāo)識(shí) prompt: str # 提示詞 metadata: Optional[Dict[str, Any]] None # 可附加額外信息如類別、預(yù)期答案片段等執(zhí)行引擎HarnessRunner是大腦。它負(fù)責(zé)加載所有任務(wù)初始化所有配置的模型適配器然后并發(fā)地執(zhí)行每個(gè)任務(wù)在所有模型上的推理。# harness/runner.py import asyncio from typing import List, Dict from tasks.task import EvaluationTask from model_adapters.base import BaseModelAdapter from result_collector import ResultCollector # 稍后實(shí)現(xiàn) class HarnessRunner: def __init__(self, model_adapters: Dict[str, BaseModelAdapter]): :param model_adapters: 字典key為模型顯示名value為初始化好的適配器實(shí)例 self.model_adapters model_adapters self.collector ResultCollector() async def run_task(self, task: EvaluationTask, model_name: str, adapter: BaseModelAdapter): 單個(gè)任務(wù)在單個(gè)模型上執(zhí)行 print(fRunning task {task.id} on model {model_name}...) response await adapter.generate_async(task.prompt) # 將結(jié)果存入收集器 self.collector.add_result(task_idtask.id, model_namemodel_name, responseresponse, task_metadatatask.metadata) async def run_all_tasks(self, tasks: List[EvaluationTask]): 并發(fā)執(zhí)行所有任務(wù)在所有模型上 semaphore asyncio.Semaphore(10) # 控制最大并發(fā)數(shù)避免被API限流 async def run_with_semaphore(task, model_name, adapter): async with semaphore: await self.run_task(task, model_name, adapter) # 創(chuàng)建所有協(xié)程任務(wù) coroutines [] for task in tasks: for model_name, adapter in self.model_adapters.items(): coroutines.append(run_with_semaphore(task, model_name, adapter)) # 并發(fā)執(zhí)行 await asyncio.gather(*coroutines, return_exceptionsTrue) # return_exceptions防止單個(gè)失敗導(dǎo)致整體崩潰 print(All tasks completed.) return self.collector這里的關(guān)鍵是使用了asyncio.Semaphore來(lái)控制最大并發(fā)量。無(wú)節(jié)制地并發(fā)調(diào)用API很容易觸發(fā)速率限制Rate Limit導(dǎo)致大量請(qǐng)求失敗。設(shè)置一個(gè)合理的信號(hào)量值比如5或10是保證穩(wěn)定運(yùn)行的重要技巧。3.3 第三步實(shí)現(xiàn)結(jié)果收集與報(bào)告生成結(jié)果收集器需要結(jié)構(gòu)化地存儲(chǔ)數(shù)據(jù)并支持導(dǎo)出。# result_collector.py from typing import List, Dict, Any from model_adapters.base import ModelResponse import pandas as pd import json class ResultCollector: def __init__(self): self.results: List[Dict[str, Any]] [] # 存儲(chǔ)所有結(jié)果的列表 def add_result(self, task_id: str, model_name: str, response: ModelResponse, task_metadata: Dict None): record { task_id: task_id, model: model_name, prompt: task_metadata.get(prompt_snippet, ) if task_metadata else , # 可存提示詞片段 response: response.content, latency: response.latency, total_tokens: response.total_tokens, cost: response.cost, error: [ERROR] in response.content # 簡(jiǎn)單錯(cuò)誤標(biāo)識(shí) } self.results.append(record) def to_dataframe(self) - pd.DataFrame: 將結(jié)果轉(zhuǎn)換為Pandas DataFrame便于分析 return pd.DataFrame(self.results) def save_to_json(self, filepath: str): with open(filepath, w, encodingutf-8) as f: json.dump(self.results, f, ensure_asciiFalse, indent2) def generate_report(self, output_path: str report.md): 生成一個(gè)簡(jiǎn)單的Markdown格式報(bào)告 df self.to_dataframe() if df.empty: report # Evaluation Report\n\nNo results collected. else: # 按任務(wù)ID分組橫向?qū)Ρ炔煌Ｐ偷捻憫?yīng) report_lines [# LLM Evaluation Harness Report\n] for task_id in df[task_id].unique(): report_lines.append(f\n## Task: {task_id}\n) task_df df[df[task_id] task_id] # 選擇需要展示的列 display_df task_df[[model, response, latency, total_tokens, cost]] report_lines.append(display_df.to_markdown(indexFalse)) report_lines.append(\n---) report \n.join(report_lines) with open(output_path, w, encodingutf-8) as f: f.write(report) print(fReport generated: {output_path})3.4 第四步組裝與運(yùn)行——一個(gè)完整的示例現(xiàn)在我們把所有部分組裝起來(lái)寫一個(gè)主程序。# main.py import asyncio import yaml from tasks.task import EvaluationTask from model_adapters.openai_adapter import OpenAIModelAdapter from model_adapters.anthropic_adapter import AnthropicModelAdapter # 假設(shè)已實(shí)現(xiàn) from harness.runner import HarnessRunner def load_config(config_path: str): with open(config_path, r) as f: return yaml.safe_load(f) def create_tasks(): 創(chuàng)建評(píng)測(cè)任務(wù)列表 tasks [ EvaluationTask( idtask_1_summarize, prompt請(qǐng)用一句話總結(jié)《三體》的核心矛盾。, metadata{category: summarization, language: zh} ), EvaluationTask( idtask_2_code, prompt用Python寫一個(gè)函數(shù)判斷一個(gè)字符串是否是回文。, metadata{category: code_generation} ), EvaluationTask( idtask_3_reasoning, prompt如果所有貓都怕水而我的寵物Socks是一只貓那么Socks怕水嗎請(qǐng)一步步推理。, metadata{category: logical_reasoning} ), ] return tasks async def main(): # 1. 加載配置API Keys等應(yīng)從環(huán)境變量或安全配置讀取 config load_config(config.yaml) # 2. 初始化模型適配器 model_adapters {} openai_config config[models][openai] model_adapters[gpt-4o] OpenAIModelAdapter( model_namegpt-4o, api_keyopenai_config[api_key] # base_url... 可配置 ) # 同理初始化Claude # model_adapters[claude-3-sonnet] AnthropicModelAdapter(...) # 3. 創(chuàng)建執(zhí)行器 runner HarnessRunner(model_adapters) # 4. 加載任務(wù) tasks create_tasks() # 5. 運(yùn)行所有任務(wù) collector await runner.run_all_tasks(tasks) # 6. 保存結(jié)果和生成報(bào)告 collector.save_to_json(results.json) collector.generate_report(evaluation_report.md) # 7. 快速查看統(tǒng)計(jì)信息 df collector.to_dataframe() print(\n 性能摘要 ) summary df.groupby(model).agg({ latency: mean, total_tokens: mean, cost: sum, error: sum }).round(3) print(summary) if __name__ __main__: asyncio.run(main())對(duì)應(yīng)的config.yaml文件示例# config.yaml models: openai: api_key: ${OPENAI_API_KEY} # 建議通過(guò)環(huán)境變量注入 # base_url: https://api.openai.com/v1 # 默認(rèn) anthropic: api_key: ${ANTHROPIC_API_KEY}運(yùn)行這個(gè)主程序你會(huì)得到一份詳細(xì)的Markdown報(bào)告里面并排列出了GPT-4和Claude如果配置了對(duì)三個(gè)不同任務(wù)的回答、耗時(shí)和Token消耗以及一個(gè)匯總的性能統(tǒng)計(jì)表。4. 進(jìn)階優(yōu)化與功能擴(kuò)展基礎(chǔ)框架跑通后就可以根據(jù)實(shí)際需求添加更多實(shí)用功能了。4.1 實(shí)現(xiàn)復(fù)雜的評(píng)估邏輯之前的評(píng)估主要靠人眼看報(bào)告。我們可以引入自動(dòng)評(píng)估。例如對(duì)于代碼生成任務(wù)可以寫一個(gè)函數(shù)來(lái)運(yùn)行生成的代碼并檢查其正確性。# evaluators/code_evaluator.py import subprocess import tempfile import os def evaluate_python_code(code_snippet: str, test_input: str, expected_output: str) - Dict[str, Any]: 評(píng)估Python代碼片段。 :param code_snippet: 模型生成的代碼字符串 :param test_input: 測(cè)試輸入以字符串形式在代碼中可能需要被讀取 :param expected_output: 期望輸出 :return: 包含通過(guò)與否、實(shí)際輸出、錯(cuò)誤信息的字典 result {passed: False, actual_output: None, error: None} # 將代碼片段包裝在一個(gè)可執(zhí)行的腳本中 full_code f import sys {code_snippet} # 假設(shè)生成的函數(shù)叫 is_palindrome if __name__ __main__: # 這里可以設(shè)計(jì)更復(fù)雜的測(cè)試 test_str \{test_input}\ output is_palindrome(test_str) print(str(output).lower()) # 統(tǒng)一輸出為小寫字符串便于比較 with tempfile.NamedTemporaryFile(modew, suffix.py, deleteFalse) as f: f.write(full_code) temp_file_path f.name try: # 執(zhí)行代碼 process subprocess.run( [sys.executable, temp_file_path], capture_outputTrue, textTrue, timeout5 # 設(shè)置超時(shí)防止死循環(huán) ) actual_output process.stdout.strip() result[actual_output] actual_output if process.returncode 0: # 簡(jiǎn)單比較輸出 result[passed] (actual_output expected_output.lower()) else: result[error] process.stderr except subprocess.TimeoutExpired: result[error] Execution timeout except Exception as e: result[error] str(e) finally: os.unlink(temp_file_path) # 清理臨時(shí)文件 return result然后可以在ResultCollector.add_result方法中根據(jù)任務(wù)元數(shù)據(jù)調(diào)用對(duì)應(yīng)的評(píng)估器并將評(píng)估結(jié)果如eval_passed: True也存入記錄中。4.2 集成更強(qiáng)大的結(jié)果分析與可視化pandas和matplotlib/seaborn是絕配。我們可以輕松地?cái)U(kuò)展報(bào)告生成功能。# result_collector.py (補(bǔ)充) import matplotlib.pyplot as plt import seaborn as sns class ResultCollector: # ... 之前的代碼 ... def generate_visual_report(self, output_dir: str ./reports): 生成可視化圖表 df self.to_dataframe() if df.empty: return os.makedirs(output_dir, exist_okTrue) # 1. 模型平均延遲對(duì)比柱狀圖 plt.figure(figsize(10, 6)) latency_df df.groupby(model)[latency].mean().sort_values() sns.barplot(xlatency_df.index, ylatency_df.values) plt.title(Average Latency per Model) plt.ylabel(Latency (seconds)) plt.xticks(rotation45) plt.tight_layout() plt.savefig(os.path.join(output_dir, avg_latency.png)) plt.close() # 2. 每個(gè)任務(wù)各模型的響應(yīng)對(duì)比分組柱狀圖例如對(duì)比Token消耗 # ... 更多圖表生成代碼 ...4.3 設(shè)計(jì)任務(wù)模板與批量導(dǎo)入手動(dòng)在代碼里寫EvaluationTask列表很麻煩?？梢栽O(shè)計(jì)一個(gè)YAML或JSON格式的任務(wù)模板文件。# tasks.yaml tasks: - id: summarize_article_1 prompt: | 請(qǐng)總結(jié)以下文章的主要內(nèi)容 《文章內(nèi)容...》 category: summarization evaluation: type: contains_keywords keywords: [量子計(jì)算, 優(yōu)勢(shì), 挑戰(zhàn)] - id: translate_sentence_1 prompt: | 將以下英文句子翻譯成中文 The rapid advancement of artificial intelligence presents both unprecedented opportunities and significant ethical challenges. category: translation然后寫一個(gè)TaskLoader來(lái)讀取這個(gè)文件批量創(chuàng)建EvaluationTask對(duì)象。這樣非開(kāi)發(fā)人員比如產(chǎn)品經(jīng)理也能通過(guò)編輯YAML文件來(lái)設(shè)計(jì)評(píng)測(cè)集。5. 避坑指南與實(shí)戰(zhàn)心得在開(kāi)發(fā)和使用的過(guò)程中我踩過(guò)不少坑也積累了一些經(jīng)驗(yàn)。5.1 網(wǎng)絡(luò)與API穩(wěn)定性處理指數(shù)退避重試網(wǎng)絡(luò)抖動(dòng)和API臨時(shí)過(guò)載很常見(jiàn)。一定要為重試邏輯設(shè)置隨機(jī)延遲Jitter和最大重試次數(shù)。tenacity庫(kù)讓這一切變得簡(jiǎn)單。from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import openai retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10), retryretry_if_exception_type((openai.APITimeoutError, openai.APIError)) ) async def robust_api_call(adapter, prompt): return await adapter.generate_async(prompt)設(shè)置合理的超時(shí)為每個(gè)API請(qǐng)求設(shè)置單獨(dú)的超時(shí)如30秒防止某個(gè)慢請(qǐng)求阻塞整個(gè)流程。asyncio.wait_for可以配合使用。監(jiān)控與熔斷如果某個(gè)模型API持續(xù)失敗可以考慮實(shí)現(xiàn)一個(gè)簡(jiǎn)單的熔斷器Circuit Breaker暫時(shí)跳過(guò)該模型避免浪費(fèi)時(shí)間和配額。5.2 成本控制與用量監(jiān)控精細(xì)化成本計(jì)算前面提到的成本估算是雛形。生產(chǎn)環(huán)境需要根據(jù)每個(gè)模型的官方定價(jià)精確計(jì)算輸入、輸出Token的費(fèi)用。最好將這部分邏輯抽象成一個(gè)CostCalculator類。設(shè)置預(yù)算上限在運(yùn)行大規(guī)模評(píng)測(cè)前根據(jù)任務(wù)數(shù)量和平均Token消耗預(yù)估總成本?？梢栽贖arnessRunner中增加一個(gè)預(yù)算計(jì)數(shù)器當(dāng)消耗接近閾值時(shí)發(fā)出警告或停止執(zhí)行。使用日志記錄詳細(xì)記錄每個(gè)請(qǐng)求的Token使用情況和估算成本方便后續(xù)對(duì)賬和分析。5.3 結(jié)果評(píng)估的客觀性挑戰(zhàn)避免主觀偏見(jiàn)人工對(duì)比結(jié)果時(shí)很容易對(duì)某個(gè)模型有先入為主的偏好。嘗試設(shè)計(jì)更客觀的評(píng)估指標(biāo)代碼執(zhí)行像上面那樣自動(dòng)運(yùn)行檢查。文本相似度對(duì)于翻譯、摘要任務(wù)使用ROUGE、BLEU或BERTScore等指標(biāo)與參考答案對(duì)比。規(guī)則檢查對(duì)于要求特定格式如JSON、XML的輸出用解析器檢查其合法性。設(shè)計(jì)多樣化的測(cè)試集測(cè)試用例要覆蓋不同難度、不同領(lǐng)域、不同指令類型創(chuàng)意寫作、邏輯推理、信息提取等避免評(píng)估結(jié)果片面。記錄“軟指標(biāo)”除了正確性在報(bào)告中也可以加入人工標(biāo)注的“流暢度”、“創(chuàng)造性”、“遵循指令程度”等評(píng)分但這些需要多人標(biāo)注來(lái)減少偏差。5.4 性能優(yōu)化技巧異步并發(fā)控制Semaphore的值不是越大越好。需要根據(jù)API的速率限制如RPM, TPM來(lái)調(diào)整?？梢韵仍O(shè)置一個(gè)保守值如5觀察請(qǐng)求成功率后再調(diào)整。連接池復(fù)用使用aiohttp.ClientSession或httpx.AsyncClient時(shí)確保在整個(gè)應(yīng)用生命周期內(nèi)復(fù)用同一個(gè)會(huì)話session以利用HTTP持久連接減少TCP握手開(kāi)銷。緩存機(jī)制對(duì)于確定性的提示詞不包含隨機(jī)數(shù)或動(dòng)態(tài)時(shí)間可以考慮將模型的響應(yīng)緩存起來(lái)例如使用diskcache或redis。這樣在多次運(yùn)行或調(diào)試時(shí)可以避免重復(fù)調(diào)用API節(jié)省成本和時(shí)間。自己動(dòng)手實(shí)現(xiàn)一個(gè)Harness的過(guò)程遠(yuǎn)比單純調(diào)用一個(gè)現(xiàn)成工具收獲大。它不僅讓你對(duì)LLM API的細(xì)節(jié)有了更深的掌控更重要的是它迫使你系統(tǒng)化地思考如何評(píng)估一個(gè)AI模型的好壞。這套框架現(xiàn)在已經(jīng)成為我日常工作流的一部分每當(dāng)有新的提示詞策略或需要選型新模型時(shí)跑一遍Harness讓數(shù)據(jù)說(shuō)話決策就變得清晰多了。 從零構(gòu)建LLM評(píng)估框架：實(shí)戰(zhàn)Harness實(shí)現(xiàn)與多模型對(duì)比測(cè)試
1. 項(xiàng)目緣起從“黑盒”到“白盒”的探索最近在折騰大模型應(yīng)用開(kāi)發(fā)時(shí)我遇到了一個(gè)挺典型的問(wèn)題手頭有好幾個(gè)不同的LLM大語(yǔ)言模型服務(wù)比如OpenAI的GPT、Claude還有幾個(gè)開(kāi)源的模型。每次想測(cè)試一個(gè)新提示詞Prompt在不同模型上的效果或者對(duì)比不同模型對(duì)同一任務(wù)的響應(yīng)質(zhì)量都得手動(dòng)一個(gè)個(gè)去調(diào)用API然后把返回結(jié)果復(fù)制粘貼到文檔里再人工去比對(duì)。這個(gè)過(guò)程不僅繁瑣低效而且很難做到標(biāo)準(zhǔn)化評(píng)估比如響應(yīng)時(shí)間、輸出格式一致性、成本消耗這些關(guān)鍵指標(biāo)全靠感覺(jué)。這讓我想起了軟件測(cè)試?yán)锏摹皽y(cè)試工具鏈”。我們不會(huì)手動(dòng)去點(diǎn)每一個(gè)按鈕來(lái)測(cè)試功能而是會(huì)用像JUnit、Pytest這樣的框架來(lái)編寫自動(dòng)化測(cè)試用例。那么對(duì)于LLM應(yīng)用是不是也應(yīng)該有這樣一個(gè)“測(cè)試框架”呢這就是我接觸到“Harness”概念的起點(diǎn)。在AI工程領(lǐng)域Harness這個(gè)詞常被用來(lái)指代一套用于評(píng)估、測(cè)試和管理AI模型特別是LLM的框架或工具集。它的核心目標(biāo)是把模型評(píng)估這個(gè)事從隨意、手工的狀態(tài)變成系統(tǒng)化、自動(dòng)化、可重復(fù)的過(guò)程。網(wǎng)上確實(shí)有一些現(xiàn)成的工具比如著名的lm-evaluation-harness現(xiàn)在常被稱為EleutherAI LM Harness。它功能強(qiáng)大覆蓋了海量的評(píng)測(cè)任務(wù)。但我在實(shí)際想用的時(shí)候發(fā)現(xiàn)了一些痛點(diǎn)一是它通常更偏向于學(xué)術(shù)界的基準(zhǔn)測(cè)試對(duì)于我這種想快速驗(yàn)證業(yè)務(wù)場(chǎng)景提示詞效果的開(kāi)發(fā)者來(lái)說(shuō)有點(diǎn)“重”二是它的配置和擴(kuò)展對(duì)于不熟悉其代碼結(jié)構(gòu)的人來(lái)說(shuō)學(xué)習(xí)成本不低三也是最重要的我想徹底搞明白這背后的機(jī)制而不是當(dāng)一個(gè)“調(diào)包俠”。知其然更要知其所以然自己動(dòng)手實(shí)現(xiàn)一個(gè)輕量化的、貼合自身需求的Harness就成了一個(gè)很有吸引力的挑戰(zhàn)。所以這個(gè)項(xiàng)目的標(biāo)題“實(shí)戰(zhàn)復(fù)盤我是如何用代碼實(shí)現(xiàn) Harness 的”就是記錄我從零開(kāi)始用Python設(shè)計(jì)和編碼構(gòu)建一個(gè)屬于我自己的、輕量級(jí)LLM評(píng)估工具鏈的全過(guò)程。它不追求大而全而是聚焦于解決我實(shí)際開(kāi)發(fā)中的痛點(diǎn)自動(dòng)化、標(biāo)準(zhǔn)化地對(duì)比不同LLM API對(duì)同一組提示詞的響應(yīng)。2. 核心設(shè)計(jì)思路一個(gè)輕量評(píng)估框架的藍(lán)圖在開(kāi)始敲代碼之前我得先想清楚這個(gè)自研Harness到底要干什么以及怎么干。大方向是自動(dòng)化評(píng)估但具體落到設(shè)計(jì)上需要拆解出幾個(gè)核心模塊。2.1 需求定義與架構(gòu)選型我的核心需求很明確多模型支持能夠方便地接入不同的LLM服務(wù)提供商如OpenAI、Anthropic、Google等以及本地模型。任務(wù)定義能夠以結(jié)構(gòu)化的方式定義一組“評(píng)測(cè)任務(wù)”每個(gè)任務(wù)包含輸入提示詞Prompt和預(yù)期的評(píng)估標(biāo)準(zhǔn)不一定是標(biāo)準(zhǔn)答案可能是評(píng)分規(guī)則。自動(dòng)化執(zhí)行框架能自動(dòng)遍歷所有任務(wù)調(diào)用配置的模型執(zhí)行推理并收集結(jié)果。結(jié)果收集與比對(duì)將不同模型對(duì)同一任務(wù)的結(jié)果并排收集方便人工或自動(dòng)比對(duì)。需要記錄關(guān)鍵指標(biāo)如響應(yīng)內(nèi)容、耗時(shí)、Token使用量、成本如果API計(jì)費(fèi)等。可擴(kuò)展性易于添加新的模型接口、新的評(píng)估指標(biāo)或新的任務(wù)類型?；谶@些需求一個(gè)典型的分層架構(gòu)浮現(xiàn)在腦海中任務(wù)層 (Task Layer)負(fù)責(zé)定義和管理具體的評(píng)測(cè)任務(wù)。一個(gè)任務(wù)就是一個(gè)(prompt, evaluation_criteria)對(duì)。執(zhí)行層 (Execution Layer)核心引擎。負(fù)責(zé)加載任務(wù)調(diào)用配置好的模型執(zhí)行推理并處理可能的錯(cuò)誤如網(wǎng)絡(luò)超時(shí)、API限流。模型適配層 (Model Adapter Layer)這是實(shí)現(xiàn)多模型支持的關(guān)鍵。為每個(gè)支持的LLM服務(wù)編寫一個(gè)統(tǒng)一的“適配器”Adapter將框架內(nèi)部的通用請(qǐng)求格式轉(zhuǎn)換為特定API所需的格式并解析其返回結(jié)果。結(jié)果層 (Result Layer)定義標(biāo)準(zhǔn)化的結(jié)果數(shù)據(jù)結(jié)構(gòu)并負(fù)責(zé)將不同模型對(duì)同一任務(wù)的結(jié)果聚合、持久化如保存為JSON或CSV文件并生成易于閱讀的報(bào)告如Markdown表格。為什么不直接用langchain這類框架langchain的核心是構(gòu)建復(fù)雜的工作流鏈Chain其LLM類雖然也統(tǒng)一了接口但它的設(shè)計(jì)重心在于鏈?zhǔn)秸{(diào)用和記憶等高級(jí)功能對(duì)于純粹的、批量的、帶對(duì)比的模型評(píng)測(cè)場(chǎng)景封裝得不夠直接。自己實(shí)現(xiàn)可以更聚焦控制每一個(gè)細(xì)節(jié)比如精確計(jì)時(shí)、自定義重試邏輯、更靈活的結(jié)果收集。2.2 關(guān)鍵技術(shù)點(diǎn)與工具選型編程語(yǔ)言毫無(wú)疑問(wèn)是Python。它在AI和數(shù)據(jù)科學(xué)領(lǐng)域的生態(tài)是無(wú)與倫比的有豐富的HTTP客戶端、數(shù)據(jù)處理和序列化庫(kù)。異步 vs 同步考慮到可能會(huì)批量測(cè)試數(shù)十上百個(gè)提示詞使用異步IO可以大幅提升效率尤其是在網(wǎng)絡(luò)請(qǐng)求成為瓶頸時(shí)。Python的asyncio庫(kù)和aiohttp是首選。配置管理模型的API Key、Base URL等敏感信息不能硬編碼。使用pydantic配合.env文件或config.yaml來(lái)管理配置既安全又靈活。數(shù)據(jù)序列化任務(wù)定義和結(jié)果存儲(chǔ)使用JSON格式因?yàn)樗强缯Z(yǔ)言、人類可讀的并且Python的json模塊支持得很好。對(duì)于最終報(bào)告可以結(jié)合pandas將結(jié)果轉(zhuǎn)為DataFrame再輸出為CSV或Markdown。錯(cuò)誤處理與重試網(wǎng)絡(luò)請(qǐng)求不穩(wěn)定API有速率限制。必須實(shí)現(xiàn)健壯的錯(cuò)誤處理和指數(shù)退避的重試機(jī)制。tenacity庫(kù)是一個(gè)很好的選擇。這個(gè)設(shè)計(jì)藍(lán)圖看起來(lái)清晰了接下來(lái)就是把這些模塊用代碼搭建起來(lái)。3. 分步實(shí)現(xiàn)從零搭建代碼Harness我將按照自底向上的順序先實(shí)現(xiàn)最基礎(chǔ)的模型適配器再構(gòu)建執(zhí)行引擎最后定義任務(wù)和結(jié)果處理。3.1 第一步構(gòu)建統(tǒng)一的模型適配器接口適配器的目標(biāo)是對(duì)外提供統(tǒng)一的generate(prompt: str) - str方法對(duì)內(nèi)處理各自API的差異。首先定義一個(gè)抽象基類規(guī)定所有適配器必須實(shí)現(xiàn)的方法。# model_adapters/base.py import abc from typing import Optional, Dict, Any from pydantic import BaseModel class ModelResponse(BaseModel): 標(biāo)準(zhǔn)化的模型響應(yīng)結(jié)構(gòu) content: str # 模型返回的文本內(nèi)容 model_name: str # 模型標(biāo)識(shí)如 gpt-4 prompt_tokens: Optional[int] None completion_tokens: Optional[int] None total_tokens: Optional[int] None latency: Optional[float] None # 請(qǐng)求耗時(shí)單位秒 cost: Optional[float] None # 估算成本單位美元或其他貨幣 class BaseModelAdapter(abc.ABC): 模型適配器抽象基類 def __init__(self, model_name: str, **kwargs): self.model_name model_name # 可以在這里初始化客戶端如OpenAI客戶端 self.client self._initialize_client(**kwargs) abc.abstractmethod def _initialize_client(self, **kwargs): 初始化特定模型的客戶端 pass abc.abstractmethod async def generate_async(self, prompt: str, **generation_params) - ModelResponse: 異步生成文本的核心方法。 :param prompt: 輸入的提示詞 :param generation_params: 模型特定的生成參數(shù)溫度、top_p等 :return: 標(biāo)準(zhǔn)化的ModelResponse對(duì)象 pass # 也可以提供一個(gè)同步版本作為備選 def generate_sync(self, prompt: str, **kwargs) - ModelResponse: import asyncio return asyncio.run(self.generate_async(prompt, **kwargs))接下來(lái)實(shí)現(xiàn)一個(gè)具體的適配器比如OpenAI的。# model_adapters/openai_adapter.py import time from typing import Optional import openai from .base import BaseModelAdapter, ModelResponse class OpenAIModelAdapter(BaseModelAdapter): def _initialize_client(self, api_key: str, base_url: Optional[str] None, **kwargs): # 使用OpenAI官方庫(kù)或直接HTTP請(qǐng)求 client openai.AsyncOpenAI(api_keyapi_key, base_urlbase_url) return client async def generate_async(self, prompt: str, **generation_params) - ModelResponse: start_time time.time() try: # 設(shè)置默認(rèn)參數(shù)并允許通過(guò)kwargs覆蓋 params { model: self.model_name, messages: [{role: user, content: prompt}], temperature: 0.7, max_tokens: 1024, **generation_params # 用戶自定義參數(shù)優(yōu)先級(jí)最高 } response await self.client.chat.completions.create(**params) end_time time.time() latency end_time - start_time # 提取響應(yīng)內(nèi)容 content response.choices[0].message.content # 提取Usage信息 usage response.usage prompt_tokens usage.prompt_tokens if usage else None completion_tokens usage.completion_tokens if usage else None total_tokens usage.total_tokens if usage else None # 簡(jiǎn)單成本估算示例實(shí)際需根據(jù)官方定價(jià)計(jì)算 cost None if total_tokens: # 這里需要根據(jù)具體模型定價(jià)設(shè)置例如 gpt-4o 的輸入輸出價(jià)格 # 僅為示例非真實(shí)計(jì)算 cost total_tokens * 0.00001 return ModelResponse( contentcontent, model_nameself.model_name, prompt_tokensprompt_tokens, completion_tokenscompletion_tokens, total_tokenstotal_tokens, latencylatency, costcost ) except Exception as e: end_time time.time() # 返回一個(gè)包含錯(cuò)誤信息的響應(yīng)而不是直接拋出異常便于框架統(tǒng)一處理 return ModelResponse( contentf[ERROR] {str(e)}, model_nameself.model_name, latencyend_time - start_time )注意成本估算是非常粗略的實(shí)際應(yīng)用中需要根據(jù)每個(gè)模型的官方定價(jià)表如每1000個(gè)輸入/輸出Token的價(jià)格來(lái)實(shí)現(xiàn)精確計(jì)算邏輯并考慮是否區(qū)分輸入和輸出Token。這里僅作演示。同理可以創(chuàng)建AnthropicModelAdapter、GoogleGeminiAdapter等。對(duì)于通過(guò)ollama運(yùn)行的本地模型可以創(chuàng)建一個(gè)OllamaModelAdapter其_initialize_client可能只是一個(gè)配置了基礎(chǔ)URL的httpx.AsyncClient而generate_async方法則向http://localhost:11434/api/generate發(fā)送特定的POST請(qǐng)求。3.2 第二步定義任務(wù)與構(gòu)建執(zhí)行引擎任務(wù)定義很簡(jiǎn)單就是一個(gè)包含提示詞和元數(shù)據(jù)的對(duì)象。# tasks/task.py from pydantic import BaseModel from typing import Optional, Dict, Any class EvaluationTask(BaseModel): id: str # 任務(wù)唯一標(biāo)識(shí) prompt: str # 提示詞 metadata: Optional[Dict[str, Any]] None # 可附加額外信息如類別、預(yù)期答案片段等執(zhí)行引擎HarnessRunner是大腦。它負(fù)責(zé)加載所有任務(wù)初始化所有配置的模型適配器然后并發(fā)地執(zhí)行每個(gè)任務(wù)在所有模型上的推理。# harness/runner.py import asyncio from typing import List, Dict from tasks.task import EvaluationTask from model_adapters.base import BaseModelAdapter from result_collector import ResultCollector # 稍后實(shí)現(xiàn) class HarnessRunner: def __init__(self, model_adapters: Dict[str, BaseModelAdapter]): :param model_adapters: 字典key為模型顯示名value為初始化好的適配器實(shí)例 self.model_adapters model_adapters self.collector ResultCollector() async def run_task(self, task: EvaluationTask, model_name: str, adapter: BaseModelAdapter): 單個(gè)任務(wù)在單個(gè)模型上執(zhí)行 print(fRunning task {task.id} on model {model_name}...) response await adapter.generate_async(task.prompt) # 將結(jié)果存入收集器 self.collector.add_result(task_idtask.id, model_namemodel_name, responseresponse, task_metadatatask.metadata) async def run_all_tasks(self, tasks: List[EvaluationTask]): 并發(fā)執(zhí)行所有任務(wù)在所有模型上 semaphore asyncio.Semaphore(10) # 控制最大并發(fā)數(shù)避免被API限流 async def run_with_semaphore(task, model_name, adapter): async with semaphore: await self.run_task(task, model_name, adapter) # 創(chuàng)建所有協(xié)程任務(wù) coroutines [] for task in tasks: for model_name, adapter in self.model_adapters.items(): coroutines.append(run_with_semaphore(task, model_name, adapter)) # 并發(fā)執(zhí)行 await asyncio.gather(*coroutines, return_exceptionsTrue) # return_exceptions防止單個(gè)失敗導(dǎo)致整體崩潰 print(All tasks completed.) return self.collector這里的關(guān)鍵是使用了asyncio.Semaphore來(lái)控制最大并發(fā)量。無(wú)節(jié)制地并發(fā)調(diào)用API很容易觸發(fā)速率限制Rate Limit導(dǎo)致大量請(qǐng)求失敗。設(shè)置一個(gè)合理的信號(hào)量值比如5或10是保證穩(wěn)定運(yùn)行的重要技巧。3.3 第三步實(shí)現(xiàn)結(jié)果收集與報(bào)告生成結(jié)果收集器需要結(jié)構(gòu)化地存儲(chǔ)數(shù)據(jù)并支持導(dǎo)出。# result_collector.py from typing import List, Dict, Any from model_adapters.base import ModelResponse import pandas as pd import json class ResultCollector: def __init__(self): self.results: List[Dict[str, Any]] [] # 存儲(chǔ)所有結(jié)果的列表 def add_result(self, task_id: str, model_name: str, response: ModelResponse, task_metadata: Dict None): record { task_id: task_id, model: model_name, prompt: task_metadata.get(prompt_snippet, ) if task_metadata else , # 可存提示詞片段 response: response.content, latency: response.latency, total_tokens: response.total_tokens, cost: response.cost, error: [ERROR] in response.content # 簡(jiǎn)單錯(cuò)誤標(biāo)識(shí) } self.results.append(record) def to_dataframe(self) - pd.DataFrame: 將結(jié)果轉(zhuǎn)換為Pandas DataFrame便于分析 return pd.DataFrame(self.results) def save_to_json(self, filepath: str): with open(filepath, w, encodingutf-8) as f: json.dump(self.results, f, ensure_asciiFalse, indent2) def generate_report(self, output_path: str report.md): 生成一個(gè)簡(jiǎn)單的Markdown格式報(bào)告 df self.to_dataframe() if df.empty: report # Evaluation Report\n\nNo results collected. else: # 按任務(wù)ID分組橫向?qū)Ρ炔煌Ｐ偷捻憫?yīng) report_lines [# LLM Evaluation Harness Report\n] for task_id in df[task_id].unique(): report_lines.append(f\n## Task: {task_id}\n) task_df df[df[task_id] task_id] # 選擇需要展示的列 display_df task_df[[model, response, latency, total_tokens, cost]] report_lines.append(display_df.to_markdown(indexFalse)) report_lines.append(\n---) report \n.join(report_lines) with open(output_path, w, encodingutf-8) as f: f.write(report) print(fReport generated: {output_path})3.4 第四步組裝與運(yùn)行——一個(gè)完整的示例現(xiàn)在我們把所有部分組裝起來(lái)寫一個(gè)主程序。# main.py import asyncio import yaml from tasks.task import EvaluationTask from model_adapters.openai_adapter import OpenAIModelAdapter from model_adapters.anthropic_adapter import AnthropicModelAdapter # 假設(shè)已實(shí)現(xiàn) from harness.runner import HarnessRunner def load_config(config_path: str): with open(config_path, r) as f: return yaml.safe_load(f) def create_tasks(): 創(chuàng)建評(píng)測(cè)任務(wù)列表 tasks [ EvaluationTask( idtask_1_summarize, prompt請(qǐng)用一句話總結(jié)《三體》的核心矛盾。, metadata{category: summarization, language: zh} ), EvaluationTask( idtask_2_code, prompt用Python寫一個(gè)函數(shù)判斷一個(gè)字符串是否是回文。, metadata{category: code_generation} ), EvaluationTask( idtask_3_reasoning, prompt如果所有貓都怕水而我的寵物Socks是一只貓那么Socks怕水嗎請(qǐng)一步步推理。, metadata{category: logical_reasoning} ), ] return tasks async def main(): # 1. 加載配置API Keys等應(yīng)從環(huán)境變量或安全配置讀取 config load_config(config.yaml) # 2. 初始化模型適配器 model_adapters {} openai_config config[models][openai] model_adapters[gpt-4o] OpenAIModelAdapter( model_namegpt-4o, api_keyopenai_config[api_key] # base_url... 可配置 ) # 同理初始化Claude # model_adapters[claude-3-sonnet] AnthropicModelAdapter(...) # 3. 創(chuàng)建執(zhí)行器 runner HarnessRunner(model_adapters) # 4. 加載任務(wù) tasks create_tasks() # 5. 運(yùn)行所有任務(wù) collector await runner.run_all_tasks(tasks) # 6. 保存結(jié)果和生成報(bào)告 collector.save_to_json(results.json) collector.generate_report(evaluation_report.md) # 7. 快速查看統(tǒng)計(jì)信息 df collector.to_dataframe() print(\n 性能摘要 ) summary df.groupby(model).agg({ latency: mean, total_tokens: mean, cost: sum, error: sum }).round(3) print(summary) if __name__ __main__: asyncio.run(main())對(duì)應(yīng)的config.yaml文件示例# config.yaml models: openai: api_key: ${OPENAI_API_KEY} # 建議通過(guò)環(huán)境變量注入 # base_url: https://api.openai.com/v1 # 默認(rèn) anthropic: api_key: ${ANTHROPIC_API_KEY}運(yùn)行這個(gè)主程序你會(huì)得到一份詳細(xì)的Markdown報(bào)告里面并排列出了GPT-4和Claude如果配置了對(duì)三個(gè)不同任務(wù)的回答、耗時(shí)和Token消耗以及一個(gè)匯總的性能統(tǒng)計(jì)表。4. 進(jìn)階優(yōu)化與功能擴(kuò)展基礎(chǔ)框架跑通后就可以根據(jù)實(shí)際需求添加更多實(shí)用功能了。4.1 實(shí)現(xiàn)復(fù)雜的評(píng)估邏輯之前的評(píng)估主要靠人眼看報(bào)告。我們可以引入自動(dòng)評(píng)估。例如對(duì)于代碼生成任務(wù)可以寫一個(gè)函數(shù)來(lái)運(yùn)行生成的代碼并檢查其正確性。# evaluators/code_evaluator.py import subprocess import tempfile import os def evaluate_python_code(code_snippet: str, test_input: str, expected_output: str) - Dict[str, Any]: 評(píng)估Python代碼片段。 :param code_snippet: 模型生成的代碼字符串 :param test_input: 測(cè)試輸入以字符串形式在代碼中可能需要被讀取 :param expected_output: 期望輸出 :return: 包含通過(guò)與否、實(shí)際輸出、錯(cuò)誤信息的字典 result {passed: False, actual_output: None, error: None} # 將代碼片段包裝在一個(gè)可執(zhí)行的腳本中 full_code f import sys {code_snippet} # 假設(shè)生成的函數(shù)叫 is_palindrome if __name__ __main__: # 這里可以設(shè)計(jì)更復(fù)雜的測(cè)試 test_str \{test_input}\ output is_palindrome(test_str) print(str(output).lower()) # 統(tǒng)一輸出為小寫字符串便于比較 with tempfile.NamedTemporaryFile(modew, suffix.py, deleteFalse) as f: f.write(full_code) temp_file_path f.name try: # 執(zhí)行代碼 process subprocess.run( [sys.executable, temp_file_path], capture_outputTrue, textTrue, timeout5 # 設(shè)置超時(shí)防止死循環(huán) ) actual_output process.stdout.strip() result[actual_output] actual_output if process.returncode 0: # 簡(jiǎn)單比較輸出 result[passed] (actual_output expected_output.lower()) else: result[error] process.stderr except subprocess.TimeoutExpired: result[error] Execution timeout except Exception as e: result[error] str(e) finally: os.unlink(temp_file_path) # 清理臨時(shí)文件 return result然后可以在ResultCollector.add_result方法中根據(jù)任務(wù)元數(shù)據(jù)調(diào)用對(duì)應(yīng)的評(píng)估器并將評(píng)估結(jié)果如eval_passed: True也存入記錄中。4.2 集成更強(qiáng)大的結(jié)果分析與可視化pandas和matplotlib/seaborn是絕配。我們可以輕松地?cái)U(kuò)展報(bào)告生成功能。# result_collector.py (補(bǔ)充) import matplotlib.pyplot as plt import seaborn as sns class ResultCollector: # ... 之前的代碼 ... def generate_visual_report(self, output_dir: str ./reports): 生成可視化圖表 df self.to_dataframe() if df.empty: return os.makedirs(output_dir, exist_okTrue) # 1. 模型平均延遲對(duì)比柱狀圖 plt.figure(figsize(10, 6)) latency_df df.groupby(model)[latency].mean().sort_values() sns.barplot(xlatency_df.index, ylatency_df.values) plt.title(Average Latency per Model) plt.ylabel(Latency (seconds)) plt.xticks(rotation45) plt.tight_layout() plt.savefig(os.path.join(output_dir, avg_latency.png)) plt.close() # 2. 每個(gè)任務(wù)各模型的響應(yīng)對(duì)比分組柱狀圖例如對(duì)比Token消耗 # ... 更多圖表生成代碼 ...4.3 設(shè)計(jì)任務(wù)模板與批量導(dǎo)入手動(dòng)在代碼里寫EvaluationTask列表很麻煩?？梢栽O(shè)計(jì)一個(gè)YAML或JSON格式的任務(wù)模板文件。# tasks.yaml tasks: - id: summarize_article_1 prompt: | 請(qǐng)總結(jié)以下文章的主要內(nèi)容 《文章內(nèi)容...》 category: summarization evaluation: type: contains_keywords keywords: [量子計(jì)算, 優(yōu)勢(shì), 挑戰(zhàn)] - id: translate_sentence_1 prompt: | 將以下英文句子翻譯成中文 The rapid advancement of artificial intelligence presents both unprecedented opportunities and significant ethical challenges. category: translation然后寫一個(gè)TaskLoader來(lái)讀取這個(gè)文件批量創(chuàng)建EvaluationTask對(duì)象。這樣非開(kāi)發(fā)人員比如產(chǎn)品經(jīng)理也能通過(guò)編輯YAML文件來(lái)設(shè)計(jì)評(píng)測(cè)集。5. 避坑指南與實(shí)戰(zhàn)心得在開(kāi)發(fā)和使用的過(guò)程中我踩過(guò)不少坑也積累了一些經(jīng)驗(yàn)。5.1 網(wǎng)絡(luò)與API穩(wěn)定性處理指數(shù)退避重試網(wǎng)絡(luò)抖動(dòng)和API臨時(shí)過(guò)載很常見(jiàn)。一定要為重試邏輯設(shè)置隨機(jī)延遲Jitter和最大重試次數(shù)。tenacity庫(kù)讓這一切變得簡(jiǎn)單。from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import openai retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10), retryretry_if_exception_type((openai.APITimeoutError, openai.APIError)) ) async def robust_api_call(adapter, prompt): return await adapter.generate_async(prompt)設(shè)置合理的超時(shí)為每個(gè)API請(qǐng)求設(shè)置單獨(dú)的超時(shí)如30秒防止某個(gè)慢請(qǐng)求阻塞整個(gè)流程。asyncio.wait_for可以配合使用。監(jiān)控與熔斷如果某個(gè)模型API持續(xù)失敗可以考慮實(shí)現(xiàn)一個(gè)簡(jiǎn)單的熔斷器Circuit Breaker暫時(shí)跳過(guò)該模型避免浪費(fèi)時(shí)間和配額。5.2 成本控制與用量監(jiān)控精細(xì)化成本計(jì)算前面提到的成本估算是雛形。生產(chǎn)環(huán)境需要根據(jù)每個(gè)模型的官方定價(jià)精確計(jì)算輸入、輸出Token的費(fèi)用。最好將這部分邏輯抽象成一個(gè)CostCalculator類。設(shè)置預(yù)算上限在運(yùn)行大規(guī)模評(píng)測(cè)前根據(jù)任務(wù)數(shù)量和平均Token消耗預(yù)估總成本?？梢栽贖arnessRunner中增加一個(gè)預(yù)算計(jì)數(shù)器當(dāng)消耗接近閾值時(shí)發(fā)出警告或停止執(zhí)行。使用日志記錄詳細(xì)記錄每個(gè)請(qǐng)求的Token使用情況和估算成本方便后續(xù)對(duì)賬和分析。5.3 結(jié)果評(píng)估的客觀性挑戰(zhàn)避免主觀偏見(jiàn)人工對(duì)比結(jié)果時(shí)很容易對(duì)某個(gè)模型有先入為主的偏好。嘗試設(shè)計(jì)更客觀的評(píng)估指標(biāo)代碼執(zhí)行像上面那樣自動(dòng)運(yùn)行檢查。文本相似度對(duì)于翻譯、摘要任務(wù)使用ROUGE、BLEU或BERTScore等指標(biāo)與參考答案對(duì)比。規(guī)則檢查對(duì)于要求特定格式如JSON、XML的輸出用解析器檢查其合法性。設(shè)計(jì)多樣化的測(cè)試集測(cè)試用例要覆蓋不同難度、不同領(lǐng)域、不同指令類型創(chuàng)意寫作、邏輯推理、信息提取等避免評(píng)估結(jié)果片面。記錄“軟指標(biāo)”除了正確性在報(bào)告中也可以加入人工標(biāo)注的“流暢度”、“創(chuàng)造性”、“遵循指令程度”等評(píng)分但這些需要多人標(biāo)注來(lái)減少偏差。5.4 性能優(yōu)化技巧異步并發(fā)控制Semaphore的值不是越大越好。需要根據(jù)API的速率限制如RPM, TPM來(lái)調(diào)整?？梢韵仍O(shè)置一個(gè)保守值如5觀察請(qǐng)求成功率后再調(diào)整。連接池復(fù)用使用aiohttp.ClientSession或httpx.AsyncClient時(shí)確保在整個(gè)應(yīng)用生命周期內(nèi)復(fù)用同一個(gè)會(huì)話session以利用HTTP持久連接減少TCP握手開(kāi)銷。緩存機(jī)制對(duì)于確定性的提示詞不包含隨機(jī)數(shù)或動(dòng)態(tài)時(shí)間可以考慮將模型的響應(yīng)緩存起來(lái)例如使用diskcache或redis。這樣在多次運(yùn)行或調(diào)試時(shí)可以避免重復(fù)調(diào)用API節(jié)省成本和時(shí)間。自己動(dòng)手實(shí)現(xiàn)一個(gè)Harness的過(guò)程遠(yuǎn)比單純調(diào)用一個(gè)現(xiàn)成工具收獲大。它不僅讓你對(duì)LLM API的細(xì)節(jié)有了更深的掌控更重要的是它迫使你系統(tǒng)化地思考如何評(píng)估一個(gè)AI模型的好壞。這套框架現(xiàn)在已經(jīng)成為我日常工作流的一部分每當(dāng)有新的提示詞策略或需要選型新模型時(shí)跑一遍Harness讓數(shù)據(jù)說(shuō)話決策就變得清晰多了。