企业官网建设流程全解析

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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_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/macOS：

export TAOTOKEN_API_KEY="sk-你的Key"

Windows PowerShell：

$env:TAOTOKEN_API_KEY="sk-你的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, model=None, temperature=0.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, data=data, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {TAOTOKEN_API_KEY}", }, method="POST", ) with urllib.request.urlopen(req, timeout=REQUEST_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_required=False): 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, timeout=30) -> ExecutionResult: tool = self.get(name) if not tool: return ExecutionResult(False, error=f"Tool {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(target=run) th.start() th.join(timeout=timeout) if th.is_alive(): return ExecutionResult(False, error=f"Tool {name} timeout after {timeout}s", duration=time.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, k=10) -> 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(begin=CALL_BEGIN, end=CALL_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(tools=json.dumps(tools, ensure_ascii=False, indent=2)) 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_callback=None): 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_required=False): return self.registry.tool(name, description, parameters, permission_required) def run(self, query, session_id="default") -> 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_ascii=False)}") 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( name="get_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( name="restart_service", description="重启指定的系统服务，属于敏感操作", parameters={ "type": "object", "properties": { "service_name": {"type": "string", "description": "服务名称"} }, "required": ["service_name"], }, permission_required=True, ) 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_call=chat, permission_callback=permission_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_id=sid))

跑起来：

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 以内，主要耗时都在模型请求上。

这里给一个对照表，方便你确认各环节是否正常：

链路跑通后，你可以把chat换成任意模型，只要改DEFAULT_MODEL就行，Harness 代码一行不用动。这就是统一 Key 通道的价值：模型可替换，业务逻辑稳定。

5. 本篇常见报错排查：401、local proxy failed、reading choices 与 OAuth

搭 Harness 的过程中，报错基本集中在模型通道和工具解析两块。我把几个高频错误和排查路径列出来，对照着看。

401 Unauthorized。最常见。原因通常是 Key 没读到、Key 写错、或者请求头格式不对。先确认环境变量：echo $TAOTOKEN_API_KEY（Windows 用echo %TAOTOKEN_API_KEY%）。如果为空，说明没 export 成功。再确认请求头是Authorization: Bearer sk-xxx，Bearer 后面有一个空格。还有一种情况是 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字段是有效模型 ID，messages是[{"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 和工具描述的时候有据可依。这个日志机制比任何调试器都好用，尤其是在模型行为不稳定的时候。