ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

Agent Harness 简述:从零搭建可复现的 Agent 运行骨架与 TaoToken 接入

Agent Harness 简述:从零搭建可复现的 Agent 运行骨架与 TaoToken 接入 1. 从一次“跑不通”的 Agent 任务说起很多人第一次写 Agent代码不到一百行一个 while 循环把用户输入丢给模型模型返回tool_calls就执行执行完把结果塞回 messages再问模型直到模型不再调用工具。逻辑上没问题但真跑起来问题一个接一个工具参数是字符串还是对象、模型返回的 JSON 被截断、循环跑了二十轮还在原地打转、上下文越堆越长最后超 token、报错信息只有一句local proxy failed完全不知道哪一层挂了。这些问题的根源不在模型而在模型外面那层“支架”太薄。这层支架就是 Agent Harness——它把只会生成文本的模型接到工具、文件、状态和执行环境上让模型能观察、决策、行动、拿到反馈再进入下一轮。模型是决策核心Harness 是运行时和边界两者合起来加上任务目标才是一个能交付结果的 Agent。我试过用最朴素的方式手搓一个循环结果在第三个工具调用就卡住了模型把file_path写成了filepath我的执行器直接抛 KeyError整个进程退出前面十几轮上下文全丢。那次之后我才认真把 Harness 拆成几个独立模块执行循环、工具注册与校验、状态与上下文管理、错误恢复、日志观测。拆开之后同一个模型的表现稳定了一大截。这篇就按这个思路从零搭一个最小可运行的 Agent Harness 骨架目录结构、依赖、启动命令都给全最后把模型 endpoint 切到 TaoToken 的统一通道跑通一次完整任务闭环。适合已经会调 API、想搞清楚 Agent 执行循环到底怎么组织的开发者。核心检索词就三个Agent Harness 是什么、执行循环怎么设计、工具调用与状态管理怎么落地。2. TaoToken 前置准备统一 Key 与 API 通道在写 Harness 之前先把模型通道固定下来。Harness 本身不应该关心你用的是哪家模型它只认一个 OpenAI 兼容的 endpoint。这样后面换模型、换供应商只改配置不动循环代码。TaoToken 在这里的角色就是统一通道一个 Key、一个 Base URL背后可以路由到不同模型。对 Harness 来说它就是一个标准的/v1/chat/completions接口支持tools参数和tool_calls返回这正是 Agent 循环需要的。你需要准备三样东西第一API Key。到控制台创建地址是 https://taotoken.net/api-keys 创建后复制保存只显示一次。第二Base URL。统一用https://taotoken.net/api注意这个地址不带任何查询参数直接作为 OpenAI SDK 的base_url。第三Model ID。在模型列表里选一个支持工具调用的比如claude-sonnet-4-5这类具体以控制台展示为准。Harness 配置里把它写成变量方便替换。这里有个容易踩的坑很多人把 Base URL 写成带/v1的完整路径然后在 SDK 里又拼一次/v1结果请求打到/v1/v1/chat/completions返回 404。正确做法是base_url只写到域名加/apiSDK 自己会补/v1/chat/completions。如果你用的是原生requests那就手动拼https://taotoken.net/api/v1/chat/completions。另外Key 不要硬编码进代码。用环境变量或者.env文件Harness 启动时读取。这样你本地调试、容器部署、CI 跑测试用的是同一套代码只换环境变量。配置验证可以先单独做一次不涉及 Harnessexport TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_MODELclaude-sonnet-4-5 curl -s $TAOTOKEN_BASE_URL/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: $TAOTOKEN_MODEL, messages: [{role: user, content: 只回复两个字连通}], max_tokens: 16 }返回里能看到choices[0].message.content就说明通道没问题。这一步先跑通后面 Harness 报错时就能排除掉“Key 或网络”这一层直接定位到循环逻辑。3. 可复制的最小 Harness 骨架与配置这一节给完整目录和代码。目标是一个能跑的最小骨架注册两个工具读文件、写文件模型能调用它们循环能正常结束状态能持久化到磁盘。目录结构agent-harness/ ├── .env ├── requirements.txt ├── config.toml ├── harness/ │ ├── __init__.py │ ├── loop.py # 执行循环 │ ├── tools.py # 工具注册与校验 │ ├── state.py # 状态与上下文管理 │ └── llm.py # 模型客户端 └── run.py # 启动入口依赖清单requirements.txtopenai1.40.0 pydantic2.7.0 python-dotenv1.0.0 tomli2.0.0配置文件config.toml路径和字段名保持和代码一致[llm] base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY model claude-sonnet-4-5 max_tokens 2048 temperature 0.2 [loop] max_turns 12 stop_on_no_tool_call true [state] workspace ./workspace history_file ./workspace/history.jsonl模型客户端harness/llm.py只做一件事把 messages 和 tools 发出去拿回 message 对象。import os from openai import OpenAI def build_client(base_url: str, api_key_env: str) - OpenAI: api_key os.environ.get(api_key_env) if not api_key: raise RuntimeError(f环境变量 {api_key_env} 未设置) return OpenAI(base_urlbase_url, api_keyapi_key) def chat(client: OpenAI, model: str, messages: list, tools: list, **kw): resp client.chat.completions.create( modelmodel, messagesmessages, toolstools, tool_choiceauto, **kw, ) return resp.choices[0].message工具注册harness/tools.py用 Pydantic 做参数校验这是 Harness 和裸循环最大的区别之一模型给的参数不可信必须校验后再执行。import json from pathlib import Path from pydantic import BaseModel, ValidationError class ReadFileArgs(BaseModel): path: str class WriteFileArgs(BaseModel): path: str content: str def read_file(args: ReadFileArgs) - str: p Path(args.path) if not p.exists(): return fERROR: 文件不存在 {args.path} return p.read_text(encodingutf-8)[:4000] def write_file(args: WriteFileArgs) - str: p Path(args.path) p.parent.mkdir(parentsTrue, exist_okTrue) p.write_text(args.content, encodingutf-8) return fOK: 已写入 {args.path}{len(args.content)} 字符 TOOL_SPECS [ { type: function, function: { name: read_file, description: 读取工作区内的文本文件, parameters: { type: object, properties: {path: {type: string}}, required: [path], }, }, }, { type: function, function: { name: write_file, description: 把内容写入工作区文件覆盖已有内容, parameters: { type: object, properties: { path: {type: string}, content: {type: string}, }, required: [path, content], }, }, }, ] REGISTRY { read_file: (ReadFileArgs, read_file), write_file: (WriteFileArgs, write_file), } def dispatch(name: str, raw_args: str) - str: if name not in REGISTRY: return fERROR: 未知工具 {name} model_cls, fn REGISTRY[name] try: parsed model_cls(**json.loads(raw_args)) except (json.JSONDecodeError, ValidationError) as e: return fERROR: 参数校验失败 {e} try: return fn(parsed) except Exception as e: return fERROR: 工具执行异常 {type(e).__name__}: {e}状态管理harness/state.py负责把每轮消息追加到 jsonl支持中断后恢复。import json from pathlib import Path class StateStore: def __init__(self, history_file: str): self.path Path(history_file) self.path.parent.mkdir(parentsTrue, exist_okTrue) def append(self, message: dict): with self.path.open(a, encodingutf-8) as f: f.write(json.dumps(message, ensure_asciiFalse) \n) def load(self) - list: if not self.path.exists(): return [] out [] with self.path.open(r, encodingutf-8) as f: for line in f: line line.strip() if line: out.append(json.loads(line)) return out执行循环harness/loop.py这是 Harness 的心脏。观察-计划-行动-反馈四步一轮直到模型不再调用工具或达到最大轮数。from .llm import chat from .tools import TOOL_SPECS, dispatch def run_agent(client, model, messages, state, max_turns12, **kw): for turn in range(max_turns): msg chat(client, model, messages, TOOL_SPECS, **kw) assistant_msg {role: assistant, content: msg.content or } if msg.tool_calls: assistant_msg[tool_calls] [ { id: tc.id, type: function, function: { name: tc.function.name, arguments: tc.function.arguments, }, } for tc in msg.tool_calls ] messages.append(assistant_msg) state.append(assistant_msg) if not msg.tool_calls: return msg.content or for tc in msg.tool_calls: result dispatch(tc.function.name, tc.function.arguments) tool_msg { role: tool, tool_call_id: tc.id, content: result, } messages.append(tool_msg) state.append(tool_msg) return 达到最大轮数任务未结束启动入口run.pyimport os import tomli from dotenv import load_dotenv from harness.llm import build_client from harness.loop import run_agent from harness.state import StateStore load_dotenv() with open(config.toml, rb) as f: cfg tomli.load(f) client build_client(cfg[llm][base_url], cfg[llm][api_key_env]) state StateStore(cfg[state][history_file]) task 在工作区创建 hello.txt内容写 harness ok然后读回来确认。 messages [{role: user, content: task}] state.append(messages[0]) result run_agent( client, cfg[llm][model], messages, state, max_turnscfg[loop][max_turns], max_tokenscfg[llm][max_tokens], temperaturecfg[llm][temperature], ) print(最终输出, result).env文件TAOTOKEN_API_KEYsk-你的key启动命令cd agent-harness python -m venv .venv source .venv/bin/activate pip install -r requirements.txt python run.py这套骨架刻意做得很小但每个模块职责清晰llm.py只管通道tools.py只管注册和校验state.py只管持久化loop.py只管编排。后面要加浏览器工具、加审批、加沙箱都是往对应模块里塞不会把循环改乱。4. 验证请求与成功结果跑通一次任务闭环配置写完后先别急着跑完整任务分两步验证。第一步验证模型通道和工具调用能力。把run.py里的 task 换成最简单的task 调用 write_file 在工作区写一个 test.txt内容为 ping。运行python run.py观察终端输出和workspace/history.jsonl。正常的话history 里会依次出现user 消息、assistant 带tool_calls的消息、tool 返回OK: 已写入 ...、最后 assistant 的总结文本。这说明执行循环、工具分发、状态追加三件事都通了。第二步跑完整闭环任务。用回原来的 task在工作区创建 hello.txt内容写 harness ok然后读回来确认。预期行为是模型先调write_file拿到 OK再调read_file拿到文件内容最后输出类似“已创建并确认内容为 harness ok”。workspace/hello.txt应该真实存在。如果这一步成功你会在history.jsonl里看到至少 5 条记录workspace/下有文件终端打印最终输出。这就是一次完整的观察-计划-行动-反馈闭环。验证模型本身是否正常可以单独开一个对话测试地址是 https://taotoken.net/api-chat 输入同样的问题看返回用来区分是 Harness 逻辑问题还是通道问题。几个实测细节值得注意。tool_choiceauto时模型有时会直接回答而不调工具尤其是任务描述模糊的时候。把任务写具体明确说“调用 write_file”命中率会高很多。另外max_turns不要设太大12 轮足够大多数短任务设太大反而会在模型卡住时浪费 token。temperature设 0.2 左右工具调用参数更稳定。还有一点read_file里我做了 4000 字符截断。真实场景里文件可能很大直接塞回上下文会爆 token。截断是 Harness 该做的事不是模型该操心的。同理后面可以加“只返回前 N 行”“只返回匹配行”这类策略都属于上下文管理的一部分。5. 本篇常见错误排查这一节按真实报错来对。你跑这套骨架大概率会遇到下面几个。401 Unauthorized。返回体里通常是invalid api key或authentication failed。原因基本是环境变量没加载或 Key 写错。检查.env是否在run.py同目录、load_dotenv()是否在读取配置之前调用、TAOTOKEN_API_KEY是否有多余空格或引号。用第 2 节的 curl 单独验证一次能快速定位是 Key 问题还是代码问题。local proxy failed / connection error。这类报错说明请求根本没出去或者被本地网络层拦了。先确认base_url是https://taotoken.net/api没有多余路径再确认机器能正常访问外网 HTTPS。如果你在容器里跑检查容器 DNS 和出网策略。这个错误和 Key 无关别去反复换 Key。reading choices 或 NoneType has no attribute choices。这是解析返回时resp.choices为空。常见原因是请求体里model字段写错或者messages格式不对比如把 tool 消息的tool_call_id漏了。还有一种情况是max_tokens设得太小模型还没生成完就被截断返回结构不完整。把max_tokens调到 2048 以上再试。OAuth / token 过期类报错。如果你用的是某些 CLI 工具自带的登录态而不是 API Key可能会遇到 OAuth 刷新失败。Harness 场景下建议统一走 API Key不要混用 CLI 的登录凭证。Codex 的auth.json、Claude Code 的登录态和这里的TAOTOKEN_API_KEY是两套东西别混。工具参数校验失败。返回ERROR: 参数校验失败。看 history 里 assistant 那条tool_calls的arguments字段通常是模型把数字写成了字符串或者漏了必填字段。解决办法是在工具 description 里把参数类型写清楚必要时在dispatch里做一次宽松转换比如把3转成3。但不要无脑吞掉错误返回明确的 ERROR 文本给模型它下一轮会自己修正。循环不结束。模型反复调同一个工具或者每轮都调工具但任务早完成了。检查stop_on_no_tool_call逻辑是否生效以及max_turns是否兜底。更根本的办法是在 system prompt 里写清楚“任务完成后直接输出结果不要再调用工具”。上下文膨胀。跑十几轮后请求体越来越大最后超 token 报错。这是 Harness 必须处理的问题。最小做法是给 history 做滑动窗口只保留最近 N 轮进阶做法是把早期工具结果压缩成摘要。骨架里state.py已经持久化了全量历史你可以在loop.py里加一个trim_messages函数在每轮请求前裁剪。对照这些报错逐个排基本能把 90% 的启动问题解决掉。排障时优先看history.jsonl它记录了每一轮的真实请求和返回比看终端日志清楚得多。6. 把 Harness 接到长期编码与 Agent 工作流骨架跑通之后下一步通常是把它接到真实工作流里让 Agent 读代码库、改文件、跑测试、根据失败结果继续修。这时候 Harness 的复杂度会上一个台阶需要处理权限边界、危险操作审批、长任务检查点、失败重试。如果你打算长期跑编码类 Agent建议用 Coding Plan 这类按周期计费的方式地址是 https://taotoken.net/coding-plan 比按 token 计费更适合高频调用的场景。接入方式不变还是同一个 Base URL 和 Key只是计费模型不同。Harness 的接入文档在 https://taotoken.net/doc 里面有完整的参数说明和示例。模型对话测试用 https://taotoken.net/api-chat API Key 管理在 https://taotoken.net/api-keys 。回到骨架本身我建议你先在最小版本上多跑几个任务观察 history 里模型的决策路径。你会发现同一个模型在不同任务描述下的工具调用策略差别很大而 Harness 的价值就是把这些不确定性框在一个可控、可观测、可恢复的循环里。模型能力决定上限Harness 设计决定下限。把下限做扎实Agent 才敢往生产环境放。
返回列表