如果你所在团队正在尝试用多个 AI Agent 协作完成真实的研发任务那么大概率会经历这样的场景需求拆解 Agent 给出了方案编码 Agent 生成了代码审查 Agent 却说代码存在越权调用测试 Agent 又在沙箱里把内存吃满。整个流水线跑下来每个 Agent 好像都“工作”了但最终交付物却很难直接信任。问题的根源并不在于某个模型能力不够而在于我们缺少一套把多个 Agent 约束在企业工程规范之内的框架。2026 年AI Agent 开发已经从“单点实验”进入“工程化落地”阶段Harness Engineering 正是解决多 Agent 协同、沙箱隔离、技能复用、人工审批和全链路审计这套问题的关键思路。本文将围绕一个“企业级多 Agent 协同项目”的完整设计和实现展开如何用 Harness 思路管理多个 Agent如何为 Agent 的执行环境加装沙箱如何设计可复用、可版本化、可持续进化的 Skill 技能体系以及如何在关键节点引入人工介入。适合正在做 AI Agent 工程化的后端开发者、算法同学和技术负责人阅读文章末尾也会给出常见问题和上线建议。1. 什么是 Harness Engineering多 Agent 项目的“缰绳”1.1 多 Agent 协同为什么容易失控先看一个多智能体系统Multi-Agent SystemMAS的典型定义用 prompt 和 workflow 把多个角色、多种工具的 Agent 组织起来共同完成一个复杂目标。听起来很理想但一旦真正跑起来问题会非常具体。上下文会漂移。不同 Agent 对同一个需求的理解不一致导致方案、代码、测试各说各话。权限会放大。每个 Agent 都有调用文件、网络、命令行工具的权限多个 Agent 串联后风险边界也会被叠加。结果不可预期。大模型生成的内容本身具有随机性多 Agent 串联之后最终结果更容易偏离用户意图。审计难以落地。哪个 Agent 在什么阶段做了什么决定调用哪个 Skill 版本执行了什么命令如果这些都没有记录出了问题很难回溯。所以进入 2026 年“让 Agent 跑起来”已经不是难点难点在于让 Agent 在可控、安全、可审计的边界内跑起来。这也是 Harness Engineering 兴起的直接原因。1.2 Harness Engineering 解决什么问题Harness 原意是马的挽具、控制装置引申到 AI 工程领域可以理解为“给 Agent 套上缰绳”的工程方法。它不关注某一个模型有多强而是关注 Agent 系统的边界能做什么、不能做什么、失败后如何恢复、关键节点由谁拍板。Harness Engineering 的核心目标可以浓缩为四点可控Agent 的执行步骤、工具调用、上下文范围都有明确限制。可信Agent 的每个结论都能关联到依据例如 Skill 规则、沙箱运行日志。可审计完整记录每一次决策和动作满足企业内部合规要求。可恢复失败或被拒之后有清晰的降级、重试和回滚路径。它与 AgentOps、LLMOps 有重叠但更偏向工程架构层面。你可以把 AgentOps 理解为“监控和运维”而 Harness Engineering 更强调“如何从上到下设计 Agent 的运行边界和控制流”。1.3 企业级 Harness 的四个支柱结合当前主流实践企业级 Harness 体系通常包含四个关键组成部分支柱作用落地方式沙箱隔离限制 Agent 代码执行的资源、网络、文件访问范围子进程资源限制、Docker、gVisor、WASM 等Skill 技能体系将可复用经验从 prompt 中抽离出来标准化、版本化SKILL.md 脚本 规则清单 版本管理人工介入在高风险节点保留人的决策权审批队列、IM 通知、超时默认拒绝可观测与审计记录全链路决策和执行日志trace_id、结构化日志、指标采集这四个部分不是可选项而是企业落地多 Agent 的底线。任何一个缺失Agent 规模越大风险越高。2. 企业级多 Agent 协同项目总体设计2.1 项目背景与业务场景为了方便讲解本文设计一个贴近实际的企业级场景代码变更交付流水线。一条需求从进入到交付需要以下 Agent 协作规划 Agent拆解需求生成可执行计划。编码 Agent根据计划生成代码应用团队编码规范 Skill。审查 Agent使用代码审查 Skill 检查风险。测试 Agent在沙箱中执行代码验证基本可运行性。人工审批发布前的最后一道人工卡点。选择这个场景有两个原因第一它足够贴近研发团队现状读者容易代入第二它的安全要求高天然需要沙箱、Skill、人工介入、审计这些 Harness 组件。2.2 系统整体架构整个系统的核心是一条单向流水线多个 Agent 通过编排引擎串联。为了保持流程清晰这里不用复杂图直接用 ASCII 示意图展示链路用户请求 ↓ 编排引擎 (Orchestrator) ↓ 分配任务并维护状态 [规划 Agent] → [编码 Agent] → [审查 Agent] → [测试 Agent] ↓ 沙箱执行环境 ↓ 人工审批节点 ↓ 输出交付物与此同时整套系统还需要两个横向能力Skill 仓库为编码、审查等 Agent 提供标准化规则和脚本。审计日志贯穿所有 Agent 节点记录调用输入、输出、Skill 版本和执行时长。2.3 Agent 角色与职责Agent职责依赖 Skill权限边界规划 Agent拆解需求输出步骤计划无仅读上下文只读不执行命令编码 Agent生成代码文件code-style只允许写入工作目录审查 Agent规则化代码审查code-review只读不修改业务代码测试 Agent沙箱执行代码并验证sandbox-policy只能在沙箱内运行要注意Agent 的权限边界应该遵循最小权限原则。规划 Agent 不需要写文件就不要给它写文件的入口编码 Agent 不需要访问生产数据库就不应该在它的上下文中出现生产连接串。2.4 人工介入点设计人工介入不能“随机”而应该基于风险进行设计。本文项目将流程划分为三个风险等级高风险动作必须人工审批发布上线、生产数据操作、外部 API 调用、权限提升。中风险动作自动执行但留痕代码审查、测试执行。低风险动作完全自动格式检查、静态扫描。对应到流水线上只有最后一个“交付节点”需要人工审批。在真实企业中发布到生产环境、调用外部付费接口时通常也需要人工确认。3. 环境准备与项目结构3.1 运行环境本文示例以 Linux 环境为主因为沙箱资源限制、namespace 等能力在 Linux 上支持最好。具体版本仅供参考实际环境需要根据你的项目调整操作系统LinuxCentOS 7、Ubuntu 20.04 均可Python3.10 及以上工具git、python3-pip不需要 GPUCPU 即可演示整套流程如果只能在 Windows 上测试SandboxRunner 中resource模块会不可用建议改用 Docker 或 WSL 环境。这不是示例代码的 bug而是系统能力差异。3.2 项目目录结构我们采用一个简洁但完整的目录结构enterprise-agent-harness/ ├── main.py # 入口 ├── orchestration/ │ └── orchestrator.py # 编排引擎 ├── agents/ │ ├── planner.py # 规划 Agent │ ├── coder.py # 编码 Agent │ ├── reviewer.py # 审查 Agent │ └── tester.py # 测试 Agent ├── sandbox/ │ ├── policy.yaml # 沙箱策略 │ └── sandbox_runner.py # 沙箱执行器 ├── skills/ │ ├── code-review/ │ │ ├── SKILL.md │ │ └── scripts/ │ │ └── main.py │ └── skill_loader.py # Skill 加载器 ├── human/ │ └── approval.py # 人工审批节点 └── audit/ └── logger.py # 审计日志这个结构把“控制能力”和“业务代码”分离sandbox、skills、human、audit属于 Harness 层agents、orchestration属于业务流程层。3.3 依赖说明为了让读者能完整跑通本文尽量使用 Python 标准库实现核心代码不依赖大型第三方框架。生产环境中你完全可以选用 LangGraph、Dify、n8n 等现成编排平台它们提供了更完善的 Agent 编排界面。这里自己用代码实现一遍是为了把多 Agent 协同、沙箱、Skill、人工介入的本质讲透。如果你用 pip 管理依赖下面是最小依赖清单pip install PyYAML只有沙箱策略读取用到 YAML其余全部使用标准库。4. 动手实现沙箱执行环境4.1 为什么 Agent 执行必须沙箱化Agent 生成的代码本质上是不可完全信任的。模型可能因为训练数据产生幻觉写出一条会删除文件的命令也可能因为 prompt 注入被恶意文本诱导去访问系统目录。只要 Agent 执行环境没有边界这些风险就会直接落到宿主机上。沙箱的本质是提供一个受限的运行环境Linux 上常见实现包括 namespace、cgroup、seccomp、LandlockWindows 上有进程隔离和沙箱 APImacOS 有 App Sandbox。更上层的容器Docker、微虚拟机Firecracker、用户态内核gVisor、WASM 运行时都是对底层隔离能力的工程封装。本文示例先实现一个“进程级资源限制”版本只演示思路。真实生产环境请结合容器或 WASM 方案下面的对比表格会说明。4.2 沙箱策略文件先定义策略文件sandbox/policy.yaml# 文件路径sandbox/policy.yaml # 沙箱执行策略 # 注意示例仅实现其中部分限制完整生产实现需要与容器/seccomp 等方案结合 timeout_seconds: 30 max_memory_mb: 256 max_cpu_time: 10 network: enabled: false filesystem: allow_write_paths: - /tmp/agent_workspace deny_paths: - /etc - /root - /home - /var几个关键说明timeout_seconds防止 Agent 生成的代码出现死循环或长时间运行。max_memory_mb限制内存避免资源耗尽拖垮宿主机。network.enabled默认关闭网络Agent 不应随便访问公网。filesystem声明哪些目录允许写入、哪些目录必须拒绝访问。真实的文件系统隔离需要靠只读挂载、namespace 或容器实现YAML 只是策略表达的入口。4.3 沙箱执行器实现下面是一个简化但可运行的沙箱执行器sandbox/sandbox_runner.py 文件路径sandbox/sandbox_runner.py 说明轻量级沙箱执行器用于演示 Agent 产物的沙箱执行思路。 生产环境请使用 Docker、gVisor、Firecracker、WASM 等隔离更强的方案。 import resource import subprocess import tempfile import time from dataclasses import dataclass from pathlib import Path dataclass class SandboxResult: returncode: int stdout: str stderr: str duration: float timeout: bool class SandboxRunner: def __init__(self, policy: dict): self.policy policy self.timeout policy.get(timeout_seconds, 30) self.max_memory_mb policy.get(max_memory_mb, 256) def run_python(self, code: str, workdir: str | None None) - SandboxResult: with tempfile.TemporaryDirectory(prefixagent_sandbox_) as tmp_dir: exec_dir Path(workdir) if workdir else Path(tmp_dir) script_path exec_dir / task_script.py script_path.write_text(code, encodingutf-8) def limit_resources(): memory_bytes self.max_memory_mb * 1024 * 1024 resource.setrlimit(resource.RLIMIT_AS, (memory_bytes, memory_bytes)) start time.time() try: proc subprocess.run( [python3, str(script_path)], capture_outputTrue, textTrue, timeoutself.timeout, cwdstr(exec_dir), preexec_fnlimit_resources, ) return SandboxResult( returncodeproc.returncode, stdoutproc.stdout, stderrproc.stderr, durationtime.time() - start, timeoutFalse, ) except subprocess.TimeoutExpired as exc: return SandboxResult( returncode-1, stdoutexc.stdout or , stderrtimeout, durationtime.time() - start, timeoutTrue, )这个实现的核心逻辑是把 Agent 生成的代码写到临时目录然后新建子进程执行。需要注意几点preexec_fnlimit_resources会在执行子进程前设置内存上限Linux 下有效。cwd指向临时目录避免 Agent 代码在项目根目录乱写文件。TemporaryDirectory会在执行结束自动清理临时文件避免 Agent 残留垃圾。这个示例只实现了资源限制完整的文件系统、网络隔离需要依靠容器或内核安全模块。4.4 沙箱方案选型对比方案隔离强度性能开销适用场景工程复杂度子进程 资源限制弱低教学演示、可信度较高的内部代码低Docker 容器中中企业内部代码执行、构建任务中gVisor较强中高多租户、不可信代码执行高Firecracker 微虚拟机强高严格隔离、公有云多租户高WASM 运行时中低插件、函数级隔离中WASM 沙箱的实现原理其实很有意思它通过编译期和运行时的边界把宿主 API 隐藏起来只暴露开发者显式导入的函数。Agent 代码无法直接访问文件系统或网络必须通过宿主的 import 函数间接调用。这种方式很适合 Skill 插件体系但生态建设比容器方案更早期一些。5. 动手实现Skill 技能体系5.1 理解 Skill从 Prompt 片段到结构化技能包很多团队最初的做法是把规则写进系统提示词比如“请遵守团队编码规范”。但一旦规则增多提示词会越来越长越来越难维护而且无法测试、无法版本化、无法按 Agent 动态加载。Skill 的出现解决了这个问题。Skill 本质上是把某类任务的经验固化成结构化技能包通常包含说明文档、规则清单、可执行脚本和示例。这样Agent 在需要时可以按名称加载技能而不是把全部规则塞进上下文。Skill 的好处标准化每个技能有固定的目录结构和元信息。版本化技能像代码一样可以打 tag、做 diff。可测试每种技能都能配合一组测试用例验证效果。权限边界每个技能可以声明自己需要访问的资源范围。5.2 Skill 目录规范本文采用最基础的 Skill 目录规范与当前主流实践保持一致skills/ ├── code-review/ │ ├── SKILL.md # 技能元信息 规则清单 使用说明 │ └── scripts/ │ └── main.py # 可选辅助脚本 └── skill_loader.py # 技能加载器关键文件是SKILL.md它同时承担说明和规则两种角色。5.3 编写一个代码审查 Skill以下是一个代码审查 Skill 的SKILL.md示例--- name: code-review description: 基于团队规范执行代码审查重点关注资源泄漏、越权访问、错误处理、可读性。 version: 0.1.0 --- # Code Review Skill ## 适用场景 - 编码 Agent 生成代码后审查 Agent 使用本技能进行规则化检查。 ## 规则清单 - 禁止出现硬编码数据库密码或 Token - 所有资源型操作必须包含 close 或 with 语句 - 网络请求必须设置超时时间 - 文件写入路径必须位于授权目录内 - 函数必须包含类型注解与 docstring ## 使用方式 1. 将待审查代码作为输入文本。 2. 按规则清单逐条判断 PASS / FAIL / WARN。 3. 输出 JSON 格式检查报告。 ## 输出格式 {rule: ..., level: PASS|FAIL|WARN, note: ...}为什么用 Markdown 而不是纯文本或 JSON因为 Markdown 既能让人阅读也能让 Agent 解析还可以放进 git 做版本管理。规则清单放在固定二级标题下是为了让加载器能稳定地提取规则。5.4 Skill 的加载、版本化与自进化先实现一个 Skill 加载器skills/skill_loader.py 文件路径skills/skill_loader.py 说明从 SKILL.md 中加载 Skill读取元信息与规则清单。 from pathlib import Path class Skill: def __init__(self, name: str, description: str, version: str, rules: list[str], script_path: str | None): self.name name self.description description self.version version self.rules rules self.script_path script_path def __repr__(self): return fSkill(name{self.name}, version{self.version}) def parse_skill_md(skill_dir: Path) - Skill: skill_md skill_dir / SKILL.md if not skill_md.exists(): raise FileNotFoundError(f缺少 SKILL.md: {skill_dir}) text skill_md.read_text(encodingutf-8) name skill_dir.name description version 0.0.0 rules: list[str] [] lines text.splitlines() in_rules False for line in lines: if line.startswith(description:): description line.split(:, 1)[1].strip() elif line.startswith(version:): version line.split(:, 1)[1].strip() elif line.strip() ## 规则清单: in_rules True continue elif line.startswith(## ) and in_rules: break elif in_rules and line.strip().startswith(- ): rules.append(line.strip()[2:].strip()) script skill_dir / scripts / main.py script_path str(script) if script.exists() else None return Skill( namename, descriptiondescription, versionversion, rulesrules, script_pathscript_path, ) def load_skills(skill_root: str | Path) - dict[str, Skill]: skill_root Path(skill_root) skills: dict[str, Skill] {} if not skill_root.exists(): return skills for child in skill_root.iterdir(): if child.is_dir() and (child / SKILL.md).exists(): skill parse_skill_md(child) skills[skill.name] skill return skills关于“自进化 Skill”这里提供一条可落地的实施思路Agent 执行任务时如果审查未通过将失败案例保存为结构化记录。规则维护者或另一个“规则提取 Agent”从失败记录中生成新的候选检查项。候选规则自动写入SKILL.md的新版本分支例如v0.2.0。新版本通过 CI 测试并由人工审核后合并。Skill 仓库按版本加载例如code-review0.2.0发布团队可以使用旧版本进行回滚。也就是说自进化不是让 Skill“自己悄悄改自己”而是让 Skill“不断积累经验并走评审流程”。人工介入是自进化链路里最重要的安全闸门。6. 动手实现多 Agent 编排工作流6.1 工作流设计我们在编排层用状态机控制流程状态流转如下INIT → PLANNING → CODING → REVIEW → TESTING → APPROVAL → DONE ↓ ↓ REVIEW_FAILED REJECTED每个状态对应一个 Agent 或一步人工操作状态流转由编排引擎维护。这样做的好处是流程清晰出错时可以快速定位是哪一步。6.2 编排核心代码以下是核心编排代码orchestration/orchestrator.py 文件路径orchestration/orchestrator.py 说明多 Agent 编排引擎使用状态机模式控制流程。 from abc import ABC, abstractmethod from dataclasses import dataclass, field dataclass class WorkContext: task_id: str title: str detail: str status: str INIT artifacts: dict field(default_factorydict) logs: list field(default_factorylist) class BaseAgent(ABC): def __init__(self, name: str): self.name name abstractmethod def execute(self, ctx: WorkContext) - None: pass class PlannerAgent(BaseAgent): def __init__(self): super().__init__(planner) def execute(self, ctx: WorkContext) - None: print(f[{self.name}] 开始拆解任务) ctx.artifacts[plan] [ 1. 解析需求, 2. 编写代码, 3. 静态审查, 4. 执行测试, 5. 人工审批, ] ctx.status PLANNING_DONE ctx.logs.append(f{self.name}: plan ready) class CoderAgent(BaseAgent): def __init__(self, code_style_skillNone): super().__init__(coder) self.code_style_skill code_style_skill def execute(self, ctx: WorkContext) - None: print(f[{self.name}] 根据计划生成代码) if self.code_style_skill: print(f[{self.name}] 应用 Skill: {self.code_style_skill.name}) # 真实项目中这里会调用大模型生成代码 # 本文直接用演示代码说明流程 ctx.artifacts[code] print(hello from agent coder)\n ctx.status CODING_DONE ctx.logs.append(f{self.name}: code generated) class ReviewerAgent(BaseAgent): def __init__(self, review_skillNone): super().__init__(reviewer) self.review_skill review_skill def execute(self, ctx: WorkContext) - None: print(f[{self.name}] 执行代码审查) code ctx.artifacts.get(code, ) report {level: PASS, issues: []} if self.review_skill: # 真实项目中可以调用大模型或脚本分析代码 # 这里只演示简单规则检查危险关键词 danger_words [password, token, eval(, subprocess] for rule in self.review_skill.rules: for word in danger_words: if word.lower() in code.lower(): report[issues].append(fhit danger word: {word}) break ctx.artifacts[review_report] report if report[issues]: ctx.status REVIEW_FAILED else: ctx.status REVIEW_PASSED ctx.logs.append(f{self.name}: review {ctx.status}) class TesterAgent(BaseAgent): def __init__(self, sandbox_runner): super().__init__(tester) self.sandbox_runner sandbox_runner def execute(self, ctx: WorkContext) - None: print(f[{self.name}] 将代码放入沙箱执行) code ctx.artifacts.get(code, ) result self.sandbox_runner.run_python(code) ctx.artifacts[test_result] { returncode: result.returncode, stdout: result.stdout, stderr: result.stderr, duration: round(result.duration, 2), } if result.returncode 0 and not result.timeout: ctx.status TEST_PASSED else: ctx.status TEST_FAILED ctx.logs.append(f{self.name}: test {ctx.status}) class Workflow: def __init__(self, sandbox_runner, skills: dict, approval_handler): self.sandbox_runner sandbox_runner self.skills skills self.approval_handler approval_handler self.agents { planner: PlannerAgent(), coder: CoderAgent(code_style_skillskills.get(code-style)), reviewer: ReviewerAgent(review_skillskills.get(code-review)), tester: TesterAgent(sandbox_runnerself.sandbox_runner), } def run(self, ctx: WorkContext) - WorkContext: ctx.status INIT ctx.logs.append(f{ctx.task_id}: workflow start) self.agents[planner].execute(ctx) self.agents[coder].execute(ctx) self.agents[reviewer].execute(ctx) if ctx.status REVIEW_FAILED: ctx.logs.append(f{ctx.task_id}: workflow end with review failed) return ctx self.agents[tester].execute(ctx) if ctx.status TEST_FAILED: ctx.logs.append(f{ctx.task_id}: workflow end with test failed) return ctx # 人工审批节点 print(f[workflow] 任务 {ctx.task_id} 进入人工审批) approved self.approval_handler.approve(ctx) if approved: ctx.status DONE ctx.logs.append(f{ctx.task_id}: workflow end with approval) else: ctx.status REJECTED ctx.logs.append(f{ctx.task_id}: workflow end with rejection) return ctx这个编排代码有几个值得注意的设计点WorkContext是贯穿全流程的数据容器记录状态、产物和日志。每个 Agent 职责单一只修改自己的产物和状态。ReviewerAgent和TesterAgent都有失败分支不会无条件继续。人工审批由外部approval_handler注入方便测试时替换。6.3 人工审批节点代码下面是人工审批节点的示例human/approval.py 文件路径human/approval.py 说明人工审批节点支持自动通过以便联调生产环境请接入 IM 通知。 import os class ConsoleApproval: def approve(self, ctx) - bool: if os.getenv(AUTO_APPROVE) 1: print([approval] AUTO_APPROVE1自动通过) return True print(f[approval] 任务标题: {ctx.title}) result input(请输入审批结果 (approve/reject): ).strip().lower() return result approve真实生产中这个节点会替换为企业内部的审批 API例如飞书、钉钉、企业微信的审批机器人同时需要记录审批人、审批时间、审批意见。6.4 运行入口与验证最后是入口文件main.py 文件路径main.py 说明企业级多 Agent 协同项目示例入口。 import json from pathlib import Path from sandbox.sandbox_runner import SandboxRunner from skills.skill_loader import load_skills from orchestration.orchestrator import Workflow, WorkContext from human.approval import ConsoleApproval import yaml def load_policy(path: str) - dict: with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def main(): policy load_policy(sandbox/policy.yaml) runner SandboxRunner(policy) skills load_skills(skills) approval ConsoleApproval() workflow Workflow( sandbox_runnerrunner, skillsskills, approval_handlerapproval, ) ctx WorkContext( task_idTASK-001, title生成 hello 程序, detail生成一个打印 hello 的 Python 程序, ) result workflow.run(ctx) print(\n 执行结果 ) print(ftask_id : {result.task_id}) print(fstatus : {result.status}) print(fplan : {result.artifacts.get(plan)}) print(fcode : {result.artifacts.get(code)}) print(freview : {result.artifacts.get(review_report)}) print(ftest : {json.dumps(result.artifacts.get(test_result), ensure_asciiFalse)}) print(flogs : {json.dumps(result.logs, ensure_asciiFalse)}) if __name__ __main__: main()运行命令cd enterprise-agent-harness export AUTO_APPROVE1 python3 main.py预期输出[planner] 开始拆解任务 [coder] 根据计划生成代码 [reviewer] 执行代码审查 [reviewer] review REVIEW_PASSED [tester] 将代码放入沙箱执行 [workflow] 任务 TASK-001 进入人工审批 [approval] AUTO_APPROVE1自动通过 执行结果 task_id : TASK-001 status : DONE plan : [1. 解析需求, 2. 编写代码, 3. 静态审查, 4. 执行测试, 5. 人工审批] code : print(hello from agent coder