从零构建轻量级 Python Agent Harness：用 TaoToken 统一 Key 打通工具调用链路

1. 从零构建 Python Agent Harness 的真实痛点与场景很多人第一次写 Agent都是从 LangChain 的initialize_agent开始的。跑通 demo 那一刻确实爽但当你把它塞进一个边缘盒子、一个只有 512MB 内存的工控机、或者一个需要冷启动 100ms 内响应的服务里问题就全冒出来了依赖装了两百多个包、启动要等好几秒、工具调用链路黑盒、想改一行调度逻辑得翻源码。我试过在一个 ARM 边缘设备上部署某主流框架光是pip install就花了十几分钟装完占用接近 600MB冷启动 2 秒起步。对于需要常驻、低延迟、可审计的场景这套东西根本没法用。于是就有了这篇文章要讲的事从零构建一个轻量级 Python Agent Harness核心代码控制在几百行依赖只有 Python 标准库加一个 HTTP 客户端把工具调用链路完全握在自己手里。所谓 Agent Harness你可以把它理解成智能体的“底盘”或者“运行时”。大模型负责思考和决策但它本身不能读文件、不能查数据库、不能调接口。Harness 的职责就是接收用户输入、维护上下文记忆、把可用工具的描述喂给模型、解析模型返回的工具调用意图、真正执行工具、把结果回填给模型、循环直到得出最终答案。它不负责模型能力只负责把模型和外部世界连起来。这篇文章适合谁如果你正在做边缘 AI、嵌入式 Agent、小型业务系统里的自动化助手或者你只是想彻底搞懂 Agent 底层到底怎么跑起来的那这篇就是写给你的。我会给出可复制的目录结构、依赖清单、最小可运行配置以及一次完整的工具调用链路验证。模型接入部分我用 TaoToken 统一 Key 来打通这样不用在多个厂商的 Key 之间来回切换一个通道就能覆盖不同模型。整个 Harness 的设计目标很明确核心代码小于 500 行、第三方依赖不超过 3 个、内存占用低于 50MB、冷启动低于 100ms、工具注册用装饰器一行搞定。下面从环境准备开始一步步把它搭起来。2. TaoToken 统一 Key 前置准备与 Python Agent Harness 接入配置在写 Harness 之前先把模型通道准备好。轻量级 Harness 的一个核心诉求是“模型可替换”今天用这个模型明天想换另一个不应该改业务代码。TaoToken 在这里的作用就是提供一个统一的 API 通道和统一的 Key你只需要维护一份 Base URL 和一份 Key模型 ID 作为参数传入即可。先拿到凭证。打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后进入控制台在 API Keys 页面创建一个 Key。控制台地址是 https://taotoken.net/console 创建 Key 的页面是 https://taotoken.net/api-keys 。创建完把 Key 复制出来形如sk-xxxxxxxx只显示一次记得存好。接下来是接入配置。TaoToken 的 API 端点是 https://taotoken.net/api 兼容 OpenAI 的 Chat Completions 协议所以任何支持自定义 Base URL 的 OpenAI SDK 都能直接用。这里我用最轻的方式不装 openai 官方 SDK直接用标准库urllib发请求这样 Harness 的依赖能压到最低。如果你更习惯 SDK装openai也可以配置方式一样。先建项目目录。结构尽量扁平方便你一眼看全light-harness/ ├── harness/ │ ├── __init__.py │ ├── core.py # Harness 核心调度 │ ├── registry.py # 工具注册中心 │ ├── memory.py # 记忆银行 │ ├── planner.py # ReAct 规划器 │ └── llm.py # 模型调用适配层 ├── tools/ │ └── ops_tools.py # 业务工具集 ├── config.py # 统一配置 └── main.py # 入口依赖清单只有一行标准库之外不需要任何东西# Python 3.8 即可无需额外依赖 python --version如果你要用 SDK 版本就pip install openai仅此一个。下面写配置。config.py里把 Base URL、Key、模型 ID 集中管理Key 从环境变量读不要硬编码进代码# config.py import os TAOTOKEN_BASE_URL https://taotoken.net/api TAOTOKEN_API_KEY os.getenv(TAOTOKEN_API_KEY, ) DEFAULT_MODEL gpt-4o-mini # 模型 ID 按需替换 REQUEST_TIMEOUT 60 MAX_AGENT_STEPS 10设置环境变量。Linux/macOSexport TAOTOKEN_API_KEYsk-你的KeyWindows PowerShell$env:TAOTOKEN_API_KEYsk-你的Key这里有个关键点Base URL 必须是https://taotoken.net/api不要多加/v1也不要少写路径拼接由适配层负责。模型 ID 是字符串你可以在模型对话页面 https://taotoken.net/models 查看当前可用的模型列表选一个适合工具调用的即可。工具调用对模型的指令遵循能力有要求建议选支持 function calling 的模型。配置好之后先单独验证一下通道是否通。写一个最小的llm.py用标准库发一次请求# harness/llm.py import json import urllib.request from config import TAOTOKEN_BASE_URL, TAOTOKEN_API_KEY, DEFAULT_MODEL, REQUEST_TIMEOUT def chat(messages, modelNone, temperature0.0): 统一的模型调用入口返回 assistant 文本内容 url f{TAOTOKEN_BASE_URL}/v1/chat/completions payload { model: model or DEFAULT_MODEL, messages: messages, temperature: temperature, } data json.dumps(payload).encode(utf-8) req urllib.request.Request( url, datadata, headers{ Content-Type: application/json, Authorization: fBearer {TAOTOKEN_API_KEY}, }, methodPOST, ) with urllib.request.urlopen(req, timeoutREQUEST_TIMEOUT) as resp: body json.loads(resp.read().decode(utf-8)) return body[choices][0][message][content]注意 URL 拼接是{BASE_URL}/v1/chat/completions因为 BASE_URL 本身不带/v1。跑一下验证python -c from harness.llm import chat; print(chat([{role:user,content:只回复两个字通了}]))预期输出就是模型返回的简短文本。如果这一步报 401说明 Key 没读到或者写错了如果报连接错误检查网络和 Base URL 拼写。通道通了再往下搭 Harness 本体。3. 可复制的 Python Agent Harness 最小配置与工具注册实现这一节是核心。我把 Harness 拆成四个文件每个文件职责单一加起来不到 500 行。先讲工具注册中心因为工具调用链路是整条主线。工具注册用装饰器模式。你写一个普通 Python 函数上面加一行registry.tool(...)填上名称、描述、参数 schema它就自动进入可用工具列表。描述和参数 schema 会被序列化后喂给模型模型据此决定调不调用、传什么参数。所以描述要写清楚“这个工具干什么、什么时候用”参数 schema 用 JSON Schema 格式。# harness/registry.py import json import time import traceback import threading from dataclasses import dataclass from typing import Callable, Dict, List, Optional, Any dataclass class Tool: name: str description: str parameters: Dict[str, Any] func: Callable permission_required: bool False dataclass class ExecutionResult: success: bool content: Any None error: str None duration: float 0.0 class ToolRegistry: def __init__(self): self.tools: Dict[str, Tool] {} def tool(self, name, description, parameters, permission_requiredFalse): def decorator(func): self.tools[name] Tool(name, description, parameters, func, permission_required) return func return decorator def list_tools(self) - List[Dict[str, Any]]: return [ {name: t.name, description: t.description, parameters: t.parameters} for t in self.tools.values() ] def get(self, name) - Optional[Tool]: return self.tools.get(name) def execute(self, name, parameters, timeout30) - ExecutionResult: tool self.get(name) if not tool: return ExecutionResult(False, errorfTool {name} not found) result, error None, None start time.time() def run(): nonlocal result, error try: result tool.func(**parameters) except Exception as e: error f{e}\n{traceback.format_exc()} th threading.Thread(targetrun) th.start() th.join(timeouttimeout) if th.is_alive(): return ExecutionResult(False, errorfTool {name} timeout after {timeout}s, durationtime.time() - start) return ExecutionResult(error is None, result, error, time.time() - start)记忆银行用字典实现按 session_id 隔离超过最大轮数自动裁剪最早的记录避免内存无限增长。默认保留最近 50 条够大多数单会话场景用。# harness/memory.py import time from dataclasses import dataclass, asdict from typing import Dict, List, Any dataclass class MemoryRecord: role: str content: str timestamp: float class MemoryBank: def __init__(self, max_len: int 50): self.sessions: Dict[str, List[MemoryRecord]] {} self.max_len max_len def add(self, session_id, role, content): self.sessions.setdefault(session_id, []).append( MemoryRecord(role, content, time.time())) if len(self.sessions[session_id]) self.max_len: self.sessions[session_id] self.sessions[session_id][-self.max_len:] def context(self, session_id, k10) - str: records self.sessions.get(session_id, [])[-k:] return \n.join(f{r.role}: {r.content} for r in records) def clear(self, session_id): self.sessions.pop(session_id, None)规划器负责把上下文和工具列表拼成 prompt让模型输出“调用工具”或“直接回答”。工具调用用一对特殊标记包裹方便正则解析。这里用|FunctionCallBegin|和|FunctionCallEnd|作为边界中间是 JSON 数组。# harness/planner.py import json import re from typing import Any, Dict, List, Optional CALL_BEGIN |FunctionCallBegin| CALL_END |FunctionCallEnd| SYSTEM_PROMPT 你是一个可以调用工具的 AI 助手。请严格按以下规则输出 1. 需要调用工具时输出{begin}[{{name:工具名,parameters:{{参数名:值}}}}]{end} 2. 不需要工具时直接输出回答文本。 3. 每次只调用一个工具不要输出多余内容。 可用工具 {tools} .format(beginCALL_BEGIN, endCALL_END, tools{tools}) class ReActPlanner: def __init__(self, llm_call): self.llm_call llm_call def parse(self, content: str) - Optional[Dict[str, Any]]: pattern re.escape(CALL_BEGIN) r(.*?) re.escape(CALL_END) m re.search(pattern, content, re.DOTALL) if not m: return None try: return json.loads(m.group(1).strip())[0] except Exception: return None def plan(self, query, context, tools) - Dict[str, Any]: prompt SYSTEM_PROMPT.format(toolsjson.dumps(tools, ensure_asciiFalse, indent2)) messages [ {role: system, content: prompt}, {role: user, content: f上下文\n{context}\n\n问题{query}}, ] resp self.llm_call(messages) call self.parse(resp) if call: return {type: tool_call, content: call} return {type: answer, content: resp}核心调度器把上面三块串起来跑一个 while 循环规划 → 判断类型 → 执行工具或返回答案 → 回填记忆 → 继续。加最大步数防止死循环加权限回调拦截敏感工具。# harness/core.py import json from harness.registry import ToolRegistry from harness.memory import MemoryBank from harness.planner import ReActPlanner from config import MAX_AGENT_STEPS class LightHarness: def __init__(self, llm_call, permission_callbackNone): self.registry ToolRegistry() self.memory MemoryBank() self.planner ReActPlanner(llm_call) self.permission_callback permission_callback self.max_steps MAX_AGENT_STEPS def tool(self, name, description, parameters, permission_requiredFalse): return self.registry.tool(name, description, parameters, permission_required) def run(self, query, session_iddefault) - str: self.memory.add(session_id, user, query) for step in range(1, self.max_steps 1): context self.memory.context(session_id) tools self.registry.list_tools() plan self.planner.plan(query, context, tools) if plan[type] answer: self.memory.add(session_id, assistant, plan[content]) return plan[content] call plan[content] name, params call[name], call.get(parameters, {}) tool self.registry.get(name) if tool and tool.permission_required and self.permission_callback: if not self.permission_callback(name, params): self.memory.add(session_id, system, f工具 {name} 被拒绝) continue result self.registry.execute(name, params) if result.success: self.memory.add(session_id, system, f工具 {name} 返回{json.dumps(result.content, ensure_asciiFalse)}) else: self.memory.add(session_id, system, f工具 {name} 失败{result.error}) msg f超过最大步数 {self.max_steps}任务终止 self.memory.add(session_id, assistant, msg) return msg到这里Harness 本体就齐了。四个文件加起来 200 行出头没有任何第三方依赖。接下来注册一个真实工具跑通整条链路。4. 验证工具调用链路一次完整的 Python Agent 请求与预期输出现在写业务工具和入口。我以运维助手为例注册两个工具一个查系统状态一个重启服务敏感操作需要确认。查状态用psutil会更真实但为了保持零依赖这里用标准库os和shutil模拟你换成真实逻辑即可。# tools/ops_tools.py import os import shutil def register_ops_tools(harness): harness.tool( nameget_system_status, description查询服务器的 CPU 核数、磁盘使用率、当前工作目录, parameters{type: object, properties: {}, required: []}, ) def get_system_status(): total, used, free shutil.disk_usage(/) return { cpu_count: os.cpu_count(), disk_used_percent: round(used / total * 100, 2), cwd: os.getcwd(), } harness.tool( namerestart_service, description重启指定的系统服务属于敏感操作, parameters{ type: object, properties: { service_name: {type: string, description: 服务名称} }, required: [service_name], }, permission_requiredTrue, ) def restart_service(service_name: str): # 生产环境替换为真实重启命令 return f服务 {service_name} 已重启入口文件把模型调用、Harness、工具串起来# main.py from harness.llm import chat from harness.core import LightHarness from tools.ops_tools import register_ops_tools def permission_callback(tool_name, params): ans input(f允许调用 {tool_name} 参数 {params}(y/n): ) return ans.lower() y harness LightHarness(llm_callchat, permission_callbackpermission_callback) register_ops_tools(harness) if __name__ __main__: sid ops_001 while True: q input(你) if q.lower() in (exit, quit): break print(助手, harness.run(q, session_idsid))跑起来python main.py输入“帮我看看这台机器的磁盘用了多少”预期链路是这样的Harness 把用户问题和工具列表发给模型模型返回|FunctionCallBegin|[{name:get_system_status,parameters:{}}]|FunctionCallEnd|规划器解析出工具调用注册中心执行get_system_status返回磁盘使用率结果回填记忆模型拿到结果后生成自然语言回答比如“当前磁盘使用率为 42.3%”。整个过程你会在终端看到最终回答。再输入“重启一下 nginx 服务”这次会触发权限回调终端弹出确认提示输入 y 后工具执行模型返回“nginx 服务已重启”。输入 n 则工具被拒绝模型会基于“被拒绝”这个系统消息重新规划通常会说“操作已取消”。如果你想验证工具调用是否真的发生可以在registry.execute里加一行打印或者在core.py的循环里打印plan。实测下来一次完整的工具调用链路从用户输入到最终回答调度开销在 1ms 以内主要耗时都在模型请求上。这里给一个对照表方便你确认各环节是否正常环节正常表现异常表现模型通道返回文本401 / 连接超时工具注册list_tools 有内容列表为空工具解析plan 返回 tool_call一直返回 answer工具执行ExecutionResult.successTrueerror 非空记忆回填context 含 system 消息上下文丢失链路跑通后你可以把chat换成任意模型只要改DEFAULT_MODEL就行Harness 代码一行不用动。这就是统一 Key 通道的价值模型可替换业务逻辑稳定。5. 本篇常见报错排查401、local proxy failed、reading choices 与 OAuth搭 Harness 的过程中报错基本集中在模型通道和工具解析两块。我把几个高频错误和排查路径列出来对照着看。401 Unauthorized。最常见。原因通常是 Key 没读到、Key 写错、或者请求头格式不对。先确认环境变量echo $TAOTOKEN_API_KEYWindows 用echo %TAOTOKEN_API_KEY%。如果为空说明没 export 成功。再确认请求头是Authorization: Bearer sk-xxxBearer 后面有一个空格。还有一种情况是 Key 被复制时带了换行或空格strip 一下。如果都正常还报 401去控制台 https://taotoken.net/api-keys 确认 Key 是否被删除或过期。local proxy failed / connection refused。这个报错说明请求根本没发出去卡在本地网络层。检查 Base URL 是不是写成了https://taotoken.net/api/v1这种多加了路径的形式正确是https://taotoken.net/api然后代码里拼/v1/chat/completions。另外检查系统代理设置有些环境变量HTTP_PROXY、HTTPS_PROXY会干扰 urllib 请求临时 unset 掉再试。如果你在容器里跑确认容器能访问外网。reading choices 报 KeyError 或 IndexError。这个错误发生在解析响应时body[choices][0]取不到。原因通常是响应体结构和你预期的不一样比如返回的是错误 JSON里面没有 choices 字段。排查方法在chat函数里把原始响应打印出来print(resp.read().decode())看看到底返回了什么。常见的是模型 ID 写错返回了错误信息或者请求体格式不对比如 messages 不是列表。确认model字段是有效模型 IDmessages是[{role:user,content:...}]这种结构。OAuth / authentication 相关报错。如果你用的是某些需要 OAuth 的客户端工具可能会遇到 token 刷新失败。但在我们这套纯 HTTP 请求的 Harness 里不涉及 OAuth只有 Bearer Key。如果你在别的工具里看到 OAuth 报错检查是不是把 API Key 和 OAuth token 搞混了。TaoToken 的 API 通道用 API Key 即可不需要走 OAuth 流程。工具调用一直不触发。模型不返回工具调用标记而是直接回答。原因通常是系统 prompt 里的工具描述不够清晰或者模型本身工具调用能力弱。解决办法把工具描述写具体明确“什么时候用这个工具”换一个支持 function calling 的模型在 prompt 里加一句“如果问题涉及实时数据或外部操作优先调用工具”。另外检查parse函数的正则是否匹配标记有没有被模型改写。工具执行超时。ExecutionResult返回 timeout。检查工具函数里是不是有阻塞操作比如input()、网络请求没设超时。把耗时操作放到线程里是对的但工具本身也要设超时。如果是真实的外部 API 调用给urllib或requests加timeout参数。记忆上下文错乱。多轮对话后模型答非所问。检查session_id是否一致不同会话用了同一个 id 会串。另外max_len太小会导致早期上下文被裁掉适当调大。如果上下文太长导致模型截断减少context的k值只取最近几轮。排查顺序建议先单独验证模型通道第 2 节的chat测试再验证工具注册打印list_tools再验证解析打印plan结果最后验证执行打印ExecutionResult。一层层往下问题定位很快。6. 语义一致收尾把 Python Agent Harness 用起来整套东西搭完你会发现 Agent 的底层逻辑其实没那么神秘。核心就是三件事把工具描述清楚、把模型输出解析准、把执行结果回填好。剩下的都是工程细节。轻量级 Harness 的价值在于它让你对整条链路有完全的掌控出问题能定位到具体哪一行想加功能不用等框架更新。如果你后面要把它用到生产有几个方向可以继续做把记忆换成向量库支持长期检索把调度改成 async 支持并发把工具执行加上审计日志把权限回调接到真实的审批系统。这些扩展都不需要动核心调度逻辑按接口实现即可。模型通道这块统一 Key 的好处在你需要切换模型做对比时会特别明显。同一份 Harness 代码改一个模型 ID 就能从轻量模型切到强模型不用改任何请求逻辑。需要看当前可用模型列表去 https://taotoken.net/models 就行。如果你打算长期跑编码类 Agent 或者多步工具调用任务可以了解下 Coding Plan https://taotoken.net/coding-plan 按用量规划更省心。接入文档在 https://taotoken.net/doc 里面有各语言的调用示例遇到协议细节可以对照。最后留一个实用技巧在core.py的循环里加一个事件回调把每一步的plan、tool_call、result都打出来存成 JSONL 日志。这样每次 Agent 跑完你都能回放整条决策链路调 prompt 和工具描述的时候有据可依。这个日志机制比任何调试器都好用尤其是在模型行为不稳定的时候。