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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 端点是 https://taotoken.net/api 。2.1 获取 API Key 与配置环境变量登录后进入控制台在 API Keys 页面创建一个新 Key。建议按项目维度创建方便后续做用量归因。拿到 Key 之后不要硬编码在代码里用环境变量管理export TAOTOKEN_API_KEYsk-你的实际Key export TAOTOKEN_BASE_URLhttps://taotoken.net/api如果你用的是 Python可以在项目根目录建一个.env文件TAOTOKEN_API_KEYsk-你的实际Key TAOTOKEN_BASE_URLhttps://taotoken.net/api然后在代码里用python-dotenv加载。这样做的好处是当你需要把项目分享给别人或者部署到服务器时只需要替换环境变量不需要改任何代码。2.2 验证 Key 是否可用在写复杂的多 Agent 逻辑之前先用一个最小请求确认通道正常import os from openai import OpenAI client OpenAI( api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL) ) response client.chat.completions.create( modelclaude-3-5-sonnet-20241022, messages[{role: user, content: 回复 OK 两个字母即可}], max_tokens10 ) 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( modelmodel_name, messages[{role: user, content: ping}], max_tokens5 ) 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_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL) ) def call_agent(agent_config, user_message, historyNone): 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( modelagent_config[model], messagesmessages, temperatureagent_config[temperature], max_tokensagent_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, encodingutf-8) as f: json.dump(result, f, ensure_asciiFalse, indent2) 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环境变量如果不需要代理直接 unsetunset 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 结构不符合预期可能是模型名称写错了导致返回了错误信息。排查方法在调用后先打印完整 responseresponse client.chat.completions.create(...) print(response.model_dump_json(indent2))确认返回结构里确实有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/apiKey 用 TaoToken 控制台生成的 KeyModel 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 流程之前先做一轮人工评估。
