1. 从单 Agent 到多 Agent 协作：为什么需要 Harness Engineering

单个 LLM Agent 能写代码、能查资料、能做数据分析，但一旦任务变成“先调研竞品、再写技术方案、最后生成可运行 Demo”这种跨阶段、跨能力的复合任务，单 Agent 就会暴露三个硬伤：上下文窗口被塞满后开始丢信息、串行执行导致整体耗时线性叠加、单点失败后整个流程直接中断。我试过让一个 Agent 从头到尾跑完一个包含 6 个子任务的项目初始化流程，结果在第 4 步时它已经忘记了第 1 步定义的接口规范。

群体智能的思路不是让一个 Agent 变得更强，而是让多个角色明确、能力互补的 Agent 通过局部交互完成全局任务。这就像蚁群找食物：没有中央指挥官告诉每只蚂蚁该走哪条路，但通过信息素的正反馈机制，整个群体能涌现出最短路径。映射到 AI Agent 场景，就是任务分配、结果聚合、冲突消解三个环节的工程化落地。

Harness Engineering 在这里扮演的角色，是把“多 Agent 协作”从论文里的概念变成你本地能跑起来的代码。它需要解决四个工程问题：Agent 角色怎么定义、任务怎么拆分和路由、多个模型的 API 调用怎么统一管理、执行结果怎么验证和聚合。其中“统一管理”这一环，如果每个 Agent 都去维护一套独立的 Key 和 Base URL，配置成本会迅速失控。TaoToken 在这里的价值就是提供一个统一的 API 通道，让不同角色的 Agent 调用不同模型时，只需要一套 Key 和一个 Base URL。

这篇文章不会停留在概念层面。我会给出可复制的 Agent 角色配置、协作流程 JSON、端到端验证脚本，帮你在本地跑通一次多 Agent 协同任务，并观察不同模型在同一个子任务上的输出差异。

2. TaoToken 前置准备：统一 Key 与多模型通道配置

在开始写 Agent 编排代码之前，先把 API 通道打通。多 Agent 协作场景下，你大概率会用到至少两种模型：一个负责规划和任务分解（比如 Claude 系列），一个负责执行和工具调用（比如 GPT 系列或国产模型）。如果每个模型都单独申请 Key、单独配置环境变量，代码里会充斥大量 if-else 分支。

TaoToken 的做法是提供一个兼容 OpenAI 格式的 API 端点，你只需要一个 Key，就能在请求里通过model字段切换不同模型。官网地址是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 端点是 https://taotoken.net/api 。

2.1 获取 API Key 与配置环境变量

登录后进入控制台，在 API Keys 页面创建一个新 Key。建议按项目维度创建，方便后续做用量归因。拿到 Key 之后，不要硬编码在代码里，用环境变量管理：

export TAOTOKEN_API_KEY="sk-你的实际Key" export TAOTOKEN_BASE_URL="https://taotoken.net/api"

如果你用的是 Python，可以在项目根目录建一个.env文件：

TAOTOKEN_API_KEY=sk-你的实际Key TAOTOKEN_BASE_URL=https://taotoken.net/api

然后在代码里用python-dotenv加载。这样做的好处是，当你需要把项目分享给别人或者部署到服务器时，只需要替换环境变量，不需要改任何代码。

2.2 验证 Key 是否可用

在写复杂的多 Agent 逻辑之前，先用一个最小请求确认通道正常：

