聊 Deepseek Harness 架构之前先说一个我自己的粗判断模型排行榜每周都在变但真正决定一个 AI 应用能不能上线、能不能稳定跑半年的往往是模型外面那一圈被称为 harness 的工程层。同一套模型权重套在简陋的脚本里就是能聊天套在打磨过的 harness 里就能自动查数据、改代码、跑测试、写报告还能在出错之后自己回滚。这中间的差距跟模型能力关系不大跟工程层的成熟度关系极大。我把这篇文章要覆盖的东西说清楚harness 到底是什么、它和 agent 的区别在哪、一个面向 DeepSeek 系模型的 harness 该怎么分层、每层的关键实现细节、从零跑起来的最小可用代码、以及我在真实项目里踩过的坑和排查手法。需要先声明一句harness 不是一个官方产品名而是一类工程层的通称——你可以把它理解成马具马模型负责跑马具负责把方向、力度、边界都约束住让跑偏这件事尽量不发生。后面所有内容都基于这个理解展开。适合读者是已经能调通模型接口、准备把 demo 变成产品的开发者以及对 agent 架构感兴趣的工程同学。零基础也能看涉及基础概念我会补一句解释。1. 先把概念掰开Harness 到底是什么很多人第一次听到 harness 会以为它是某个具体软件或者在找一个官网去下载。我先把最容易混淆的部分讲透后面谈架构才不会卡壳。1.1 从模型能说话到系统能干活之间那条沟模型本身只做一件事给定一段 token 序列预测接下来的 token。它没有记忆、没有手、没有眼睛也叫不动任何外部系统。你在对话框里让它查一下昨天的订单量它只能编一个看似合理的数字给你因为它的输出是概率分布的采样结果不是数据库查询结果。harness 存在的意义就是给模型补上这几样东西。工具调用让它能伸手去查、去写上下文管理让它能记住超过一次对话轮次的信息执行循环让它能多步推理、失败重试沙箱保证它伸出去的手不会砸坏别的东西可观测性让人类工程师能看懂它每一步到底干了什么。少了任何一环系统就会以某种方式崩掉——要么胡编数据要么把上下文塞爆要么在死循环里烧掉你的账单。我习惯用一个类比解释给非技术的同学听模型是一个能力很强但刚入职的实习生知识面广、反应快但他不知道公司内部的系统怎么登、不知道哪些操作需要审批、不知道文件放哪个目录。harness 就是工位、权限、操作手册和主管的审批流程加起来的那一套环境。实习生还是那个实习生有没有这套环境产出完全是两个量级。这也是为什么我从来不看某模型支持多少种工具调用这种宣传语而是看它的 harness 能不能把工具调用的失败路径处理干净。真实环境里工具调用成功的比例远低于 demo 里的展示异常分支的工程质量才是分水岭。1.2 Harness 和 Agent 的边界到底在哪这两个词经常被混用我在项目里会给它们划一条明确的线harness 是跑道和夹具agent 是跑在上面的那个决策主体。agent 关心的是策略层面的事——我现在该调哪个工具、要不要先澄清需求、任务完成了没有、下一步计划是什么。它体现为一段提示词、一套循环规则、一套停止条件。harness 关心的是机制层面的事——请求怎么发出去、流式响应怎么拼、工具怎么注册和鉴权、上下文超预算了砍哪一段、进程崩了怎么恢复。agent 是想做什么harness 是怎么做到。这条线画清楚之后很多工程决策就顺了。你会发现 agent 的策略可以频繁迭代改提示词、改停止条件、换模型而 harness 的机制应该尽量稳定因为它承载着所有 agent 的公共依赖。我在实际项目里的做法是把 agent 策略做成配置甚至数据把 harness 做成有测试覆盖的代码。前者一天可以改十次后者改一次要跑一遍回归。打个更具体的比方。你让 agent 去分析这个仓库的测试覆盖率并给出改进建议agent 会规划出读目录结构、找测试文件、跑覆盖率命令、读结果、写总结。这套规划可以写在提示词里。而跑覆盖率命令这个动作在 harness 里需要落实为命令白名单校验、超时控制、输出截断、退出码解析、失败时的错误信息归一化。agent 说我要跑测试harness 说好我给你 120 秒、只允许在这个目录下、输出超过 8KB 就截断并提示我。职责完全不同。1.3 为什么 DeepSeek 系模型的 Harness 需要单独设计理论上 harness 可以做成模型无关的接口层统一到 OpenAI 兼容格式就能应付大多数情况。但实操下来针对特定模型族做适配仍然是值得的原因有三个。第一是上下文窗口和成本结构的差异。不同模型的窗口大小、计费方式、缓存折扣策略都不一样这直接决定了你的预算分配算法。如果你按 128K 窗口设计换到一个窗口更小的模型上就会频繁触发截断反之则是浪费。预算分配不是常量是要算出来的。第二是工具调用的输出格式细节。即使都是 OpenAI 兼容各家在并行工具调用、流式增量拼接、参数为空的处理上仍有差别。有的模型会在参数里塞进多余换行有的会把多个工具调用拆成十几个 delta 分片有的在参数非法时直接返回空字符串而不是报错。这些细节不处理你的工具层会随机报 JSON 解析错误而且很难复现。第三是推理模型的特殊行为。带思维链的模型会输出一段推理内容这段内容要不要存进历史、要不要展示给用户、要不要计入 token 预算都需要在 harness 里单独处理。我见过太多项目把推理内容原样塞回上下文结果下一轮对话的 token 消耗直接翻倍。我的建议是harness 的外层接口保持模型无关内层留一个薄薄的适配层把上面三类差异全部收敛在这一层里。适配层大概两三百行代码但能省掉后面无数的为什么换个模型就炸了。2. 整体架构拆解一套可落地的分层设计概念讲完开始上结构。下面这套分层是我在几个项目里反复调整后稳定下来的版本不一定是最优解但每一层的划分都有明确理由。2.1 六层结构各自的职责我把 harness 拆成六层从上到下依次是接入层、会话层、编排层、模型适配层、工具执行层、基础设施层。每层的职责必须单一跨层调用要严格禁止否则很快会变成一团互相依赖的泥巴。层级核心职责关键产出常见踩坑点接入层协议转换、鉴权、限流统一内部请求对象把业务逻辑写进路由处理函数会话层状态存储、历史维护、并发隔离会话快照用全局变量存会话状态编排层循环控制、停止条件、降级执行轨迹停止条件只靠模型自己说完成模型适配层请求构造、流式解析、错误归一结构化响应把厂商错误码直接抛给上层工具执行层注册、校验、沙箱、幂等工具结果对象不做参数校验直接执行基础设施层日志、指标、追踪、存储可观测数据只在出错时打日志这张表我建议贴在项目 README 里因为它是团队协作的边界约定。新人最容易犯的错是把工具调用的业务逻辑写在接入层的路由函数里一开始省事等到要加第二个入口比如接一个内部消息平台时就得把所有逻辑复制一遍。六层里最容易被低估的是会话层。很多教程把会话状态存在一个字典里键是用户 ID。单机单进程跑起来没问题一旦开多进程或者做水平扩展用户就会发现自己上一句说的话被忘掉了因为下一次请求打到了另一个进程。这个 bug 的排查成本极高因为它是概率性的。我的做法是会话状态一律外置存储进程内只保留请求生命周期内的临时数据。2.2 一次请求在 Harness 里的完整旅程光看分层还是抽象走一遍数据流会清楚很多。假设用户输入帮我把订单表里上周的异常订单导出来下面是它在系统里经过的路径。接入层收到请求做鉴权、限流、参数校验构造出一个内部统一的Request对象里面包含会话 ID、用户输入、模型选择、工具权限集。这一层只做转换不做决策。会话层根据会话 ID 拉取历史快照。快照不是简单的一串消息而是包含消息列表、已注入的检索片段、用户画像标签、当前 token 预算使用情况的结构化对象。拉取的同时要加锁或者用乐观并发控制防止同一会话的两个请求互相覆盖。编排层进入主循环。第一轮把系统提示词、历史消息、可用工具的 schema、本轮用户输入拼成完整请求交给模型适配层。适配层负责把它转成具体厂商的请求格式发出去解析流式响应返回一个归一化的结果对象字段包括文本内容、推理内容如果有、工具调用列表、token 用量。回到编排层如果结果里有工具调用就交给工具执行层。工具执行层先按名称查注册表确认这个工具在当前会话的权限集里然后用 JSON Schema 校验参数校验通过才真正执行。执行结果统一包装成ToolResult带上成功标记、内容、耗时、是否可重试。工具结果被追加进消息列表进入下一轮循环。同时编排层要更新 token 预算如果超出阈值就触发上下文压缩。循环一直持续到满足停止条件模型返回了不含工具调用的最终答案、达到最大步数、超时、或者预算耗尽。最后编排层输出最终答案给接入层同时把完整执行轨迹写入可观测系统。会话层把更新后的历史快照落盘。这条链路上有两个设计点值得强调。一是所有跨层的数据结构都应该是显式的对象而不是裸字典。用字典的代价是加字段时没人知道还有谁在消费这个字段改一个键名可能炸掉三个模块。二是循环的每一次迭代都要生成一个可追溯的步骤记录包括这一轮消耗了多少 token、调了哪个工具、花了多少时间。没有这层记录线上问题基本只能靠猜。2.3 部署形态选型单进程、单机多进程还是微服务关于要不要上微服务和分布式架构我的态度非常明确harness 在绝大多数场景下不需要微服务。这个话题被过度讨论了很多团队在只有几十个日活用户的时候就拆了七八个服务结果调试一个工具调用要翻五六份日志。我给出一个信号清单出现其中任何一条才考虑拆分。第一条是工具执行需要独立伸缩比如代码执行、视频转码这类重资源工具和主流程的资源曲线完全不同。第二条是不同工具需要不同的安全域比如访问生产数据库的工具必须跑在隔离环境里。第三条是团队规模超过十人且模块边界已经稳定。三条都不满足的时候单机多进程加一个外置状态存储能扛住相当可观的量级。单机形态下我推荐的进程模型是一个主进程负责接入和编排工具执行走子进程池重任务丢给独立的工作进程通过队列通信。这样崩溃隔离和超时控制都好做运维复杂度也低。真正需要横向扩展的时候你会发现因为状态已经外置了加一层负载均衡就能平滑扩上去不用重写。3. 核心模块实现细节与实操要点架构图谁都会画难点在每一层的细节。这一节挑四个最容易出问题的模块展开每个都给出具体做法和参数。3.1 模型适配层请求构造、流式解析与错误归一适配层的核心价值是把厂商差异挡在外面。上层永远只看到一套字段不管底层是哪家模型。请求构造上我建议把模型参数集中在一个配置对象里而不是散落在调用点。温度、最大输出长度、工具选择策略、是否开启流式这些都应该是配置项。最容易被忽略的是最大输出长度——很多模型默认值是 4096如果你不显式设置长回答会被硬截断而且截断时往往不报错只是内容突然中断排查起来很痛苦。流式解析是适配层最容易写出 bug 的地方。工具调用在流式响应里是按分片增量返回的你必须按索引累积拼接不能当成完整对象处理。下面是一段我实际在用的解析骨架import json import httpx def stream_chat(messages, toolsNone, modeldeepseek-chat, base_url, api_key): payload {model: model, messages: messages, stream: True} if tools: payload[tools] tools headers {Authorization: fBearer {api_key}, Content-Type: application/json} buffer, tool_calls [], {} with httpx.stream(POST, f{base_url}/chat/completions, jsonpayload, headersheaders, timeout180) as resp: resp.raise_for_status() for line in resp.iter_lines(): if not line or not line.startswith(data:): continue data line[5:].strip() if data [DONE]: break chunk json.loads(data) if not chunk.get(choices): continue delta chunk[choices][0].get(delta, {}) if delta.get(content): yield {type: text, value: delta[content]} for tc in delta.get(tool_calls) or []: idx tc.get(index, 0) slot tool_calls.setdefault(idx, {id: , name: , args: }) if tc.get(id): slot[id] tc[id] fn tc.get(function) or {} if fn.get(name): slot[name] fn[name] if fn.get(arguments): slot[args] fn[arguments] if tool_calls: yield {type: tool_calls, value: [tool_calls[k] for k in sorted(tool_calls)]}这段代码里有三个细节值得单独说。第一index字段必须用上并行工具调用时多个工具的分片会交叉出现不按索引分桶就会拼成乱码。第二name用而不是因为少数情况下函数名本身也会被分片。第三参数拼接完必须做一次 JSON 解析校验解析失败说明模型吐出了非法 JSON这时候不要直接抛异常而应该把原始字符串截断后作为错误信息回传给模型让它自己修正。我实测下来这种让模型看到自己的错误的做法纠正成功率比直接重试高不少。错误归一这块我建议至少区分四类可重试错误网络超时、限流、不可重试错误参数非法、鉴权失败、配额错误余额不足、超额、未知错误。前两类处理方式完全不同前者按指数退避重试后者立刻失败并给出明确提示。把所有错误当成一类处理是新手最常见的反模式——要么对鉴权失败也重试三次浪费时间要么对网络抖动直接报错给用户。3.2 工具注册与调用协议让模型真的会用工具工具层的设计目标只有一个模型看到的工具描述必须和实际执行的函数严格一致。任何偏差都会导致模型调用出错。我用装饰器做注册好处是定义和使用在同一个地方不容易出现文档和执行不一致的情况import json import inspect REGISTRY {} def tool(nameNone, descNone, scopesNone, timeout30): def deco(fn): key name or fn.__name__ sig inspect.signature(fn) props, required {}, [] type_map {str: string, int: integer, float: number, bool: boolean} for pname, param in sig.parameters.items(): py_type param.annotation if param.annotation is not inspect._empty else str props[pname] {type: type_map.get(py_type, string)} if param.default is inspect._empty: required.append(pname) REGISTRY[key] { fn: fn, timeout: timeout, scopes: set(scopes or []), schema: { type: function, function: { name: key, description: desc or (fn.__doc__ or ).strip()[:512], parameters: {type: object, properties: props, required: required}, }, }, } return fn return deco tool(desc按订单号查询订单详情返回状态、金额、创建时间, scopes[order:read], timeout10) def query_order(order_id: str) - dict: ...这段代码有两点经验。参数类型从函数签名自动推导避免了手写 JSON Schema 时忘记同步改文档的经典问题我见过太多次文档说传字符串、代码实现要整数的事故。scopes权限集是我强烈建议加上的它让这个会话能不能调这个工具变成一个显式声明而不是散落在业务代码里的 if 判断。工具描述description的写法直接决定模型的调用准确率。我的经验是一句话说清做什么紧接着一句说清什么情况下用。比如查询订单详情不如按订单号查询订单详情当用户提供了订单号且需要订单状态或金额时使用。后半句能让模型在多个相似工具之间做出正确选择。另外描述里不要写实现细节模型不需要知道你是走 SQL 还是走 HTTP那些信息只会占用 token 预算且干扰判断。参数的边界也要写进描述。比如一个工具接收limit参数描述里应该写最多返回条数取值 1 到 100默认 20。我踩过一次坑模型传了limit10000后端没做上限校验直接把整张表拉进内存服务当场 OOM。永远不要相信模型传的参数在合理范围内校验必须在执行前做。3.3 上下文预算与压缩别等塞爆了才想这件事上下文管理是 harness 里最像资源调度的部分也是纯工程活儿跟模型能力无关。我的做法是把窗口看成一个预算池每个消费者按优先级分配额度。先算一笔具体的账。假设目标模型的上下文窗口是 65536 token我要预留这几个部分输出预留 4096应对长回答系统提示词实测约 700十二个工具的 schema 序列化后约 1900本轮检索注入的知识片段 2000安全余量 1000用来吸收 token 估算误差。把这五项减掉留给历史对话的额度是65536 - 4096 - 700 - 1900 - 2000 - 1000 55840 token这个数字就是历史消息的硬上限。接下来要解决的是怎么估算当前用了多少。精确做法是调用对应模型的 tokenizer但很多接口不开放工程上更常见的是估算再乘以安全系数。我的经验值是中文按字符数乘 0.6 估算 token英文和代码按字符数除以 3.5然后用实测数据校准这个系数。不同模型的系数会差 10% 到 20%所以安全余量一定要留。超出预算时的压缩策略我是分三档递进的。第一档是截断把最早的几轮对话整体丢掉代价是丢失早期上下文好处是无损、无额外调用。第二档是摘要把要丢弃的段落交给模型做一次摘要用摘要替换原文压缩比通常在 5 到 10 倍。第三档是事实卡从历史里抽取关键事实存成结构化字段比如用户所在城市杭州偏好简洁回复已确认订单号A123。事实卡不占多少 token但能保留最关键的个性化信息。值得单独强调的是工具返回结果必须做截断。一个查询接口返回 5 万字符的 JSON直接塞进上下文能吃掉几万 token而且里面大部分字段模型根本用不上。我的做法是给每个工具配一个输出处理器只保留必要字段并且硬性限制单次返回不超过 4000 字符超出就截断并在结尾标注结果已截断如需更多请缩小查询范围。这个设计让模型自己学会问得更精确而不是无脑拉全量数据。3.4 沙箱与权限让模型伸手但不让它拆家只要 harness 具备执行能力——跑命令、写文件、调数据库——沙箱就是必选项不是可选项。我见过太多演示项目直接在宿主进程里exec模型生成的代码演示效果好上线就是灾难。沙箱要控制的东西有明确的优先级排序。第一是资源用量CPU 时间和内存必须设上限否则一个死循环就能把机器拖垮。第二是时间硬性超时超时直接杀进程。第三是文件系统只挂载必要的临时目录其余只读或不可见。第四是网络默认关闭需要联网的工具单独声明。第五是权限用独立的低权限账号运行绝不使用管理员权限。下面是一个 Unix 环境下跑 Python 片段的沙箱骨架用的是标准库的资源限制import os import resource import subprocess def run_python_snippet(code: str, timeout: int 10) - dict: def _limits(): cpu max(1, timeout) resource.setrlimit(resource.RLIMIT_CPU, (cpu, cpu)) resource.setrlimit(resource.RLIMIT_AS, (512 * 1024 * 1024, 512 * 1024 * 1024)) resource.setrlimit(resource.RLIMIT_FSIZE, (8 * 1024 * 1024, 8 * 1024 * 1024)) resource.setrlimit(resource.RLIMIT_NPROC, (32, 32)) try: proc subprocess.run( [python, -I, -S, -c, code], capture_outputTrue, timeouttimeout, preexec_fn_limits, cwd/tmp/sandbox, env{PATH: /usr/bin:/bin, HOME: /tmp/sandbox}, ) out proc.stdout.decode(utf-8, replace)[:8000] err proc.stderr.decode(utf-8, replace)[:2000] return {ok: proc.returncode 0, stdout: out, stderr: err, code: proc.returncode} except subprocess.TimeoutExpired: return {ok: False, stdout: , stderr: f执行超时{timeout}s, code: -1}几个关键点说明一下。-I参数让 Python 忽略环境变量和用户目录的配置避免被预先植入的配置影响-S跳过 site 初始化减少启动开销和被注入的可能。RLIMIT_NPROC防止代码 fork 出大量子进程。cwd指向独立的临时目录配合容器时这个目录应该是一个临时卷。preexec_fn在 Unix 上可用Windows 下需要用别的机制跨平台场景建议直接用容器方案。注意这只是一种轻量隔离能挡住误操作挡不住刻意构造的攻击。面向不可信输入的场景必须上进程级或容器级隔离并且运行在独立的宿主环境里。权限设计上我的做法是工具级权限集 会话级授权的组合。每个工具声明自己需要的权限标识每个会话在创建时确定拥有哪些标识调用时做交集判断。这样既能做到细粒度控制又不需要在每次调用时做复杂的鉴权计算。把权限做在工具层还有一个好处当你要接入新的用户群体时只需要配置权限集不用改任何业务代码。4. 实操从零把 Harness 跑起来前面讲了很多设计层面的东西这一节直接给能跑的代码。为了让内容可复现我会按照最小版本 → 加工具 → 加流式 → 加预算的顺序逐步叠加每一步都是完整可运行的。4.1 环境准备与最小可运行版本依赖很简单一个 HTTP 客户端加一个异步运行时就够了。我用httpx是因为它同时支持同步和流式pydantic用来做配置和参数校验。如果你要对接本地推理服务绝大多数推理框架都提供 OpenAI 兼容端点把base_url指过去就行harness 层不用改。python -m venv .venv source .venv/bin/activate pip install httpx pydantic配置我统一从环境变量读绝不写死在代码里import os from pydantic import BaseModel class Config(BaseModel): base_url: str os.getenv(LLM_BASE_URL, ) api_key: str os.getenv(LLM_API_KEY, ) model: str os.getenv(LLM_MODEL, deepseek-chat) max_steps: int 8 max_output_tokens: int 4096 step_timeout: int 60 total_timeout: int 300 CFG Config()max_steps这个参数看着不起眼但它是我认为最重要的一个安全阀。没有它一个逻辑设计不当的 agent 会无限循环每一步都消耗 token。我设的经验值是简单任务 3 到 5 步中等复杂度 8 到 12 步超过 15 步基本说明提示词或者工具设计有问题应该去修设计而不是调大上限。最小版本的主循环长这样先不考虑工具import httpx def chat_once(messages): payload { model: CFG.model, messages: messages, max_tokens: CFG.max_output_tokens, } resp httpx.post( f{CFG.base_url}/chat/completions, jsonpayload, headers{Authorization: fBearer {CFG.api_key}}, timeoutCFG.step_timeout, ) resp.raise_for_status() data resp.json() return data[choices][0][message][content], data.get(usage, {})跑到这一步你应该先验证一件事接口通不通、模型名对不对、返回结构符不符合预期。我见过不少人直接跳到多轮工具调用结果调试半天发现是基础请求参数写错了。先让最简单的版本跑通再往上加这是省时间的做法。4.2 加入工具调用循环有了最小版本把工具循环加上。核心逻辑就三件事模型说要调工具我们执行把结果喂回去。import json def run_agent(user_input: str, session_messagesNone, max_stepsNone): max_steps max_steps or CFG.max_steps messages list(session_messages or []) messages.append({role: user, content: user_input}) schemas [v[schema] for v in REGISTRY.values()] trace [] for step in range(max_steps): payload {model: CFG.model, messages: messages, max_tokens: CFG.max_output_tokens} if schemas: payload[tools] schemas resp httpx.post( f{CFG.base_url}/chat/completions, jsonpayload, headers{Authorization: fBearer {CFG.api_key}}, timeoutCFG.step_timeout, ) resp.raise_for_status() msg resp.json()[choices][0][message] messages.append(msg) calls msg.get(tool_calls) or [] if not calls: trace.append({step: step, type: final, content: msg.get(content, )}) return msg.get(content, ), messages, trace for call in calls: cid call.get(id) name call[function][name] raw_args call[function].get(arguments) or {} try: args json.loads(raw_args) assert isinstance(args, dict) except Exception: result {ok: False, error: 参数不是合法 JSON 对象请检查后重新调用} else: result execute_tool(name, args) messages.append({ role: tool, tool_call_id: cid, content: json.dumps(result, ensure_asciiFalse)[:4000], }) trace.append({step: step, type: tool, name: name, args: args, result: result}) return 已达到最大执行步数任务未完成。, messages, trace配套的执行函数是权限和沙箱的汇合点def execute_tool(name: str, args: dict): entry REGISTRY.get(name) if not entry: return {ok: False, error: f工具 {name} 不存在} try: import inspect sig inspect.signature(entry[fn]) sig.bind(**args) except TypeError as e: return {ok: False, error: f参数不匹配{e}} try: return {ok: True, data: entry[fn](**args)} except Exception as e: return {ok: False, error: f执行失败{type(e).__name__}: {e}}这里有个我反复强调的原则工具异常必须被捕获并转成结构化结果返回给模型而不是往上抛。原因很简单模型看到参数不匹配缺少 order_id这类信息之后下一轮通常能自己改对。如果你直接抛异常中断整个会话用户看到的就是一句无用的报错。把错误路径也当成正常流程来设计是 agent 系统能不能用得起来的关键分界。4.3 流式输出与可观测性用户等 30 秒什么都看不到体验会很差。加流式之后文本可以边生成边推给前端。实现上就是把 3.1 节的解析器接进循环文本分片直接转发工具调用分片累积到末尾再处理。可观测性我建议在开发阶段就做不要留到上线。每次执行输出一份 trace包含每一步的耗时、该步的 token 用量、调用了哪个工具、参数和结果摘要。这份 trace 是排查问题的唯一依据。我自己的习惯是把它打到标准输出格式化成人类可读的文本同时异步写一份 JSON 到文件。开发的时候看文本线上出问题的时候查 JSON。有一个指标我强烈建议单独统计工具调用的失败率按工具名分组。这个数字能直接告诉你哪个工具的描述写得有问题。失败率高的工具八成是描述太模糊导致模型传错参数而不是模型能力不够。我做过统计一个描述写得含糊的工具失败率能到 40%把描述改成什么情况下用 参数怎么填 边界值之后能降到 5% 以内。4.4 上下文预算的计算过程把预算逻辑落成代码思路就是前面算的那笔账def estimate_tokens(text: str) - int: if not text: return 0 cjk sum(1 for ch in text if \u4e00 ch \u9fff) other len(text) - cjk return int(cjk * 0.6 other / 3.5) 4 def trim_history(messages, window65536, output_reserve4096, fixed_cost5600, safety1000): budget window - output_reserve - fixed_cost - safety total sum(estimate_tokens(json.dumps(m, ensure_asciiFalse)) for m in messages) if total budget: return messages, {trimmed: False, used: total, budget: budget} head [m for m in messages if m.get(role) system] body [m for m in messages if m.get(role) ! system] used sum(estimate_tokens(json.dumps(m, ensure_asciiFalse)) for m in head) kept [] for m in reversed(body): cost estimate_tokens(json.dumps(m, ensure_asciiFalse)) if used cost budget: break kept.append(m) used cost kept.reverse() return head kept, {trimmed: True, used: used, budget: budget, dropped: len(body) - len(kept)}这段逻辑里有几个判断要说清楚。系统提示词和工具结果永远不会被裁掉它们在head或者被单独保留因为这两类是任务前提丢了之后模型会彻底跑偏。裁剪从最旧的消息开始保留最近的对话因为最近的内容与当前任务相关性最高。裁剪后必须检查第一条保留的消息是不是工具结果——如果是说明它对应的助手消息被裁掉了这会导致部分接口报错需要在代码里补齐一条占位消息。这个细节我踩过一次表现为偶发的请求 400排查了很久才定位到。裁剪发生的时候我建议在系统提示词末尾追加一条提示更早的对话内容已被裁剪如需必要信息请向用户询问。这句话能显著降低模型基于残缺上下文胡乱推断的概率。5. 常见问题与排查技巧实录下面是这几年我实际遇到过、并且不止遇到一次的问题。每一条都附上排查路径照着走基本能定位。5.1 高频问题速查表现象最可能原因排查动作处理方式模型完全不调工具工具描述含糊或缺何时使用打印实际发送的 tools 字段重写描述补充使用场景参数 JSON 解析失败流式分片拼接错误打印拼接后的原始字符串按 index 分桶累积非法时回传错误同一工具反复调用提示词没定义停止条件看 trace 里的重复调用加相同参数连续两次则终止规则上下文超限报错工具返回未截断统计单次工具返回字符数限制单次返回 4000 字符多轮后回答变差历史裁剪把关键信息丢了检查裁剪后保留的首条消息引入事实卡保留关键字段并发请求串味会话状态存在进程内检查状态存储位置状态外置进程只留临时数据偶发请求 400裁剪后孤儿工具消息检查消息角色顺序补齐占位消息或整组裁剪超时重试导致重复下单工具没有幂等控制查看是否有重复副作用用调用 ID 做幂等键这张表里我最想强调的是最后一行。所有有副作用的工具都必须幂等这是硬要求。做法是用工具调用的 ID 作为幂等键服务端记录已处理的键重复请求直接返回上次结果。我见过真实事故网络抖动导致工具调用超时harness 按策略重试了一次结果用户被扣了两次款。技术上说得通业务上不可接受。5.2 三个最容易被忽略的坑第一个坑是停止条件完全交给模型。很多实现里循环的唯一退出条件是模型返回不含工具调用的消息。这在模型稳定的时候能用但一旦模型陷入某种模式比如反复确认我是否需要更多信息循环就会一直跑下去。我的做法是三重保护最大步数、总 token 预算、墙上时钟超时任意一个触发就强制退出并返回一个明确的任务中断结果而不是静默失败。用户看到任务在 8 步内未完成请把问题拆得更具体一些比看到转圈半天然后什么都没有要好得多。第二个坑是工具结果不做语义化处理。假设工具返回一个包含 50 个字段的 JSON模型很难从中抓住重点。我习惯在工具层做一次结果摘要把最关键的字段放在最前面其余字段用一句完整字段列表a, b, c...带过。这样模型拿到的信息密度高很多决策质量也明显提升。这个改动的性价比极高几乎是零成本换来可感知的效果提升。第三个坑是不做回归测试。harness 里的任何改动——改提示词、加工具、调预算——都可能让某些场景变差。我维护了一份二十来条的小测试集覆盖典型任务和各条失败路径每次改完跑一遍看通过率有没有下降。这个习惯是从吃了亏之后开始养的有一次改了一句系统提示词主流程变好了但一个边缘场景从能用变成完全不可用上线三天后才发现。5.3 成本与延迟的调优杠杆调优这件事我按投入产出比排了序。第一优先级是减少不必要的步数因为每一步都是一次完整的模型调用步数减半基本等于成本减半。减少步数的办法是让工具更厚——把多个小工具合并成一个能一次返回完整信息的工具减少往返。第二优先级是压缩工具结果前面讲过效果直接且明显。第三优先级是缓存系统提示词和工具 schema 在每轮里都是重复的很多接口对前缀相同的内容有缓存优惠把不变的部分放前面、变化的部分放后面能吃到这部分折扣。第四优先级是分级模型简单任务用好用便宜的模型复杂任务才上能力更强的模型。harness 层做成可配置的切换成本很低。延迟方面最大的杠杆是并行工具调用。如果一轮里有三个互不依赖的工具调用串行执行要花三倍时间。做法是把同一轮返回的多个工具调用并发执行用一个带超时的并发池控制。要注意的是如果工具之间存在依赖关系比如先查 ID 再查详情就不能并行这需要工具设计阶段就把依赖关系理清楚不要指望运行时判断。6. 插件化与多模型路由把 Harness 做成可复用的底座做到这一步你的 harness 已经能跑通完整链路了。如果要长期维护还有两件事值得投入。6.1 插件化让工具和模型都能热插拔我采用的方案是把工具定义、模型适配器、输出处理器都做成可注册的插件。启动时扫描指定目录加载所有符合约定的模块注册到统一的注册表里。这样新增一个工具只需要写一个文件不需要改动 harness 主体代码。插件化的关键约束是插件不能直接访问 harness 的内部状态。插件只能通过传入的上下文对象拿到它需要的东西比如会话 ID、权限集、日志器。这条约束听起来很烦但它保证了插件的失败不会波及主体也让插件可以被独立测试。我见过没做这个约束的项目一个插件的内存泄漏把整个服务拖垮。模型路由也是同理。给每个模型适配器声明它的能力和成本档位路由层根据任务复杂度选择。复杂度的判断可以很简单用户输入长度、是否需要工具、历史轮次。我用的是一个三档的粗暴规则实测效果已经够了不需要上更复杂的策略。6.2 评测集与灰度让改动可控harness 的迭代有个特点改动的影响面很大但反馈很慢。你改了系统提示词可能要几天后才发现某个场景变差了。解决这个问题的唯一办法是建立离线评测集把反馈周期从天缩短到分钟。我的评测集分三层。第一层是单元级针对单个工具的参数校验、异常处理、输出截断。第二层是流程级给定输入检查是否走了预期的工具调用路径用的是轨迹匹配而不是答案匹配因为路径对了答案差不到哪去。第三层是端到端人工标注的输入输出对用模型或者规则打分。上线策略上我用的是影子流量加灰度。新版本先跑影子不影响真实用户对比两条链路的轨迹差异差异在可接受范围内再切一小部分流量。这个流程听起来重但实际做下来只有几十行代码比起线上翻车的代价非常划算。我个人在实际操作中的体会是harness 这个层级的工程80% 的工作量花在异常路径和边界条件上20% 花在正常路径。新手容易反过来分配精力把 happy path 打磨得很漂亮结果一遇到工具报错、上下文超限、并发冲突就全线崩溃。真正让系统稳定运行的恰恰是那些看起来用不上的防御代码。我现在的习惯是每写一个工具第一件事不是写实现而是先想清楚它可能怎么失败把失败路径的处理写完再去写正常逻辑——顺序反过来之后线上事故率下降得非常明显。 DeepSeek Harness 架构：大模型 Agent 工程化落地实践
聊 Deepseek Harness 架构之前先说一个我自己的粗判断模型排行榜每周都在变但真正决定一个 AI 应用能不能上线、能不能稳定跑半年的往往是模型外面那一圈被称为 harness 的工程层。同一套模型权重套在简陋的脚本里就是能聊天套在打磨过的 harness 里就能自动查数据、改代码、跑测试、写报告还能在出错之后自己回滚。这中间的差距跟模型能力关系不大跟工程层的成熟度关系极大。我把这篇文章要覆盖的东西说清楚harness 到底是什么、它和 agent 的区别在哪、一个面向 DeepSeek 系模型的 harness 该怎么分层、每层的关键实现细节、从零跑起来的最小可用代码、以及我在真实项目里踩过的坑和排查手法。需要先声明一句harness 不是一个官方产品名而是一类工程层的通称——你可以把它理解成马具马模型负责跑马具负责把方向、力度、边界都约束住让跑偏这件事尽量不发生。后面所有内容都基于这个理解展开。适合读者是已经能调通模型接口、准备把 demo 变成产品的开发者以及对 agent 架构感兴趣的工程同学。零基础也能看涉及基础概念我会补一句解释。1. 先把概念掰开Harness 到底是什么很多人第一次听到 harness 会以为它是某个具体软件或者在找一个官网去下载。我先把最容易混淆的部分讲透后面谈架构才不会卡壳。1.1 从模型能说话到系统能干活之间那条沟模型本身只做一件事给定一段 token 序列预测接下来的 token。它没有记忆、没有手、没有眼睛也叫不动任何外部系统。你在对话框里让它查一下昨天的订单量它只能编一个看似合理的数字给你因为它的输出是概率分布的采样结果不是数据库查询结果。harness 存在的意义就是给模型补上这几样东西。工具调用让它能伸手去查、去写上下文管理让它能记住超过一次对话轮次的信息执行循环让它能多步推理、失败重试沙箱保证它伸出去的手不会砸坏别的东西可观测性让人类工程师能看懂它每一步到底干了什么。少了任何一环系统就会以某种方式崩掉——要么胡编数据要么把上下文塞爆要么在死循环里烧掉你的账单。我习惯用一个类比解释给非技术的同学听模型是一个能力很强但刚入职的实习生知识面广、反应快但他不知道公司内部的系统怎么登、不知道哪些操作需要审批、不知道文件放哪个目录。harness 就是工位、权限、操作手册和主管的审批流程加起来的那一套环境。实习生还是那个实习生有没有这套环境产出完全是两个量级。这也是为什么我从来不看某模型支持多少种工具调用这种宣传语而是看它的 harness 能不能把工具调用的失败路径处理干净。真实环境里工具调用成功的比例远低于 demo 里的展示异常分支的工程质量才是分水岭。1.2 Harness 和 Agent 的边界到底在哪这两个词经常被混用我在项目里会给它们划一条明确的线harness 是跑道和夹具agent 是跑在上面的那个决策主体。agent 关心的是策略层面的事——我现在该调哪个工具、要不要先澄清需求、任务完成了没有、下一步计划是什么。它体现为一段提示词、一套循环规则、一套停止条件。harness 关心的是机制层面的事——请求怎么发出去、流式响应怎么拼、工具怎么注册和鉴权、上下文超预算了砍哪一段、进程崩了怎么恢复。agent 是想做什么harness 是怎么做到。这条线画清楚之后很多工程决策就顺了。你会发现 agent 的策略可以频繁迭代改提示词、改停止条件、换模型而 harness 的机制应该尽量稳定因为它承载着所有 agent 的公共依赖。我在实际项目里的做法是把 agent 策略做成配置甚至数据把 harness 做成有测试覆盖的代码。前者一天可以改十次后者改一次要跑一遍回归。打个更具体的比方。你让 agent 去分析这个仓库的测试覆盖率并给出改进建议agent 会规划出读目录结构、找测试文件、跑覆盖率命令、读结果、写总结。这套规划可以写在提示词里。而跑覆盖率命令这个动作在 harness 里需要落实为命令白名单校验、超时控制、输出截断、退出码解析、失败时的错误信息归一化。agent 说我要跑测试harness 说好我给你 120 秒、只允许在这个目录下、输出超过 8KB 就截断并提示我。职责完全不同。1.3 为什么 DeepSeek 系模型的 Harness 需要单独设计理论上 harness 可以做成模型无关的接口层统一到 OpenAI 兼容格式就能应付大多数情况。但实操下来针对特定模型族做适配仍然是值得的原因有三个。第一是上下文窗口和成本结构的差异。不同模型的窗口大小、计费方式、缓存折扣策略都不一样这直接决定了你的预算分配算法。如果你按 128K 窗口设计换到一个窗口更小的模型上就会频繁触发截断反之则是浪费。预算分配不是常量是要算出来的。第二是工具调用的输出格式细节。即使都是 OpenAI 兼容各家在并行工具调用、流式增量拼接、参数为空的处理上仍有差别。有的模型会在参数里塞进多余换行有的会把多个工具调用拆成十几个 delta 分片有的在参数非法时直接返回空字符串而不是报错。这些细节不处理你的工具层会随机报 JSON 解析错误而且很难复现。第三是推理模型的特殊行为。带思维链的模型会输出一段推理内容这段内容要不要存进历史、要不要展示给用户、要不要计入 token 预算都需要在 harness 里单独处理。我见过太多项目把推理内容原样塞回上下文结果下一轮对话的 token 消耗直接翻倍。我的建议是harness 的外层接口保持模型无关内层留一个薄薄的适配层把上面三类差异全部收敛在这一层里。适配层大概两三百行代码但能省掉后面无数的为什么换个模型就炸了。2. 整体架构拆解一套可落地的分层设计概念讲完开始上结构。下面这套分层是我在几个项目里反复调整后稳定下来的版本不一定是最优解但每一层的划分都有明确理由。2.1 六层结构各自的职责我把 harness 拆成六层从上到下依次是接入层、会话层、编排层、模型适配层、工具执行层、基础设施层。每层的职责必须单一跨层调用要严格禁止否则很快会变成一团互相依赖的泥巴。层级核心职责关键产出常见踩坑点接入层协议转换、鉴权、限流统一内部请求对象把业务逻辑写进路由处理函数会话层状态存储、历史维护、并发隔离会话快照用全局变量存会话状态编排层循环控制、停止条件、降级执行轨迹停止条件只靠模型自己说完成模型适配层请求构造、流式解析、错误归一结构化响应把厂商错误码直接抛给上层工具执行层注册、校验、沙箱、幂等工具结果对象不做参数校验直接执行基础设施层日志、指标、追踪、存储可观测数据只在出错时打日志这张表我建议贴在项目 README 里因为它是团队协作的边界约定。新人最容易犯的错是把工具调用的业务逻辑写在接入层的路由函数里一开始省事等到要加第二个入口比如接一个内部消息平台时就得把所有逻辑复制一遍。六层里最容易被低估的是会话层。很多教程把会话状态存在一个字典里键是用户 ID。单机单进程跑起来没问题一旦开多进程或者做水平扩展用户就会发现自己上一句说的话被忘掉了因为下一次请求打到了另一个进程。这个 bug 的排查成本极高因为它是概率性的。我的做法是会话状态一律外置存储进程内只保留请求生命周期内的临时数据。2.2 一次请求在 Harness 里的完整旅程光看分层还是抽象走一遍数据流会清楚很多。假设用户输入帮我把订单表里上周的异常订单导出来下面是它在系统里经过的路径。接入层收到请求做鉴权、限流、参数校验构造出一个内部统一的Request对象里面包含会话 ID、用户输入、模型选择、工具权限集。这一层只做转换不做决策。会话层根据会话 ID 拉取历史快照。快照不是简单的一串消息而是包含消息列表、已注入的检索片段、用户画像标签、当前 token 预算使用情况的结构化对象。拉取的同时要加锁或者用乐观并发控制防止同一会话的两个请求互相覆盖。编排层进入主循环。第一轮把系统提示词、历史消息、可用工具的 schema、本轮用户输入拼成完整请求交给模型适配层。适配层负责把它转成具体厂商的请求格式发出去解析流式响应返回一个归一化的结果对象字段包括文本内容、推理内容如果有、工具调用列表、token 用量。回到编排层如果结果里有工具调用就交给工具执行层。工具执行层先按名称查注册表确认这个工具在当前会话的权限集里然后用 JSON Schema 校验参数校验通过才真正执行。执行结果统一包装成ToolResult带上成功标记、内容、耗时、是否可重试。工具结果被追加进消息列表进入下一轮循环。同时编排层要更新 token 预算如果超出阈值就触发上下文压缩。循环一直持续到满足停止条件模型返回了不含工具调用的最终答案、达到最大步数、超时、或者预算耗尽。最后编排层输出最终答案给接入层同时把完整执行轨迹写入可观测系统。会话层把更新后的历史快照落盘。这条链路上有两个设计点值得强调。一是所有跨层的数据结构都应该是显式的对象而不是裸字典。用字典的代价是加字段时没人知道还有谁在消费这个字段改一个键名可能炸掉三个模块。二是循环的每一次迭代都要生成一个可追溯的步骤记录包括这一轮消耗了多少 token、调了哪个工具、花了多少时间。没有这层记录线上问题基本只能靠猜。2.3 部署形态选型单进程、单机多进程还是微服务关于要不要上微服务和分布式架构我的态度非常明确harness 在绝大多数场景下不需要微服务。这个话题被过度讨论了很多团队在只有几十个日活用户的时候就拆了七八个服务结果调试一个工具调用要翻五六份日志。我给出一个信号清单出现其中任何一条才考虑拆分。第一条是工具执行需要独立伸缩比如代码执行、视频转码这类重资源工具和主流程的资源曲线完全不同。第二条是不同工具需要不同的安全域比如访问生产数据库的工具必须跑在隔离环境里。第三条是团队规模超过十人且模块边界已经稳定。三条都不满足的时候单机多进程加一个外置状态存储能扛住相当可观的量级。单机形态下我推荐的进程模型是一个主进程负责接入和编排工具执行走子进程池重任务丢给独立的工作进程通过队列通信。这样崩溃隔离和超时控制都好做运维复杂度也低。真正需要横向扩展的时候你会发现因为状态已经外置了加一层负载均衡就能平滑扩上去不用重写。3. 核心模块实现细节与实操要点架构图谁都会画难点在每一层的细节。这一节挑四个最容易出问题的模块展开每个都给出具体做法和参数。3.1 模型适配层请求构造、流式解析与错误归一适配层的核心价值是把厂商差异挡在外面。上层永远只看到一套字段不管底层是哪家模型。请求构造上我建议把模型参数集中在一个配置对象里而不是散落在调用点。温度、最大输出长度、工具选择策略、是否开启流式这些都应该是配置项。最容易被忽略的是最大输出长度——很多模型默认值是 4096如果你不显式设置长回答会被硬截断而且截断时往往不报错只是内容突然中断排查起来很痛苦。流式解析是适配层最容易写出 bug 的地方。工具调用在流式响应里是按分片增量返回的你必须按索引累积拼接不能当成完整对象处理。下面是一段我实际在用的解析骨架import json import httpx def stream_chat(messages, toolsNone, modeldeepseek-chat, base_url, api_key): payload {model: model, messages: messages, stream: True} if tools: payload[tools] tools headers {Authorization: fBearer {api_key}, Content-Type: application/json} buffer, tool_calls [], {} with httpx.stream(POST, f{base_url}/chat/completions, jsonpayload, headersheaders, timeout180) as resp: resp.raise_for_status() for line in resp.iter_lines(): if not line or not line.startswith(data:): continue data line[5:].strip() if data [DONE]: break chunk json.loads(data) if not chunk.get(choices): continue delta chunk[choices][0].get(delta, {}) if delta.get(content): yield {type: text, value: delta[content]} for tc in delta.get(tool_calls) or []: idx tc.get(index, 0) slot tool_calls.setdefault(idx, {id: , name: , args: }) if tc.get(id): slot[id] tc[id] fn tc.get(function) or {} if fn.get(name): slot[name] fn[name] if fn.get(arguments): slot[args] fn[arguments] if tool_calls: yield {type: tool_calls, value: [tool_calls[k] for k in sorted(tool_calls)]}这段代码里有三个细节值得单独说。第一index字段必须用上并行工具调用时多个工具的分片会交叉出现不按索引分桶就会拼成乱码。第二name用而不是因为少数情况下函数名本身也会被分片。第三参数拼接完必须做一次 JSON 解析校验解析失败说明模型吐出了非法 JSON这时候不要直接抛异常而应该把原始字符串截断后作为错误信息回传给模型让它自己修正。我实测下来这种让模型看到自己的错误的做法纠正成功率比直接重试高不少。错误归一这块我建议至少区分四类可重试错误网络超时、限流、不可重试错误参数非法、鉴权失败、配额错误余额不足、超额、未知错误。前两类处理方式完全不同前者按指数退避重试后者立刻失败并给出明确提示。把所有错误当成一类处理是新手最常见的反模式——要么对鉴权失败也重试三次浪费时间要么对网络抖动直接报错给用户。3.2 工具注册与调用协议让模型真的会用工具工具层的设计目标只有一个模型看到的工具描述必须和实际执行的函数严格一致。任何偏差都会导致模型调用出错。我用装饰器做注册好处是定义和使用在同一个地方不容易出现文档和执行不一致的情况import json import inspect REGISTRY {} def tool(nameNone, descNone, scopesNone, timeout30): def deco(fn): key name or fn.__name__ sig inspect.signature(fn) props, required {}, [] type_map {str: string, int: integer, float: number, bool: boolean} for pname, param in sig.parameters.items(): py_type param.annotation if param.annotation is not inspect._empty else str props[pname] {type: type_map.get(py_type, string)} if param.default is inspect._empty: required.append(pname) REGISTRY[key] { fn: fn, timeout: timeout, scopes: set(scopes or []), schema: { type: function, function: { name: key, description: desc or (fn.__doc__ or ).strip()[:512], parameters: {type: object, properties: props, required: required}, }, }, } return fn return deco tool(desc按订单号查询订单详情返回状态、金额、创建时间, scopes[order:read], timeout10) def query_order(order_id: str) - dict: ...这段代码有两点经验。参数类型从函数签名自动推导避免了手写 JSON Schema 时忘记同步改文档的经典问题我见过太多次文档说传字符串、代码实现要整数的事故。scopes权限集是我强烈建议加上的它让这个会话能不能调这个工具变成一个显式声明而不是散落在业务代码里的 if 判断。工具描述description的写法直接决定模型的调用准确率。我的经验是一句话说清做什么紧接着一句说清什么情况下用。比如查询订单详情不如按订单号查询订单详情当用户提供了订单号且需要订单状态或金额时使用。后半句能让模型在多个相似工具之间做出正确选择。另外描述里不要写实现细节模型不需要知道你是走 SQL 还是走 HTTP那些信息只会占用 token 预算且干扰判断。参数的边界也要写进描述。比如一个工具接收limit参数描述里应该写最多返回条数取值 1 到 100默认 20。我踩过一次坑模型传了limit10000后端没做上限校验直接把整张表拉进内存服务当场 OOM。永远不要相信模型传的参数在合理范围内校验必须在执行前做。3.3 上下文预算与压缩别等塞爆了才想这件事上下文管理是 harness 里最像资源调度的部分也是纯工程活儿跟模型能力无关。我的做法是把窗口看成一个预算池每个消费者按优先级分配额度。先算一笔具体的账。假设目标模型的上下文窗口是 65536 token我要预留这几个部分输出预留 4096应对长回答系统提示词实测约 700十二个工具的 schema 序列化后约 1900本轮检索注入的知识片段 2000安全余量 1000用来吸收 token 估算误差。把这五项减掉留给历史对话的额度是65536 - 4096 - 700 - 1900 - 2000 - 1000 55840 token这个数字就是历史消息的硬上限。接下来要解决的是怎么估算当前用了多少。精确做法是调用对应模型的 tokenizer但很多接口不开放工程上更常见的是估算再乘以安全系数。我的经验值是中文按字符数乘 0.6 估算 token英文和代码按字符数除以 3.5然后用实测数据校准这个系数。不同模型的系数会差 10% 到 20%所以安全余量一定要留。超出预算时的压缩策略我是分三档递进的。第一档是截断把最早的几轮对话整体丢掉代价是丢失早期上下文好处是无损、无额外调用。第二档是摘要把要丢弃的段落交给模型做一次摘要用摘要替换原文压缩比通常在 5 到 10 倍。第三档是事实卡从历史里抽取关键事实存成结构化字段比如用户所在城市杭州偏好简洁回复已确认订单号A123。事实卡不占多少 token但能保留最关键的个性化信息。值得单独强调的是工具返回结果必须做截断。一个查询接口返回 5 万字符的 JSON直接塞进上下文能吃掉几万 token而且里面大部分字段模型根本用不上。我的做法是给每个工具配一个输出处理器只保留必要字段并且硬性限制单次返回不超过 4000 字符超出就截断并在结尾标注结果已截断如需更多请缩小查询范围。这个设计让模型自己学会问得更精确而不是无脑拉全量数据。3.4 沙箱与权限让模型伸手但不让它拆家只要 harness 具备执行能力——跑命令、写文件、调数据库——沙箱就是必选项不是可选项。我见过太多演示项目直接在宿主进程里exec模型生成的代码演示效果好上线就是灾难。沙箱要控制的东西有明确的优先级排序。第一是资源用量CPU 时间和内存必须设上限否则一个死循环就能把机器拖垮。第二是时间硬性超时超时直接杀进程。第三是文件系统只挂载必要的临时目录其余只读或不可见。第四是网络默认关闭需要联网的工具单独声明。第五是权限用独立的低权限账号运行绝不使用管理员权限。下面是一个 Unix 环境下跑 Python 片段的沙箱骨架用的是标准库的资源限制import os import resource import subprocess def run_python_snippet(code: str, timeout: int 10) - dict: def _limits(): cpu max(1, timeout) resource.setrlimit(resource.RLIMIT_CPU, (cpu, cpu)) resource.setrlimit(resource.RLIMIT_AS, (512 * 1024 * 1024, 512 * 1024 * 1024)) resource.setrlimit(resource.RLIMIT_FSIZE, (8 * 1024 * 1024, 8 * 1024 * 1024)) resource.setrlimit(resource.RLIMIT_NPROC, (32, 32)) try: proc subprocess.run( [python, -I, -S, -c, code], capture_outputTrue, timeouttimeout, preexec_fn_limits, cwd/tmp/sandbox, env{PATH: /usr/bin:/bin, HOME: /tmp/sandbox}, ) out proc.stdout.decode(utf-8, replace)[:8000] err proc.stderr.decode(utf-8, replace)[:2000] return {ok: proc.returncode 0, stdout: out, stderr: err, code: proc.returncode} except subprocess.TimeoutExpired: return {ok: False, stdout: , stderr: f执行超时{timeout}s, code: -1}几个关键点说明一下。-I参数让 Python 忽略环境变量和用户目录的配置避免被预先植入的配置影响-S跳过 site 初始化减少启动开销和被注入的可能。RLIMIT_NPROC防止代码 fork 出大量子进程。cwd指向独立的临时目录配合容器时这个目录应该是一个临时卷。preexec_fn在 Unix 上可用Windows 下需要用别的机制跨平台场景建议直接用容器方案。注意这只是一种轻量隔离能挡住误操作挡不住刻意构造的攻击。面向不可信输入的场景必须上进程级或容器级隔离并且运行在独立的宿主环境里。权限设计上我的做法是工具级权限集 会话级授权的组合。每个工具声明自己需要的权限标识每个会话在创建时确定拥有哪些标识调用时做交集判断。这样既能做到细粒度控制又不需要在每次调用时做复杂的鉴权计算。把权限做在工具层还有一个好处当你要接入新的用户群体时只需要配置权限集不用改任何业务代码。4. 实操从零把 Harness 跑起来前面讲了很多设计层面的东西这一节直接给能跑的代码。为了让内容可复现我会按照最小版本 → 加工具 → 加流式 → 加预算的顺序逐步叠加每一步都是完整可运行的。4.1 环境准备与最小可运行版本依赖很简单一个 HTTP 客户端加一个异步运行时就够了。我用httpx是因为它同时支持同步和流式pydantic用来做配置和参数校验。如果你要对接本地推理服务绝大多数推理框架都提供 OpenAI 兼容端点把base_url指过去就行harness 层不用改。python -m venv .venv source .venv/bin/activate pip install httpx pydantic配置我统一从环境变量读绝不写死在代码里import os from pydantic import BaseModel class Config(BaseModel): base_url: str os.getenv(LLM_BASE_URL, ) api_key: str os.getenv(LLM_API_KEY, ) model: str os.getenv(LLM_MODEL, deepseek-chat) max_steps: int 8 max_output_tokens: int 4096 step_timeout: int 60 total_timeout: int 300 CFG Config()max_steps这个参数看着不起眼但它是我认为最重要的一个安全阀。没有它一个逻辑设计不当的 agent 会无限循环每一步都消耗 token。我设的经验值是简单任务 3 到 5 步中等复杂度 8 到 12 步超过 15 步基本说明提示词或者工具设计有问题应该去修设计而不是调大上限。最小版本的主循环长这样先不考虑工具import httpx def chat_once(messages): payload { model: CFG.model, messages: messages, max_tokens: CFG.max_output_tokens, } resp httpx.post( f{CFG.base_url}/chat/completions, jsonpayload, headers{Authorization: fBearer {CFG.api_key}}, timeoutCFG.step_timeout, ) resp.raise_for_status() data resp.json() return data[choices][0][message][content], data.get(usage, {})跑到这一步你应该先验证一件事接口通不通、模型名对不对、返回结构符不符合预期。我见过不少人直接跳到多轮工具调用结果调试半天发现是基础请求参数写错了。先让最简单的版本跑通再往上加这是省时间的做法。4.2 加入工具调用循环有了最小版本把工具循环加上。核心逻辑就三件事模型说要调工具我们执行把结果喂回去。import json def run_agent(user_input: str, session_messagesNone, max_stepsNone): max_steps max_steps or CFG.max_steps messages list(session_messages or []) messages.append({role: user, content: user_input}) schemas [v[schema] for v in REGISTRY.values()] trace [] for step in range(max_steps): payload {model: CFG.model, messages: messages, max_tokens: CFG.max_output_tokens} if schemas: payload[tools] schemas resp httpx.post( f{CFG.base_url}/chat/completions, jsonpayload, headers{Authorization: fBearer {CFG.api_key}}, timeoutCFG.step_timeout, ) resp.raise_for_status() msg resp.json()[choices][0][message] messages.append(msg) calls msg.get(tool_calls) or [] if not calls: trace.append({step: step, type: final, content: msg.get(content, )}) return msg.get(content, ), messages, trace for call in calls: cid call.get(id) name call[function][name] raw_args call[function].get(arguments) or {} try: args json.loads(raw_args) assert isinstance(args, dict) except Exception: result {ok: False, error: 参数不是合法 JSON 对象请检查后重新调用} else: result execute_tool(name, args) messages.append({ role: tool, tool_call_id: cid, content: json.dumps(result, ensure_asciiFalse)[:4000], }) trace.append({step: step, type: tool, name: name, args: args, result: result}) return 已达到最大执行步数任务未完成。, messages, trace配套的执行函数是权限和沙箱的汇合点def execute_tool(name: str, args: dict): entry REGISTRY.get(name) if not entry: return {ok: False, error: f工具 {name} 不存在} try: import inspect sig inspect.signature(entry[fn]) sig.bind(**args) except TypeError as e: return {ok: False, error: f参数不匹配{e}} try: return {ok: True, data: entry[fn](**args)} except Exception as e: return {ok: False, error: f执行失败{type(e).__name__}: {e}}这里有个我反复强调的原则工具异常必须被捕获并转成结构化结果返回给模型而不是往上抛。原因很简单模型看到参数不匹配缺少 order_id这类信息之后下一轮通常能自己改对。如果你直接抛异常中断整个会话用户看到的就是一句无用的报错。把错误路径也当成正常流程来设计是 agent 系统能不能用得起来的关键分界。4.3 流式输出与可观测性用户等 30 秒什么都看不到体验会很差。加流式之后文本可以边生成边推给前端。实现上就是把 3.1 节的解析器接进循环文本分片直接转发工具调用分片累积到末尾再处理。可观测性我建议在开发阶段就做不要留到上线。每次执行输出一份 trace包含每一步的耗时、该步的 token 用量、调用了哪个工具、参数和结果摘要。这份 trace 是排查问题的唯一依据。我自己的习惯是把它打到标准输出格式化成人类可读的文本同时异步写一份 JSON 到文件。开发的时候看文本线上出问题的时候查 JSON。有一个指标我强烈建议单独统计工具调用的失败率按工具名分组。这个数字能直接告诉你哪个工具的描述写得有问题。失败率高的工具八成是描述太模糊导致模型传错参数而不是模型能力不够。我做过统计一个描述写得含糊的工具失败率能到 40%把描述改成什么情况下用 参数怎么填 边界值之后能降到 5% 以内。4.4 上下文预算的计算过程把预算逻辑落成代码思路就是前面算的那笔账def estimate_tokens(text: str) - int: if not text: return 0 cjk sum(1 for ch in text if \u4e00 ch \u9fff) other len(text) - cjk return int(cjk * 0.6 other / 3.5) 4 def trim_history(messages, window65536, output_reserve4096, fixed_cost5600, safety1000): budget window - output_reserve - fixed_cost - safety total sum(estimate_tokens(json.dumps(m, ensure_asciiFalse)) for m in messages) if total budget: return messages, {trimmed: False, used: total, budget: budget} head [m for m in messages if m.get(role) system] body [m for m in messages if m.get(role) ! system] used sum(estimate_tokens(json.dumps(m, ensure_asciiFalse)) for m in head) kept [] for m in reversed(body): cost estimate_tokens(json.dumps(m, ensure_asciiFalse)) if used cost budget: break kept.append(m) used cost kept.reverse() return head kept, {trimmed: True, used: used, budget: budget, dropped: len(body) - len(kept)}这段逻辑里有几个判断要说清楚。系统提示词和工具结果永远不会被裁掉它们在head或者被单独保留因为这两类是任务前提丢了之后模型会彻底跑偏。裁剪从最旧的消息开始保留最近的对话因为最近的内容与当前任务相关性最高。裁剪后必须检查第一条保留的消息是不是工具结果——如果是说明它对应的助手消息被裁掉了这会导致部分接口报错需要在代码里补齐一条占位消息。这个细节我踩过一次表现为偶发的请求 400排查了很久才定位到。裁剪发生的时候我建议在系统提示词末尾追加一条提示更早的对话内容已被裁剪如需必要信息请向用户询问。这句话能显著降低模型基于残缺上下文胡乱推断的概率。5. 常见问题与排查技巧实录下面是这几年我实际遇到过、并且不止遇到一次的问题。每一条都附上排查路径照着走基本能定位。5.1 高频问题速查表现象最可能原因排查动作处理方式模型完全不调工具工具描述含糊或缺何时使用打印实际发送的 tools 字段重写描述补充使用场景参数 JSON 解析失败流式分片拼接错误打印拼接后的原始字符串按 index 分桶累积非法时回传错误同一工具反复调用提示词没定义停止条件看 trace 里的重复调用加相同参数连续两次则终止规则上下文超限报错工具返回未截断统计单次工具返回字符数限制单次返回 4000 字符多轮后回答变差历史裁剪把关键信息丢了检查裁剪后保留的首条消息引入事实卡保留关键字段并发请求串味会话状态存在进程内检查状态存储位置状态外置进程只留临时数据偶发请求 400裁剪后孤儿工具消息检查消息角色顺序补齐占位消息或整组裁剪超时重试导致重复下单工具没有幂等控制查看是否有重复副作用用调用 ID 做幂等键这张表里我最想强调的是最后一行。所有有副作用的工具都必须幂等这是硬要求。做法是用工具调用的 ID 作为幂等键服务端记录已处理的键重复请求直接返回上次结果。我见过真实事故网络抖动导致工具调用超时harness 按策略重试了一次结果用户被扣了两次款。技术上说得通业务上不可接受。5.2 三个最容易被忽略的坑第一个坑是停止条件完全交给模型。很多实现里循环的唯一退出条件是模型返回不含工具调用的消息。这在模型稳定的时候能用但一旦模型陷入某种模式比如反复确认我是否需要更多信息循环就会一直跑下去。我的做法是三重保护最大步数、总 token 预算、墙上时钟超时任意一个触发就强制退出并返回一个明确的任务中断结果而不是静默失败。用户看到任务在 8 步内未完成请把问题拆得更具体一些比看到转圈半天然后什么都没有要好得多。第二个坑是工具结果不做语义化处理。假设工具返回一个包含 50 个字段的 JSON模型很难从中抓住重点。我习惯在工具层做一次结果摘要把最关键的字段放在最前面其余字段用一句完整字段列表a, b, c...带过。这样模型拿到的信息密度高很多决策质量也明显提升。这个改动的性价比极高几乎是零成本换来可感知的效果提升。第三个坑是不做回归测试。harness 里的任何改动——改提示词、加工具、调预算——都可能让某些场景变差。我维护了一份二十来条的小测试集覆盖典型任务和各条失败路径每次改完跑一遍看通过率有没有下降。这个习惯是从吃了亏之后开始养的有一次改了一句系统提示词主流程变好了但一个边缘场景从能用变成完全不可用上线三天后才发现。5.3 成本与延迟的调优杠杆调优这件事我按投入产出比排了序。第一优先级是减少不必要的步数因为每一步都是一次完整的模型调用步数减半基本等于成本减半。减少步数的办法是让工具更厚——把多个小工具合并成一个能一次返回完整信息的工具减少往返。第二优先级是压缩工具结果前面讲过效果直接且明显。第三优先级是缓存系统提示词和工具 schema 在每轮里都是重复的很多接口对前缀相同的内容有缓存优惠把不变的部分放前面、变化的部分放后面能吃到这部分折扣。第四优先级是分级模型简单任务用好用便宜的模型复杂任务才上能力更强的模型。harness 层做成可配置的切换成本很低。延迟方面最大的杠杆是并行工具调用。如果一轮里有三个互不依赖的工具调用串行执行要花三倍时间。做法是把同一轮返回的多个工具调用并发执行用一个带超时的并发池控制。要注意的是如果工具之间存在依赖关系比如先查 ID 再查详情就不能并行这需要工具设计阶段就把依赖关系理清楚不要指望运行时判断。6. 插件化与多模型路由把 Harness 做成可复用的底座做到这一步你的 harness 已经能跑通完整链路了。如果要长期维护还有两件事值得投入。6.1 插件化让工具和模型都能热插拔我采用的方案是把工具定义、模型适配器、输出处理器都做成可注册的插件。启动时扫描指定目录加载所有符合约定的模块注册到统一的注册表里。这样新增一个工具只需要写一个文件不需要改动 harness 主体代码。插件化的关键约束是插件不能直接访问 harness 的内部状态。插件只能通过传入的上下文对象拿到它需要的东西比如会话 ID、权限集、日志器。这条约束听起来很烦但它保证了插件的失败不会波及主体也让插件可以被独立测试。我见过没做这个约束的项目一个插件的内存泄漏把整个服务拖垮。模型路由也是同理。给每个模型适配器声明它的能力和成本档位路由层根据任务复杂度选择。复杂度的判断可以很简单用户输入长度、是否需要工具、历史轮次。我用的是一个三档的粗暴规则实测效果已经够了不需要上更复杂的策略。6.2 评测集与灰度让改动可控harness 的迭代有个特点改动的影响面很大但反馈很慢。你改了系统提示词可能要几天后才发现某个场景变差了。解决这个问题的唯一办法是建立离线评测集把反馈周期从天缩短到分钟。我的评测集分三层。第一层是单元级针对单个工具的参数校验、异常处理、输出截断。第二层是流程级给定输入检查是否走了预期的工具调用路径用的是轨迹匹配而不是答案匹配因为路径对了答案差不到哪去。第三层是端到端人工标注的输入输出对用模型或者规则打分。上线策略上我用的是影子流量加灰度。新版本先跑影子不影响真实用户对比两条链路的轨迹差异差异在可接受范围内再切一小部分流量。这个流程听起来重但实际做下来只有几十行代码比起线上翻车的代价非常划算。我个人在实际操作中的体会是harness 这个层级的工程80% 的工作量花在异常路径和边界条件上20% 花在正常路径。新手容易反过来分配精力把 happy path 打磨得很漂亮结果一遇到工具报错、上下文超限、并发冲突就全线崩溃。真正让系统稳定运行的恰恰是那些看起来用不上的防御代码。我现在的习惯是每写一个工具第一件事不是写实现而是先想清楚它可能怎么失败把失败路径的处理写完再去写正常逻辑——顺序反过来之后线上事故率下降得非常明显。