从单Agent到多Agent协同：Harness Engineering工程化实践

这次我们来看一个偏工程向但实战价值很高的方向Harness Engineering。简单说它就是一套“怎么把 AI Agent 从单点能用变成多 Agent 可控、可交付、可维护”的系统工程方法。2026 年这个时间点单 Agent 单任务基本已经不够用了真正落地到企业场景更多是多个 Agent 分工、协作、做复杂流程。而 Harness 正是承载这套协同机制的“底座”它负责 Agent 的启动、调度、权限控制、上下文管理、工具调用、Skill 加载、日志追踪以及主从模式下的任务分发与结果回收。很多人问“harness 和 agent 到底什么区别”。这里先给出一个比较直接的理解Agent 是你的“大脑”负责推理、决策、生成下一步指令Harness 是给这个大脑配的“身体 办公楼”负责让大脑能安全地使用工具、调用外部 Skill、与其他大脑协同、并且全程留痕。没有 Harness 的 Agent 只是一次性对话有了 HarnessAgent 才能成为稳定运行的“业务系统组件”。本文会围绕 Harness Engineering 的底层原理、核心组件、多 Agent 协同实战和 Skill 开发展开带你把一个企业级多 Agent 项目的骨架搭起来。内容包括Harness 与 Agent 的核心边界与关系企业级 Agent Harness 的架构分层与核心组件多 Agent 主从协同模式的设计方式Subagent 如何被当作“特殊 Tool”组织进流程Skill 开发怎么写、怎么挂载、怎么管理版本一套可运行的最小 Harness 工程示例接口 API 与批量任务接入思路资源占用观察、常见问题排查与工程化最佳实践如果你是做 AI 应用开发、企业级系统集成或者已经在用 Claude Code、Codex、OpenCode 等工具做自动化任务但对“多 Agent 协同”和“Skill 体系”还没有形成系统认知这篇文章可以直接收藏。1. 核心能力速览能力项说明项目类型工程方法论 参考实现面向企业级多 Agent 协同系统核心概念Agent智能体、Harness运行框架/底座、Tool工具、Skill可复用技能包主要解决问题多 Agent 调度、可控执行、权限管理、上下文共享、Skill 扩展、任务追踪多 Agent 模式主从模式Supervisor/Worker、Subagent 即 Tool 模式Skill 能力可插拔技能包与 MCP 互补支持团队共享与版本管理启动方式命令行启动 / Python 代码内嵌 / API 服务API 支持推荐暴露标准 HTTP 接口便于外部系统接入批量任务支持任务队列、目录扫描批量处理、失败重试资源占用取决于模型大小与并发数需要实测适合场景企业流程自动化、文档处理、代码生成、数据挖掘、多步业务编排这里说明一下Harness Engineering 不是一个单一的开源项目名而是过去两年里逐步被社区提炼出来的一类工程实践。你可以在 Claude Code、Codex、OpenCode、Coze、Dify 等不同产品里看到它的影子。本文会站在“如果让我从零搭一套企业级多 Agent 系统我会怎么设计”的角度给出通用方法。2. Harness Engineering 到底是什么2.1 从 Agent 到 Harness 的演进最早我们聊 AI Agent基本就是说给大模型一个 System Prompt再加几个工具函数让模型循环调用工具完成任务。这在 Demo 阶段没问题但一旦放到企业环境问题会立刻暴露出来Agent 怎么被安全地触发多个 Agent 之间怎么协作谁来决定哪个 Agent 执行哪步任务Agent 的工具调用权限怎么控制如果 Agent 陷入死循环怎么办一次复杂任务涉及 50 个步骤怎么追踪、审计、回滚沉淀下来的能力怎么复用给其他团队这些问题单靠 Agent 本身解决不了。Harness Engineering 就是把 Agent 放进一个可控运行环境里的工程化方案。Harness 解决的不是“模型聪明不聪明”而是“模型产出能不能被稳定地交付到业务系统里”。2.2 Harness 与 Agent 的区别用一个比较直观的比喻Agent 是员工Harness 是公司 办公系统 管理制度。Agent 负责思考和执行比如“读取这份合同提取关键条款”。Harness 负责让 Agent 能干活并且干得合规分配给它权限、给它工具清单、记录它每一步操作、限制它的资源使用时间、在它失败时安排重试。Tool 是 Agent 的“双手”比如文件读取、数据库查询、HTTP 请求。Skill 是“标准化作业手册”把某类任务的完整操作方法提示词 工具 参数规范 校验流程打包让 Agent 直接复用。所以当你写一个企业级多 Agent 项目时真正的工程量不在 Agent 本身而在 Harness 层。这也是为什么 2026 年社区越来越强调 Harness Engineering因为大家都发现单条 Agent 链路跑通容易多条链路稳定跑一年很难。2.3 从热词里读出的真实需求从社区搜索热词来看大家最关心的几个问题非常集中harness 和 agent 区别上面已经回答了。多 Agent 协作主流模式是什么答案是主从Supervisor/Worker模式最成熟也有对等Peer-to-Peer和协作组模式但企业落地优先主从。将 subagent 视作另类 Tool 进行调用这是一个非常关键的设计观念。既然 Tool 本质上是一个“输入参数、返回结果”的函数那么 Subagent 也一样只是函数内部是另一个 Agent 在跑。把 Subagent 抽象成 Tool可以大大简化 Harness 的设计。Skill 怎么写这是大家最想实操的部分。Skill 核心不是“提示词模板”而是一套结构化的能力包。Skill 和 MCP 有什么区别MCP 解决的是“给 Agent 提供标准化的外部工具通道”Skill 解决的是“把某类任务的执行方法沉淀下来”。两者互补Skill 可以用也可以调用 MCP 工具。3. 企业级 Agent Harness 架构设计3.1 总体分层一个可落地的多 Agent Harness 架构我建议按下面五层设计层级职责关键组件接入层接收外部任务请求HTTP API、消息队列、定时触发器、Webhook编排层决定任务如何分解、哪个 Agent 执行Supervisor Agent、任务路由器、上下文管理器Agent 层具体推理与执行多个 Worker Agent、Subagent工具与技能层提供可复用能力Tool 注册中心、Skill 仓库、MCP Client基础设施层运行环境与稳定性日志追踪、权限控制、重试机制、显存/CPU 监控3.2 核心组件清单一个企业级 Harness 至少需要以下组件第一个是 Agent Registry也就是 Agent 注册表。它负责登记当前系统里有哪些 Agent每个 Agent 的职责、提示词版本、可用工具、允许调用的 Skill、并发上限。不能允许一个 Worker 随便调用所有工具那样会失控。第二个是 Task Scheduler任务调度器。它接收上游任务拆分成子任务按依赖关系决定执行顺序并把子任务分配给合适的 Worker。调度器还需要记录每个子任务的执行状态pending、running、success、failed、retry。第三个是 Context Manager上下文管理器。多 Agent 场景下最棘手的问题就是上下文隔离与共享。主 Agent 和多个子 Agent 之间哪些信息要共享哪些信息必须隔离我建议按任务粒度隔离父任务上下文只下发摘要和必要数据避免把全量上下文传给每个子 Agent否则 Token 消耗会非常夸张。第四个是 Tool Registry 与 Skill Repository。Tool 是原子操作Skill 是复合能力包。开发团队平时主要维护 Skill 仓库。第五个是 Observability可观测性模块。每一步执行都要有 trace_id包括谁调用了谁、Prompt 是什么、模型返回了什么、工具结果如何、耗时多少、Token 消耗多少。没有这套东西多 Agent 系统就是黑盒线上出问题无法排查。3.3 主从模式下 Subagent 作为 Tool 调用前面提到过社区里有一个被高频讨论的设计把 Subagent 当作另类 Tool 来调用。这个思路非常实用。实现上就是这样在 Harness 里Tool 的接口定义是统一的接收一个 JSON 参数返回一个 JSON 结果。Subagent 也遵循这个接口。当你注册一个 Subagent 时它其实就是一个特殊的 Tool{ name: contract_review_agent, type: subagent, description: 调用合同审查子代理输入合同文本或文件路径返回审查意见列表, parameters: { type: object, properties: { file_path: { type: string, description: 合同文件路径 }, focus_points: { type: array, items: { type: string }, description: 需要重点审查的内容 } }, required: [file_path] } }主 Agent 在编排阶段不需要关心这个 Tool 内部是代码还是另一个 Agent它只按照 Tool 调用协议传入参数、获得结果。这种方式的好处是编排逻辑统一主 Agent 无需感知 Subagent 的存在。系统可以通过同一个入口做权限控制、审计跟踪。后续把一个 Subagent 替换成普通代码函数对上层无感。扩展性极强新增一个业务 Agent 只需要注册一个新的 Subagent Tool。这个设计应该成为企业级多 Agent 系统的默认姿势。4. 环境准备与前置条件接下来进入可落地的部分。本文给出的参考实现基于 Python核心依赖是 FastAPI、Pydantic 和一个 LLM 客户端接口。4.1 推荐环境以下是一套通用环境检查清单具体版本以你实际安装为准# 操作系统 # Windows 10/11、macOS 12、Ubuntu 20.04 均可 # Python 版本建议使用 3.10 或 3.11 python --version # 包管理 pip install --upgrade pip4.2 核心依赖安装参考实现需要以下依赖pip install fastapi uvicorn pydantic requests openai如果你用的是本地模型例如通过 Ollama、vLLM 或者 LM Studio 暴露的 OpenAI 兼容接口只需要把 base_url 指向本地服务即可。4.3 目录结构设计一个清晰的项目目录结构是多 Agent 系统可维护性的起点。参考结构如下harness-demo/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── harness/ │ │ ├── __init__.py │ │ ├── registry.py # Agent 注册表 │ │ ├── scheduler.py # 任务调度器 │ │ ├── context.py # 上下文管理器 │ │ └── executor.py # 工具/Subagent 执行器 │ ├── agents/ │ │ ├── base.py # BaseAgent 抽象类 │ │ ├── supervisor.py # 主 Agent │ │ └── workers.py # 多个 Worker Agent │ ├── tools/ │ │ ├── registry.py # Tool 注册 │ │ └── subagent_tool.py # Subagent 包装为 Tool │ ├── skills/ │ │ ├── skill_loader.py # Skill 加载器 │ │ └── builtin_skills/ │ └── api/ │ └── routes.py # HTTP 接口 ├── skills/ # 外部 Skill 目录 ├── tests/ └── requirements.txt5. 从零搭建 Agent Harness 核心骨架这一节我们直接写代码。目标不是做一个完整生产系统而是搭一个能跑通的最小 Harness让你理解核心机制。5.1 定义 Tool 与 Subagent 的统一接口首先定义一个基础工具协议所有工具和 Subagent 都实现这个协议from typing import Any, Dict, Optional from pydantic import BaseModel class ToolResult(BaseModel): success: bool data: Optional[Any] None error: Optional[str] None trace_id: Optional[str] None class BaseTool: 所有工具和 Subagent 的统一接口 name: str base_tool description: str async def execute(self, params: Dict[str, Any]) - ToolResult: raise NotImplementedError这里的关键点是Tool 和 Subagent 使用同一个BaseTool基类。这样子 Agent 就可以被包装成一个SubagentTool对上层完全透明。5.2 编写 BaseAgentAgent 的核心职责是“推理 调用工具”。我们用一个基类统一这个流程import json from typing import Dict, Any, List from openai import AsyncOpenAI class BaseAgent: 基础 Agent。 llm_client: OpenAI 兼容客户端。 system_prompt: 系统提示词。 tools: 该 Agent 可用的工具列表。 def __init__( self, name: str, llm_client: AsyncOpenAI, system_prompt: str, tools: List[Any], model: str gpt-4o-mini ): self.name name self.llm_client llm_client self.system_prompt system_prompt self.tools tools self.model model def _build_tool_schemas(self) - List[Dict[str, Any]]: schemas [] for tool in self.tools: schemas.append( { type: function, function: { name: tool.name, description: tool.description, parameters: getattr(tool, parameters, {}), }, } ) return schemas async def run(self, task: str, context: Dict[str, Any] None) - Dict[str, Any]: messages [] if self.system_prompt: messages.append({role: system, content: self.system_prompt}) messages.append({role: user, content: task}) if context: messages.append({role: user, content: f[上下文]\n{json.dumps(context, ensure_asciiFalse)}}) max_rounds 10 final_answer None for _ in range(max_rounds): response await self.llm_client.chat.completions.create( modelself.model, messagesmessages, toolsself._build_tool_schemas(), tool_choiceauto, ) message response.choices[0].message if message.tool_calls: messages.append(message) for tool_call in message.tool_calls: tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments or {}) result await self._execute_tool(tool_name, tool_args) messages.append( { role: tool, tool_call_id: tool_call.id, content: json.dumps(result, ensure_asciiFalse), } ) else: final_answer message.content break return {agent: self.name, answer: final_answer} async def _execute_tool(self, tool_name: str, tool_args: Dict[str, Any]): for tool in self.tools: if tool.name tool_name: result await tool.execute(tool_args) return result.dict() return {success: False, error: ftool not found: {tool_name}}这个基类的设计思路是Agent 不关心工具内部实现只按 schema 调用。每个 Agent 能访问哪些工具在初始化时通过tools参数注入。5.3 将 Subagent 包装为 Tool这是整个 Harness 设计里最精彩的一步。一个 Subagent 本质上就是一个 BaseAgent但它要被包装成 BaseToolfrom typing import Any, Dict class SubagentTool(BaseTool): 将子 Agent 包装成工具。 这样主 Agent 可以像调用普通工具一样调用子 Agent。 def __init__(self, agent: BaseAgent, name: str None, description: str None, parameters: Dict None): self.agent agent self.name name or f{agent.name}_tool self.description description or f调用子代理 {agent.name} 完成子任务 self.parameters parameters or { type: object, properties: { task: {type: string, description: 子任务描述}, context: {type: object, description: 传递给子代理的上下文}, }, required: [task], } async def execute(self, params: Dict[str, Any]) - ToolResult: try: task params.get(task, ) context params.get(context, {}) result await self.agent.run(tasktask, contextcontext) return ToolResult(successTrue, dataresult) except Exception as e: return ToolResult(successFalse, errorstr(e))有了这个SubagentTool主 Agent 调子 Agent 就跟调用普通函数一样编排层得到极大简化。5.4 调度器与执行流程调度器负责把任务分配给合适的 Agent。这里给一个简单的注册表 调度实现from typing import Dict, Any class AgentRegistry: Agent 注册表管理所有 Agent 与工具的注册和查找 def __init__(self): self._agents: Dict[str, BaseAgent] {} self._tools: Dict[str, BaseTool] {} def register_agent(self, agent: BaseAgent): self._agents[agent.name] agent def register_tool(self, tool: BaseTool): self._tools[tool.name] tool def get_agent(self, name: str) - BaseAgent: return self._agents[name] def get_tool(self, name: str) - BaseTool: return self._tools[name] def list_agents(self): return list(self._agents.keys()) def list_tools(self): return list(self._tools.keys())调度逻辑在业务层可以很灵活可以按关键词路由可以按 Agent 描述让模型路由也可以固定写死流程。企业落地前期我建议先写死编排逻辑等流程稳定后再尝试让模型参与路由决策。先确定性再智能性。6. 多 Agent 协同实战合同审查场景下面用一个具体案例把多 Agent 协同串起来合同审查 风险总结。6.1 场景拆解在真实业务里合同审查不是一句话就能完成的。合理分解为多 Agent 协作文件解析 Agent读取合同文件把 PDF/DOCX 转成纯文本。条款提取 Agent分析文本提取付款条款、违约责任、知识产权等关键信息。风险审查 Agent根据提取结果评估风险点并给出风险等级。总结 Agent汇总所有输出生成简明的审查意见。主 Agent 在这里承担“协调者”角色但它不需要自己处理所有步骤而是按顺序调用上述子 Agent。这些子 Agent 对主 Agent 来说就是几个 SubagentTool。6.2 代码实现首先初始化 LLM 客户端和一个主 Agentimport asyncio from openai import AsyncOpenAI client AsyncOpenAI( base_urlhttp://localhost:8000/v1, # 如果使用本地模型 api_keyEMPTY, )注意如果环境里没有本地模型服务可以直接把 base_url 改为 OpenAI 官方兼容接口。生产环境更推荐接入公司内部模型网关。接着注册多个 Worker Agent# worker 1: 文件解析 Agent parse_agent BaseAgent( namefile_parser, llm_clientclient, system_prompt你是一个文件解析助手。根据文件路径读取文件内容并输出纯文本。, tools[file_read_tool], ) # worker 2: 条款提取 Agent extract_agent BaseAgent( nameclause_extractor, llm_clientclient, system_prompt你是一个合同条款提取助手。从合同文本中提取付款、违约、知识产权、保密等关键条款输出结构化 JSON。, tools[], ) # worker 3: 风险审查 Agent risk_agent BaseAgent( namerisk_reviewer, llm_clientclient, system_prompt你是一个法律风险审查助手。根据合同条款分析潜在风险点输出风险等级和整改建议。, tools[], )然后把这些 Worker 包装成 Tool注册给主 Agentparse_tool SubagentTool( agentparse_agent, nameparse_document, description解析合同文件返回纯文本内容, ) extract_tool SubagentTool( agentextract_agent, nameextract_clauses, description提取合同关键条款返回结构化 JSON, ) risk_tool SubagentTool( agentrisk_agent, namereview_risks, description审查合同风险返回风险意见, ) supervisor BaseAgent( namesupervisor, llm_clientclient, system_prompt( 你是一个合同审查主管。你的工作流程是 第一步调用 parse_document 解析文件 第二步调用 extract_clauses 提取条款 第三步调用 review_risks 进行风险审查 最后汇总输出审查报告。 ), tools[parse_tool, extract_tool, risk_tool], )执行主 Agentasync def main(): result await supervisor.run(task请审查 contracts/sample_contract.pdf) print(result[answer]) asyncio.run(main())这个流程没有复杂的图编排但已经是一个完整的多 Agent 协同链路。它验证了核心设计主 Agent SubagentTool 统一执行接口。6.3 无工具场景下的 Subagent注意上面的 extract_agent 和 risk_agent 没有绑定任何工具它们本质上就是用不同 System Prompt 驱动的子 Agent。这说明一个容易被忽略的事实多 Agent 协同的价值不只在“多工具调用”更在于“多个专业上下文隔离”。每个 Worker 可以使用不同的 System Prompt、不同的模型参数、不同的上下文窗口策略而主 Agent 只需要负责调度。这与“单 Agent 长 Prompt”相比模块化程度高得多。7. Skill 开发实战将能力沉淀为可复用包7.1 Skill 与 MCP 的分工先回答热词里的高频困惑Skill 和 MCP 到底什么关系MCPModel Context Protocol解决的是“模型如何标准化地访问外部工具和数据源”。它定义了一套协议Model 通过 MCP Client 连接 MCP ServerMCP Server 暴露 Tool、Resource、Prompt。Skill 则是一份“可复用的任务执行方法包”。它通常包含任务说明、执行步骤、Prompt 模板、所需工具列表、参数定义、校验规则。Skill 可以调用 MCP 工具也可以调用普通本地函数还可以编排多个子步骤。最简单的一句话区分MCP 是“工具通道”Skill 是“作业手册”。7.2 Skill 包的标准结构推荐使用目录 配置文件的组织方式skills/ └── contract_summary/ ├── SKILL.md # 是什么、什么时候用、怎么用 ├── instructions.md # 给 Agent 的详细系统提示词 ├── tools.json # 需要挂载的工具列表 ├── params_schema.json # 输入参数定义 ├── examples/ │ └── sample_input.json └── validate.py # 可选输出校验脚本SKILL.md 示例内容# Contract Summary Skill ## 描述 从合同文本中生成结构化的摘要包括合同主体、标的、金额、付款方式、有效期、违约责任等核心信息。 ## 适用场景 - 合同归档时快速生成摘要 - 合同比对前的预处理 - 风控系统前期数据提取 ## 前置条件 需要文件解析工具和 LLM 客户端。 ## 使用步骤 1. 接收文本或文件路径。 2. 调用文件解析工具获取纯文本。 3. 按 instructions.md 中的提示词提取摘要。 4. 输出 JSON结构参考 params_schema.json。7.3 Skill Loader 实现Skill 加载器的职责是读取目录下的配置文件把 Skill 注册成一个可被 Agent 调用的 Tool。import json from pathlib import Path from typing import Dict, Any class SkillLoader: def __init__(self, skills_dir: str ./skills): self.skills_dir Path(skills_dir) def load_all(self) - Dict[str, Dict[str, Any]]: skills {} for skill_dir in self.skills_dir.iterdir(): if skill_dir.is_dir(): skill self._load_single(skill_dir) if skill: skills[skill[name]] skill return skills def _load_single(self, skill_dir: Path) - Dict[str, Any] | None: skill_md skill_dir / SKILL.md params_schema_file skill_dir / params_schema.json if not skill_md.exists(): return None params {} if params_schema_file.exists(): with open(params_schema_file, r, encodingutf-8) as f: params json.load(f) return { name: skill_dir.name, description: skill_md.read_text(encodingutf-8)[:200], params_schema: params, path: str(skill_dir), }7.4 Skill 如何被 Agent 使用在 Harness 内部的推荐做法是Skill 加载后包装成一个SkillTool注册进 Agent 的工具列表。class SkillTool(BaseTool): def __init__(self, skill: dict, llm_client: Any): self.name skill[name] self.description skill[description] self.parameters skill[params_schema] self.llm_client llm_client async def execute(self, params: dict) - ToolResult: # 读取 skill 目录下的指令文件 # 组合提示词调用 LLM # 返回结果 try: # 这里根据你的 skill 格式动态加载指令 system_prompt f你正在执行 Skill: {self.name}\n请依据技能包说明处理用户请求。 response await self.llm_client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: system_prompt}, {role: user, content: json.dumps(params, ensure_asciiFalse)}, ], ) return ToolResult(successTrue, dataresponse.choices[0].message.content) except Exception as e: return ToolResult(successFalse, errorstr(e))从企业工程化的角度看Skill 的真正价值是“沉淀与复用”。每个团队把常用的任务执行方式固化为 Skill 包通过 Git 仓库管理不同项目共享同一套技能库。这就是 Harness Engineering 强调的“可控”与“沉淀”。8. 接口 API 与批量任务8.1 提供 HTTP 接口企业系统接入多 Agent 能力通常不会直接调用 Python 类而是通过 HTTP API。用 FastAPI 封装一层即可。from fastapi import FastAPI from pydantic import BaseModel app FastAPI(titleAgent Harness API) class TaskRequest(BaseModel): task: str context: dict {} class TaskResponse(BaseModel): status: str result: str trace_id: str app.post(/v1/agents/supervisor/run, response_modelTaskResponse) async def run_supervisor(req: TaskRequest): result await supervisor.run(taskreq.task, contextreq.context) return TaskResponse(statussuccess, resultresult[answer], trace_idtrace-123)启动服务uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload8.2 curl 调用示例curl -X POST http://127.0.0.1:8000/v1/agents/supervisor/run \ -H Content-Type: application/json \ -d { task: 请审查 contracts/sample_contract.pdf, context: { company: 某科技有限公司, risk_threshold: medium } }8.3 批量任务处理企业场景下经常需要一次性处理大量文件。推荐的做法是目录扫描 任务队列 结果落盘。import asyncio import json from pathlib import Path async def batch_process(input_dir: str, output_dir: str): input_path Path(input_dir) output_path Path(output_dir) output_path.mkdir(parentsTrue, exist_okTrue) tasks [] for file in input_path.glob(*.pdf): task supervisor.run(taskf请审查 contracts/{file.name}) tasks.append((file.name, task)) results {} for file_name, task in tasks: try: result await task results[file_name] {status: success, output: result[answer]} except Exception as e: results[file_name] {status: failed, error: str(e)} with open(output_path / batch_results.json, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) asyncio.run(batch_process(contracts, outputs))注意生产环境的批量任务不能像上面这样直接并发全部任务。你需要用队列控制并发数一般先设为 12 个并发跑稳定后再逐步加大。还要设计失败重试与中间状态持久化否则进程一崩整个批次就断了。9. 资源占用与性能观察9.1 主要性能敏感点多 Agent 系统的资源占用与传统 Web 服务完全不同核心消耗集中在LLM 推理每个 Agent 的每一步推理都在消耗计算资源。本地模型看显存API 模型看请求数与 Token 量。上下文累积多 Agent 会放大 Token 消耗。主 Agent 的上下文越长成本越高。工具执行文件解析、PDF OCR、数据库查询等操作会消耗 CPU 和内存。并发数量同时运行的 Agent 数量直接决定总资源占用。9.2 如何观察资源占用启动服务后建议通过以下方式观察# Linux/Mac 查看进程资源 top -p $(pgrep -f uvicorn) # 查看 GPU 显存使用适用于本地模型 nvidia-smi如果你用的是本地模型可以在系统提示词里要求 Agent 每次返回结果时附带“输出长度”或“花费时间”但更可靠的是在 Harness 层做统计。每完成一次 Agent 调用记录耗时、Token 数、成功失败状态。9.3 降低资源占用的策略第一控制上下文传递。主 Agent 给 Subagent 传递的 context 要精简不要直接把全量历史对话都传过去只传结构化关键信息。第二使用轻量模型做路由和提取使用大模型做关键生成。不同 Agent 可以配置不同模型。比如条款提取 Agent 用 7B 模型就够了风险审查 Agent 可能需要更强的模型。第三设置最大迭代轮数。BaseAgent 中 max_rounds 参数非常重要避免 Agent 陷入工具调用死循环。生产环境建议设置 610 轮上限。第四批处理时控制并发必要时加入流式输出避免一次请求长时间占用大块计算资源。10. 常见问题与排查方法问题现象可能原因排查方式解决方案Agent 反复调用同一个工具输出不收敛max_rounds 设置过大模型陷入死循环查看日志中工具调用轮次降低 max_rounds检查提示词是否让 Agent 明确知道何时停止子 Agent 返回结果为空子 Agent 上下文被截断或参数传递错误检查 SubagentTool 传入的 task 与 context输出入参日志确认子 Agent 收到正确内容调用 API 超时LLM 推理时间长或模型负载高检查 API 网关监控和模型推理队列增加超时时间或改用异步任务 轮询Skill 加载失败目录结构不符合约定params_schema 解析失败查看 SkillLoader 日志检查 SKILL.md 和 params_schema.json 是否存在JSON 是否合法本地模型显存不足并发 Agent 过多查看 nvidia-smi 显存占用降低并发数或改用更小参数模型端口被占用之前服务未退出或其他程序占用了 8000 端口查看端口占用情况更换端口或结束残留进程批量任务中途失败单条任务异常导致进程退出查看 batch_results.json 中失败状态增加 try/except、失败重试、断点续跑多 Agent 上下文串扰上下文管理器未做好隔离检查子 Agent 收到的 context 是否包含无关数据明确按任务隔离上下文只传必要字段11. 最佳实践与使用建议多 Agent 系统不是一上来就搭复杂架构。建议按以下节奏推进11.1 从单 Agent 跑通开始先不搞多 Agent把主流程用单 Agent 跑通确认每个工具的返回格式、模型响应质量、失败场景。多 Agent 协同的复杂性比单 Agent 高一个量级先不要叠加复杂度。11.2 用确定性编排替代自由路由在多 Agent 落地初期不要用“让大模型决定下一个调用谁”这种自由编排。建议先用代码写死流程第一步做什么第二步做什么。这对排查问题和控制成本都有好处。模型自由路由适合流程探索阶段不适合生产环境。11.3 沉淀 Skill 前先问自己三个问题不是所有任务都值得做成 Skill。值得沉淀的任务通常满足三个条件重复出现频率高。执行步骤相对固定。结果需要标准化输出。如果一个任务每次的执行方法都不一样做成 Skill 反而增加维护成本。11.4 日志追踪必须从第一天做多 Agent 系统一旦上线你一定会遇到“用户说结果不对但不知道是哪个 Agent 的问题”。所以从第一天就要在每个环节埋 trace_id记录每个 Agent 的输入、输出、耗时、Token 数。没有这套观测体系多 Agent 系统无法达到企业级稳定要求。11.5 安全与合规边界多 Agent 系统通常需要访问文档、数据库、业务系统权限控制比单 Agent 更重要。要遵循最小权限原则每个 Agent 只授予完成自身任务所需的权限不要所有 Agent 共享一个超级账号。涉及合同、财务、个人信息等敏感数据时需要明确数据使用授权和审计要求不能因为 Agent 是“AI”就放松合规要求。任何自动化操作都应该能在事后进行完整审计和回溯。12. 总结与下一步Harness Engineering 解决的核心问题不是单点 Agent 的“智能程度”而是多 Agent 在真实业务环境中的“可控与可维护”。从本文的实战可以看到把 Subagent 包装成 Tool并用统一的 Harness 框架管理调度、上下文、Skill 和日志能让多 Agent 系统的复杂度大幅下降。如果你想从零开始实践建议按这个顺序推进第一步搭建一个最小的 BaseAgent能循环调用工具并返回结果。第二步实现 SubagentTool把第二个 Agent 包装成第一个 Agent 的工具。第三步给系统加上 FastAPI 接口跑通一个 HTTP 调用。第四步把常用的任务方法沉淀为第一个 Skill 包。第五步再考虑批量任务、并发控制、权限管理和监控告警。这个领域目前还在快速发展中但有一点可以确定掌握 Harness 和 Skill 的设计思路比学会某个特定框架更重要。建议你先跑通本文的最小示例再结合自己团队的业务场景做扩展。后面可以继续关注 Skill 与 MCP 的深度结合、多 Agent 上下文传输优化、以及模型路由策略演进等方向。