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让数据说话决策就变得清晰多了。