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