Agent Harness 从零手写：用 Python + openai SDK 搭一个最小可用 Agent 并改到 TaoToken

1. 为什么我要手写一个 Agent Harness先说清楚这篇要解决什么问题。Agent Harness 这个词听起来唬人拆开看就是「让大模型能自己决定调用工具、拿到结果、再继续推理」的那套外壳代码。Python 加 openai SDK 就能写出来不需要 LangChain不需要任何框架。适合谁会一点 Python、想搞明白 Agent 内部到底怎么转、但被各种框架封装劝退的人。我试过直接上框架结果出了 bug 只能对着层层封装干瞪眼根本不知道消息是怎么编排的、工具调用是怎么回传的。后来反着来只用 openai SDK 直连从零写一个最小可用的 Harness写完才发现核心循环只有十几行剩下的全是工程活会话隔离、上下文裁剪、异常兜底、流式输出。这篇会给你一条完整可跟做的路径先定义工具调用循环与消息编排再接入统一 Key/API 通道跑通一次真实对话。目录结构、依赖清单、配置片段全部可复制最后附三步验证动作——本地启动、单轮工具调用、多轮上下文回传帮你确认 Harness 最小闭环真的可用。核心检索词就三个Agent Harness、Python、openai SDK全文围绕它们展开。2. 前置准备用 TaoToken 统一 Key 与 API 通道写 Harness 之前得先解决「模型从哪来」的问题。openai SDK 本身只是个客户端它需要一个兼容 OpenAI 协议的 Base URL 和一把 Key。你可以用 TaoToken 作为统一通道好处是同一个 SDK、同一套代码换模型只改一个 Model ID不用动业务逻辑。先拿 Key。打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后在控制台创建 API Key。控制台地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite Key 管理页在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。拿到形如sk-xxxx的字符串后别硬编码进代码用环境变量。API 的基础地址是 https://taotoken.net/api 注意这个地址后面不加任何查询参数openai SDK 会自动在它后面拼/chat/completions。如果你用的是 Anthropic 风格的客户端对应的接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有 Base URL 和鉴权头的写法。这里有个关键点openai SDK 认的是base_url和api_key两个参数。TaoToken 兼容 OpenAI 协议所以你把base_url指向https://taotoken.net/apiSDK 就会把请求发到正确的位置。Model ID 填你在控制台看到的模型名比如gpt-4o-mini或claude-3-5-sonnet这类具体以控制台列表为准。依赖清单很简单一个文件搞定openai1.30.0 python-dotenv1.0.0python-dotenv用来读.env文件避免 Key 泄露到代码仓库。装依赖pip install -r requirements.txt目录结构建议这样后面每一步都对应一个文件agent-harness/ ├── .env ├── requirements.txt ├── config.py ├── tools.py ├── harness.py └── main.pyconfig.py负责读环境变量并初始化 clienttools.py放工具定义和 JSON Schemaharness.py是核心循环main.py是入口。这样分层的好处是以后想换模型只改config.py想加工具只改tools.py核心循环不动。3. 可复制配置client 初始化与工具 Schema先写config.py。这里把 Base URL、Key、Model ID 三件套集中管理后面所有文件都从这里导入。# config.py import os from dotenv import load_dotenv from openai import OpenAI load_dotenv() API_KEY os.getenv(TAOTOKEN_API_KEY) BASE_URL https://taotoken.net/api MODEL_ID os.getenv(TAOTOKEN_MODEL, gpt-4o-mini) if not API_KEY: raise RuntimeError(缺少 TAOTOKEN_API_KEY请检查 .env 文件) client OpenAI( api_keyAPI_KEY, base_urlBASE_URL, )对应的.env文件长这样注意不要提交到 gitTAOTOKEN_API_KEYsk-你的真实key TAOTOKEN_MODELgpt-4o-mini接下来是tools.py。工具注册的本质是给模型一份 JSON Schema 说明书模型靠它决定「该不该调、传什么参数」。我放两个工具一个查天气一个算表达式方便验证单轮和多轮。# tools.py import json def get_weather(city: str) - str: fake_db {北京: 晴20°C, 上海: 多云24°C} return fake_db.get(city, f{city}暂无数据) def calculate(expression: str) - str: try: return str(eval(expression, {__builtins__: {}}, {})) except Exception as e: return f计算失败{e} TOOL_REGISTRY { get_weather: get_weather, calculate: calculate, } TOOLS_SCHEMA [ { type: function, function: { name: get_weather, description: 查询指定城市的当前天气。当用户询问天气时调用。, parameters: { type: object, properties: { city: {type: string, description: 城市名如 北京} }, required: [city], }, }, }, { type: function, function: { name: calculate, description: 计算一个数学表达式如 12*83。当用户需要算术时调用。, parameters: { type: object, properties: { expression: {type: string, description: 合法的 Python 算术表达式} }, required: [expression], }, }, }, ]注意description写得越清楚模型选错工具的概率越低。很多人抱怨 Agent 表现差第一件事就该回来检查工具描述是不是写糊了。eval这里做了沙箱限制只允许算术实际生产别直接用eval换成ast.literal_eval或专门的解析器。4. 核心循环harness.py 的消息编排这是整篇的心脏。Agent 的本质就是一个带次数上限的for循环加上「模型要不要调工具」的判断。消息编排的顺序是system 提示 历史消息 本轮用户输入一起发给模型模型返回tool_calls就执行工具把结果作为roletool的消息追加回去再问一次直到模型不再调工具返回最终文本。# harness.py import json from config import client, MODEL_ID from tools import TOOLS_SCHEMA, TOOL_REGISTRY MAX_ROUNDS 5 def run_agent(user_input: str, history: list | None None) - tuple[str, list]: messages history or [ {role: system, content: 你是一个会调用工具的助手需要时调用工具不需要时直接回答。} ] messages.append({role: user, content: user_input}) for round_idx in range(MAX_ROUNDS): response client.chat.completions.create( modelMODEL_ID, messagesmessages, toolsTOOLS_SCHEMA, tool_choiceauto, ) msg response.choices[0].message messages.append(msg) if not msg.tool_calls: return msg.content, messages for call in msg.tool_calls: name call.function.name args json.loads(call.function.arguments or {}) print(f[TRACE] 第{round_idx1}轮调用工具 {name} 参数 {args}) fn TOOL_REGISTRY.get(name) result fn(**args) if fn else f未知工具{name} messages.append({ role: tool, tool_call_id: call.id, content: str(result), }) return 达到最大轮次仍未收敛请检查工具描述或输入。, messages几个容易踩的点。第一msg必须原样追加回messages因为里面带着tool_calls的id后面roletool的消息要靠tool_call_id和它配对少一个模型就报错。第二json.loads(call.function.arguments)里arguments是字符串不是字典忘了解析会直接抛异常。第三MAX_ROUNDS是防死循环的保险丝模型偶尔会反复调同一个工具没有上限就会一直烧 token。main.py负责串起来并演示多轮上下文回传# main.py from harness import run_agent if __name__ __main__: history None while True: text input(你) if text in (exit, quit): break answer, history run_agent(text, history) print(fAgent{answer})这里history在循环外持有每轮把更新后的messages传回去就实现了多轮上下文。业务参数城市、表达式交给模型决定但会话状态由程序持有这个边界要划清楚。5. 三步验证与常见报错排查写完不算完得验证最小闭环真的通了。我按三步走每步都有明确的成功标志。第一步本地启动。运行python main.py如果.env配错最常见的是openai.AuthenticationError: 401。对照检查Key 是不是复制时带了空格、base_url是不是写成了https://taotoken.net/api/末尾多斜杠有时会导致路径拼接异常、环境变量有没有被load_dotenv正确读到。401 基本就是 Key 或 Base URL 的问题和代码逻辑无关。第二步单轮工具调用。输入「北京天气怎么样」预期看到[TRACE] 第1轮调用工具 get_weather 参数 {city: 北京}然后 Agent 回复「北京晴20°C」。如果模型没调工具直接瞎编说明工具description不够明确回去改 Schema。如果报KeyError或TypeError多半是arguments没解析或工具函数签名对不上。第三步多轮上下文回传。先问「北京天气」再问「那上海呢」。第二轮模型应该能结合上下文知道你在问天气直接调get_weather(city上海)。如果它反问「你问的是哪个城市」说明history没传对检查run_agent的返回值有没有被正确接住。其他常见报错对照报错信息原因处理local proxy failed/ 连接超时网络层问题或 Base URL 写错确认base_url为https://taotoken.net/api不要加多余路径reading choices/NoneType响应结构异常通常是鉴权或模型名错误打印完整response看返回体核对 Model IDOAuth/invalid api keyKey 失效或格式不对去控制台重新生成确认以sk-开头json.decoder.JSONDecodeErrorarguments不是合法 JSON打印原始字符串检查模型是否返回了空参数如果你用的是 Claude Code 或 Cline 这类工具接 TaoToken配置三件套要写全Base URL 填https://taotoken.net/apiKey 填控制台生成的Model ID 填控制台列表里的模型名。缺任何一个都会连不上。Cline 的 MCP 配置里如果出现command和args记得把环境变量也带上否则子进程读不到 Key。排障时优先看[TRACE]日志它能还原「程序到底干了啥」。大部分问题不是模型笨而是消息编排少了一条、工具 Schema 写错了、或者 Key 没读到。6. 继续往下走从最小闭环到可用 Harness跑通三步验证后你已经有了一个真正属于自己的 Agent Harness。接下来可以按需加肌肉会话隔离用session_id分字典存历史上下文裁剪只保留 system 加最近 N 条流式输出用streamTrue逐片yield工具执行出错时把错误信息当结果喂回模型让它自己修正。想验证不同模型的表现可以去模型对话页 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchatutm_campaignrewrite 直接对比同一段 prompt 的输出。长期做编码或 Agent 任务Coding Plan 在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite Key 管理在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。最后留一个我踩过的坑流式模式下tool_calls的参数是分片到达的id、name、arguments都会拆成好几块必须按index归组、用拼接不能覆盖。我第一次就是只拼到最后一片json.loads直接失败。这个细节教程里很少讲清楚但你只要自己写一遍就再也不会忘。