1. 为什么你的 Agent 需要一个 Harness从“只会聊天”到“能干活”大模型本身很聪明能推理、能写代码、能规划步骤但它有一个绕不开的限制它活在文字里碰不到真实世界。你让它“帮我看看项目里有没有未提交的改动”它只能回你一段听起来很合理的猜测因为它读不了你的磁盘、跑不了git status、也看不到命令行的报错输出。Harness 就是补上这一环的东西。你可以把它理解成给模型装上的“手脚和神经系统”模型负责决策Harness 负责把决策翻译成真实动作再把动作结果喂回给模型让它继续下一步。Claude Code、Cursor 这类工具之所以能读文件、跑测试、迭代修改靠的正是背后这套执行环境。一个最小可用的 AI Agent Harness核心就三块Agent Loop循环调度、工具注册表Tool Registry、任务编排Task Orchestration。循环负责“想—做—看—再想”工具注册表负责“能做什么”任务编排负责“先做什么后做什么”。本文会带你从零把这套骨架搭起来并用 TaoToken 的统一 Key 作为模型调用入口让你不用在多个厂商的 Key 之间来回切换。适合谁看写过一点 Python、想搞明白 Agent 底层怎么跑的人被各种框架的抽象层绕晕、想自己掌控执行流程的人以及想给自己的项目加一个“能动手”的 AI 助手的人。全程可复制跑通一个端到端任务就算成功。2. 用 TaoToken 统一 Key 打通模型调用入口Base URL、Key 与 Model ID 三件套在写循环之前先把模型调用这条线理顺。很多新手卡在第一步不同厂商的接口格式、鉴权方式、模型名都不一样代码里到处是 if-else。TaoToken 的思路是提供一个统一的 API 通道你只需要记住三样东西Base URL、API Key、Model ID。Base URL 用https://taotoken.net/api这是所有请求的根地址。API Key 在控制台的 API Keys 页面创建创建后复制保存它只显示一次。Model ID 就是你要调用的模型标识比如gpt-4o、claude-3-5-sonnet这类具体以文档里的模型列表为准。我习惯用环境变量管理这些配置避免把 Key 写死在代码里。在项目根目录建一个.env文件# .env TAOTOKEN_API_KEYsk-你的key TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODELgpt-4o然后用python-dotenv加载。如果你用的是 OpenAI 官方 SDK只需要在初始化客户端时把base_url指过去其余调用方式完全不变# src/core/client.py import os from dotenv import load_dotenv from openai import OpenAI load_dotenv() def build_client() - OpenAI: 构建统一的模型客户端所有模块共用 api_key os.getenv(TAOTOKEN_API_KEY) base_url os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) if not api_key: raise RuntimeError(缺少 TAOTOKEN_API_KEY请检查 .env 文件) return OpenAI(api_keyapi_key, base_urlbase_url) MODEL_ID os.getenv(TAOTOKEN_MODEL, gpt-4o)这里有个容易踩的坑base_url末尾不要多加/v1或斜杠SDK 会自己拼接路径。如果你写成了https://taotoken.net/api/v1/请求路径就会变成/api/v1/v1/chat/completions直接 404。实测下来保持https://taotoken.net/api这个形式最稳。配置好之后先跑一个最小验证确认通道是通的# scripts/check_connection.py from src.core.client import build_client, MODEL_ID client build_client() resp client.chat.completions.create( modelMODEL_ID, messages[{role: user, content: 只回复两个字通了}], ) print(resp.choices[0].message.content)如果输出“通了”说明 Base URL、Key、Model ID 三件套都对。这一步别跳过后面所有模块都依赖它。如果报 401多半是 Key 没加载进来或者复制时带了空格如果报连接错误检查base_url是否写错。把这条线打通后面的循环和工具才有意义。3. 可复制的 Harness 骨架Agent Loop、工具注册表与任务编排配置现在进入正题。我把 Harness 拆成三个文件loop.py管循环tools.py管工具注册orchestrator.py管任务编排。先看目录结构my_harness/ ├── src/ │ ├── core/ │ │ ├── client.py # 模型客户端上一节 │ │ ├── loop.py # Agent Loop │ │ ├── tools.py # 工具注册表 │ │ └── orchestrator.py # 任务编排 │ └── agent.py # 组装入口 ├── workspace/ # Agent 的工作区 ├── .env └── requirements.txt依赖装这几个就够pip install openai python-dotenv pydantic3.1 Agent Loop一个 while 循环驱动“想—做—看”循环的本质很简单把用户输入和工具定义发给模型模型要么直接回答要么返回工具调用请求如果有工具调用就执行、把结果塞回消息列表、再问一次模型直到模型不再请求工具为止。# src/core/loop.py import json from typing import List, Dict, Any from .client import build_client, MODEL_ID from .tools import ToolRegistry class AgentLoop: Agent 循环模型思考 - 调用工具 - 观察结果 - 继续思考 def __init__(self, workspace_root: str ./workspace, max_iterations: int 15): self.client build_client() self.model MODEL_ID self.max_iterations max_iterations self.tools ToolRegistry(workspace_rootworkspace_root) self.messages: List[Dict[str, Any]] [] def run(self, user_input: str) - str: self.messages [ {role: system, content: 你是一个能调用工具的助手需要动手时请调用工具不要凭空猜测。}, {role: user, content: user_input}, ] for step in range(1, self.max_iterations 1): print(f[Loop {step}] 请求模型...) resp self.client.chat.completions.create( modelself.model, messagesself.messages, toolsself.tools.openai_definitions(), tool_choiceauto, ) msg resp.choices[0].message self.messages.append(msg.model_dump()) if not msg.tool_calls: return msg.content or 任务结束 for call in msg.tool_calls: name call.function.name args json.loads(call.function.arguments) print(f - 调用工具 {name} 参数 {args}) result self.tools.execute(name, args) self.messages.append({ role: tool, tool_call_id: call.id, content: str(result), }) return 达到最大迭代次数任务未完成关键点在于什么时候停由模型自己决定。你不需要写“如果包含‘完成’就退出”这种脆弱逻辑。模型不再返回tool_calls循环自然结束。3.2 工具注册表加工具不改循环代码工具注册表要解决两件事一是把 Python 函数包装成模型能理解的 JSON Schema二是执行时做安全校验。路径限制和危险命令拦截是底线。# src/core/tools.py import subprocess from pathlib import Path from typing import Dict, Any, Callable, List class ToolRegistry: def __init__(self, workspace_root: str ./workspace): self.root Path(workspace_root).resolve() self.root.mkdir(parentsTrue, exist_okTrue) self._tools: Dict[str, Dict[str, Any]] {} self._register_defaults() def _register_defaults(self): self.register( nameread_file, description读取工作区内文件内容, parameters{ type: object, properties: {path: {type: string, description: 相对路径}}, required: [path], }, handlerself._read_file, ) self.register( namewrite_file, description向工作区内文件写入内容, parameters{ type: object, properties: { path: {type: string}, content: {type: string}, }, required: [path, content], }, handlerself._write_file, ) self.register( namerun_command, description在工作区执行 shell 命令并返回输出, parameters{ type: object, properties: {command: {type: string}}, required: [command], }, handlerself._run_command, ) def register(self, name: str, description: str, parameters: Dict, handler: Callable): self._tools[name] {description: description, parameters: parameters, handler: handler} def _safe_path(self, path: str): target (self.root / path).resolve() if target ! self.root and self.root not in target.parents: return None return target def _read_file(self, path: str) - str: target self._safe_path(path) if not target: return f拒绝{path} 超出工作区 if not target.exists(): return f文件不存在{path} return target.read_text(encodingutf-8) def _write_file(self, path: str, content: str) - str: target self._safe_path(path) if not target: return f拒绝{path} 超出工作区 target.parent.mkdir(parentsTrue, exist_okTrue) target.write_text(content, encodingutf-8) return f已写入 {path}{len(content)} 字符 def _run_command(self, command: str) - str: blocked [rm -rf, sudo, mkfs, dd if, shutdown] if any(b in command.lower() for b in blocked): return f拒绝执行危险命令{command} try: r subprocess.run(command, shellTrue, cwdself.root, capture_outputTrue, textTrue, timeout30) return (r.stdout or r.stderr or 命令执行完成无输出).strip() except subprocess.TimeoutExpired: return 命令超时30 秒 def openai_definitions(self) - List[Dict]: return [ {type: function, function: { name: n, description: i[description], parameters: i[parameters]}} for n, i in self._tools.items() ] def execute(self, name: str, args: Dict) - str: if name not in self._tools: return f未知工具{name} return self._tools[name][handler](**args)加新工具时只要在_register_defaults里多写一条register循环代码一行都不用动。这就是注册表的价值。3.3 任务编排把大目标拆成有序步骤单轮循环能处理“读文件并总结”这类任务但“先建目录、再写脚本、再运行验证”这种多步骤任务需要一个编排层。最简单的做法是用一个规划提示词让模型输出步骤列表然后逐步喂给循环。# src/core/orchestrator.py from .loop import AgentLoop class Orchestrator: def __init__(self, loop: AgentLoop): self.loop loop def plan(self, goal: str) - list: prompt ( f把下面的目标拆成 2-5 个可独立执行的步骤每行一个以 - 开头不要编号\n{goal} ) resp self.loop.client.chat.completions.create( modelself.loop.model, messages[{role: user, content: prompt}], ) text resp.choices[0].message.content return [l.strip(- ).strip() for l in text.splitlines() if l.strip().startswith(-)] def run(self, goal: str) - dict: steps self.plan(goal) print(f规划出 {len(steps)} 个步骤{steps}) results [] for i, step in enumerate(steps, 1): print(f\n 执行步骤 {i}/{len(steps)}{step} ) out self.loop.run(step) results.append({step: step, result: out}) return {goal: goal, steps: results}3.4 组装入口# src/agent.py from src.core.loop import AgentLoop from src.core.orchestrator import Orchestrator class MyAgent: def __init__(self, workspace: str ./workspace): self.loop AgentLoop(workspace_rootworkspace) self.orchestrator Orchestrator(self.loop) def chat(self, text: str) - str: return self.loop.run(text) def workflow(self, goal: str) - dict: return self.orchestrator.run(goal)到这里骨架就齐了。三个模块各司其职配置集中在.env工具通过注册表扩展编排层负责拆解目标。4. 端到端验证让 Harness 跑通一个真实任务并检查结果光有代码不算跑通得让它真的动起来。准备一个测试目标“在工作区创建一个 hello.py内容是打印当前时间然后运行它并把输出保存到 result.txt”。先写一个入口脚本# run_demo.py from src.agent import MyAgent agent MyAgent(workspace./workspace) result agent.workflow( 在工作区创建 hello.py内容为打印当前时间然后运行它把输出写入 result.txt ) for item in result[steps]: print(f\n步骤{item[step]}\n结果{item[result]})运行python run_demo.py你会看到类似这样的过程规划出 3 个步骤循环里模型先调用write_file写入hello.py再调用run_command执行python hello.py最后调用write_file把输出写进result.txt。每一步的[Loop n]日志和工具调用参数都会打印出来。验证成功的标志有三个workspace/hello.py文件存在且内容正确workspace/result.txt里有一行时间戳控制台没有出现“达到最大迭代次数”。你可以手动cat workspace/result.txt确认。如果模型没有按预期调用工具而是直接编了一段回答通常是系统提示词不够明确。把loop.py里的 system 消息改成“你必须通过工具完成文件操作禁止假设文件已存在”再跑一次。另一个常见情况是模型把路径写成了绝对路径被_safe_path拦下返回“超出工作区”这时看日志里的参数就能定位。实测下来这个最小骨架能稳定处理“读写文件 执行命令”这类任务。复杂任务失败时先看是规划步骤不合理还是某一步的工具参数不对两者排查方向完全不同。5. 常见报错排查401、local proxy failed、reading choices 与 OAuth 问题跑通之后你大概率会在不同环境里遇到下面几类报错。我把它们和真实原因对应起来方便你快速定位。401 Unauthorized最常见。先确认.env里的TAOTOKEN_API_KEY是否被正确加载。可以在build_client里临时打印api_key[:8]看前几位。如果 Key 是从控制台复制的注意别把首尾空格带进去。还有一种情况是环境变量名写错比如写成了TAOTOKEN_KEY代码里读的是TAOTOKEN_API_KEY自然取到None。local proxy failed / connection error这类报错通常指向网络层。检查TAOTOKEN_BASE_URL是否写成了https://taotoken.net/api/末尾多了斜杠或https://taotoken.net/api/v1多了版本段。正确的形式是https://taotoken.net/api。另外确认本机没有残留的HTTP_PROXY/HTTPS_PROXY环境变量干扰请求可以在终端echo $HTTPS_PROXY看一眼如果有值且不是你要的先unset再跑。reading choices of undefined这个报错说明resp.choices是空的或resp结构不对。常见原因是模型名写错了接口返回了一个错误对象而不是正常的 completion 结构。检查TAOTOKEN_MODEL是否在文档的模型列表里。另一个原因是请求体里messages为空比如循环里消息列表被意外清空。在loop.py里加一行assert self.messages能提前暴露。OAuth / 鉴权相关报错如果你在 Claude Code 或类似工具里配置注意区分“API Key 模式”和“OAuth 登录模式”。用 TaoToken 的 Key 时应该走 API Key 配置而不是触发 OAuth 流程。以 Claude Code 为例配置项要写全三件套Base URL 填https://taotoken.net/apiKey 填你的TAOTOKEN_API_KEYModel ID 填你要用的模型。三者缺一工具就会回退到默认的 OAuth 或报鉴权失败。工具调用参数解析失败如果看到json.loads抛异常说明模型返回的arguments不是合法 JSON。这通常发生在模型输出被截断时。把max_iterations调小、或者在 system 提示里强调“工具参数必须是合法 JSON”能降低概率。更稳妥的做法是在json.loads外面包一层 try失败时把原始字符串作为错误信息塞回消息列表让模型自己纠正。排查时记住一个顺序先确认 Key 和 Base URL再确认 Model ID最后看消息列表结构。这三层都对了剩下的基本是提示词和工具参数的问题。6. 把 Harness 接进你的项目从最小骨架到可持续迭代骨架跑通之后接下来是让它长成你能长期用的东西。几个方向值得投入。第一把工具注册表做成插件式。现在工具都写在_register_defaults里项目一大就乱。可以改成扫描tools/目录下的模块每个模块暴露一个register(registry)函数启动时自动加载。这样加工具就是加文件不用动核心代码。第二给循环加流式输出。现在run是等模型完整返回才继续用户看不到中间过程。把create换成streamTrue边收边打印体验会好很多。注意流式模式下工具调用的拼接方式不同需要按delta累积。第三把记忆和上下文管理补上。最小骨架里messages每轮重置多轮对话会丢上下文。可以加一个Memory类把历史消息持久化到 JSON 或 SQLite循环启动时加载最近 N 条。上下文太长时用模型自己生成摘要替换早期消息这就是所谓的滑动窗口加摘要。第四接入 Coding Plan 这类长期编码场景。如果你打算让 Agent 持续处理代码任务用按量计费的 Key 可能成本不好控。TaoToken 的 Coding Plan 适合这种长期、高频的编码场景配合 Harness 的循环调度可以做成一个常驻的编码助手。配置方式同样是 Base URL Key Model ID 三件套在对应的工具里填好即可。最后提醒一点Harness 的复杂度应该跟着你的需求走。如果只是想让 Agent 读写文件、跑跑命令本文这套三百行左右的骨架足够了。不要一上来就上多 Agent 协作、向量数据库、消息队列那些是规模上来之后才需要的东西。先把单循环跑稳再逐步加能力踩的坑会少很多。 从零手写AI Agent Harness：用TaoToken统一Key搭建你自己的AI执行环境
1. 为什么你的 Agent 需要一个 Harness从“只会聊天”到“能干活”大模型本身很聪明能推理、能写代码、能规划步骤但它有一个绕不开的限制它活在文字里碰不到真实世界。你让它“帮我看看项目里有没有未提交的改动”它只能回你一段听起来很合理的猜测因为它读不了你的磁盘、跑不了git status、也看不到命令行的报错输出。Harness 就是补上这一环的东西。你可以把它理解成给模型装上的“手脚和神经系统”模型负责决策Harness 负责把决策翻译成真实动作再把动作结果喂回给模型让它继续下一步。Claude Code、Cursor 这类工具之所以能读文件、跑测试、迭代修改靠的正是背后这套执行环境。一个最小可用的 AI Agent Harness核心就三块Agent Loop循环调度、工具注册表Tool Registry、任务编排Task Orchestration。循环负责“想—做—看—再想”工具注册表负责“能做什么”任务编排负责“先做什么后做什么”。本文会带你从零把这套骨架搭起来并用 TaoToken 的统一 Key 作为模型调用入口让你不用在多个厂商的 Key 之间来回切换。适合谁看写过一点 Python、想搞明白 Agent 底层怎么跑的人被各种框架的抽象层绕晕、想自己掌控执行流程的人以及想给自己的项目加一个“能动手”的 AI 助手的人。全程可复制跑通一个端到端任务就算成功。2. 用 TaoToken 统一 Key 打通模型调用入口Base URL、Key 与 Model ID 三件套在写循环之前先把模型调用这条线理顺。很多新手卡在第一步不同厂商的接口格式、鉴权方式、模型名都不一样代码里到处是 if-else。TaoToken 的思路是提供一个统一的 API 通道你只需要记住三样东西Base URL、API Key、Model ID。Base URL 用https://taotoken.net/api这是所有请求的根地址。API Key 在控制台的 API Keys 页面创建创建后复制保存它只显示一次。Model ID 就是你要调用的模型标识比如gpt-4o、claude-3-5-sonnet这类具体以文档里的模型列表为准。我习惯用环境变量管理这些配置避免把 Key 写死在代码里。在项目根目录建一个.env文件# .env TAOTOKEN_API_KEYsk-你的key TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODELgpt-4o然后用python-dotenv加载。如果你用的是 OpenAI 官方 SDK只需要在初始化客户端时把base_url指过去其余调用方式完全不变# src/core/client.py import os from dotenv import load_dotenv from openai import OpenAI load_dotenv() def build_client() - OpenAI: 构建统一的模型客户端所有模块共用 api_key os.getenv(TAOTOKEN_API_KEY) base_url os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) if not api_key: raise RuntimeError(缺少 TAOTOKEN_API_KEY请检查 .env 文件) return OpenAI(api_keyapi_key, base_urlbase_url) MODEL_ID os.getenv(TAOTOKEN_MODEL, gpt-4o)这里有个容易踩的坑base_url末尾不要多加/v1或斜杠SDK 会自己拼接路径。如果你写成了https://taotoken.net/api/v1/请求路径就会变成/api/v1/v1/chat/completions直接 404。实测下来保持https://taotoken.net/api这个形式最稳。配置好之后先跑一个最小验证确认通道是通的# scripts/check_connection.py from src.core.client import build_client, MODEL_ID client build_client() resp client.chat.completions.create( modelMODEL_ID, messages[{role: user, content: 只回复两个字通了}], ) print(resp.choices[0].message.content)如果输出“通了”说明 Base URL、Key、Model ID 三件套都对。这一步别跳过后面所有模块都依赖它。如果报 401多半是 Key 没加载进来或者复制时带了空格如果报连接错误检查base_url是否写错。把这条线打通后面的循环和工具才有意义。3. 可复制的 Harness 骨架Agent Loop、工具注册表与任务编排配置现在进入正题。我把 Harness 拆成三个文件loop.py管循环tools.py管工具注册orchestrator.py管任务编排。先看目录结构my_harness/ ├── src/ │ ├── core/ │ │ ├── client.py # 模型客户端上一节 │ │ ├── loop.py # Agent Loop │ │ ├── tools.py # 工具注册表 │ │ └── orchestrator.py # 任务编排 │ └── agent.py # 组装入口 ├── workspace/ # Agent 的工作区 ├── .env └── requirements.txt依赖装这几个就够pip install openai python-dotenv pydantic3.1 Agent Loop一个 while 循环驱动“想—做—看”循环的本质很简单把用户输入和工具定义发给模型模型要么直接回答要么返回工具调用请求如果有工具调用就执行、把结果塞回消息列表、再问一次模型直到模型不再请求工具为止。# src/core/loop.py import json from typing import List, Dict, Any from .client import build_client, MODEL_ID from .tools import ToolRegistry class AgentLoop: Agent 循环模型思考 - 调用工具 - 观察结果 - 继续思考 def __init__(self, workspace_root: str ./workspace, max_iterations: int 15): self.client build_client() self.model MODEL_ID self.max_iterations max_iterations self.tools ToolRegistry(workspace_rootworkspace_root) self.messages: List[Dict[str, Any]] [] def run(self, user_input: str) - str: self.messages [ {role: system, content: 你是一个能调用工具的助手需要动手时请调用工具不要凭空猜测。}, {role: user, content: user_input}, ] for step in range(1, self.max_iterations 1): print(f[Loop {step}] 请求模型...) resp self.client.chat.completions.create( modelself.model, messagesself.messages, toolsself.tools.openai_definitions(), tool_choiceauto, ) msg resp.choices[0].message self.messages.append(msg.model_dump()) if not msg.tool_calls: return msg.content or 任务结束 for call in msg.tool_calls: name call.function.name args json.loads(call.function.arguments) print(f - 调用工具 {name} 参数 {args}) result self.tools.execute(name, args) self.messages.append({ role: tool, tool_call_id: call.id, content: str(result), }) return 达到最大迭代次数任务未完成关键点在于什么时候停由模型自己决定。你不需要写“如果包含‘完成’就退出”这种脆弱逻辑。模型不再返回tool_calls循环自然结束。3.2 工具注册表加工具不改循环代码工具注册表要解决两件事一是把 Python 函数包装成模型能理解的 JSON Schema二是执行时做安全校验。路径限制和危险命令拦截是底线。# src/core/tools.py import subprocess from pathlib import Path from typing import Dict, Any, Callable, List class ToolRegistry: def __init__(self, workspace_root: str ./workspace): self.root Path(workspace_root).resolve() self.root.mkdir(parentsTrue, exist_okTrue) self._tools: Dict[str, Dict[str, Any]] {} self._register_defaults() def _register_defaults(self): self.register( nameread_file, description读取工作区内文件内容, parameters{ type: object, properties: {path: {type: string, description: 相对路径}}, required: [path], }, handlerself._read_file, ) self.register( namewrite_file, description向工作区内文件写入内容, parameters{ type: object, properties: { path: {type: string}, content: {type: string}, }, required: [path, content], }, handlerself._write_file, ) self.register( namerun_command, description在工作区执行 shell 命令并返回输出, parameters{ type: object, properties: {command: {type: string}}, required: [command], }, handlerself._run_command, ) def register(self, name: str, description: str, parameters: Dict, handler: Callable): self._tools[name] {description: description, parameters: parameters, handler: handler} def _safe_path(self, path: str): target (self.root / path).resolve() if target ! self.root and self.root not in target.parents: return None return target def _read_file(self, path: str) - str: target self._safe_path(path) if not target: return f拒绝{path} 超出工作区 if not target.exists(): return f文件不存在{path} return target.read_text(encodingutf-8) def _write_file(self, path: str, content: str) - str: target self._safe_path(path) if not target: return f拒绝{path} 超出工作区 target.parent.mkdir(parentsTrue, exist_okTrue) target.write_text(content, encodingutf-8) return f已写入 {path}{len(content)} 字符 def _run_command(self, command: str) - str: blocked [rm -rf, sudo, mkfs, dd if, shutdown] if any(b in command.lower() for b in blocked): return f拒绝执行危险命令{command} try: r subprocess.run(command, shellTrue, cwdself.root, capture_outputTrue, textTrue, timeout30) return (r.stdout or r.stderr or 命令执行完成无输出).strip() except subprocess.TimeoutExpired: return 命令超时30 秒 def openai_definitions(self) - List[Dict]: return [ {type: function, function: { name: n, description: i[description], parameters: i[parameters]}} for n, i in self._tools.items() ] def execute(self, name: str, args: Dict) - str: if name not in self._tools: return f未知工具{name} return self._tools[name][handler](**args)加新工具时只要在_register_defaults里多写一条register循环代码一行都不用动。这就是注册表的价值。3.3 任务编排把大目标拆成有序步骤单轮循环能处理“读文件并总结”这类任务但“先建目录、再写脚本、再运行验证”这种多步骤任务需要一个编排层。最简单的做法是用一个规划提示词让模型输出步骤列表然后逐步喂给循环。# src/core/orchestrator.py from .loop import AgentLoop class Orchestrator: def __init__(self, loop: AgentLoop): self.loop loop def plan(self, goal: str) - list: prompt ( f把下面的目标拆成 2-5 个可独立执行的步骤每行一个以 - 开头不要编号\n{goal} ) resp self.loop.client.chat.completions.create( modelself.loop.model, messages[{role: user, content: prompt}], ) text resp.choices[0].message.content return [l.strip(- ).strip() for l in text.splitlines() if l.strip().startswith(-)] def run(self, goal: str) - dict: steps self.plan(goal) print(f规划出 {len(steps)} 个步骤{steps}) results [] for i, step in enumerate(steps, 1): print(f\n 执行步骤 {i}/{len(steps)}{step} ) out self.loop.run(step) results.append({step: step, result: out}) return {goal: goal, steps: results}3.4 组装入口# src/agent.py from src.core.loop import AgentLoop from src.core.orchestrator import Orchestrator class MyAgent: def __init__(self, workspace: str ./workspace): self.loop AgentLoop(workspace_rootworkspace) self.orchestrator Orchestrator(self.loop) def chat(self, text: str) - str: return self.loop.run(text) def workflow(self, goal: str) - dict: return self.orchestrator.run(goal)到这里骨架就齐了。三个模块各司其职配置集中在.env工具通过注册表扩展编排层负责拆解目标。4. 端到端验证让 Harness 跑通一个真实任务并检查结果光有代码不算跑通得让它真的动起来。准备一个测试目标“在工作区创建一个 hello.py内容是打印当前时间然后运行它并把输出保存到 result.txt”。先写一个入口脚本# run_demo.py from src.agent import MyAgent agent MyAgent(workspace./workspace) result agent.workflow( 在工作区创建 hello.py内容为打印当前时间然后运行它把输出写入 result.txt ) for item in result[steps]: print(f\n步骤{item[step]}\n结果{item[result]})运行python run_demo.py你会看到类似这样的过程规划出 3 个步骤循环里模型先调用write_file写入hello.py再调用run_command执行python hello.py最后调用write_file把输出写进result.txt。每一步的[Loop n]日志和工具调用参数都会打印出来。验证成功的标志有三个workspace/hello.py文件存在且内容正确workspace/result.txt里有一行时间戳控制台没有出现“达到最大迭代次数”。你可以手动cat workspace/result.txt确认。如果模型没有按预期调用工具而是直接编了一段回答通常是系统提示词不够明确。把loop.py里的 system 消息改成“你必须通过工具完成文件操作禁止假设文件已存在”再跑一次。另一个常见情况是模型把路径写成了绝对路径被_safe_path拦下返回“超出工作区”这时看日志里的参数就能定位。实测下来这个最小骨架能稳定处理“读写文件 执行命令”这类任务。复杂任务失败时先看是规划步骤不合理还是某一步的工具参数不对两者排查方向完全不同。5. 常见报错排查401、local proxy failed、reading choices 与 OAuth 问题跑通之后你大概率会在不同环境里遇到下面几类报错。我把它们和真实原因对应起来方便你快速定位。401 Unauthorized最常见。先确认.env里的TAOTOKEN_API_KEY是否被正确加载。可以在build_client里临时打印api_key[:8]看前几位。如果 Key 是从控制台复制的注意别把首尾空格带进去。还有一种情况是环境变量名写错比如写成了TAOTOKEN_KEY代码里读的是TAOTOKEN_API_KEY自然取到None。local proxy failed / connection error这类报错通常指向网络层。检查TAOTOKEN_BASE_URL是否写成了https://taotoken.net/api/末尾多了斜杠或https://taotoken.net/api/v1多了版本段。正确的形式是https://taotoken.net/api。另外确认本机没有残留的HTTP_PROXY/HTTPS_PROXY环境变量干扰请求可以在终端echo $HTTPS_PROXY看一眼如果有值且不是你要的先unset再跑。reading choices of undefined这个报错说明resp.choices是空的或resp结构不对。常见原因是模型名写错了接口返回了一个错误对象而不是正常的 completion 结构。检查TAOTOKEN_MODEL是否在文档的模型列表里。另一个原因是请求体里messages为空比如循环里消息列表被意外清空。在loop.py里加一行assert self.messages能提前暴露。OAuth / 鉴权相关报错如果你在 Claude Code 或类似工具里配置注意区分“API Key 模式”和“OAuth 登录模式”。用 TaoToken 的 Key 时应该走 API Key 配置而不是触发 OAuth 流程。以 Claude Code 为例配置项要写全三件套Base URL 填https://taotoken.net/apiKey 填你的TAOTOKEN_API_KEYModel ID 填你要用的模型。三者缺一工具就会回退到默认的 OAuth 或报鉴权失败。工具调用参数解析失败如果看到json.loads抛异常说明模型返回的arguments不是合法 JSON。这通常发生在模型输出被截断时。把max_iterations调小、或者在 system 提示里强调“工具参数必须是合法 JSON”能降低概率。更稳妥的做法是在json.loads外面包一层 try失败时把原始字符串作为错误信息塞回消息列表让模型自己纠正。排查时记住一个顺序先确认 Key 和 Base URL再确认 Model ID最后看消息列表结构。这三层都对了剩下的基本是提示词和工具参数的问题。6. 把 Harness 接进你的项目从最小骨架到可持续迭代骨架跑通之后接下来是让它长成你能长期用的东西。几个方向值得投入。第一把工具注册表做成插件式。现在工具都写在_register_defaults里项目一大就乱。可以改成扫描tools/目录下的模块每个模块暴露一个register(registry)函数启动时自动加载。这样加工具就是加文件不用动核心代码。第二给循环加流式输出。现在run是等模型完整返回才继续用户看不到中间过程。把create换成streamTrue边收边打印体验会好很多。注意流式模式下工具调用的拼接方式不同需要按delta累积。第三把记忆和上下文管理补上。最小骨架里messages每轮重置多轮对话会丢上下文。可以加一个Memory类把历史消息持久化到 JSON 或 SQLite循环启动时加载最近 N 条。上下文太长时用模型自己生成摘要替换早期消息这就是所谓的滑动窗口加摘要。第四接入 Coding Plan 这类长期编码场景。如果你打算让 Agent 持续处理代码任务用按量计费的 Key 可能成本不好控。TaoToken 的 Coding Plan 适合这种长期、高频的编码场景配合 Harness 的循环调度可以做成一个常驻的编码助手。配置方式同样是 Base URL Key Model ID 三件套在对应的工具里填好即可。最后提醒一点Harness 的复杂度应该跟着你的需求走。如果只是想让 Agent 读写文件、跑跑命令本文这套三百行左右的骨架足够了。不要一上来就上多 Agent 协作、向量数据库、消息队列那些是规模上来之后才需要的东西。先把单循环跑稳再逐步加能力踩的坑会少很多。