这次我们来看一个偏工程向但实战价值很高的方向:Harness Engineering。
简单说,它就是一套“怎么把 AI Agent 从单点能用,变成多 Agent 可控、可交付、可维护”的系统工程方法。2026 年这个时间点,单 Agent 单任务基本已经不够用了,真正落地到企业场景,更多是多个 Agent 分工、协作、做复杂流程。而 Harness 正是承载这套协同机制的“底座”:它负责 Agent 的启动、调度、权限控制、上下文管理、工具调用、Skill 加载、日志追踪,以及主从模式下的任务分发与结果回收。
很多人问“harness 和 agent 到底什么区别”。这里先给出一个比较直接的理解:Agent 是你的“大脑”,负责推理、决策、生成下一步指令;Harness 是给这个大脑配的“身体 + 办公楼”,负责让大脑能安全地使用工具、调用外部 Skill、与其他大脑协同、并且全程留痕。没有 Harness 的 Agent 只是一次性对话,有了 Harness,Agent 才能成为稳定运行的“业务系统组件”。
本文会围绕 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 编写 BaseAgent
Agent 的核心职责是“推理 + 调用工具”。我们用一个基类统一这个流程:
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_ascii=False)}"}) max_rounds = 10 final_answer = None for _ in range(max_rounds): response = await self.llm_client.chat.completions.create( model=self.model, messages=messages, tools=self._build_tool_schemas(), tool_choice="auto", ) 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_ascii=False), } ) 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": f"tool not found: {tool_name}"}这个基类的设计思路是:Agent 不关心工具内部实现,只按 schema 调用。每个 Agent 能访问哪些工具,在初始化时通过tools参数注入。
5.3 将 Subagent 包装为 Tool
这是整个 Harness 设计里最精彩的一步。一个 Subagent 本质上就是一个 BaseAgent,但它要被包装成 BaseTool:
from 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(task=task, context=context) return ToolResult(success=True, data=result) except Exception as e: return ToolResult(success=False, error=str(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 客户端和一个主 Agent:
import asyncio from openai import AsyncOpenAI client = AsyncOpenAI( base_url="http://localhost:8000/v1", # 如果使用本地模型 api_key="EMPTY", )注意:如果环境里没有本地模型服务,可以直接把 base_url 改为 OpenAI 官方兼容接口。生产环境更推荐接入公司内部模型网关。
接着注册多个 Worker Agent:
# worker 1: 文件解析 Agent parse_agent = BaseAgent( name="file_parser", llm_client=client, system_prompt="你是一个文件解析助手。根据文件路径读取文件内容并输出纯文本。", tools=[file_read_tool], ) # worker 2: 条款提取 Agent extract_agent = BaseAgent( name="clause_extractor", llm_client=client, system_prompt="你是一个合同条款提取助手。从合同文本中提取付款、违约、知识产权、保密等关键条款,输出结构化 JSON。", tools=[], ) # worker 3: 风险审查 Agent risk_agent = BaseAgent( name="risk_reviewer", llm_client=client, system_prompt="你是一个法律风险审查助手。根据合同条款,分析潜在风险点,输出风险等级和整改建议。", tools=[], )然后把这些 Worker 包装成 Tool,注册给主 Agent:
parse_tool = SubagentTool( agent=parse_agent, name="parse_document", description="解析合同文件,返回纯文本内容", ) extract_tool = SubagentTool( agent=extract_agent, name="extract_clauses", description="提取合同关键条款,返回结构化 JSON", ) risk_tool = SubagentTool( agent=risk_agent, name="review_risks", description="审查合同风险,返回风险意见", ) supervisor = BaseAgent( name="supervisor", llm_client=client, system_prompt=( "你是一个合同审查主管。你的工作流程是:" "第一步,调用 parse_document 解析文件;" "第二步,调用 extract_clauses 提取条款;" "第三步,调用 review_risks 进行风险审查;" "最后,汇总输出审查报告。" ), tools=[parse_tool, extract_tool, risk_tool], )执行主 Agent:
async 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 到底什么关系?
MCP(Model Context Protocol)解决的是“模型如何标准化地访问外部工具和数据源”。它定义了一套协议:Model 通过 MCP Client 连接 MCP Server,MCP 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", encoding="utf-8") as f: params = json.load(f) return { "name": skill_dir.name, "description": skill_md.read_text(encoding="utf-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( model="gpt-4o-mini", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": json.dumps(params, ensure_ascii=False)}, ], ) return ToolResult(success=True, data=response.choices[0].message.content) except Exception as e: return ToolResult(success=False, error=str(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(title="Agent 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_model=TaskResponse) async def run_supervisor(req: TaskRequest): result = await supervisor.run(task=req.task, context=req.context) return TaskResponse(status="success", result=result["answer"], trace_id="trace-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(parents=True, exist_ok=True) tasks = [] for file in input_path.glob("*.pdf"): task = supervisor.run(task=f"请审查 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", encoding="utf-8") as f: json.dump(results, f, ensure_ascii=False, indent=2) asyncio.run(batch_process("contracts", "outputs"))注意:生产环境的批量任务不能像上面这样直接并发全部任务。你需要用队列控制并发数,一般先设为 1~2 个并发,跑稳定后再逐步加大。还要设计失败重试与中间状态持久化,否则进程一崩,整个批次就断了。
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 陷入工具调用死循环。生产环境建议设置 6~10 轮上限。
第四,批处理时控制并发,必要时加入流式输出,避免一次请求长时间占用大块计算资源。
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 上下文传输优化、以及模型路由策略演进等方向。