在大型语言模型推理任务中如何以极低的成本实现接近SOTA的性能是许多开发者和研究者关注的焦点。传统的思路往往聚焦于模型本身的微调或架构改进但近期一项名为“Harness Scaling”的技术通过优化推理流程而非模型本身仅用15美元的成本就在Terminal-Bench 2.1基准测试上达到了95.3%的准确率刷新了该基准的SOTA记录。这为资源受限的团队或个人开发者提供了一条极具性价比的优化路径。本文将从工程实践的角度深入解析Harness Scaling的核心思想、实现原理并提供一个可复现的实践指南。我们将从理解基准测试和推理流程开始逐步拆解Harness Scaling的关键步骤包括如何构建高效的推理“马具”、设计缩放策略、编写核心代码并最终验证效果。无论你是希望优化现有大模型推理服务性能的工程师还是对低成本AI研究感兴趣的研究者都能通过本文获得一套清晰、可操作的方法论。1. 理解核心概念Terminal-Bench与Harness Scaling在动手实践之前必须厘清两个核心概念Terminal-Bench基准测试和Harness Scaling技术本身。这决定了我们优化的目标和方向。1.1 Terminal-Bench 2.1评估推理能力的“考场”Terminal-Bench是一个专门设计用于评估大型语言模型在终端用户任务上表现的基准测试集其2.1版本包含了更复杂、更多样的现实世界推理问题。与侧重于知识问答或代码生成的基准不同Terminal-Bench模拟的是用户与模型交互的完整闭环例如多轮对话理解与规划根据历史对话规划下一步行动。工具使用与API调用理解自然语言指令并转化为正确的工具调用序列。复杂指令分解将一个模糊的用户请求分解为一系列可执行的具体步骤。这个基准的关键在于它评估的不是模型“知道”什么而是模型在特定“流程”或“上下文”中“做对”事情的能力。因此优化模型在Terminal-Bench上的表现本质上是优化模型在特定应用流程中的表现。1.2 Harness Scaling优化“流程”而非“引擎”Harness中文可译为“马具”或“挽具”在这里是一个精妙的比喻。我们可以将大型语言模型LLM比作一匹强大的“赛马”引擎而Harness则是驾驭这匹赛马、使其在特定赛道上发挥最佳水平的“缰绳、鞍具和策略”流程。Harness Scaling的核心思想是在不修改模型权重即不微调模型的前提下通过精心设计和规模化地优化调用模型的前后处理流程来显著提升最终任务的表现。这个流程通常包括输入工程如何将原始问题转化为最适合模型理解的提示词Prompt。包括添加系统指令、思维链CoT示例、上下文信息等。采样策略如何设置温度Temperature、Top-p等参数以平衡输出的创造性和确定性。输出解析与后处理如何从模型生成的文本中准确、结构化地提取出任务所需的答案或动作。自我验证与重试让模型对自身输出进行检查或在不确定时设计重试机制。流程编排对于复杂任务如何将大问题拆解为多个子问题并编排多次模型调用。“Scaling”缩放在这里意味着不是设计一个固定的好流程而是系统地探索流程中各个组件的不同配置组合即不同的“Harness”并在大量任务上评估这些组合最终找到一个或一组在统计意义上最优的流程配置。简而言之Harness Scaling把优化重点从“让模型变得更聪明”转移到了“让使用模型的方法变得更聪明”上。15美元的成本主要就花在了自动化搜索和评估大量不同Harness配置的计算上。2. 环境准备与项目结构要实现Harness Scaling我们需要搭建一个能够自动化进行提示词工程、模型调用、结果评估和配置搜索的实验环境。2.1 基础环境与依赖假设我们使用Python作为主要开发语言并利用OpenAI API或兼容API作为模型服务。以下是核心依赖# 创建虚拟环境可选但推荐 python -m venv harness_env source harness_env/bin/activate # Linux/macOS # harness_env\Scripts\activate # Windows # 安装核心依赖 pip install openai1.0.0 # 官方SDK pip install tqdm # 进度条 pip install pandas # 数据处理 pip install numpy # 数值计算 pip install scikit-learn # 用于评估指标计算如准确率 pip install tenacity # 用于实现API调用的重试机制 pip install pyyaml # 用于管理Harness配置如果使用其他模型API如Anthropic Claude, Google Gemini或本地部署的vLLM服务需安装相应的SDK。2.2 项目目录结构一个清晰的项目结构有助于管理大量的Harness配置和实验结果。terminal_bench_harness_scaling/ ├── configs/ # 存放各种Harness配置 │ ├── harness_baseline.yaml │ ├── harness_cot.yaml │ └── harness_self_consistency.yaml ├── data/ # 存放Terminal-Bench数据集或你的自定义数据集 │ ├── train.jsonl │ └── test.jsonl ├── src/ │ ├── __init__.py │ ├── harness.py # Harness核心类定义流程 │ ├── evaluator.py # 评估逻辑 │ ├── scaler.py # Scaling搜索策略 │ └── utils.py # 工具函数如API调用封装 ├── results/ # 存放每次实验的详细结果 │ └── 20240520_experiment_1/ ├── scripts/ │ └── run_experiment.py # 实验启动脚本 ├── requirements.txt └── README.md3. 构建可配置的推理HarnessHarness的核心是一个可配置的类它定义了从原始问题到最终答案的完整处理流程。3.1 定义Harness配置YAML我们使用YAML文件来灵活定义不同的Harness。configs/harness_template.yamlname: cot_with_self_check # Harness名称 model: gpt-3.5-turbo # 使用的模型 temperature: 0.7 max_tokens: 1024 # 输入工程模块 input_engineering: system_prompt: | 你是一个严谨的推理助手。请逐步思考并在最终答案前标明“答案是”。 few_shot_examples: # 少量示例Few-Shot - question: 如果小明有5个苹果吃了2个又买了3个他现在有几个苹果 thought: 首先初始有5个。吃掉2个剩余5-23个。然后买来3个总共336个。 answer: 答案是6 cot_enabled: true # 是否启用思维链 cot_trigger: 请逐步思考 # 输出解析模块 output_parsing: answer_prefix: 答案是 # 用于从模型输出中定位答案 fallback_strategy: retry # 解析失败时的策略retry, use_raw, fail # 自我验证模块 self_verification: enabled: true verification_prompt: | 请检查以下推理过程和答案是否正确。如果正确回复“正确”如果错误指出错误并给出正确推理。 问题{question} 推理{reasoning} 答案{extracted_answer} max_retries: 1 # 流程编排针对复杂任务 orchestration: max_steps: 3 # 最大分解步数 aggregation_method: last_step # 如何聚合多步结果3.2 实现Harness核心类src/harness.pyimport yaml import re from tenacity import retry, stop_after_attempt, wait_exponential from openai import OpenAI from typing import Dict, Any, Optional, List class ReasoningHarness: def __init__(self, config_path: str, api_key: str): 初始化Harness加载配置。 :param config_path: Harness配置文件的路径 :param api_key: OpenAI API密钥 with open(config_path, r, encodingutf-8) as f: self.config yaml.safe_load(f) self.client OpenAI(api_keyapi_key) self.model self.config.get(model, gpt-3.5-turbo) def _build_messages(self, question: str) - List[Dict[str, str]]: 根据配置构建发送给模型的messages列表。 messages [] # 1. 添加系统提示 if sys_prompt : self.config[input_engineering].get(system_prompt): messages.append({role: system, content: sys_prompt}) # 2. 添加Few-Shot示例 if examples : self.config[input_engineering].get(few_shot_examples, []): for ex in examples: # 示例中的对话历史模拟 messages.append({role: user, content: ex[question]}) # 如果启用了COT将思考过程作为assistant的一部分内容 cot_content ex.get(thought, ) \n ex[answer] if self.config[input_engineering].get(cot_enabled) else ex[answer] messages.append({role: assistant, content: cot_content}) # 3. 构建当前问题的提示 user_content question if self.config[input_engineering].get(cot_enabled): cot_trigger self.config[input_engineering].get(cot_trigger, 请逐步思考) user_content f{user_content}\n\n{cot_trigger} messages.append({role: user, content: user_content}) return messages retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def _call_llm(self, messages: List[Dict[str, str]]) - str: 调用LLM API包含重试机制。 response self.client.chat.completions.create( modelself.model, messagesmessages, temperatureself.config.get(temperature, 0.7), max_tokensself.config.get(max_tokens, 1024) ) return response.choices[0].message.content def _extract_answer(self, llm_output: str) - Optional[str]: 从模型输出中解析出最终答案。 parsing_config self.config[output_parsing] prefix parsing_config.get(answer_prefix) if prefix and prefix in llm_output: # 简单查找前缀后的内容作为答案 pattern re.escape(prefix) r\s*(.) match re.search(pattern, llm_output, re.DOTALL) if match: return match.group(1).strip() # 如果解析失败根据策略处理 strategy parsing_config.get(fallback_strategy, fail) if strategy use_raw: return llm_output.strip() elif strategy retry: return None # 返回None外部逻辑可触发重试 else: # fail raise ValueError(f无法从输出中解析答案: {llm_output[:200]}...) def _self_verify(self, question: str, reasoning: str, answer: str) - bool: 自我验证步骤。 if not self.config[self_verification].get(enabled, False): return True # 未启用验证默认通过 verify_prompt_template self.config[self_verification][verification_prompt] verify_prompt verify_prompt_template.format( questionquestion, reasoningreasoning, extracted_answeranswer ) messages [{role: user, content: verify_prompt}] verification_response self._call_llm(messages) return 正确 in verification_response # 简单的关键词判断生产环境需更鲁棒 def run(self, question: str) - Dict[str, Any]: 执行完整的Harness流程。 返回包含原始输出、解析后的答案、元数据等信息的字典。 result { question: question, harness: self.config[name], success: False, answer: None, raw_output: None, reasoning: None, retries: 0 } max_retries self.config[output_parsing].get(max_parsing_retries, 0) 1 if self.config[self_verification].get(enabled): max_retries max(max_retries, self.config[self_verification].get(max_retries, 0) 1) for attempt in range(max_retries): try: # 1. 构建输入 messages self._build_messages(question) # 2. 调用模型 raw_output self._call_llm(messages) result[raw_output] raw_output # 3. 提取答案 extracted_answer self._extract_answer(raw_output) if extracted_answer is None and attempt max_retries - 1: # 解析失败触发重试可能通过修改提示 continue elif extracted_answer is None: raise ValueError(答案解析失败已达最大重试次数。) # 4. 自我验证如果启用 # 简单地从原始输出中分离“推理过程”和“答案”实际项目需更精细 result[reasoning] raw_output # 此处简化处理 if self._self_verify(question, raw_output, extracted_answer): result[answer] extracted_answer result[success] True break else: # 验证失败可能触发重试或标记为失败 if attempt max_retries - 1: # 可以在这里修改提示词加入验证失败的反馈 continue else: result[answer] extracted_answer # 即使验证失败也记录答案 # result[success] 保持 False except Exception as e: result[error] str(e) if attempt max_retries - 1: break result[retries] attempt return result这个ReasoningHarness类封装了从配置加载、提示词构建、模型调用到输出解析和自我验证的完整流程。通过修改YAML配置文件我们可以轻松定义不同的Harness策略。4. 实现Harness Scaling搜索策略有了基础的Harness下一步是实现Scaling——自动化地搜索最优配置。我们采用一种简单的网格搜索Grid Search与随机搜索Random Search结合的策略作为示例。src/scaler.pyimport itertools import random import json from pathlib import Path from typing import List, Dict, Any from .harness import ReasoningHarness from .evaluator import evaluate_predictions # 假设有一个评估函数 class HarnessScaler: def __init__(self, base_config: Dict[str, Any], search_space: Dict[str, List]): :param base_config: 基础Harness配置字典 :param search_space: 搜索空间例如 { temperature: [0.1, 0.5, 0.9], input_engineering.cot_enabled: [True, False], input_engineering.few_shot_examples: [None, example_list_1, example_list_2] } self.base_config base_config self.search_space search_space def _generate_configs_grid(self) - List[Dict[str, Any]]: 生成网格搜索的所有配置组合。 keys list(self.search_space.keys()) value_lists [self.search_space[key] for key in keys] configs [] for combination in itertools.product(*value_lists): new_config self._deep_update(self.base_config.copy(), dict(zip(keys, combination))) configs.append(new_config) return configs def _generate_configs_random(self, num_samples: int) - List[Dict[str, Any]]: 随机采样指定数量的配置。 configs [] for _ in range(num_samples): new_config self.base_config.copy() for key, values in self.search_space.items(): chosen_value random.choice(values) new_config self._deep_update(new_config, {key: chosen_value}) # 确保每个配置有唯一名称 new_config[name] fharness_random_{random.randint(1000,9999)} configs.append(new_config) return configs def _deep_update(self, base: Dict, update: Dict) - Dict: 深度更新字典支持嵌套键如 input_engineering.cot_enabled。 for key, value in update.items(): if . in key: # 处理嵌套键 parts key.split(.) d base for part in parts[:-1]: d d.setdefault(part, {}) d[parts[-1]] value else: base[key] value return base def run_scaling(self, train_data: List[Dict], eval_data: List[Dict], search_mode: str random, num_samples: int 10, budget_usd: float 15.0) - Dict[str, Any]: 执行Scaling搜索。 :param train_data: 用于搜索/开发的数据集 :param eval_data: 用于最终评估的数据集 :param search_mode: grid 或 random :param num_samples: 随机搜索时的采样数 :param budget_usd: 预算美元用于估算成本和控制搜索范围 :return: 包含最佳配置和结果详情的字典 # 1. 根据预算和搜索模式生成待测试的配置列表 if search_mode grid: candidate_configs self._generate_configs_grid() # 如果网格组合太多根据预算进行粗略裁剪 # 简单估算每个数据点调用一次模型假设平均token成本 estimated_cost_per_config len(train_data) * 0.002 # 示例估算值 max_configs_by_budget int(budget_usd / estimated_cost_per_config) if len(candidate_configs) max_configs_by_budget: print(f网格组合数({len(candidate_configs)})超出预算限制({max_configs_by_budget})将进行随机子采样。) candidate_configs random.sample(candidate_configs, max_configs_by_budget) else: # random # 根据预算计算最大可采样数 estimated_cost_per_config len(train_data) * 0.002 max_samples_by_budget int(budget_usd / estimated_cost_per_config) num_samples min(num_samples, max_samples_by_budget) candidate_configs self._generate_configs_random(num_samples) print(f将在 {len(candidate_configs)} 个Harness配置上进行搜索预计成本约 ${len(candidate_configs)*estimated_cost_per_config:.2f}) best_score -1 best_config None all_results [] # 2. 遍历配置在训练集上评估 for i, config in enumerate(candidate_configs): print(f\n评估配置 [{i1}/{len(candidate_configs)}]: {config.get(name, unnamed)}) # 临时保存配置到文件供Harness加载 config_path Path(ftemp_config_{i}.yaml) with open(config_path, w, encodingutf-8) as f: yaml.dump(config, f, allow_unicodeTrue) try: # 初始化Harness (API_KEY需从环境变量获取) import os api_key os.getenv(OPENAI_API_KEY) if not api_key: raise ValueError(请设置 OPENAI_API_KEY 环境变量) harness ReasoningHarness(str(config_path), api_key) # 在训练子集上运行评估为了节省成本可以采样部分数据 sample_data random.sample(train_data, min(50, len(train_data))) # 示例采样50条 predictions [] for item in sample_data: result harness.run(item[question]) predictions.append({ id: item.get(id), prediction: result[answer], ground_truth: item.get(answer) }) # 计算得分例如准确率 score evaluate_predictions(predictions) # 假设该函数返回准确率 print(f 得分: {score:.4f}) all_results.append({ config: config, score: score, sample_size: len(sample_data) }) if score best_score: best_score score best_config config.copy() # 深拷贝最佳配置 except Exception as e: print(f 配置评估失败: {e}) all_results.append({ config: config, score: -1, error: str(e) }) finally: # 清理临时文件 if config_path.exists(): config_path.unlink() # 3. 用最佳配置在完整的评估集上做最终测试 print(f\n找到最佳配置在训练子集上得分为: {best_score:.4f}) print(开始在独立评估集上进行最终测试...) final_config_path Path(best_harness_config.yaml) with open(final_config_path, w, encodingutf-8) as f: yaml.dump(best_config, f, allow_unicodeTrue) final_harness ReasoningHarness(str(final_config_path), os.getenv(OPENAI_API_KEY)) final_predictions [] for item in eval_data: result final_harness.run(item[question]) final_predictions.append({ id: item.get(id), prediction: result[answer], ground_truth: item.get(answer), success: result[success] }) final_score evaluate_predictions(final_predictions) success_rate sum(1 for p in final_predictions if p[success]) / len(final_predictions) scaling_result { best_config: best_config, best_config_score_on_train_subset: best_score, final_score_on_eval_set: final_score, success_rate: success_rate, num_configs_evaluated: len(candidate_configs), all_results: all_results, final_predictions: final_predictions } # 保存结果 result_dir Path(results) / fscaling_run_{int(time.time())} result_dir.mkdir(parentsTrue, exist_okTrue) with open(result_dir / scaling_result.json, w, encodingutf-8) as f: json.dump(scaling_result, f, indent2, ensure_asciiFalse) with open(result_dir / best_config.yaml, w, encodingutf-8) as f: yaml.dump(best_config, f, allow_unicodeTrue) print(f\n Scaling 完成 ) print(f最佳配置已保存至: {result_dir / best_config.yaml}) print(f在评估集上的最终准确率: {final_score:.4f}) print(f任务执行成功率: {success_rate:.4f}) return scaling_result这个HarnessScaler类负责自动化地探索配置空间。它通过网格搜索或随机搜索生成不同的Harness配置在较小的训练数据子集上进行快速评估选出表现最好的配置最后在独立的测试集上给出最终性能报告。15美元的成本主要消耗在调用API评估不同配置的过程中。5. 运行验证与结果分析5.1 准备数据与运行实验首先你需要准备符合Terminal-Bench格式的数据。假设我们有一个简化的JSONL格式数据文件data/train.jsonl{id: 1, question: 用户说帮我订一张明天从北京飞往上海的机票要上午的。 接下来我应该调用哪个API, answer: flight_booking_api} {id: 2, question: 对话历史用户问天气。你回复了今天天气。用户又说那明天呢 用户现在想要什么, answer: 查询明天的天气}然后编写一个启动脚本scripts/run_experiment.pyimport sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from src.scaler import HarnessScaler import yaml import json def load_data(file_path): data [] with open(file_path, r, encodingutf-8) as f: for line in f: data.append(json.loads(line.strip())) return data def main(): # 0. 设置API密钥 os.environ[OPENAI_API_KEY] your-api-key-here # 请替换为你的密钥 # 1. 加载基础配置 with open(configs/harness_baseline.yaml, r, encodingutf-8) as f: base_config yaml.safe_load(f) # 2. 定义搜索空间 search_space { temperature: [0.1, 0.3, 0.7, 1.0], max_tokens: [512, 1024, 2048], input_engineering.cot_enabled: [True, False], input_engineering.few_shot_examples: [None], # 可以定义不同的示例集 output_parsing.fallback_strategy: [retry, use_raw], self_verification.enabled: [True, False] } # 3. 加载数据 train_data load_data(data/train.jsonl) eval_data load_data(data/test.jsonl) # 4. 初始化Scaler并运行 scaler HarnessScaler(base_config, search_space) # 假设我们只有15美元预算使用随机搜索 result scaler.run_scaling( train_datatrain_data, eval_dataeval_data, search_moderandom, num_samples20, # 尝试20种随机配置 budget_usd15.0 ) print(f\n实验结束。最佳准确率: {result[final_score_on_eval_set]}) if __name__ __main__: main()5.2 结果分析与解读运行脚本后你会在results/目录下得到详细的实验报告。关键结果包括最佳配置 (best_config.yaml)记录了达到最高分数的Harness参数组合。例如可能发现temperature0.3、cot_enabledTrue、self_verification.enabledTrue的组合在给定预算下最优。最终评估分数 (scaling_result.json中的final_score_on_eval_set)这就是在独立测试集上达到的准确率如95.3%。这个分数衡量了Harness Scaling方法对最终任务表现的提升。配置-分数关系通过分析all_results你可以了解哪些参数对性能影响最大。例如可能发现启用思维链CoT对复杂推理任务提升显著而对简单分类任务影响不大。成本分析15美元的成本是如何构成的假设每条数据调用一次模型平均输入输出共500个token。使用gpt-3.5-turbo模型每1K tokens成本约$0.002。在训练集上评估了N个配置每个配置用50条数据评估。最终在1000条的测试集上用最佳配置评估一次。 总成本 ≈(N * 50 1000) * 500 / 1000 * 0.002美元。通过调整N配置搜索数量和评估数据量可以将总成本控制在15美元以内。6. 常见问题与排查路径在实际运行Harness Scaling实验时你可能会遇到以下典型问题。6.1 API调用相关错误问题现象可能原因检查与解决方式AuthenticationErrorAPI密钥错误或未设置1. 检查OPENAI_API_KEY环境变量是否正确设置。2. 确保密钥有余额且未过期。3. 如果使用代理检查网络设置。RateLimitError请求频率超限1. 在代码中增加重试间隔tenacity库的wait参数。2. 降低并行请求数量如果使用了并发。3. 申请提高API速率限制。APIConnectionError或超时网络问题或服务不稳定1. 检查本地网络连接。2. 增加超时设置timeout参数。3. 实现指数退避的重试机制。代码示例增强的API调用封装from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from openai import APIConnectionError, RateLimitError, APIStatusError retry( stopstop_after_attempt(5), waitwait_exponential(multiplier1, min4, max60), retryretry_if_exception_type((APIConnectionError, RateLimitError, APIStatusError)) ) def robust_llm_call(client, messages, model, temperature, max_tokens): try: response client.chat.completions.create( modelmodel, messagesmessages, temperaturetemperature, max_tokensmax_tokens, timeout30.0 # 设置超时 ) return response.choices[0].message.content except RateLimitError as e: print(f速率限制等待后重试: {e}) raise except APIConnectionError as e: print(f网络连接错误: {e}) raise6.2 准确率没有提升或波动大问题现象可能原因检查与解决方式不同配置间分数差异很小搜索空间定义不合理或评估数据太少1. 检查搜索参数是否真的影响输出如temperature从0.1到1.0。2. 增加评估数据量如从50条到200条减少随机噪声。3. 尝试对模型输出影响更大的参数如system_prompt内容。分数波动大不稳定模型生成具有随机性或数据本身有歧义1. 对于关键实验对同一配置多次运行取平均分。2. 检查数据标注质量是否存在模糊或错误的标准答案。3. 在评估时使用更稳定的采样策略如temperature0。最佳配置在训练集有效测试集失效过拟合了训练数据的特定模式1. 确保训练集和测试集来自同一分布且无信息泄露。2. 减少Harness的复杂度如减少Few-Shot示例数量。3. 使用交叉验证或在更大的开发集上选择配置。6.3 输出解析失败率高问题现象可能原因检查与解决方式_extract_answer频繁返回None或报错答案前缀answer_prefix与模型输出不匹配1. 打印大量raw_output观察模型实际输出格式。2. 调整answer_prefix或使用更鲁棒的正则表达式如允许前缀前后有空格、换行。3. 考虑使用LLM本身进行输出解析即让模型以指定格式如JSON输出答案。自我验证步骤总是失败验证提示词设计不佳或验证标准过于严格1. 检查验证提示词是否清晰无歧义。2. 分析验证失败案例看是模型推理错误还是验证器误判。3. 可以尝试让验证器输出“是/否”或置信度分数而不是简单的关键词匹配。7. 最佳实践与扩展方向7.1 生产环境部署建议配置管理不要将API密钥等敏感信息硬编码在配置文件中。使用环境变量或专门的密钥管理服务。性能与成本监控记录每次API调用的token消耗、耗时和费用设置预算告警。缓存机制对于相同的输入和Harness配置缓存模型输出避免重复调用显著降低成本和延迟。降级与熔断当主要模型API不可用时应有备选模型或简化流程作为降级方案。日志与可观测性详细记录每个请求的输入、输出、中间步骤和异常便于问题排查和效果分析。7.2 扩展与优化方向更高效的搜索算法本文示例使用了随机搜索。可以引入贝叶斯优化Bayesian Optimization或基于梯度的优化方法来更智能地探索高维配置空间用更少的尝试找到更优解。分层Scaling不是所有参数都同等重要。可以先对system_prompt、cot_enabled等高影响力参数进行粗粒度搜索再对temperature等参数进行细粒度调优。集成学习不局限于寻找一个最优Harness可以训练一个元模型Meta-Model根据输入问题的特征如长度、关键词、类型动态选择或组合不同的Harness。自动化提示词工程将提示词模板也参数化并纳入搜索空间。例如使用遗传算法或LLM本身来迭代优化提示词模板。应用于其他模型和任务此框架不限于OpenAI模型或Terminal-Bench任务。可以轻松适配其他提供Chat Completion API的模型如Claude、Gemini、本地部署的LLaMA等以及其他需要复杂推理的评测基准如Big-Bench Hard, MMLU等。Harness Scaling的核心启示在于对于许多现实世界的AI应用投资于优化“如何使用模型”的流程其性价比可能远高于追求更大、更昂贵的模型。通过系统化的实验和自动化搜索即使预算有限也能在特定任务上取得令人瞩目的性能提升。 15美元成本实现SOTA：Harness Scaling优化大模型推理流程实践
在大型语言模型推理任务中如何以极低的成本实现接近SOTA的性能是许多开发者和研究者关注的焦点。传统的思路往往聚焦于模型本身的微调或架构改进但近期一项名为“Harness Scaling”的技术通过优化推理流程而非模型本身仅用15美元的成本就在Terminal-Bench 2.1基准测试上达到了95.3%的准确率刷新了该基准的SOTA记录。这为资源受限的团队或个人开发者提供了一条极具性价比的优化路径。本文将从工程实践的角度深入解析Harness Scaling的核心思想、实现原理并提供一个可复现的实践指南。我们将从理解基准测试和推理流程开始逐步拆解Harness Scaling的关键步骤包括如何构建高效的推理“马具”、设计缩放策略、编写核心代码并最终验证效果。无论你是希望优化现有大模型推理服务性能的工程师还是对低成本AI研究感兴趣的研究者都能通过本文获得一套清晰、可操作的方法论。1. 理解核心概念Terminal-Bench与Harness Scaling在动手实践之前必须厘清两个核心概念Terminal-Bench基准测试和Harness Scaling技术本身。这决定了我们优化的目标和方向。1.1 Terminal-Bench 2.1评估推理能力的“考场”Terminal-Bench是一个专门设计用于评估大型语言模型在终端用户任务上表现的基准测试集其2.1版本包含了更复杂、更多样的现实世界推理问题。与侧重于知识问答或代码生成的基准不同Terminal-Bench模拟的是用户与模型交互的完整闭环例如多轮对话理解与规划根据历史对话规划下一步行动。工具使用与API调用理解自然语言指令并转化为正确的工具调用序列。复杂指令分解将一个模糊的用户请求分解为一系列可执行的具体步骤。这个基准的关键在于它评估的不是模型“知道”什么而是模型在特定“流程”或“上下文”中“做对”事情的能力。因此优化模型在Terminal-Bench上的表现本质上是优化模型在特定应用流程中的表现。1.2 Harness Scaling优化“流程”而非“引擎”Harness中文可译为“马具”或“挽具”在这里是一个精妙的比喻。我们可以将大型语言模型LLM比作一匹强大的“赛马”引擎而Harness则是驾驭这匹赛马、使其在特定赛道上发挥最佳水平的“缰绳、鞍具和策略”流程。Harness Scaling的核心思想是在不修改模型权重即不微调模型的前提下通过精心设计和规模化地优化调用模型的前后处理流程来显著提升最终任务的表现。这个流程通常包括输入工程如何将原始问题转化为最适合模型理解的提示词Prompt。包括添加系统指令、思维链CoT示例、上下文信息等。采样策略如何设置温度Temperature、Top-p等参数以平衡输出的创造性和确定性。输出解析与后处理如何从模型生成的文本中准确、结构化地提取出任务所需的答案或动作。自我验证与重试让模型对自身输出进行检查或在不确定时设计重试机制。流程编排对于复杂任务如何将大问题拆解为多个子问题并编排多次模型调用。“Scaling”缩放在这里意味着不是设计一个固定的好流程而是系统地探索流程中各个组件的不同配置组合即不同的“Harness”并在大量任务上评估这些组合最终找到一个或一组在统计意义上最优的流程配置。简而言之Harness Scaling把优化重点从“让模型变得更聪明”转移到了“让使用模型的方法变得更聪明”上。15美元的成本主要就花在了自动化搜索和评估大量不同Harness配置的计算上。2. 环境准备与项目结构要实现Harness Scaling我们需要搭建一个能够自动化进行提示词工程、模型调用、结果评估和配置搜索的实验环境。2.1 基础环境与依赖假设我们使用Python作为主要开发语言并利用OpenAI API或兼容API作为模型服务。以下是核心依赖# 创建虚拟环境可选但推荐 python -m venv harness_env source harness_env/bin/activate # Linux/macOS # harness_env\Scripts\activate # Windows # 安装核心依赖 pip install openai1.0.0 # 官方SDK pip install tqdm # 进度条 pip install pandas # 数据处理 pip install numpy # 数值计算 pip install scikit-learn # 用于评估指标计算如准确率 pip install tenacity # 用于实现API调用的重试机制 pip install pyyaml # 用于管理Harness配置如果使用其他模型API如Anthropic Claude, Google Gemini或本地部署的vLLM服务需安装相应的SDK。2.2 项目目录结构一个清晰的项目结构有助于管理大量的Harness配置和实验结果。terminal_bench_harness_scaling/ ├── configs/ # 存放各种Harness配置 │ ├── harness_baseline.yaml │ ├── harness_cot.yaml │ └── harness_self_consistency.yaml ├── data/ # 存放Terminal-Bench数据集或你的自定义数据集 │ ├── train.jsonl │ └── test.jsonl ├── src/ │ ├── __init__.py │ ├── harness.py # Harness核心类定义流程 │ ├── evaluator.py # 评估逻辑 │ ├── scaler.py # Scaling搜索策略 │ └── utils.py # 工具函数如API调用封装 ├── results/ # 存放每次实验的详细结果 │ └── 20240520_experiment_1/ ├── scripts/ │ └── run_experiment.py # 实验启动脚本 ├── requirements.txt └── README.md3. 构建可配置的推理HarnessHarness的核心是一个可配置的类它定义了从原始问题到最终答案的完整处理流程。3.1 定义Harness配置YAML我们使用YAML文件来灵活定义不同的Harness。configs/harness_template.yamlname: cot_with_self_check # Harness名称 model: gpt-3.5-turbo # 使用的模型 temperature: 0.7 max_tokens: 1024 # 输入工程模块 input_engineering: system_prompt: | 你是一个严谨的推理助手。请逐步思考并在最终答案前标明“答案是”。 few_shot_examples: # 少量示例Few-Shot - question: 如果小明有5个苹果吃了2个又买了3个他现在有几个苹果 thought: 首先初始有5个。吃掉2个剩余5-23个。然后买来3个总共336个。 answer: 答案是6 cot_enabled: true # 是否启用思维链 cot_trigger: 请逐步思考 # 输出解析模块 output_parsing: answer_prefix: 答案是 # 用于从模型输出中定位答案 fallback_strategy: retry # 解析失败时的策略retry, use_raw, fail # 自我验证模块 self_verification: enabled: true verification_prompt: | 请检查以下推理过程和答案是否正确。如果正确回复“正确”如果错误指出错误并给出正确推理。 问题{question} 推理{reasoning} 答案{extracted_answer} max_retries: 1 # 流程编排针对复杂任务 orchestration: max_steps: 3 # 最大分解步数 aggregation_method: last_step # 如何聚合多步结果3.2 实现Harness核心类src/harness.pyimport yaml import re from tenacity import retry, stop_after_attempt, wait_exponential from openai import OpenAI from typing import Dict, Any, Optional, List class ReasoningHarness: def __init__(self, config_path: str, api_key: str): 初始化Harness加载配置。 :param config_path: Harness配置文件的路径 :param api_key: OpenAI API密钥 with open(config_path, r, encodingutf-8) as f: self.config yaml.safe_load(f) self.client OpenAI(api_keyapi_key) self.model self.config.get(model, gpt-3.5-turbo) def _build_messages(self, question: str) - List[Dict[str, str]]: 根据配置构建发送给模型的messages列表。 messages [] # 1. 添加系统提示 if sys_prompt : self.config[input_engineering].get(system_prompt): messages.append({role: system, content: sys_prompt}) # 2. 添加Few-Shot示例 if examples : self.config[input_engineering].get(few_shot_examples, []): for ex in examples: # 示例中的对话历史模拟 messages.append({role: user, content: ex[question]}) # 如果启用了COT将思考过程作为assistant的一部分内容 cot_content ex.get(thought, ) \n ex[answer] if self.config[input_engineering].get(cot_enabled) else ex[answer] messages.append({role: assistant, content: cot_content}) # 3. 构建当前问题的提示 user_content question if self.config[input_engineering].get(cot_enabled): cot_trigger self.config[input_engineering].get(cot_trigger, 请逐步思考) user_content f{user_content}\n\n{cot_trigger} messages.append({role: user, content: user_content}) return messages retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def _call_llm(self, messages: List[Dict[str, str]]) - str: 调用LLM API包含重试机制。 response self.client.chat.completions.create( modelself.model, messagesmessages, temperatureself.config.get(temperature, 0.7), max_tokensself.config.get(max_tokens, 1024) ) return response.choices[0].message.content def _extract_answer(self, llm_output: str) - Optional[str]: 从模型输出中解析出最终答案。 parsing_config self.config[output_parsing] prefix parsing_config.get(answer_prefix) if prefix and prefix in llm_output: # 简单查找前缀后的内容作为答案 pattern re.escape(prefix) r\s*(.) match re.search(pattern, llm_output, re.DOTALL) if match: return match.group(1).strip() # 如果解析失败根据策略处理 strategy parsing_config.get(fallback_strategy, fail) if strategy use_raw: return llm_output.strip() elif strategy retry: return None # 返回None外部逻辑可触发重试 else: # fail raise ValueError(f无法从输出中解析答案: {llm_output[:200]}...) def _self_verify(self, question: str, reasoning: str, answer: str) - bool: 自我验证步骤。 if not self.config[self_verification].get(enabled, False): return True # 未启用验证默认通过 verify_prompt_template self.config[self_verification][verification_prompt] verify_prompt verify_prompt_template.format( questionquestion, reasoningreasoning, extracted_answeranswer ) messages [{role: user, content: verify_prompt}] verification_response self._call_llm(messages) return 正确 in verification_response # 简单的关键词判断生产环境需更鲁棒 def run(self, question: str) - Dict[str, Any]: 执行完整的Harness流程。 返回包含原始输出、解析后的答案、元数据等信息的字典。 result { question: question, harness: self.config[name], success: False, answer: None, raw_output: None, reasoning: None, retries: 0 } max_retries self.config[output_parsing].get(max_parsing_retries, 0) 1 if self.config[self_verification].get(enabled): max_retries max(max_retries, self.config[self_verification].get(max_retries, 0) 1) for attempt in range(max_retries): try: # 1. 构建输入 messages self._build_messages(question) # 2. 调用模型 raw_output self._call_llm(messages) result[raw_output] raw_output # 3. 提取答案 extracted_answer self._extract_answer(raw_output) if extracted_answer is None and attempt max_retries - 1: # 解析失败触发重试可能通过修改提示 continue elif extracted_answer is None: raise ValueError(答案解析失败已达最大重试次数。) # 4. 自我验证如果启用 # 简单地从原始输出中分离“推理过程”和“答案”实际项目需更精细 result[reasoning] raw_output # 此处简化处理 if self._self_verify(question, raw_output, extracted_answer): result[answer] extracted_answer result[success] True break else: # 验证失败可能触发重试或标记为失败 if attempt max_retries - 1: # 可以在这里修改提示词加入验证失败的反馈 continue else: result[answer] extracted_answer # 即使验证失败也记录答案 # result[success] 保持 False except Exception as e: result[error] str(e) if attempt max_retries - 1: break result[retries] attempt return result这个ReasoningHarness类封装了从配置加载、提示词构建、模型调用到输出解析和自我验证的完整流程。通过修改YAML配置文件我们可以轻松定义不同的Harness策略。4. 实现Harness Scaling搜索策略有了基础的Harness下一步是实现Scaling——自动化地搜索最优配置。我们采用一种简单的网格搜索Grid Search与随机搜索Random Search结合的策略作为示例。src/scaler.pyimport itertools import random import json from pathlib import Path from typing import List, Dict, Any from .harness import ReasoningHarness from .evaluator import evaluate_predictions # 假设有一个评估函数 class HarnessScaler: def __init__(self, base_config: Dict[str, Any], search_space: Dict[str, List]): :param base_config: 基础Harness配置字典 :param search_space: 搜索空间例如 { temperature: [0.1, 0.5, 0.9], input_engineering.cot_enabled: [True, False], input_engineering.few_shot_examples: [None, example_list_1, example_list_2] } self.base_config base_config self.search_space search_space def _generate_configs_grid(self) - List[Dict[str, Any]]: 生成网格搜索的所有配置组合。 keys list(self.search_space.keys()) value_lists [self.search_space[key] for key in keys] configs [] for combination in itertools.product(*value_lists): new_config self._deep_update(self.base_config.copy(), dict(zip(keys, combination))) configs.append(new_config) return configs def _generate_configs_random(self, num_samples: int) - List[Dict[str, Any]]: 随机采样指定数量的配置。 configs [] for _ in range(num_samples): new_config self.base_config.copy() for key, values in self.search_space.items(): chosen_value random.choice(values) new_config self._deep_update(new_config, {key: chosen_value}) # 确保每个配置有唯一名称 new_config[name] fharness_random_{random.randint(1000,9999)} configs.append(new_config) return configs def _deep_update(self, base: Dict, update: Dict) - Dict: 深度更新字典支持嵌套键如 input_engineering.cot_enabled。 for key, value in update.items(): if . in key: # 处理嵌套键 parts key.split(.) d base for part in parts[:-1]: d d.setdefault(part, {}) d[parts[-1]] value else: base[key] value return base def run_scaling(self, train_data: List[Dict], eval_data: List[Dict], search_mode: str random, num_samples: int 10, budget_usd: float 15.0) - Dict[str, Any]: 执行Scaling搜索。 :param train_data: 用于搜索/开发的数据集 :param eval_data: 用于最终评估的数据集 :param search_mode: grid 或 random :param num_samples: 随机搜索时的采样数 :param budget_usd: 预算美元用于估算成本和控制搜索范围 :return: 包含最佳配置和结果详情的字典 # 1. 根据预算和搜索模式生成待测试的配置列表 if search_mode grid: candidate_configs self._generate_configs_grid() # 如果网格组合太多根据预算进行粗略裁剪 # 简单估算每个数据点调用一次模型假设平均token成本 estimated_cost_per_config len(train_data) * 0.002 # 示例估算值 max_configs_by_budget int(budget_usd / estimated_cost_per_config) if len(candidate_configs) max_configs_by_budget: print(f网格组合数({len(candidate_configs)})超出预算限制({max_configs_by_budget})将进行随机子采样。) candidate_configs random.sample(candidate_configs, max_configs_by_budget) else: # random # 根据预算计算最大可采样数 estimated_cost_per_config len(train_data) * 0.002 max_samples_by_budget int(budget_usd / estimated_cost_per_config) num_samples min(num_samples, max_samples_by_budget) candidate_configs self._generate_configs_random(num_samples) print(f将在 {len(candidate_configs)} 个Harness配置上进行搜索预计成本约 ${len(candidate_configs)*estimated_cost_per_config:.2f}) best_score -1 best_config None all_results [] # 2. 遍历配置在训练集上评估 for i, config in enumerate(candidate_configs): print(f\n评估配置 [{i1}/{len(candidate_configs)}]: {config.get(name, unnamed)}) # 临时保存配置到文件供Harness加载 config_path Path(ftemp_config_{i}.yaml) with open(config_path, w, encodingutf-8) as f: yaml.dump(config, f, allow_unicodeTrue) try: # 初始化Harness (API_KEY需从环境变量获取) import os api_key os.getenv(OPENAI_API_KEY) if not api_key: raise ValueError(请设置 OPENAI_API_KEY 环境变量) harness ReasoningHarness(str(config_path), api_key) # 在训练子集上运行评估为了节省成本可以采样部分数据 sample_data random.sample(train_data, min(50, len(train_data))) # 示例采样50条 predictions [] for item in sample_data: result harness.run(item[question]) predictions.append({ id: item.get(id), prediction: result[answer], ground_truth: item.get(answer) }) # 计算得分例如准确率 score evaluate_predictions(predictions) # 假设该函数返回准确率 print(f 得分: {score:.4f}) all_results.append({ config: config, score: score, sample_size: len(sample_data) }) if score best_score: best_score score best_config config.copy() # 深拷贝最佳配置 except Exception as e: print(f 配置评估失败: {e}) all_results.append({ config: config, score: -1, error: str(e) }) finally: # 清理临时文件 if config_path.exists(): config_path.unlink() # 3. 用最佳配置在完整的评估集上做最终测试 print(f\n找到最佳配置在训练子集上得分为: {best_score:.4f}) print(开始在独立评估集上进行最终测试...) final_config_path Path(best_harness_config.yaml) with open(final_config_path, w, encodingutf-8) as f: yaml.dump(best_config, f, allow_unicodeTrue) final_harness ReasoningHarness(str(final_config_path), os.getenv(OPENAI_API_KEY)) final_predictions [] for item in eval_data: result final_harness.run(item[question]) final_predictions.append({ id: item.get(id), prediction: result[answer], ground_truth: item.get(answer), success: result[success] }) final_score evaluate_predictions(final_predictions) success_rate sum(1 for p in final_predictions if p[success]) / len(final_predictions) scaling_result { best_config: best_config, best_config_score_on_train_subset: best_score, final_score_on_eval_set: final_score, success_rate: success_rate, num_configs_evaluated: len(candidate_configs), all_results: all_results, final_predictions: final_predictions } # 保存结果 result_dir Path(results) / fscaling_run_{int(time.time())} result_dir.mkdir(parentsTrue, exist_okTrue) with open(result_dir / scaling_result.json, w, encodingutf-8) as f: json.dump(scaling_result, f, indent2, ensure_asciiFalse) with open(result_dir / best_config.yaml, w, encodingutf-8) as f: yaml.dump(best_config, f, allow_unicodeTrue) print(f\n Scaling 完成 ) print(f最佳配置已保存至: {result_dir / best_config.yaml}) print(f在评估集上的最终准确率: {final_score:.4f}) print(f任务执行成功率: {success_rate:.4f}) return scaling_result这个HarnessScaler类负责自动化地探索配置空间。它通过网格搜索或随机搜索生成不同的Harness配置在较小的训练数据子集上进行快速评估选出表现最好的配置最后在独立的测试集上给出最终性能报告。15美元的成本主要消耗在调用API评估不同配置的过程中。5. 运行验证与结果分析5.1 准备数据与运行实验首先你需要准备符合Terminal-Bench格式的数据。假设我们有一个简化的JSONL格式数据文件data/train.jsonl{id: 1, question: 用户说帮我订一张明天从北京飞往上海的机票要上午的。 接下来我应该调用哪个API, answer: flight_booking_api} {id: 2, question: 对话历史用户问天气。你回复了今天天气。用户又说那明天呢 用户现在想要什么, answer: 查询明天的天气}然后编写一个启动脚本scripts/run_experiment.pyimport sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from src.scaler import HarnessScaler import yaml import json def load_data(file_path): data [] with open(file_path, r, encodingutf-8) as f: for line in f: data.append(json.loads(line.strip())) return data def main(): # 0. 设置API密钥 os.environ[OPENAI_API_KEY] your-api-key-here # 请替换为你的密钥 # 1. 加载基础配置 with open(configs/harness_baseline.yaml, r, encodingutf-8) as f: base_config yaml.safe_load(f) # 2. 定义搜索空间 search_space { temperature: [0.1, 0.3, 0.7, 1.0], max_tokens: [512, 1024, 2048], input_engineering.cot_enabled: [True, False], input_engineering.few_shot_examples: [None], # 可以定义不同的示例集 output_parsing.fallback_strategy: [retry, use_raw], self_verification.enabled: [True, False] } # 3. 加载数据 train_data load_data(data/train.jsonl) eval_data load_data(data/test.jsonl) # 4. 初始化Scaler并运行 scaler HarnessScaler(base_config, search_space) # 假设我们只有15美元预算使用随机搜索 result scaler.run_scaling( train_datatrain_data, eval_dataeval_data, search_moderandom, num_samples20, # 尝试20种随机配置 budget_usd15.0 ) print(f\n实验结束。最佳准确率: {result[final_score_on_eval_set]}) if __name__ __main__: main()5.2 结果分析与解读运行脚本后你会在results/目录下得到详细的实验报告。关键结果包括最佳配置 (best_config.yaml)记录了达到最高分数的Harness参数组合。例如可能发现temperature0.3、cot_enabledTrue、self_verification.enabledTrue的组合在给定预算下最优。最终评估分数 (scaling_result.json中的final_score_on_eval_set)这就是在独立测试集上达到的准确率如95.3%。这个分数衡量了Harness Scaling方法对最终任务表现的提升。配置-分数关系通过分析all_results你可以了解哪些参数对性能影响最大。例如可能发现启用思维链CoT对复杂推理任务提升显著而对简单分类任务影响不大。成本分析15美元的成本是如何构成的假设每条数据调用一次模型平均输入输出共500个token。使用gpt-3.5-turbo模型每1K tokens成本约$0.002。在训练集上评估了N个配置每个配置用50条数据评估。最终在1000条的测试集上用最佳配置评估一次。 总成本 ≈(N * 50 1000) * 500 / 1000 * 0.002美元。通过调整N配置搜索数量和评估数据量可以将总成本控制在15美元以内。6. 常见问题与排查路径在实际运行Harness Scaling实验时你可能会遇到以下典型问题。6.1 API调用相关错误问题现象可能原因检查与解决方式AuthenticationErrorAPI密钥错误或未设置1. 检查OPENAI_API_KEY环境变量是否正确设置。2. 确保密钥有余额且未过期。3. 如果使用代理检查网络设置。RateLimitError请求频率超限1. 在代码中增加重试间隔tenacity库的wait参数。2. 降低并行请求数量如果使用了并发。3. 申请提高API速率限制。APIConnectionError或超时网络问题或服务不稳定1. 检查本地网络连接。2. 增加超时设置timeout参数。3. 实现指数退避的重试机制。代码示例增强的API调用封装from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from openai import APIConnectionError, RateLimitError, APIStatusError retry( stopstop_after_attempt(5), waitwait_exponential(multiplier1, min4, max60), retryretry_if_exception_type((APIConnectionError, RateLimitError, APIStatusError)) ) def robust_llm_call(client, messages, model, temperature, max_tokens): try: response client.chat.completions.create( modelmodel, messagesmessages, temperaturetemperature, max_tokensmax_tokens, timeout30.0 # 设置超时 ) return response.choices[0].message.content except RateLimitError as e: print(f速率限制等待后重试: {e}) raise except APIConnectionError as e: print(f网络连接错误: {e}) raise6.2 准确率没有提升或波动大问题现象可能原因检查与解决方式不同配置间分数差异很小搜索空间定义不合理或评估数据太少1. 检查搜索参数是否真的影响输出如temperature从0.1到1.0。2. 增加评估数据量如从50条到200条减少随机噪声。3. 尝试对模型输出影响更大的参数如system_prompt内容。分数波动大不稳定模型生成具有随机性或数据本身有歧义1. 对于关键实验对同一配置多次运行取平均分。2. 检查数据标注质量是否存在模糊或错误的标准答案。3. 在评估时使用更稳定的采样策略如temperature0。最佳配置在训练集有效测试集失效过拟合了训练数据的特定模式1. 确保训练集和测试集来自同一分布且无信息泄露。2. 减少Harness的复杂度如减少Few-Shot示例数量。3. 使用交叉验证或在更大的开发集上选择配置。6.3 输出解析失败率高问题现象可能原因检查与解决方式_extract_answer频繁返回None或报错答案前缀answer_prefix与模型输出不匹配1. 打印大量raw_output观察模型实际输出格式。2. 调整answer_prefix或使用更鲁棒的正则表达式如允许前缀前后有空格、换行。3. 考虑使用LLM本身进行输出解析即让模型以指定格式如JSON输出答案。自我验证步骤总是失败验证提示词设计不佳或验证标准过于严格1. 检查验证提示词是否清晰无歧义。2. 分析验证失败案例看是模型推理错误还是验证器误判。3. 可以尝试让验证器输出“是/否”或置信度分数而不是简单的关键词匹配。7. 最佳实践与扩展方向7.1 生产环境部署建议配置管理不要将API密钥等敏感信息硬编码在配置文件中。使用环境变量或专门的密钥管理服务。性能与成本监控记录每次API调用的token消耗、耗时和费用设置预算告警。缓存机制对于相同的输入和Harness配置缓存模型输出避免重复调用显著降低成本和延迟。降级与熔断当主要模型API不可用时应有备选模型或简化流程作为降级方案。日志与可观测性详细记录每个请求的输入、输出、中间步骤和异常便于问题排查和效果分析。7.2 扩展与优化方向更高效的搜索算法本文示例使用了随机搜索。可以引入贝叶斯优化Bayesian Optimization或基于梯度的优化方法来更智能地探索高维配置空间用更少的尝试找到更优解。分层Scaling不是所有参数都同等重要。可以先对system_prompt、cot_enabled等高影响力参数进行粗粒度搜索再对temperature等参数进行细粒度调优。集成学习不局限于寻找一个最优Harness可以训练一个元模型Meta-Model根据输入问题的特征如长度、关键词、类型动态选择或组合不同的Harness。自动化提示词工程将提示词模板也参数化并纳入搜索空间。例如使用遗传算法或LLM本身来迭代优化提示词模板。应用于其他模型和任务此框架不限于OpenAI模型或Terminal-Bench任务。可以轻松适配其他提供Chat Completion API的模型如Claude、Gemini、本地部署的LLaMA等以及其他需要复杂推理的评测基准如Big-Bench Hard, MMLU等。Harness Scaling的核心启示在于对于许多现实世界的AI应用投资于优化“如何使用模型”的流程其性价比可能远高于追求更大、更昂贵的模型。通过系统化的实验和自动化搜索即使预算有限也能在特定任务上取得令人瞩目的性能提升。