import os from openai import OpenAI client = OpenAI( api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL") ) response = client.chat.completions.create( model="claude-3-5-sonnet-20241022", messages=[{"role": "user", "content": "回复 OK 两个字母即可"}], max_tokens=10 ) print(response.choices[0].message.content)

如果输出OK，说明 Key 和 Base URL 配置正确。如果报 401，检查 Key 是否复制完整、是否有多余空格。如果报local proxy failed，检查你的网络环境是否直接访问了https://taotoken.net/api，不要经过任何本地代理配置。

2.3 多模型可用性检查

多 Agent 协作的前提是你能在同一个通道里调用不同模型。写一个简单的探测脚本：

models_to_test = [ "claude-3-5-sonnet-20241022", "gpt-4o", "gpt-4o-mini" ] for model_name in models_to_test: try: resp = client.chat.completions.create( model=model_name, messages=[{"role": "user", "content": "ping"}], max_tokens=5 ) print(f"{model_name}: OK") except Exception as e: print(f"{model_name}: FAILED - {e}")

这个脚本会告诉你哪些模型在当前 Key 下可用。实测下来，规划类任务用 Claude 系列在任务分解的粒度控制上更稳，执行类任务用 GPT-4o-mini 在成本和速度上更划算。

3. 可复制的多 Agent 协作配置：角色、流程与统一调用

这一节给出完整的配置文件。整个多 Agent 系统由三个角色组成：Planner（规划者）、Executor（执行者）、Reviewer（审核者）。Planner 负责把用户任务拆成子任务列表，Executor 负责逐个执行，Reviewer 负责检查执行结果并决定是否需要重试或调整。

3.1 Agent 角色定义 JSON

把角色定义写成 JSON，方便后续用代码加载和动态调整：

{ "agents": [ { "name": "planner", "model": "claude-3-5-sonnet-20241022", "system_prompt": "你是一个任务规划专家。用户会给你一个复杂任务，你需要将其拆解为 3-5 个可独立执行的子任务。每个子任务必须包含：任务描述、预期输出格式、依赖的前置子任务编号（如果没有则为空数组）。只输出 JSON 数组，不要输出其他内容。", "temperature": 0.3, "max_tokens": 2000 }, { "name": "executor", "model": "gpt-4o-mini", "system_prompt": "你是一个任务执行专家。你会收到一个具体的子任务描述和预期输出格式，请严格按照要求完成并输出结果。如果任务涉及代码，请给出完整可运行的代码。", "temperature": 0.5, "max_tokens": 3000 }, { "name": "reviewer", "model": "claude-3-5-sonnet-20241022", "system_prompt": "你是一个质量审核专家。你会收到一个子任务的原始要求、执行者的输出结果。请判断输出是否满足要求，输出 JSON：{\"pass\": true/false, \"reason\": \"判断理由\", \"suggestion\": \"如果不通过，给出修改建议\"}。", "temperature": 0.2, "max_tokens": 1000 } ] }

三个角色使用不同的模型和温度参数：Planner 需要结构化输出，温度调低；Executor 需要一定创造性，温度适中；Reviewer 需要严格判断，温度最低。

3.2 协作流程配置

协作流程定义任务如何在三个角色之间流转：

{ "workflow": { "entry": "planner", "max_retry": 2, "steps": [ { "from": "planner", "to": "executor", "condition": "always", "data_mapping": { "subtask": "planner.output[i]" } }, { "from": "executor", "to": "reviewer", "condition": "always", "data_mapping": { "original_requirement": "planner.output[i]", "execution_result": "executor.output" } }, { "from": "reviewer", "to": "executor", "condition": "reviewer.output.pass == false && retry_count < max_retry", "data_mapping": { "subtask": "planner.output[i]", "previous_result": "executor.output", "suggestion": "reviewer.output.suggestion" } }, { "from": "reviewer", "to": "aggregator", "condition": "reviewer.output.pass == true || retry_count >= max_retry", "data_mapping": { "final_result": "executor.output" } } ] } }

这个流程的核心逻辑是：Planner 输出子任务列表后，Executor 逐个执行，Reviewer 逐个审核。审核不通过时，把修改建议回传给 Executor 重试，最多重试 2 次。超过重试次数后，无论是否通过都进入聚合阶段，但会在最终结果里标记该子任务的状态。

3.3 统一调用封装

把 TaoToken 的调用封装成一个统一的函数，所有 Agent 都通过这个函数发请求：

import os import json from openai import OpenAI client = OpenAI( api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL") ) def call_agent(agent_config, user_message, history=None): messages = [ {"role": "system", "content": agent_config["system_prompt"]} ] if history: messages.extend(history) messages.append({"role": "user", "content": user_message}) response = client.chat.completions.create( model=agent_config["model"], messages=messages, temperature=agent_config["temperature"], max_tokens=agent_config["max_tokens"] ) return response.choices[0].message.content

这个函数接收 agent_config 和用户消息，返回模型输出。所有 Agent 共用同一个 client 实例，Key 和 Base URL 只在初始化时读取一次。

3.4 编排主循环

把上面的配置串起来：

def run_workflow(user_task, agents_config, workflow_config): agent_map = {a["name"]: a for a in agents_config["agents"]} planner_output = call_agent( agent_map["planner"], f"请拆解以下任务：{user_task}" ) try: subtasks = json.loads(planner_output) except json.JSONDecodeError: return {"error": "Planner 输出不是合法 JSON", "raw": planner_output} results = [] for idx, subtask in enumerate(subtasks): retry_count = 0 execution_result = None review_result = None while retry_count <= workflow_config["workflow"]["max_retry"]: exec_prompt = f"子任务：{subtask['description']}\n预期输出格式：{subtask['output_format']}" if review_result and not review_result.get("pass"): exec_prompt += f"\n上次审核建议：{review_result['suggestion']}" execution_result = call_agent(agent_map["executor"], exec_prompt) review_prompt = f"原始要求：{subtask['description']}\n执行结果：{execution_result}" review_raw = call_agent(agent_map["reviewer"], review_prompt) try: review_result = json.loads(review_raw) except json.JSONDecodeError: review_result = {"pass": False, "reason": "Reviewer 输出解析失败", "suggestion": "请重新输出"} if review_result.get("pass"): break retry_count += 1 results.append({ "subtask_index": idx, "subtask": subtask, "result": execution_result, "review": review_result, "retry_count": retry_count }) return {"subtasks": results}

这段代码就是多 Agent 协作的最小可运行版本。Planner 拆任务，Executor 执行，Reviewer 审核，不通过就带着建议重试。

4. 端到端验证：跑通一次多 Agent 协同任务

配置写好了，现在用一个真实任务验证整条链路。任务设定为：“为一个 Python 命令行工具项目生成项目结构说明、核心模块代码、以及单元测试文件。”

4.1 执行脚本

if __name__ == "__main__": with open("agents_config.json", "r") as f: agents_config = json.load(f) with open("workflow_config.json", "r") as f: workflow_config = json.load(f) task = "为一个 Python 命令行工具项目生成项目结构说明、核心模块代码、以及单元测试文件。" result = run_workflow(task, agents_config, workflow_config) with open("workflow_result.json", "w", encoding="utf-8") as f: json.dump(result, f, ensure_ascii=False, indent=2) print(f"共完成 {len(result['subtasks'])} 个子任务") for item in result["subtasks"]: status = "通过" if item["review"].get("pass") else "未通过" print(f"子任务 {item['subtask_index']}: {status}, 重试 {item['retry_count']} 次")

4.2 预期输出与观察点

运行后你会看到类似输出：

共完成 3 个子任务 子任务 0: 通过, 重试 0 次 子任务 1: 通过, 重试 1 次 子任务 2: 通过, 重试 0 次

打开workflow_result.json，重点观察三个地方：

第一，Planner 拆出的子任务粒度是否合理。如果某个子任务描述过于宽泛（比如“实现整个项目”），说明 Planner 的 system prompt 需要加约束。

第二，Reviewer 在什么情况下判定不通过。如果重试次数集中在某个子任务上，说明 Executor 对该类任务的处理能力不足，可以考虑换模型或调整 prompt。

第三，不同模型在同一个子任务上的输出差异。你可以把 Executor 的模型从gpt-4o-mini换成claude-3-5-sonnet-20241022，重新跑一次，对比代码风格和边界条件处理。

4.3 观察输出差异的对比方法

写一个简单的对比脚本：

def compare_models(subtask, models): results = {} for model_name in models: agent_config = { "model": model_name, "system_prompt": "你是一个任务执行专家。请完成以下子任务。", "temperature": 0.5, "max_tokens": 3000 } output = call_agent(agent_config, subtask["description"]) results[model_name] = output return results

用同一个子任务分别调用不同模型，把输出并排保存。实测下来，Claude 系列在代码注释和文档生成上更细致，GPT-4o-mini 在速度上明显更快但偶尔会省略边界条件处理。这个差异观察本身就是群体智能“多样性”的价值来源——不同 Agent 的输出差异，正是冲突消解和结果聚合的输入。

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

多 Agent 协作跑通的过程中，最容易卡在 API 调用环节。下面按报错类型逐个排查。

5.1 401 Unauthorized

报错信息：

openai.AuthenticationError: Error code: 401 - {'error': {'message': 'Invalid API key', 'type': 'invalid_request_error'}}

排查步骤：检查TAOTOKEN_API_KEY环境变量是否设置成功，在 Python 里执行print(os.getenv("TAOTOKEN_API_KEY"))确认输出不是None。如果 Key 是从控制台复制的，注意不要带前后空格。如果 Key 之前能用现在突然 401，去控制台确认 Key 是否被禁用或额度是否耗尽。

5.2 local proxy failed

报错信息：

openai.APIConnectionError: Connection error: local proxy failed

这个报错通常是因为你的运行环境里配置了本地代理，但代理没有正常运行。检查HTTP_PROXY和HTTPS_PROXY环境变量，如果不需要代理，直接 unset：

unset HTTP_PROXY unset HTTPS_PROXY

然后在代码里显式指定 base_url 为https://taotoken.net/api，不要走任何中间层。

5.3 reading choices 相关报错

报错信息：

KeyError: 'choices' 或 TypeError: 'NoneType' object is not subscriptable

这种报错通常发生在response.choices[0]这一行。原因是 API 返回的 JSON 结构不符合预期，可能是模型名称写错了导致返回了错误信息。排查方法：在调用后先打印完整 response：

response = client.chat.completions.create(...) print(response.model_dump_json(indent=2))

确认返回结构里确实有choices字段。如果返回的是{"error": {...}}，说明请求本身有问题，检查 model 名称是否在可用列表里。

5.4 OAuth 相关报错

如果你在 Claude Code 或类似工具里配置 TaoToken，可能会遇到 OAuth 报错。这类工具通常需要三件套：Base URL、API Key、Model ID。以 Claude Code 为例，配置文件里需要写全：

{ "base_url": "https://taotoken.net/api", "api_key": "sk-你的实际Key", "model": "claude-3-5-sonnet-20241022" }

三个字段缺一不可。如果只填了 Base URL 和 Key，但 Model ID 留空或写错，就会报 OAuth 或模型不存在的错误。Cline MCP 和 Codex 的 auth.json 也是同样的逻辑，Base URL 指向https://taotoken.net/api，Key 用 TaoToken 控制台生成的 Key，Model ID 写你实际要调用的模型名称。

5.5 重试逻辑导致的死循环

多 Agent 协作里如果 Reviewer 一直判定不通过，而 Executor 每次重试都没有实质性改进，就会陷入死循环。上面的配置里用max_retry做了硬限制，但更好的做法是在 Reviewer 的 prompt 里加一条规则：如果连续两次重试的输出相似度超过 90%，直接判定为“无法通过自动审核”，标记为人工介入。相似度可以用简单的 difflib 计算：

import difflib def similarity(a, b): return difflib.SequenceMatcher(None, a, b).ratio()

在重试循环里记录上一次的输出，如果相似度超过阈值，直接跳出循环并标记状态。

6. 从验证到落地：多 Agent 协作的工程化建议

跑通一次多 Agent 协同任务只是起点。要把它变成日常可用的工具，还有几个工程细节值得注意。

第一，把 Agent 配置和 workflow 配置从代码里抽离成独立 JSON 文件，这样调整角色 prompt 或换模型时不需要改代码。上面的例子已经这么做了，你可以进一步把配置放到数据库或配置中心，支持运行时热更新。

第二，给每个 Agent 的调用加上日志记录。记录请求时间、模型名称、token 消耗、响应时间、是否重试。这些数据积累起来之后，你可以分析哪个子任务最耗时、哪个模型在哪个角色上表现最好。TaoToken 控制台本身有用量统计，但细粒度的 per-agent 日志还是需要自己在代码里埋点。

第三，冲突消解不要只依赖 Reviewer 的单点判断。当多个 Executor 对同一个子任务给出不同结果时，可以让 Reviewer 做一次“多结果对比审核”，把多个候选结果一起传给它，让它选择最优的或给出融合建议。这比单结果审核更接近群体智能的“多样性聚合”思路。

第四，任务分配环节可以引入简单的负载感知。上面的例子是串行执行子任务，如果子任务之间没有依赖关系，可以并行调用多个 Executor。但并行时要注意 API 的速率限制，建议在代码里加一个简单的信号量控制并发数。

如果你想把多 Agent 协作用到长期编码任务或 Agent 工作流里，可以了解 TaoToken 的 Coding Plan，它针对高频调用场景做了通道优化。需要查看可用模型列表和详细接入文档的话，从 API Keys 页面和接入文档入口进去看最新说明。模型对话入口可以用来快速验证单个模型的输出质量，在正式接入到多 Agent 流程之前先做一轮人工评估。