☰
从零手写AI Agent Harness:用TaoToken统一Key搭建你自己的AI执行环境
2026/10/1 7:17:14 网站建设 项目流程

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_KEY=sk-你的key TAOTOKEN_BASE_URL=https://taotoken.net/api TAOTOKEN_MODEL=gpt-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_key=api_key, base_url=base_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( model=MODEL_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 pydantic

3.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_root=workspace_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( model=self.model, messages=self.messages, tools=self.tools.openai_definitions(), tool_choice="auto", ) 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(parents=True, exist_ok=True) self._tools: Dict[str, Dict[str, Any]] = {} self._register_defaults() def _register_defaults(self): self.register( name="read_file", description="读取工作区内文件内容", parameters={ "type": "object", "properties": {"path": {"type": "string", "description": "相对路径"}}, "required": ["path"], }, handler=self._read_file, ) self.register( name="write_file", description="向工作区内文件写入内容", parameters={ "type": "object", "properties": { "path": {"type": "string"}, "content": {"type": "string"}, }, "required": ["path", "content"], }, handler=self._write_file, ) self.register( name="run_command", description="在工作区执行 shell 命令并返回输出", parameters={ "type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"], }, handler=self._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(encoding="utf-8") def _write_file(self, path: str, content: str) -> str: target = self._safe_path(path) if not target: return f"拒绝:{path} 超出工作区" target.parent.mkdir(parents=True, exist_ok=True) target.write_text(content, encoding="utf-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, shell=True, cwd=self.root, capture_output=True, text=True, timeout=30) 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( model=self.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_root=workspace) 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/api,Key 填你的TAOTOKEN_API_KEY,Model 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换成stream=True,边收边打印,体验会好很多。注意流式模式下工具调用的拼接方式不同,需要按delta累积。

第三,把记忆和上下文管理补上。最小骨架里messages每轮重置,多轮对话会丢上下文。可以加一个Memory类,把历史消息持久化到 JSON 或 SQLite,循环启动时加载最近 N 条。上下文太长时,用模型自己生成摘要替换早期消息,这就是所谓的滑动窗口加摘要。

第四,接入 Coding Plan 这类长期编码场景。如果你打算让 Agent 持续处理代码任务,用按量计费的 Key 可能成本不好控。TaoToken 的 Coding Plan 适合这种长期、高频的编码场景,配合 Harness 的循环调度,可以做成一个常驻的编码助手。配置方式同样是 Base URL + Key + Model ID 三件套,在对应的工具里填好即可。

最后提醒一点:Harness 的复杂度应该跟着你的需求走。如果只是想让 Agent 读写文件、跑跑命令,本文这套三百行左右的骨架足够了。不要一上来就上多 Agent 协作、向量数据库、消息队列,那些是规模上来之后才需要的东西。先把单循环跑稳,再逐步加能力,踩的坑会少很多。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询