1. 从一次“Agent 跑飞”说起Harness 到底解决什么问题你可能已经用 LangChain 或某个 Agent 框架搭过一个能查天气、能读文件的 demo感觉“Agent 不过如此”。但一旦把它放到真实任务里——比如让它排查一个失败的测试、跟进一个客户、处理一条退款——它就开始出问题读了一堆无关文件把上下文撑爆、反复调用同一个工具、改错了代码没法回滚、连续失败十几次还在硬撑。这时候你调 prompt 基本没用因为问题不在模型“想不想对”而在模型外面的运行系统没给它设好边界。这个运行系统就是 AI Agent Harness。一句话概括模型负责判断下一步想做什么Harness 负责决定它能看到什么、能调用什么、怎么执行、怎么记录、怎么验证、什么时候必须停。它不是一个具体的库而是一类系统设计——把 Agent 从“会说话的推理器”变成“可上线、可审计、可迭代的软件系统”。大厂现在招 Harness 工程师招的就是能把上下文管理、工具注册、权限隔离、执行沙箱、状态追踪、验证反馈这一整套工程骨架搭起来的人。这篇不聊概念空转直接交付一套可复制的 Agent Harness 配置骨架工具注册表、上下文管理器、带步数上限和超时控制的执行循环最后用 TaoToken 的 API 跑通一个最小可用的代码排查 Agent。适合已经写过 Agent demo、想把它推进到“能稳定跑长任务”的开发者。2. 前置准备用 TaoToken 统一模型接入层Harness 的第一层是模型接入。如果你在 Harness 里硬编码某一家模型的 SDK后面换模型、做 A/B 对比、控制成本都会很痛苦。我的做法是把模型调用抽象成一个统一的 client所有请求走同一个入口。这里用 TaoToken 的 API 做接入层它兼容 OpenAI 风格的接口改 base_url 和 key 就能切换模型Harness 上层代码不用动。你需要先拿到 API Key。打开 https://taotoken.net/api-keys 登录后在控制台创建密钥复制保存。注意 Key 只显示一次丢了就重新生成。如果你还没注册从 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 进官网注册即可。拿到 Key 后把它写进环境变量不要硬编码进代码export TAOTOKEN_API_KEYsk-你的密钥 export TAOTOKEN_BASE_URLhttps://taotoken.net/api注意base_url 用 https://taotoken.net/api 不要加多余的路径后缀OpenAI SDK 会自动拼接 /v1/chat/completions。模型选择上Harness 里的“大脑”建议用推理能力强的模型工具调用格式遵循度高的优先。你可以在 https://taotoken.net/models 看当前可用的模型列表选一个支持 function calling 的。我实测下来工具调用参数格式的稳定性比单纯的对话能力更影响 Harness 的可靠性——参数格式一乱schema 校验就拦整个循环就卡住。3. 可复制的 Harness 骨架工具注册 上下文管理 执行循环下面这套骨架分三个模块你可以直接拿去改。核心设计原则是模型只能通过注册表里的工具碰外部世界每次调用都过 schema 校验和权限检查循环有步数上限和超时所有动作落日志。3.1 工具注册表白名单 schema 校验 权限分级工具不是越多越好。工具越多模型选错的概率越高工具描述本身也占上下文。Harness 要维护一个注册表控制当前任务能用哪些工具并且给每个工具标权限等级。import json from typing import Callable, Any class ToolRegistry: def __init__(self): self._tools {} def register(self, name: str, description: str, parameters: dict, handler: Callable, permission: str read): # permission: read / write / dangerous self._tools[name] { schema: { type: function, function: { name: name, description: description, parameters: parameters, }, }, handler: handler, permission: permission, } def get_schemas(self, allowed_permissions(read,)): return [ t[schema] for t in self._tools.values() if t[permission] in allowed_permissions ] def validate_and_call(self, name: str, args: dict) - str: if name not in self._tools: return f[ERROR] 工具 {name} 未注册 tool self._tools[name] # 简易 schema 校验检查必填参数 required tool[schema][function][parameters].get(required, []) for key in required: if key not in args: return f[ERROR] 缺少必填参数 {key} try: return str(tool[handler](**args)) except Exception as e: return f[ERROR] 工具执行异常: {e}注册三个代码排查工具权限分级读文件、搜代码是 read跑测试是 write因为它有副作用。import subprocess, os WORKDIR os.path.abspath(./sandbox) def safe_path(p): full os.path.abspath(os.path.join(WORKDIR, p)) if not full.startswith(WORKDIR): raise ValueError(路径越界) return full registry ToolRegistry() registry.register( read_file, 读取工作目录下的文件内容, {type: object, properties: {path: {type: string}}, required: [path]}, lambda path: open(safe_path(path), encodingutf-8).read()[:4000], permissionread, ) registry.register( search_code, 在工作目录中搜索关键词返回匹配行, {type: object, properties: {keyword: {type: string}}, required: [keyword]}, lambda keyword: subprocess.run( [grep, -rn, keyword, WORKDIR], capture_outputTrue, textTrue, timeout10 ).stdout[:3000], permissionread, ) registry.register( run_test, 运行指定测试文件返回输出, {type: object, properties: {target: {type: string}}, required: [target]}, lambda target: subprocess.run( [python, -m, pytest, target, -q], capture_outputTrue, textTrue, timeout30, cwdWORKDIR ).stdout[-3000:], permissionwrite, )3.2 上下文管理器保留、摘要、截断三档策略上下文管理要决定哪些内容保留、哪些摘要、哪些只给引用。这里给一个简化版系统规则和用户目标永远保留工具返回超过阈值就截断历史动作只保留最近 N 轮。class ContextManager: def __init__(self, system_prompt: str, max_tool_output: int 2000, max_history: int 12): self.system system_prompt self.max_tool_output max_tool_output self.max_history max_history self.history [] def add_user(self, content: str): self.history.append({role: user, content: content}) def add_assistant(self, msg: dict): self.history.append(msg) def add_tool_result(self, tool_call_id: str, content: str): if len(content) self.max_tool_output: content content[:self.max_tool_output] \n...[已截断] self.history.append({ role: tool, tool_call_id: tool_call_id, content: content, }) def build(self): trimmed self.history[-self.max_history:] return [{role: system, content: self.system}] trimmed3.3 执行循环步数上限 超时 失败即停这是 Harness 的心脏。关键控制点最大步数、单步超时、连续失败计数、危险动作拦截。import time from openai import OpenAI client OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], ) SYSTEM_PROMPT 你是一个代码排查 Agent。 规则 1. 先收集证据再下结论。 2. 每次只调用一个工具。 3. 修改前必须说明理由。 4. 连续两次工具失败就停止并汇报。 def run_harness(user_goal: str, max_steps: int 8): ctx ContextManager(SYSTEM_PROMPT) ctx.add_user(user_goal) fail_streak 0 for step in range(max_steps): start time.time() resp client.chat.completions.create( modelgpt-4o-mini, # 换成你在 TaoToken 选的模型 messagesctx.build(), toolsregistry.get_schemas(allowed_permissions(read, write)), tool_choiceauto, timeout30, ) msg resp.choices[0].message ctx.add_assistant(msg.model_dump(exclude_noneTrue)) if not msg.tool_calls: print(f[第 {step1} 步] 模型给出最终结论\n{msg.content}) return msg.content for call in msg.tool_calls: name call.function.name args json.loads(call.function.arguments) print(f[第 {step1} 步] 调用 {name}({args})) result registry.validate_and_call(name, args) if result.startswith([ERROR]): fail_streak 1 else: fail_streak 0 ctx.add_tool_result(call.id, result) if fail_streak 2: print([HARNESS] 连续失败两次停止执行) return 任务中止工具连续失败 if time.time() - start 25: print([HARNESS] 单步超时) print([HARNESS] 达到最大步数停止) return 任务中止步数超限这套骨架不到 150 行但已经包含了 Harness 的核心控制点工具白名单、schema 校验、权限分级、上下文截断、步数上限、失败即停。你可以把它当成起点后面按需加沙箱、审批、trace。4. 本地验证跑通一个最小代码排查 Agent准备一个沙箱目录放一个故意写错的测试mkdir -p sandbox cd sandbox cat calc.py EOF def add(a, b): return a - b # 故意写错 EOF cat test_calc.py EOF from calc import add def test_add(): assert add(2, 3) 5 EOF回到 Harness 脚本所在目录运行run_harness(帮我看看 test_calc.py 为什么失败并说明原因)预期输出大致是模型先调用run_test拿到失败信息再调用read_file读calc.py发现add里写的是减法最后给出结论“add函数实现错误应为a b”。整个过程你能在终端看到每一步调用了什么工具、传了什么参数、返回了什么。如果你想让 Agent 真正改文件可以再加一个write_file工具权限设为dangerous并在validate_and_call里加人工确认逻辑。这就是 Harness 和普通 Agent 框架的差异普通框架给你一个循环Harness 给你一个带边界的循环。验证模型本身是否正常响应可以先用 https://taotoken.net/chat 做一次对话测试确认 Key 和模型都通再跑 Harness 脚本能省掉不少排查时间。5. 本篇常见错排查报错 401 UnauthorizedKey 没读到或写错了。检查echo $TAOTOKEN_API_KEY是否有值注意别把引号也复制进去。如果是在 IDE 里跑确认环境变量是在同一个终端会话里 export 的。报错 model not found模型名写错了。去 https://taotoken.net/models 复制准确的模型 ID别凭记忆写。工具调用参数解析失败json.loads报错模型返回的 arguments 不是合法 JSON。这通常发生在模型不支持 function calling 或工具描述太模糊时。换一个工具调用能力强的模型并把工具 description 写清楚——参数含义、格式、示例都写上。Agent 反复调用同一个工具上下文里工具结果被截断得太狠模型看不到之前已经拿到的信息。把max_tool_output调大或者在add_tool_result里对重复结果做去重标记。路径越界报错safe_path拦住了。这是 Harness 在正常工作说明模型试图访问工作目录外的文件。检查你的任务描述是否让模型误解了工作范围。循环跑满 max_steps 还没结论任务太复杂或工具不够。先看 trace 里模型卡在哪一步是缺工具还是缺上下文。别急着加步数上限先补工具或改系统提示。6. 把 Harness 用起来从最小骨架到长期运行这套骨架跑通后你手里就有了一个可复现的 Agent 运行环境。接下来按需扩展加 trace 日志把每步的上下文、工具调用、返回结果落盘方便事后审计加沙箱把run_test放进容器里跑隔离文件系统加审批环节让dangerous权限的工具必须人工确认加评估脚本对同一任务跑多次看成功率。如果你打算长期跑编码类 Agent比如让它持续排查 issue、跑测试、提 PR建议用 Coding Plan 这类按周期计费的方式控制成本比按 token 计费更适合长任务场景。入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。Harness 工程最忌讳一上来做大平台。先让一个具体 Agent 稳定工作再把共性抽出来。你现在这套 150 行的骨架已经比大多数 demo 更接近生产了——因为它有边界。