上一篇分享过后台任务编排的思路很多朋友留言问既然云端推理要花钱、要联网有没有办法让本地模型也能跑出一套“可复用、可评测、可扩展”的推理流程答案就是给端侧模型套一个专属 Harness。本文就基于 Qwen3 的 8B/27B 这一档端侧友好模型把一套轻量级 Harness 从设计到落地完整拆开讲清楚代码可以直接复制运行。1. 背景端侧模型为什么需要专属 Harness1.1 从一次本地推理实验说起我们平时做本地推理实验通常是下面这样三步下载一个量化后的模型权重用 Ollama、llama.cpp 或者 vLLM 把模型加载起来通过 HTTP 接口发一条 prompt拿到返回结果。看起来流程很简单但一旦你把“跑通一次对话”升级成“稳定跑完一批任务”问题就出现了prompt 格式不统一、结果输出类型不稳定、遇到 bad case 无法定位是模型问题还是上游数据问题、缺乏评测指标来判断模型改动后是变好还是变坏。这些都是 Harness 要解决的事。1.2 Harness 到底是什么Harness 直译是“背带、线束”在软件工程里通常指一套“工作负载编排框架”。放到大语言模型场景下Harness 的含义可以这样理解Harness 是包裹在模型外部的一层工程框架负责把提示词构造、模型调用、输出解析、评测指标、错误重试等环节标准化让模型能力可以稳定地被上层业务调用。注意Harness 不是模型本身也不是推理引擎而是连接“模型能力”和“业务需求”之间的那层胶水。目前社区里经常听到的 deepseek harness、codex harness就是在不同模型或不同业务场景下对 Harness 做的具体实现。它们有一个共同点把模型当成一个可以调度的“计算单元”通过 Harness 去统一管理和使用。1.3 端侧模型为什么比云端模型更需要 Harness云端模型走 HTTP 调用厂商会帮你处理并发、鉴权、限流你拿到的是稳定的 SDK 和标准接口。但端侧模型完全不一样部署环境差异大可能是 Windows 笔记本、Mac、Linux 服务器显存从 8GB 到 32GB 不等模型格式不统一有 GGUF、AWQ、GPTQ、MLX 等多种格式推理引擎分散Ollama、llama.cpp、transformers、vLLM 各有各的调用方式结果不稳定本地模型参数量小同一个问题换个说法结果可能差很多缺少评测闭环没有评测你根本无法判断换一个量化级别是否值得。所以端侧模型一定要有一个专属的 Harness 层把这些问题在工程层面屏蔽掉。这也是本文标题里“专属 Harness”的核心含义。2. 零成本推理路线的可行性分析2.1 “零成本推理”到底指什么需要先澄清一个概念零成本推理不是说不需要算力而是说你不需要按 token 付费也不需要购买云端 GPU 实例。如果你的电脑有一块 8GB 显存的显卡或者一台 16GB 内存的 Mac就可以跑 Qwen3 8B 量化的端侧推理。如果你有 32GB 以上内存或 24GB 显存那么 27B 这一档模型也能跑起来。和云端 API 相比端侧推理的成本结构是这样的成本项云端模型端侧模型GPU 实例费用按小时或按 token 计费无网络流量费用需要不需要数据隐私数据出设备数据不出设备硬件成本无但长期使用累计一次性投入运维成本较低需要自己负责所以这里说的“零成本”指的是边际推理成本接近零模型部署好之后你在本地想跑多少次就跑多少次不产生额外费用。2.2 端侧模型选择Qwen3 8B 与 27B 的定位Qwen3 系列开源模型中8B 和 27B 这两个尺寸是典型的端侧模型。它们的定位有明显差异Qwen3-8B适合 8GB 显存或 16GB 内存设备量化后占用约 5-6GB推理速度快适合日常问答、信息抽取、文本分类、简单工具调用Qwen3-27B适合 24GB 显存或 32GB 内存设备量化后占用约 15-18GB推理质量有明显提升可以承担更复杂的推理任务、代码生成、结构化输出。选择建议如果主要做结构化文本处理8B 就够了如果希望模型有更强的逻辑推理和指令遵循能力优先选 27B如果机器跑不动 27B可以先用 AWQ 或 GGUF 的 Q4 量化版本再迭代到更高精度。2.3 推理引擎选型端侧推理引擎我比较推荐 Ollama原因有三个接口简单启动后提供了兼容 OpenAI 格式的 HTTP 接口业务代码几乎没有迁移成本模型管理方便一条命令就能拉取和切换模型跨平台Windows、macOS、Linux 都能跑还支持 GGUF 格式的量化模型。如果你的场景需要更高吞吐量可以考虑 vLLM 或 SGLang如果机器性能很弱可以用 llama.cpp 的llama-server。但本文的 Harness 代码不绑定具体引擎只需要一个兼容 OpenAI Chat API 的接口。3. 轻量级 Harness 的核心设计3.1 Harness 应该包含哪些模块设计一个 Harness不是写一个chat()函数就完事。一个能用于实际项目的端侧模型 Harness通常需要包含下面五个模块模型封装层负责加载模型配置、初始化客户端、管理模型名和 API 地址。上层业务不需要关心模型是部署在 Ollama 还是 vLLM 上。提示词管理模块把系统提示词、用户输入、历史消息拼装成标准消息列表并对输入长度做校验和截断。输出解析模块负责把模型返回的文本解析成业务需要的格式比如 JSON、代码片段、分类标签。解析失败时能触发重试。评测模块对模型输出做自动化校验包括必填字段检查、JSON 合法性检查、格式规范检查。这一步是做模型迭代的基础。运行编排模块支持批量跑 prompt、并发控制、失败重试、结果落盘。3.2 设计原则设计端侧 Harness 时建议遵循下面几个原则接口统一不管底层换了什么模型、什么引擎Harness 对外暴露的接口应该是一致的。这样上层业务不用因为模型调整而改代码。可重试端侧模型偶发性比较强一次调用可能因为显存不足、上下文超长而失败。Harness 需要内置重试机制而不是把错误直接抛给上层。可观测每条请求最好记录耗时、输入 token 数、输出内容、解析结果、是否重试。这些日志是排查问题的关键。资源可控端侧环境算力有限Harness 要支持批量任务的并发数限制防止同时发起太多请求导致 OOM。4. 完整实战用 Qwen3 跑一套端侧推理 Harness接下来是本文的核心部分。我会从零搭建一套轻量级端侧推理 Harness并用它在本地跑 Qwen3 模型完成一批实际推理任务。4.1 环境准备本文示例的演示环境操作系统Windows 11 / Ubuntu 22.04代码跨平台Python3.10推理引擎Ollama 最新版模型qwen3:8b 或 qwen3:27b按你的显存/内存选择依赖库openai、pandas可选用于结果整理先安装 Ollama# Linux / macOS 安装 curl -fsSL https://ollama.com/install.sh | sh # Windows 直接下载安装包 # https://ollama.com/download启动 Ollama 服务并拉取模型ollama serve拉取 Qwen3 8B 模型ollama pull qwen3:8b如果你的机器配置更高可以拉取 27Bollama pull qwen3:27b确认模型可以正常对话ollama run qwen3:8b 你好请简单介绍你自己4.2 创建项目结构我们创建一个独立的 Python 项目结构如下local-harness/ ├── config.py # 模型配置 ├── llm_client.py # 模型客户端封装 ├── prompt_manager.py # 提示词管理 ├── output_parser.py # 输出解析 ├── evaluator.py # 评测模块 ├── harness.py # Harness 主类 ├── main.py # 演示入口 └── results/ # 结果输出目录4.3 编写模型客户端先写config.py统一管理模型配置# config.py from dataclasses import dataclass dataclass class ModelConfig: model_name: str qwen3:8b base_url: str http://localhost:11434/v1 api_key: str ollama # Ollama 不校验 key保留字段即可 temperature: float 0.2 # 端侧任务建议低温度保证稳定 max_tokens: int 2048 timeout: int 120 # 端侧推理较慢超时时间要宽松再写llm_client.py把 OpenAI SDK 封装成统一的客户端接口# llm_client.py from openai import OpenAI from config import ModelConfig class LLMClient: 统一模型调用客户端支持同步和流式输出 def __init__(self, config: ModelConfig): self.config config self.client OpenAI( base_urlconfig.base_url, api_keyconfig.api_key, timeoutconfig.timeout, ) def chat( self, messages: list[dict], temperature: float | None None, max_tokens: int | None None, ) - str: 发送对话请求返回完整文本结果 resp self.client.chat.completions.create( modelself.config.model_name, messagesmessages, temperaturetemperature or self.config.temperature, max_tokensmax_tokens or self.config.max_tokens, ) return resp.choices[0].message.content def chat_stream(self, messages: list[dict]) - str: 流式返回结果适合交互式场景 stream self.client.chat.completions.create( modelself.config.model_name, messagesmessages, temperatureself.config.temperature, max_tokensself.config.max_tokens, streamTrue, ) parts [] for chunk in stream: if chunk.choices[0].delta.content: parts.append(chunk.choices[0].delta.content) return .join(parts)注意这里使用的是 Ollama 的 OpenAI 兼容接口如果你的本地服务部署在别的端口或使用其他引擎只需要修改base_url和model_name即可llm_client.py不需要改动。4.4 编写提示词管理模块prompt_manager.py负责拼装消息列表。这里我加入了一个简单但实用的设计支持系统提示词模板和多轮会话拼接。# prompt_manager.py from typing import Any class PromptManager: 提示词管理负责构造标准消息列表 def __init__(self, system_prompt: str, max_history: int 10): self.system_prompt system_prompt self.max_history max_history def build_messages( self, user_input: str, history: list[dict] | None None, ) - list[dict]: messages [{role: system, content: self.system_prompt}] # 拼接历史消息超出长度则丢弃最早的 if history: messages.extend(history[-self.max_history:]) messages.append({role: user, content: user_input}) return messages这里有一个容易被忽略的点system prompt 对端侧模型的影响很大。Qwen3 8B 这种规模的模型指令遵循能力有限如果 system prompt 写得太长真正有用的指令会被淹没。所以后续配置 system prompt 时尽量做到简洁、分点、明确。4.5 编写输出解析模块输出解析是 Harness 里最容易被低估的模块。端侧模型生成 JSON 时经常出现多余文字、前后缀、单双引号混用等问题所以解析需要做容错处理。# output_parser.py import json import re from typing import Any class OutputParser: 模型输出解析器重点解决 JSON 解析容错 staticmethod def parse_json(text: str) - dict: 尝试将模型输出解析为 JSON支持多种容错处理 # 方法 1直接解析 try: return json.loads(text) except json.JSONDecodeError: pass # 方法 2提取第一个 { 到最后一个 } 之间的内容 try: start text.find({) end text.rfind(}) if start ! -1 and end ! -1 and end start: return json.loads(text[start : end 1]) except json.JSONDecodeError: pass # 方法 3去掉 Markdown 代码块标记后解析 try: cleaned re.sub(rjson|, , text).strip() return json.loads(cleaned) except json.JSONDecodeError: pass raise ValueError(f无法解析模型输出为 JSON: {text[:200]}) staticmethod def parse_classify(text: str, candidates: list[str]) - str: 将模型输出解析为候选分类中的一个 for candidate in candidates: if candidate.lower() in text.lower(): return candidate return unknown这里的思路是“由严格到宽松”先尝试标准解析再逐步放宽条件。如果所有方法都失败就抛出异常由 Harness 主类决定是否重试。4.6 编写评测模块评测模块的作用是在跑完推理后自动判断哪条结果是成功的、哪条需要人工检查。这里我实现了一个最小可用的评测器# evaluator.py from typing import Any class Evaluator: 评测模块对模型输出做结构化校验 staticmethod def evaluate( output: Any, expected_keys: list[str] | None None, expected_type: str | None None, ) - dict: 返回评测结果 expected_keys: 期望输出 dict 中必须存在的字段 expected_type: 期望输出类型如 dict / list / str result {passed: False, errors: []} if expected_type dict: if not isinstance(output, dict): result[errors].append(f输出类型错误期望 dict实际 {type(output).__name__}) else: missing [k for k in (expected_keys or []) if k not in output] if missing: result[errors].append(f缺少字段: {missing}) else: result[passed] True elif expected_type list: if not isinstance(output, list): result[errors].append(f输出类型错误期望 list实际 {type(output).__name__}) else: result[passed] True else: result[passed] True return result4.7 编写 Harness 主类harness.py是核心入口它把前面几个模块串成一个完整流程# harness.py import json import time from typing import Any, Callable from config import ModelConfig from evaluator import Evaluator from llm_client import LLMClient from output_parser import OutputParser from prompt_manager import PromptManager class LocalHarness: 端侧模型 Harness 主类 def __init__( self, model_config: ModelConfig, system_prompt: str, max_retries: int 3, ): self.model_config model_config self.client LLMClient(model_config) self.prompt_manager PromptManager(system_prompt) self.parser OutputParser() self.evaluator Evaluator() self.max_retries max_retries def run( self, user_input: str, parse_mode: str text, expected_keys: list[str] | None None, expected_type: str | None None, history: list[dict] | None None, ) - dict: 执行一次完整的推理任务 parse_mode: text / json / classify messages self.prompt_manager.build_messages(user_input, history) errors [] for attempt in range(self.max_retries): start time.time() try: raw_output self.client.chat(messages) # 根据模式解析 if parse_mode json: parsed self.parser.parse_json(raw_output) elif parse_mode classify: candidates expected_keys or [] parsed self.parser.parse_classify(raw_output, candidates) else: parsed raw_output # 评测 eval_result self.evaluator.evaluate( parsed, expected_keysexpected_keys, expected_typeexpected_type, ) return { task_id: f{int(start)}-{attempt}, success: eval_result[passed], raw_output: raw_output, parsed_output: parsed, errors: eval_result[errors], elapsed: round(time.time() - start, 2), } except Exception as e: errors.append(str(e)) # 简单退避重试 time.sleep(1 * (attempt 1)) return { task_id: f{int(time.time())}-final, success: False, raw_output: , parsed_output: None, errors: errors, elapsed: 0, }4.8 跑一批真实任务最后写一个main.py构造一个真实的业务场景对一组客服工单做意图分类并输出 JSON 结果。# main.py import json import os from config import ModelConfig from harness import LocalHarness SYSTEM_PROMPT 你是客服系统的意图识别助手。用户会输入一段工单描述你需要 1. 判断用户的意图分类 2. 提取关键实体 请严格按照以下 JSON 格式输出不要输出额外内容 {category: 退款|换货|物流|咨询, entity: 关键实体, confidence: 0.0-1.0} .strip() def main(): os.makedirs(results, exist_okTrue) # 根据你的机器配置选择模型 config ModelConfig(model_nameqwen3:8b) harness LocalHarness( model_configconfig, system_promptSYSTEM_PROMPT, max_retries3, ) test_cases [ 我买了一个手机壳订单号是20240815想申请退款可以吗, 我的包裹已经三天没有物流信息了麻烦帮我查一下, 你们这个充电宝可以换一个颜色吗我要黑色的, 请问你们支持哪些支付方式, ] all_results [] for case in test_cases: result harness.run( user_inputcase, parse_modejson, expected_keys[category, entity, confidence], expected_typedict, ) # 打印关键信息 print(fInput: {case}) print(fSuccess: {result[success]}, Elapsed: {result[elapsed]}s) print(fParsed: {json.dumps(result[parsed_output], ensure_asciiFalse)}) print(- * 60) all_results.append( { input: case, **result, } ) # 保存结果到文件 with open(results/output.json, w, encodingutf-8) as f: json.dump(all_results, f, ensure_asciiFalse, indent2) print(f\n结果已保存到 results/output.json) print(f成功数: {sum(1 for r in all_results if r[success])} / {len(all_results)}) if __name__ __main__: main()运行命令python main.py在 Ollama 已经启动、模型已拉取的前提下预期输出类似Input: 我买了一个手机壳订单号是20240815想申请退款可以吗 Success: True, Elapsed: 12.35s Parsed: {category: 退款, entity: 手机壳, confidence: 0.91} ------------------------------------------------------------第一批跑通后你会发现 Harness 的价值它不关心业务层怎么调用只负责保证每次请求都走标准流程并且把失败原因记录得明明白白。5. 进阶把 Harness 扩展成 Agent 工具编排层5.1 函数调用场景在上面代码的基础上可以很自然地把 Harness 升级为 Agent 工具编排层。核心思路是在 system prompt 中声明可用工具让模型输出 JSON 形式的工具调用参数Harness 负责解析并执行。举个例子声明一个查询订单状态的工具TOOLS { query_order_status: { description: 查询订单状态, params: {order_id: string}, }, create_refund: { description: 创建退款申请, params: {order_id: string, amount: float}, }, }然后在 system prompt 中追加说明你可以调用以下工具 1. query_order_status(order_id: str) - 查询订单状态 2. create_refund(order_id: str, amount: float) - 创建退款申请 如果需要调用工具请输出 {tool: 工具名, args: {参数: 值}}之后在 Harness 里增加一个工具执行器def execute_tool(self, parsed_output: dict) - dict: 执行模型选择的工具 tool_name parsed_output.get(tool) args parsed_output.get(args, {}) if tool_name query_order_status: return self._query_order_status(args[order_id]) elif tool_name create_refund: return self._create_refund(args[order_id], args[amount]) else: raise ValueError(f未知工具: {tool_name})这里的重点是模型只负责“决策”工具执行仍然在你本地代码里这样安全性是可控的。社区中类似 deepseek harness 的工具编排方案也是这种模式只是工具集和应用场景更复杂。5.2 任务流编排场景更复杂一点可以把多个模型调用串联成一条任务流水线。比如第一轮意图分类第二轮根据意图生成回复文案第三轮对文案做合规校验。每一轮都是独立的 Harness 调用但上一轮的输出会作为下一轮的输入。为了让这个流程可复现每个环节的结果都要带上 task_id 和耗时方便回溯。def run_pipeline(self, user_input: str) - dict: # 步骤 1意图分类 step1 self.run( user_inputuser_input, parse_modejson, expected_keys[category], expected_typedict, ) # 步骤 2按分类生成回复 category step1[parsed_output][category] reply_prompt f用户咨询: {user_input}\n意图分类: {category}\n请生成一段客服回复。 step2 self.run(user_inputreply_prompt, parse_modetext) return { category: category, reply: step2[parsed_output], step1_elapsed: step1[elapsed], step2_elapsed: step2[elapsed], }这样设计的收益是任何一个环节出问题都可以单独定位和重跑不需要把整个链路重新跑一遍。6. 常见问题与排查思路端侧推理的坑不少这里把我实际踩过的问题整理成表。问题现象常见原因解决思路请求超时模型首次加载需要时间或 8B 模型在纯 CPU 环境推理慢增加timeout到 120s 以上提前调用一次模型完成预热显存/内存不足模型量化等级不够或并发请求太多换更小模型如 8B 换 Q4 量化降低并发数关闭其他占用显存的应用JSON 解析失败模型在 JSON 前后加了额外文字或者使用了中文标点使用 OutputParser 做容错解析在 system prompt 中强调“只输出 JSON”输出结果不稳定温度设置过高任务类场景温度设置为 0.1-0.3需要创意生成时才调高模型回答经常偏离指令system prompt 太长或表达不明确精简 prompt使用分点、编号、示例输出格式运行一段时间后变慢上下文越来越长或者模型缓存未清理定期清空历史消息重启 Ollama 服务排查时建议按这个顺序走先看 Ollama 日志再看请求超时时间最后看输出解析错误的原始文本样本。7. 最佳实践与工程建议7.1 模型与量化选择端侧部署要记住一个原则不是模型越大越好而是“够用 可跑”。在 8GB 显存设备上硬跑 27B 全量模型结果是直接 OOM连一次推理都跑不完。量化级别建议Q8质量最好但显存占用高Q5_K_M质量和显存的平衡点Q4_K_M端侧首选质量损失可接受。7.2 结构化输出的可靠性端侧模型输出 JSON 的不稳定性是最大的工程障碍。除了在代码里做容错解析还可以在 prompt 里给出一个正例你是信息抽取模型请输出 JSON {name: 张三, age: 30}模型看到具体格式示例后输出命中率会明显提升。7.3 结果缓存同一份输入可能被多次请求。如果你的数据里有大量重复文本建议增加一层缓存key 是输入文本的哈希值value 是上次的推理结果和评测状态。这一步优化对批量任务帮助极大尤其是端侧机器本身算力有限。7.4 安全与合规端侧模型的好处是数据不出设备但这也意味着你要对模型输出的内容负全部责任。建议不要直接用未经测试的 prompt 处理用户敏感信息线上使用前先在测试环境跑通一批回归用例涉及退款、支付、账号操作等场景模型只做辅助决策最终需要人工确认或规则兜底定期更新模型版本并对比评测指标避免“升级后反而变差”的情况。7.5 日志与可观测性每次推理都建议记录一条结构化日志格式可以这样{ task_id: 1723800000-0, model: qwen3:8b, elapsed: 12.35, success: true, retries: 0, input_length: 45, output_length: 120 }积累一段时间后你就能清楚地看到模型在哪些任务上耗时长、在哪些 prompt 上频繁解析失败。这些数据对后续优化价值很大。7.6 不要过度设计最后一条建议Harness 的复杂度要和你的业务规模匹配。如果你的业务只是每天跑几十条文本一个几百行的 Harness 就足够如果你的业务是面向大量用户的服务再考虑加入并发控制、分布式任务队列、监控告警。不要一上来就引入重量级框架否则维护成本会超过模型本身带来的收益。8. 总结与下一步学习方向本文从端侧模型的痛点出发解释了 Harness 在大模型应用中的定位并基于 Qwen3-8B/27B 这一档端侧友好模型完整实现了一套轻量级 Harness。你可以直接使用这份代码来完成文本分类、信息抽取、JSON 结构化输出等任务也可以把它扩展成 Agent 工具编排层。下一步可以尝试的方向在 Harness 中接入更多工具调用做成一个本地 Agent引入数据集批量跑评测并对比不同量化级别模型的差异把 Harness 封装成 FastAPI 服务给上层业务提供统一推理接口尝试 Qwen3-27B对比同一条 prompt 在两个模型上的效果差异。如果你也在做端侧模型的应用开发建议先拿本文的代码跑通第一条任务再逐步加入你自己的业务逻辑。本地推理的核心价值在于“自己掌握全部链路”从这个角度看Harness 不是可选项而是必选项。
