尧图网页设计 
从“调 Prompt”到“建 Harness”企业级 AI Agent 工程化真正缺的是这一层过去两年很多团队做 AI Agent 的方式其实是一样的把 Claude、GPT 或开源模型的 API 接进来写一大堆 system prompt然后给 Agent 挂上几个工具函数就开始宣称“我们已经完成 AI 落地”。等真正上线才发现完全不是那么回事。需要保证 Agent 拿不到不该访问的数据库需要约束它不要在一次循环里把上千个临时文件写到宿主机需要让沉淀下来的一段有效技能能被团队里其他 Agent 复用而不是锁在某个人的聊天记录里更要命的是当 Agent 判断错误、连续重试、甚至开始“自由发挥”的时候你根本不知道从哪里打断它。所以 2026 年再谈 AI Agent核心词已经从“能不能跑通”变成了“能不能管控”。而 Harness Engineering直译是“驾驭工程”或“约束工程”正是为了解决这批问题出现的工程方法论。它强调的从来不是把模型能力做得更大而是通过一套工程结构把不可控的模型行为装进可控的边界里沙箱隔离执行环境、自进化 Skill 沉淀能力、人工介入保留最终决策权。这篇文章会先讲清楚 Harness Engineering 到底解决什么问题再给出一个企业级多 Agent 协同项目的完整工程骨架包括沙箱设计、Skill 的自进化机制、人机协同的介入节点以及一套可以直接落地的最小实现。文章不会停留在概念层面你可以照着代码把一个带沙箱约束、Skill 热更新、人工审批点的多 Agent 骨架搭起来。1. 为什么 2026 年 AI Agent 工程化绕不开 Harness Engineering先说一个很容易被忽视的判断当前 AI Agent 的瓶颈不在模型能力而在工程侧对行为的约束能力。模型能力这半年提升得很快长上下文、复杂推理、工具调用都已经逐步成熟。多 Agent 协同也早就不是论文里的概念Coze、Dify、n8n 这类平台已经让 Agent 工作流搭建变得足够简单。但你如果去问那些真正把 Agent 推到生产环境的团队他们最痛苦的不是“模型不够聪明”而是以下三个问题。第一个问题Agent 的执行边界不可控。模型会“自由发挥”。你让它读取一份 CSV它可能顺手把同目录下的配置文件也读进来你让它执行一段 Python 代码它可能在宿主机上创建文件、连接网络、读取环境变量。在本地开发环境这可能只是一个小意外但在企业内网、生产数据库、核心代码仓库面前这种不可控就是事故。第二个问题能力沉淀不下来。你在一个 Agent 的 prompt 里调出一套很有效的 JSON 处理逻辑怎么迁移到另一个 Agent复制 prompt那个 Agent 未必有同样的工具集。写成代码那又回到了传统软件开发模型的灵活性全丢了。很多团队每个 Agent 都是“独立烟囱”经验无法复用效率越做越低。第三个问题没有人机边界。Agent 出错时是让它自己重试还是让人介入如果人介入在哪个环节介入、能修改什么、能审批什么大部分项目根本没有设计过这个边界结果就是两个极端要么全程无人值守出事才发现要么每个步骤都弹确认框Agent 变成了“自动点下一步”的工具毫无效率可言。Harness Engineering 对这三个问题的回答是一致的Agent 的能力可以复杂但 Agent 的行为边界必须简单。它要求把环境隔离、技能沉淀、人机审批当作和模型选型同等重要的一等工程构件来设计。这也是为什么它会被越来越多企业级 AI 项目列为必选项。2. Harness Engineering 核心概念Skill、Sandbox、介入机制在进入代码之前先把 Harness Engineering 的三个关键概念讲清楚。后面所有设计和代码都围绕这三个概念展开。2.1 Skill不是工具函数是“可被模型自主调用的经验单元”很多材料把 Skill 翻译成“技能”这个翻译容易让人误以为它只是工具函数的另一个名字。实际上Skill 和普通工具函数有本质区别。普通工具函数是开发者写死的能力例如“计算两个日期之间的天数”“调用某个 HTTP 接口”。它由开发者控制Agent 做不了任何自主选择。Skill 则是一个带描述、带校验、带版本、可被模型自主学习调用的经验单元。它通常包含一段自然语言描述告诉模型“什么时候该用这个 Skill”一个可执行实现可以是 Python 函数、API 调用甚至是一段 Shell 脚本一组参数定义和输入校验逻辑元信息包括版本号、作者、运行所需的权限级别可选的“自进化记录”用于保存该 Skill 的迭代历史。一个典型的 Skill 示例analyze_nginx_log分析 Nginx 日志。它不只是“解析日志文件”这么简单它可能封装了你团队平时分析 Nginx 日志时的完整套路先看 5xx 分布再看上游响应时间最后按 IP 归类失败请求。这个套路沉淀成 Skill 后任何 Agent 遇到 Nginx 日志相关任务都可以直接调用而不是从零开始设计分析思路。2.2 Sandbox把执行环境变成“一次性的”Sandbox沙箱是 Harness Engineering 里最核心的安全机制。它的目标非常明确让 Agent 的执行代码跑在一个受限、可丢弃、可审计的环境里。具体来说企业级 Agent 沙箱需要具备以下能力能力说明典型实现方式文件系统隔离Agent 只能访问指定目录不能读取宿主机敏感文件Docker volume、容器只读根文件系统网络隔离默认禁止外网访问按白名单放行Docker network、iptables、代理白名单系统调用限制禁止危险 syscall如 mount、ptracegVisor、Firecracker、seccomp资源限制限制 CPU、内存、磁盘、执行时长cgroup、Docker--memory、超时机制可丢弃性执行结束后环境销毁不保留中间状态容器生命周期管理沙箱的实现技术栈从轻到重依次是进程级隔离如 Firecracker、容器级隔离Docker、系统调用级隔离gVisor。对于大多数企业应用Docker 容器加资源限制是性价比最高的方案。如果你的场景涉及不可信代码执行才需要考虑 gVisor 或 Firecracker 这类更重的隔离。2.3 人工介入不是打断是“可控的授权节点”人工介入机制是 Harness Engineering 区别于纯自动化 AI 系统的重要标志。它的设计原则是不是所有步骤都需要人审批但所有高影响操作都必须有人的授权点。典型需要人工介入的场景包括执行写数据库操作尤其是 UPDATE、DELETE、DROP向外部发送消息邮件、企业微信、短信合并代码、创建发布分支申请超过预算的资源访问标记了敏感级别的数据。人工介入不应该是“弹窗打断”而应该是一个异步审批任务Agent 发出审批请求挂起当前流程等待人在审批界面做出决定然后再继续执行。这一点放到多 Agent 架构里尤其重要主 Agent 可以继续处理其他事务只有相关子任务被挂起。3. 企业级多 Agent 协同的整体架构设计把三个核心概念组合起来一个企业级多 Agent 协同系统的架构就清晰了。下面是一个完整的参考架构包含五个层次。┌─────────────────────────────────────────────────────────┐ │ 用户层 / 接入层 │ │ Web 控制台 | 企业微信 / 钉钉 Bot | API 网关 │ └─────────────────────────────────────────────────────────┘ │ ┌─────────────────────────────────────────────────────────┐ │ 编排层 / Orchestration │ │ Supervisor Agent主控 Agent │ │ 任务规划 | 子任务分发 | Skill 推荐 | 介入决策 │ └─────────────────────────────────────────────────────────┘ │ ┌─────────────────────────────────────────────────────────┐ │ 执行层 / Workers │ │ 数据分析 Agent | 代码生成 Agent | 运维操作 Agent │ │ Worker Agent执行 Agent可水平扩展 │ └─────────────────────────────────────────────────────────┘ │ ┌─────────────────────────────────────────────────────────┐ │ Harness 层 │ │ Skill 注册中心 | 沙箱管理 | 审批中心 | 审计日志 │ └─────────────────────────────────────────────────────────┘ │ ┌─────────────────────────────────────────────────────────┐ │ 基础设施层 │ │ Docker / K8s | 向量数据库 | 对象存储 | 消息队列 │ └─────────────────────────────────────────────────────────┘这个架构的关键是把决策、执行、约束三个维度拆开编排层负责规划它决定“做什么”“谁来做”“需要什么 Skill”但它自己不做具体执行执行层负责干活每个 Worker Agent 处理一项具体任务Harness 层负责约束它不关心任务本身只关心“执行是否越界”“Skill 是否有效”“是否需要在某个节点暂停等人审批”。传统单体 Agent 把所有职责揉在一起导致安全约束难以独立演进。而 Harness Engineering 的架构天然支持安全团队独立维护沙箱规则、业务团队独立维护 Skill、运维团队独立维护审批流各司其职。3.1 多 Agent 的任务流转示例用一个具体场景说明这套架构的运转流程业务方提出一个任务——“分析最近一周订单量下降的原因并生成一份报告”。主控 Agent 接收任务拆解为数据提取、趋势分析、竞品信息检索、报告生成四个子任务。主控 Agent 查询 Skill 注册中心发现“订单数据查询 Skill”“时序趋势分析 Skill”“报告生成 Skill”可用。数据提取子任务被分发到数据分析 Worker Agent。Worker 在沙箱中执行 SQL 查询脚本沙箱只允许连接数据仓库的只读账号。趋势分析子任务涉及调用外部 API 获取行业数据。Harness 层检测到该 Worker 的沙箱默认禁止外网访问于是发起人工审批请求。审批人确认后白名单临时放行该 API。各子任务完成后主控 Agent 汇总结果调用报告生成 Skill 输出最终报告。整个流程的每一次 Skill 调用、每一次沙箱操作、每一轮人工审批都写入审计日志。可以看到多 Agent 协同并不神秘。它的难点在于当一个任务被拆成多个子任务后安全边界、权限边界、审批边界仍然要清晰。这正是 Harness 层存在的意义。4. 沙箱机制的设计与实现沙箱是整个 Harness Engineering 体系里最偏底层、也最容易踩坑的部分。下面分为设计要点和代码实现两部分讲解。4.1 沙箱设计四个关键决策决策一选择 Docker 还是 Kubernetes 运行沙箱。单机或小团队场景直接用 Docker Engine API 即可。需要大规模并发、自动伸缩时再上 Kubernetes。Docker 的好处是简单直接一个容器一个任务跑完即删。决策二镜像要最小化并且不包含生产凭证。沙箱镜像只需要 Python 运行时、常用工具库、网络调试工具。生产环境的数据库密码、API Key 一律通过环境变量在运行时注入且按最小权限分配。这里真正容易踩坑的是很多团队图省事把包含凭证的“全家桶镜像”直接当成沙箱镜像用等于给 Agent 送上了所有钥匙。决策三网络默认关闭需要外网时走白名单或审批。沙箱容器创建时默认不使用 bridge 网络或使用隔离网络。Agent 需要访问某个外部 API 时由 Harness 层动态添加规则或修改代理配置。决策四强制超时和资源限制。即使逻辑完全正确Agent 也可能因为循环失控而长时间运行。所有沙箱任务必须设置执行超时通常建议 30 秒到 5 分钟以及明确的 CPU 和内存上限。4.2 基于 Docker SDK 的沙箱核心代码下面用 Python 的dockerSDK实现一个最简沙箱管理器。它的职责是接收一段代码在隔离容器里执行返回标准输出、错误信息和执行指标。# 文件路径harness/sandbox/docker_sandbox.py import docker import uuid import time class DockerSandbox: 基于 Docker 的 Agent 沙箱执行器 def __init__(self, imagepython:3.11-slim, network_disabledTrue): self.client docker.from_env() self.image image self.network_disabled network_disabled def run_code(self, code: str, timeout: int 30, memory_limit: str 256m, env: dict None, workdir: str /workspace): 在沙箱中执行 Python 代码 :param code: 要执行的代码 :param timeout: 超时秒数 :param memory_limit: 内存限制Docker 格式如 256m :param env: 注入的环境变量最小权限 :param workdir: 容器内工作目录 container_name fagent-sandbox-{uuid.uuid4().hex[:8]} exec_id None try: # 创建隔离容器 container self.client.containers.run( imageself.image, command[/bin/sh, -c, sleep 5], namecontainer_name, detachTrue, network_disabledself.network_disabled, # 默认禁用网络 mem_limitmemory_limit, pids_limit100, working_dirworkdir, environmentenv or {}, read_onlyTrue, # 根文件系统只读 tmpfs{/workspace: rw,noexec,nosuid,size64m} ) # 将代码写入容器内的可写临时目录 # 注意因为根文件系统只读这里使用 tmpfs 挂载的 /workspace code_path f/workspace/main.py # 使用 exec 写入代码 write_cmd fcat {code_path} EOF\n{code}\nEOF container.exec_run(write_cmd, userroot) # 在容器内执行代码 result container.exec_run( cmd[python, /workspace/main.py], userroot, demuxTrue, timeouttimeout ) stdout result.output[0].decode() if result.output[0] else stderr result.output[1].decode() if result.output[1] else exit_code result.exit_code return { exit_code: exit_code, stdout: stdout, stderr: stderr, container_name: container_name } except docker.errors.ContainerError as e: return {exit_code: -1, stderr: str(e)} except Exception as e: return {exit_code: -1, stderr: fSandbox execution failed: {e}} finally: # 无论成功失败容器都销毁 try: container self.client.containers.get(container_name) container.remove(forceTrue) except Exception: pass这段代码有几个值得注意的设计read_onlyTrue把镜像根文件系统设为只读Agent 无法修改系统文件tmpfs挂载/workspace提供临时可写目录但noexec禁止在该目录直接执行二进制文件network_disabledTrue默认禁网finally块确保容器无论成功失败都被删除不残留中间状态。实际生产环境还可以继续强化把代码写入改为通过 Docker API 的put_archive避免exec_run拼接命令带来的注入风险把网络白名单做成持久化配置而不是每次手动设置。但上面的最小实现已经足够演示沙箱机制的核心思路。4.3 沙箱管理的核心原则默认拒绝 临时授权沙箱配置不是“尽量严一点”而是“默认拒绝一切按需临时授权”。这个原则具体到实现上是默认无网络不对全部端口开放默认无宿主机目录挂载需要读写的数据通过 API 显式传入默认无环境变量需要的凭证单独注入默认无特权模式不用--privileged。任何“为方便调试而放开”的配置都应该在代码评审阶段被打回。Sandbox 是安全边界不是开发环境。5. 自进化 Skill让 Agent 的能力在运行中沉淀如果说沙箱是 Harness Engineering 的“安全带”那自进化 Skill 就是它的“发动机”。自进化 Skill 是指Skill 能够在实际运行后基于效果反馈自动或半自动地更新自身版本。5.1 为什么需要自进化 Skill没有自进化机制时Skill 是一次性开发的软件组件写完后固定不变效果如何完全取决于最初的开发者思考得是否周全。但 Agent 的调用场景是开放的。一个“报告生成 Skill”第一次运行时可能只会输出纯文本报告第二次运行时用户反馈“希望加上数据可视化图表”第三次运行时又发现“PDF 格式更适合汇报”。如果每次需求变化都要开发者手动改代码Skill 就和普通程序没有区别了。自进化 Skill 的流程是运行 → 收集反馈 → 评估 → 更新版本 → 重新注册。5.2 自进化 Skill 的架构与代码一个实用的自进化 Skill 系统包含两个部分Skill 存储和进化管线。下面给出 Skill 定义模型和更新逻辑示例。# 文件路径harness/skill/skill_registry.py from dataclasses import dataclass, field from typing import Dict, Callable, Any import datetime import json dataclass class Skill: Skill 数据模型 name: str # Skill 名称 version: str # 语义化版本 description: str # 自然语言描述用于模型判断何时调用 parameters: Dict[str, Any] # 参数 schema implement: Callable # 实际执行函数 author: str system tags: list field(default_factorylist) create_time: str field(default_factorylambda: datetime.datetime.now().isoformat()) success_count: int 0 # 成功调用次数 fail_count: int 0 # 失败调用次数 feedback: list field(default_factorylist) # 反馈记录 def to_desc(self) - str: 生成给模型的结构化描述 return json.dumps({ name: self.name, description: self.description, parameters: self.parameters, version: self.version, tags: self.tags }, ensure_asciiFalse) class SkillRegistry: Skill 注册中心支持版本管理和自进化 def __init__(self): self._skills: Dict[str, Skill] {} def register(self, skill: Skill): 注册新 Skill 或覆盖旧版本 self._skills[skill.name] skill def get(self, name: str) - Skill: return self._skills[name] def list_all(self) - list: return [s.to_desc() for s in self._skills.values()] def record_result(self, name: str, success: bool, feedback: str ): 记录一次调用结果用于进化评估 skill self._skills.get(name) if not skill: return if success: skill.success_count 1 else: skill.fail_count 1 # 失败日志可以作为 Skill 进化的输入 skill.feedback.append({ time: datetime.datetime.now().isoformat(), success: success, detail: feedback }) def evolve(self, name: str, new_impl: Callable, new_version: str): 发布 Skill 新版本 old self._skills.get(name) if not old: raise ValueError(fSkill {name} not found) # 保留历史统计替换实现和版本 new_skill Skill( nameold.name, versionnew_version, descriptionold.description, parametersold.parameters, implementnew_impl, authorold.author, tagsold.tags, success_countold.success_count, fail_countold.fail_count, feedbackold.feedback ) self._skills[name] new_skill这套模型并不复杂但它定义了一条关键规则Skill 的进化必须是版本化的旧的统计信息不能丢。如果新版本效果反而变差你需要有能力回滚到旧版本并且知道旧版本之前的成功率是多少。5.3 自进化 Skill 的三种进化模式根据人工参与程度自进化 Skill 可以分为三档第一档人工编辑式。Agent 运行后系统把效果不佳的调用日志和用户反馈汇集成建议推送给人。人修改 Skill 代码后以新版本号发布。这是最稳妥的模式适合对准确性要求极高的场景。第二档模型辅助编辑式。系统把失败的调用参数、输出、用户反馈发给大模型让大模型生成新的 Skill 实现代码然后由人确认后发布。这是当前生产环境里最推荐的折中方案既有进化速度又有安全兜底。第三档全自动编辑式。系统根据反馈自动修改 Skill 代码并发布。这种模式适合低风险、高频迭代的 Skill例如日志格式化、数据清洗。全自动模式需要额外的测试套件否则很容易把错误版本发布上线。5.4 Skill 自进化的关键成功因素实际落地时比代码更重要的是一套评估机制。没有评估的进化等于盲目更新。评估机制至少要包含三个维度成功率调用过程中是否报错、结果质量输出和期望的匹配度通常需要人工或另一个模型评估、效率执行时间、Token 消耗。只有当三个维度都满足阈值时新版本才允许上线。比如成功率大于 95%结果质量评分大于 4.0/5.0执行耗时不高于旧版本的 1.2 倍。这里真正容易踩坑的地方是很多团队只看“模型觉得这个 Skill 效果不错”就发布了。但“模型觉得不错”不等于“用户觉得不错”。在生产环境必须采集用户侧的最终反馈而不是只依赖模型自评。6. 人工介入设计人机协同的审批体系人工介入是 Harness Engineering 中最容易被低估的部分。很多人觉得“有人审批”就安全了但实际上如果审批节点设计得不好要么审批流形同虚设要么严重拖慢流程。6.1 三层介入模型一个合理的人工介入体系通常分为三层。第一层任务级介入。发生在 Agent 开始执行任务之前。例如用户在创建任务时明确选择“这个任务需要人工审批每个写操作”。任务级介入适合那些本来就需要严格控制的业务场景。第二层操作级介入。发生在 Agent 执行某个具体操作之前。这是最常见的介入类型。Harness 层会检测即将执行的命令是否命中高风险规则写数据库、发外网请求、删除文件命中则暂停执行生成审批任务。第三层异常级介入。发生在 Agent 执行出现异常或长时间没有进展时。系统根据告警阈值通知人工介入处理。这一层通常容易被忽略它解决的是“Agent 卡住了没人发现”的问题。6.2 人工介入的代码示例下面用消息队列 审批状态机的思路实现一个操作级介入的最小示例。核心思路Agent 在沙箱中执行高风险命令时不直接执行而是发送审批请求等待审批结果。# 文件路径harness/approval/approval_flow.py import enum import time import uuid from dataclasses import dataclass class ApprovalStatus(enum.Enum): PENDING pending APPROVED approved REJECTED rejected TIMEOUT timeout dataclass class ApprovalRequest: request_id: str agent_name: str action_desc: str # 要执行的操作描述 target_resource: str # 目标资源 requested_by: str # 请求人/触发 Agent status: ApprovalStatus ApprovalStatus.PENDING comment: str expire_at: float 0.0 class ApprovalCenter: 审批中心负责创建审批请求、处理审批结果 def __init__(self, timeout_seconds: int 300): self._requests {} self.timeout_seconds timeout_seconds def create_request(self, agent_name: str, action_desc: str, target_resource: str) - ApprovalRequest: req ApprovalRequest( request_iduuid.uuid4().hex[:12], agent_nameagent_name, action_descaction_desc, target_resourcetarget_resource, expire_attime.time() self.timeout_seconds ) self._requests[req.request_id] req # 实际生产环境将审批请求推送到企业微信/钉钉/邮件 # self._push_notification(req) return req def approve(self, request_id: str, comment: str ) - bool: 人工审批通过 req self._requests.get(request_id) if not req or req.status ! ApprovalStatus.PENDING: return False req.status ApprovalStatus.APPROVED req.comment comment return True def reject(self, request_id: str, comment: str ) - bool: 人工审批拒绝 req self._requests.get(request_id) if not req or req.status ! ApprovalStatus.PENDING: return False req.status ApprovalStatus.REJECTED req.comment comment return True def wait_for_result(self, request_id: str, timeout: int 60) - ApprovalStatus: 阻塞等待审批结果。 实际生产环境建议用异步回调或 websocket而不是轮询。 start time.time() while time.time() - start timeout: req self._requests.get(request_id) if req.status ! ApprovalStatus.PENDING: return req.status if time.time() req.expire_at: req.status ApprovalStatus.TIMEOUT return ApprovalStatus.TIMEOUT time.sleep(1) return ApprovalStatus.TIMEOUT这段代码展示了审批状态机的核心流转。集成到 Agent 执行流程时逻辑是这样的# 在高风险操作前调用审批 if is_high_risk_action(action): req approval_center.create_request( agent_namedata_agent, action_desc写数据库更新订单状态, target_resourceorder_db.orders ) status approval_center.wait_for_result(req.request_id) if status ApprovalStatus.APPROVED: execute_action(action) else: abort_action(action)6.3 实现人工介入的三个工程原则原则一审批请求必须携带足够上下文。审批人需要看到的不只是“Agent 请求写数据库”这句话还应该看到完整的 SQL、影响行数预估、涉及哪些表、由哪个任务触发、Agent 的判断依据。否则审批就是走过场。原则二审批必须有超时机制。如果没有超时Agent 会无限期等待任务队列会堆积。超时后的默认策略通常有两种自动拒绝或自动降级。建议默认自动拒绝把超时当作失败处理。原则三审批过程必须全量审计。谁审批的、什么时候审批的、审批时看到了什么内容、最终结果是什么全部留存。这是企业合规的基础要求也是后续优化审批效率的数据来源。7. 完整落地一个多 Agent 协同项目的工程骨架前面已经拆解了沙箱、Skill、审批三个核心模块。这一节把它们组装成一个可运行的最小工程骨架。假设我们要实现一个简单的“数据分析多 Agent 系统”一个 Planner Agent接收任务拆解成子任务两个 Worker Agent一个负责数据查询一个负责报告生成Harness 层统一托管沙箱执行和 Skill 调用。7.1 工程目录结构ai-harness-project/ ├── main.py # 入口启动编排 ├── requirements.txt # 依赖 ├── config/ │ ├── agents.yaml # Agent 配置 │ └── sandbox.yaml # 沙箱配置 ├── harness/ │ ├── sandbox/ │ │ └── docker_sandbox.py # 沙箱执行器 │ ├── skill/ │ │ └── skill_registry.py # Skill 注册中心 │ └── approval/ │ └── approval_flow.py # 审批中心 ├── agents/ │ ├── planner.py # 主控 Agent │ ├── data_worker.py # 数据分析 Worker │ └── report_worker.py # 报告生成 Worker ├── skills/ │ ├── sql_query.py # SQL 查询 Skill │ └── markdown_report.py # 报告生成 Skill └── logs/ └── audit.log # 审计日志7.2 主控 Agent 的编排逻辑主控 Agent 的职责是拆解任务、分发任务、汇总结果。这里演示一个最简单的 Planner它的核心逻辑是根据任务类型选择 Worker然后调用 Worker 的执行接口。实际生产环境这个“选择 Worker”的动作通常由大模型完成但控制流分发、等待、汇总应该是代码逻辑不能完全依赖模型自由发挥。# 文件路径agents/planner.py import json from harness.sandbox.docker_sandbox import DockerSandbox from harness.skill.skill_registry import SkillRegistry class PlannerAgent: 主控 Agent负责任务分发和结果汇总 def __init__(self, sandbox: DockerSandbox, registry: SkillRegistry): self.sandbox sandbox self.registry registry self.workers {} def register_worker(self, name: str, worker): self.workers[name] worker def run_task(self, task_desc: str): 任务入口。 实际生产环境这里会调用 LLM 做任务拆解。 示例中为了演示控制流直接按固定逻辑分发。 print(f[Planner] 接收任务: {task_desc}) # 步骤 1任务规划演示逻辑实际用 LLM 拆解 plan self._plan(task_desc) print(f[Planner] 任务拆解: {json.dumps(plan, ensure_asciiFalse)}) # 步骤 2分发子任务 results {} for step in plan: worker_name step[worker] worker self.workers.get(worker_name) if not worker: print(f[Planner] 未找到 Worker: {worker_name}) continue print(f[Planner] 分发任务到 {worker_name}: {step[action]}) result worker.execute(step[action]) results[worker_name] result # 步骤 3汇总结果 print([Planner] 所有子任务完成开始汇总) return results def _plan(self, task_desc: str): 示意性规划逻辑。生产环境替换为 LLM 规划。 if 订单 in task_desc or 数据 in task_desc: return [ {worker: data_worker, action: query_order_data}, {worker: report_worker, action: generate_report} ] return [{worker: data_worker, action: unknown}]7.3 Worker Agent 与沙箱执行的集成Worker Agent 是实际执行者。它的执行过程要经过 Harness 层先查 Skill 注册中心再通过沙箱执行执行前检查审批条件。# 文件路径agents/data_worker.py from harness.sandbox.docker_sandbox import DockerSandbox from harness.skill.skill_registry import SkillRegistry class DataWorker: 数据分析 Worker负责执行 SQL 查询等数据操作 def __init__(self, sandbox: DockerSandbox, registry: SkillRegistry): self.sandbox sandbox self.registry registry def execute(self, action: str): if action query_order_data: # 从 Skill 注册中心获取查询 Skill skill self.registry.get(sql_query) # 把 Skill 的执行函数需要的信息打包 code f import sqlite3 # 模拟订单数据查询 conn sqlite3.connect(/workspace/order.db) cursor conn.cursor() cursor.execute(SELECT region, COUNT(*) as cnt FROM orders GROUP BY region) rows cursor.fetchall() # 注意实际项目不要把凭据写死在代码里应通过环境变量注入 for row in rows: print(f{{row[0]}},{{row[1]}}) conn.close() # 在沙箱中执行 result self.sandbox.run_code( codecode, timeout15, memory_limit256m ) return { worker: data_worker, action: action, sandbox_result: result } return {worker: data_worker, action: action, error: unknown action}7.4 入口主程序最后把所有模块组装起来。# 文件路径main.py from harness.sandbox.docker_sandbox import DockerSandbox from harness.skill.skill_registry import SkillRegistry, Skill from agents.planner import PlannerAgent from agents.data_worker import DataWorker from agents.report_worker import ReportWorker def demo_sql_query(): SkillSQL 查询用 sqlite 模拟 import sqlite3 conn sqlite3.connect(/workspace/order.db) cursor conn.cursor() cursor.execute(SELECT region, COUNT(*) as cnt FROM orders GROUP BY region) for row in cursor.fetchall(): print(f{row[0]},{row[1]}) conn.close() def demo_report(): Skill生成 Markdown 报告 print(# 订单数据分析报告) print() def main(): # 初始化沙箱 sandbox DockerSandbox( imagepython:3.11-slim, network_disabledTrue ) # 初始化 Skill 注册中心并注册 Skill registry SkillRegistry() registry.register(Skill( namesql_query, version1.0.0, description执行 SQL 查询语句返回结果集, parameters{query: {type: string}}, implementdemo_sql_query, authorteam )) registry.register(Skill( namemarkdown_report, version1.0.0, description生成 Markdown 格式报告, parameters{content: {type: string}}, implementdemo_report, authorteam )) # 创建 Worker data_worker DataWorker(sandbox, registry) report_worker ReportWorker(sandbox, registry) # 创建主控 Agent 并注册 Worker planner PlannerAgent(sandbox, registry) planner.register_worker(data_worker, data_worker) planner.register_worker(report_worker, report_worker) # 执行任务 results planner.run_task(分析最近订单数据并生成报告) # 打印结果 for worker_name, result in results.items(): print(f\n[{worker_name}] 执行结果:) if sandbox_result in result: sandbox_result result[sandbox_result] print(fexit_code: {sandbox_result[exit_code]}) print(fstdout: {sandbox_result[stdout]}) if sandbox_result[stderr]: print(fstderr: {sandbox_result[stderr]}) if __name__ __main__: main()7.5 运行与验证要运行这个骨架首先确保本机 Docker 可用然后安装依赖pip install docker运行主程序python main.py预期输出效果如下关键部分[Planner] 接收任务: 分析最近订单数据并生成报告 [Planner] 任务拆解: [{worker: data_worker, action: query_order_data}, {worker: report_worker, action: generate_report}] [Planner] 分发任务到 data_worker: query_order_data [data_worker] 执行结果: exit_code: 0 stdout: 华东,120 华南,80 华北,65需要注意示例中的demo_sql_query直接封装在 Python 函数里并没有真正放到沙箱容器中执行。实际生产环境Skill 的implement应该是一个存到 Docker 容器内执行的脚本或函数序列而不是宿主机函数的引用。这个骨架演示的是控制流和数据流隔离细节需要结合第 4 节的沙箱代码一起实现。如果运行失败优先检查三点Docker 服务是否启动docker ps能否正常执行沙箱镜像是否存在docker images | grep python查看是否有python:3.11-slim权限是否足够当前用户是否在docker用户组中。8. 生产环境落地常见问题与排查思路从示例骨架到生产环境中间还有很多坑。下面整理几个高频问题按表格形式给出排查思路。问题现象可能原因排查方式解决方案沙箱容器启动失败提示read-only file system代码在初始化时尝试写系统目录查看容器启动日志和代码路径修改代码只允许写入/workspace等 tmpfs 目录Agent 执行 SQL 时提示连接数据库失败沙箱网络被禁用或数据库账号权限不足先确认数据库账号是否仅包含只读权限再检查网络策略将数据库账号权限改为最小只读需要外网时走白名单Skill 调用时模型选错参数Skill 的 parameters schema 描述不清晰查看模型调用 Skill 时实际传入的参数优化参数 schema 的 description增加示例值人工审批超时大量任务被杀掉审批通道通知不及时审批人未看到请求查看审批中心推送日志增加多通道通知企业微信 短信缩短轮询间隔容器执行任务后残留大量容器异常退出时未清理资源执行docker ps -a看残留容器用finally强制清理增加定时清理任务Skill 进化后效果反而变差新版本没有经过充分评估就上线对比新旧版本的 success_count 和反馈增加灰度发布新版本先给 10% 流量多 Agent 并发任务过多Docker 资源耗尽没有并发上限控制查看docker stats和宿主机资源监控增加信号量控制并发数设置全局容器数量上限9. 企业级落地的关键架构决策很多团队看到前面这些模块后会立刻开始写代码。但真正决定系统能否长期稳定运行的往往是几个架构层的决策。这些决策直接决定了你的 Harness Engineering 系统是“能用”还是“好用”。决策一审批中心必须独立于 Agent 运行。不要把审批逻辑写进某个 Worker Agent 的代码里而是做成独立服务。原因是审批是一个贯穿多个 Agent 的横切关注点它需要统一管理而不是被分散到每个 Agent 中各自实现。独立的审批中心还能积累全系统的审批历史便于后续分析和优化。决策二Skill 仓库最好与代码仓库分离。Skill 是运行时产物它包含的不只是代码还有描述、参数 schema、使用统计、反馈记录。把 Skill 存到独立的数据库或 Git 仓库可以独立于主应用做版本管理和发布。尤其是在自进化机制开启后Skill 的更新频率会远高于主应用代码。决策三审计日志不能被 Agent 自己擦除。这是一个容易被忽略但非常致命的问题。如果日志和沙箱在同一个环境Agent 可以通过执行代码来修改日志文件审计就失去意义。正确做法是日志通过只写通道如专用的消息队列发送到独立的日志系统Agent 所在沙箱没有写日志系统的权限。决策四沙箱镜像和 Skill 的执行环境保持一致。很多团队在开发环境用本机 Python 测试 Skill然后部署到沙箱镜像里结果出现“本地能跑容器里报错”的问题。解决方案是要求 Skill 开发者默认在沙箱镜像环境中做开发验证或者至少保证依赖版本一致。决策五为大模型调用设计“安全兜底”。即使有沙箱、审批、Skill 注册中心大模型仍然可能在调用 Skill 时给出不合理的参数。比如 SQL 查询 Skill 接收到DROP TABLE类型的语句。这种风险不能只靠沙箱解决还需要在 SQL 解析层做语义校验只允许 SELECT 开头的语句禁止多语句执行。其他类型的 Skill 同理应该在输入侧做严格校验。10. 从一个 Harness 骨架到真正可运维的系统回到文章开头的判断2026 年再谈 AI Agent真正的工程门槛在于“把能力关进笼子”。Harness Engineering 不是一种新框架而是一套工程纪律。它不是某个可以 npm install 的包而是你设计 Agent 系统时必须遵守的约束原则默认拒绝、最小权限、全量审计、能力版本化、人工兜底。如果你正在从零开始设计企业级 AI Agent建议按照以下路径逐步推进。第一步先完成最小闭环一个 Planner、一个 Worker、一个沙箱、三个 Skill。不要一上来就追求复杂的多 Agent 编排。先验证“模型能不能在沙箱里稳定调用 Skill”。第二步上线审批流。只接入真正的高风险操作不要全链路加审批。观察审批延迟逐步优化通知渠道。第三步采集数据和反馈。为每个 Skill 增加成功率和反馈记录建立评估基线。这一步是开启自进化的前提。第四步用评估数据驱动 Skill 迭代。新 Skill 先灰度发布成功后放大流量失败则回滚。第五步完善可观测性和审计。把所有 Agent 的执行记录、审批记录、Skill 调用记录打通确保系统出了任何问题都能定位到具体环节。最后回到工程实践本身Harness Engineering 的价值不是让 AI Agent 变得更“聪明”而是让 AI Agent 在复杂的生产环境里变得更可靠、更可控、更可追溯。这三件事恰恰是企业级 AI 系统最难做到的。如果你能把沙箱、Skill、审批这三个机制真正落地你的 Agent 系统就已经超过了大多数还停留在“调 Prompt”阶段的团队。 企业级AI Agent工程化：从Prompt到Harness的落地实践
从“调 Prompt”到“建 Harness”企业级 AI Agent 工程化真正缺的是这一层过去两年很多团队做 AI Agent 的方式其实是一样的把 Claude、GPT 或开源模型的 API 接进来写一大堆 system prompt然后给 Agent 挂上几个工具函数就开始宣称“我们已经完成 AI 落地”。等真正上线才发现完全不是那么回事。需要保证 Agent 拿不到不该访问的数据库需要约束它不要在一次循环里把上千个临时文件写到宿主机需要让沉淀下来的一段有效技能能被团队里其他 Agent 复用而不是锁在某个人的聊天记录里更要命的是当 Agent 判断错误、连续重试、甚至开始“自由发挥”的时候你根本不知道从哪里打断它。所以 2026 年再谈 AI Agent核心词已经从“能不能跑通”变成了“能不能管控”。而 Harness Engineering直译是“驾驭工程”或“约束工程”正是为了解决这批问题出现的工程方法论。它强调的从来不是把模型能力做得更大而是通过一套工程结构把不可控的模型行为装进可控的边界里沙箱隔离执行环境、自进化 Skill 沉淀能力、人工介入保留最终决策权。这篇文章会先讲清楚 Harness Engineering 到底解决什么问题再给出一个企业级多 Agent 协同项目的完整工程骨架包括沙箱设计、Skill 的自进化机制、人机协同的介入节点以及一套可以直接落地的最小实现。文章不会停留在概念层面你可以照着代码把一个带沙箱约束、Skill 热更新、人工审批点的多 Agent 骨架搭起来。1. 为什么 2026 年 AI Agent 工程化绕不开 Harness Engineering先说一个很容易被忽视的判断当前 AI Agent 的瓶颈不在模型能力而在工程侧对行为的约束能力。模型能力这半年提升得很快长上下文、复杂推理、工具调用都已经逐步成熟。多 Agent 协同也早就不是论文里的概念Coze、Dify、n8n 这类平台已经让 Agent 工作流搭建变得足够简单。但你如果去问那些真正把 Agent 推到生产环境的团队他们最痛苦的不是“模型不够聪明”而是以下三个问题。第一个问题Agent 的执行边界不可控。模型会“自由发挥”。你让它读取一份 CSV它可能顺手把同目录下的配置文件也读进来你让它执行一段 Python 代码它可能在宿主机上创建文件、连接网络、读取环境变量。在本地开发环境这可能只是一个小意外但在企业内网、生产数据库、核心代码仓库面前这种不可控就是事故。第二个问题能力沉淀不下来。你在一个 Agent 的 prompt 里调出一套很有效的 JSON 处理逻辑怎么迁移到另一个 Agent复制 prompt那个 Agent 未必有同样的工具集。写成代码那又回到了传统软件开发模型的灵活性全丢了。很多团队每个 Agent 都是“独立烟囱”经验无法复用效率越做越低。第三个问题没有人机边界。Agent 出错时是让它自己重试还是让人介入如果人介入在哪个环节介入、能修改什么、能审批什么大部分项目根本没有设计过这个边界结果就是两个极端要么全程无人值守出事才发现要么每个步骤都弹确认框Agent 变成了“自动点下一步”的工具毫无效率可言。Harness Engineering 对这三个问题的回答是一致的Agent 的能力可以复杂但 Agent 的行为边界必须简单。它要求把环境隔离、技能沉淀、人机审批当作和模型选型同等重要的一等工程构件来设计。这也是为什么它会被越来越多企业级 AI 项目列为必选项。2. Harness Engineering 核心概念Skill、Sandbox、介入机制在进入代码之前先把 Harness Engineering 的三个关键概念讲清楚。后面所有设计和代码都围绕这三个概念展开。2.1 Skill不是工具函数是“可被模型自主调用的经验单元”很多材料把 Skill 翻译成“技能”这个翻译容易让人误以为它只是工具函数的另一个名字。实际上Skill 和普通工具函数有本质区别。普通工具函数是开发者写死的能力例如“计算两个日期之间的天数”“调用某个 HTTP 接口”。它由开发者控制Agent 做不了任何自主选择。Skill 则是一个带描述、带校验、带版本、可被模型自主学习调用的经验单元。它通常包含一段自然语言描述告诉模型“什么时候该用这个 Skill”一个可执行实现可以是 Python 函数、API 调用甚至是一段 Shell 脚本一组参数定义和输入校验逻辑元信息包括版本号、作者、运行所需的权限级别可选的“自进化记录”用于保存该 Skill 的迭代历史。一个典型的 Skill 示例analyze_nginx_log分析 Nginx 日志。它不只是“解析日志文件”这么简单它可能封装了你团队平时分析 Nginx 日志时的完整套路先看 5xx 分布再看上游响应时间最后按 IP 归类失败请求。这个套路沉淀成 Skill 后任何 Agent 遇到 Nginx 日志相关任务都可以直接调用而不是从零开始设计分析思路。2.2 Sandbox把执行环境变成“一次性的”Sandbox沙箱是 Harness Engineering 里最核心的安全机制。它的目标非常明确让 Agent 的执行代码跑在一个受限、可丢弃、可审计的环境里。具体来说企业级 Agent 沙箱需要具备以下能力能力说明典型实现方式文件系统隔离Agent 只能访问指定目录不能读取宿主机敏感文件Docker volume、容器只读根文件系统网络隔离默认禁止外网访问按白名单放行Docker network、iptables、代理白名单系统调用限制禁止危险 syscall如 mount、ptracegVisor、Firecracker、seccomp资源限制限制 CPU、内存、磁盘、执行时长cgroup、Docker--memory、超时机制可丢弃性执行结束后环境销毁不保留中间状态容器生命周期管理沙箱的实现技术栈从轻到重依次是进程级隔离如 Firecracker、容器级隔离Docker、系统调用级隔离gVisor。对于大多数企业应用Docker 容器加资源限制是性价比最高的方案。如果你的场景涉及不可信代码执行才需要考虑 gVisor 或 Firecracker 这类更重的隔离。2.3 人工介入不是打断是“可控的授权节点”人工介入机制是 Harness Engineering 区别于纯自动化 AI 系统的重要标志。它的设计原则是不是所有步骤都需要人审批但所有高影响操作都必须有人的授权点。典型需要人工介入的场景包括执行写数据库操作尤其是 UPDATE、DELETE、DROP向外部发送消息邮件、企业微信、短信合并代码、创建发布分支申请超过预算的资源访问标记了敏感级别的数据。人工介入不应该是“弹窗打断”而应该是一个异步审批任务Agent 发出审批请求挂起当前流程等待人在审批界面做出决定然后再继续执行。这一点放到多 Agent 架构里尤其重要主 Agent 可以继续处理其他事务只有相关子任务被挂起。3. 企业级多 Agent 协同的整体架构设计把三个核心概念组合起来一个企业级多 Agent 协同系统的架构就清晰了。下面是一个完整的参考架构包含五个层次。┌─────────────────────────────────────────────────────────┐ │ 用户层 / 接入层 │ │ Web 控制台 | 企业微信 / 钉钉 Bot | API 网关 │ └─────────────────────────────────────────────────────────┘ │ ┌─────────────────────────────────────────────────────────┐ │ 编排层 / Orchestration │ │ Supervisor Agent主控 Agent │ │ 任务规划 | 子任务分发 | Skill 推荐 | 介入决策 │ └─────────────────────────────────────────────────────────┘ │ ┌─────────────────────────────────────────────────────────┐ │ 执行层 / Workers │ │ 数据分析 Agent | 代码生成 Agent | 运维操作 Agent │ │ Worker Agent执行 Agent可水平扩展 │ └─────────────────────────────────────────────────────────┘ │ ┌─────────────────────────────────────────────────────────┐ │ Harness 层 │ │ Skill 注册中心 | 沙箱管理 | 审批中心 | 审计日志 │ └─────────────────────────────────────────────────────────┘ │ ┌─────────────────────────────────────────────────────────┐ │ 基础设施层 │ │ Docker / K8s | 向量数据库 | 对象存储 | 消息队列 │ └─────────────────────────────────────────────────────────┘这个架构的关键是把决策、执行、约束三个维度拆开编排层负责规划它决定“做什么”“谁来做”“需要什么 Skill”但它自己不做具体执行执行层负责干活每个 Worker Agent 处理一项具体任务Harness 层负责约束它不关心任务本身只关心“执行是否越界”“Skill 是否有效”“是否需要在某个节点暂停等人审批”。传统单体 Agent 把所有职责揉在一起导致安全约束难以独立演进。而 Harness Engineering 的架构天然支持安全团队独立维护沙箱规则、业务团队独立维护 Skill、运维团队独立维护审批流各司其职。3.1 多 Agent 的任务流转示例用一个具体场景说明这套架构的运转流程业务方提出一个任务——“分析最近一周订单量下降的原因并生成一份报告”。主控 Agent 接收任务拆解为数据提取、趋势分析、竞品信息检索、报告生成四个子任务。主控 Agent 查询 Skill 注册中心发现“订单数据查询 Skill”“时序趋势分析 Skill”“报告生成 Skill”可用。数据提取子任务被分发到数据分析 Worker Agent。Worker 在沙箱中执行 SQL 查询脚本沙箱只允许连接数据仓库的只读账号。趋势分析子任务涉及调用外部 API 获取行业数据。Harness 层检测到该 Worker 的沙箱默认禁止外网访问于是发起人工审批请求。审批人确认后白名单临时放行该 API。各子任务完成后主控 Agent 汇总结果调用报告生成 Skill 输出最终报告。整个流程的每一次 Skill 调用、每一次沙箱操作、每一轮人工审批都写入审计日志。可以看到多 Agent 协同并不神秘。它的难点在于当一个任务被拆成多个子任务后安全边界、权限边界、审批边界仍然要清晰。这正是 Harness 层存在的意义。4. 沙箱机制的设计与实现沙箱是整个 Harness Engineering 体系里最偏底层、也最容易踩坑的部分。下面分为设计要点和代码实现两部分讲解。4.1 沙箱设计四个关键决策决策一选择 Docker 还是 Kubernetes 运行沙箱。单机或小团队场景直接用 Docker Engine API 即可。需要大规模并发、自动伸缩时再上 Kubernetes。Docker 的好处是简单直接一个容器一个任务跑完即删。决策二镜像要最小化并且不包含生产凭证。沙箱镜像只需要 Python 运行时、常用工具库、网络调试工具。生产环境的数据库密码、API Key 一律通过环境变量在运行时注入且按最小权限分配。这里真正容易踩坑的是很多团队图省事把包含凭证的“全家桶镜像”直接当成沙箱镜像用等于给 Agent 送上了所有钥匙。决策三网络默认关闭需要外网时走白名单或审批。沙箱容器创建时默认不使用 bridge 网络或使用隔离网络。Agent 需要访问某个外部 API 时由 Harness 层动态添加规则或修改代理配置。决策四强制超时和资源限制。即使逻辑完全正确Agent 也可能因为循环失控而长时间运行。所有沙箱任务必须设置执行超时通常建议 30 秒到 5 分钟以及明确的 CPU 和内存上限。4.2 基于 Docker SDK 的沙箱核心代码下面用 Python 的dockerSDK实现一个最简沙箱管理器。它的职责是接收一段代码在隔离容器里执行返回标准输出、错误信息和执行指标。# 文件路径harness/sandbox/docker_sandbox.py import docker import uuid import time class DockerSandbox: 基于 Docker 的 Agent 沙箱执行器 def __init__(self, imagepython:3.11-slim, network_disabledTrue): self.client docker.from_env() self.image image self.network_disabled network_disabled def run_code(self, code: str, timeout: int 30, memory_limit: str 256m, env: dict None, workdir: str /workspace): 在沙箱中执行 Python 代码 :param code: 要执行的代码 :param timeout: 超时秒数 :param memory_limit: 内存限制Docker 格式如 256m :param env: 注入的环境变量最小权限 :param workdir: 容器内工作目录 container_name fagent-sandbox-{uuid.uuid4().hex[:8]} exec_id None try: # 创建隔离容器 container self.client.containers.run( imageself.image, command[/bin/sh, -c, sleep 5], namecontainer_name, detachTrue, network_disabledself.network_disabled, # 默认禁用网络 mem_limitmemory_limit, pids_limit100, working_dirworkdir, environmentenv or {}, read_onlyTrue, # 根文件系统只读 tmpfs{/workspace: rw,noexec,nosuid,size64m} ) # 将代码写入容器内的可写临时目录 # 注意因为根文件系统只读这里使用 tmpfs 挂载的 /workspace code_path f/workspace/main.py # 使用 exec 写入代码 write_cmd fcat {code_path} EOF\n{code}\nEOF container.exec_run(write_cmd, userroot) # 在容器内执行代码 result container.exec_run( cmd[python, /workspace/main.py], userroot, demuxTrue, timeouttimeout ) stdout result.output[0].decode() if result.output[0] else stderr result.output[1].decode() if result.output[1] else exit_code result.exit_code return { exit_code: exit_code, stdout: stdout, stderr: stderr, container_name: container_name } except docker.errors.ContainerError as e: return {exit_code: -1, stderr: str(e)} except Exception as e: return {exit_code: -1, stderr: fSandbox execution failed: {e}} finally: # 无论成功失败容器都销毁 try: container self.client.containers.get(container_name) container.remove(forceTrue) except Exception: pass这段代码有几个值得注意的设计read_onlyTrue把镜像根文件系统设为只读Agent 无法修改系统文件tmpfs挂载/workspace提供临时可写目录但noexec禁止在该目录直接执行二进制文件network_disabledTrue默认禁网finally块确保容器无论成功失败都被删除不残留中间状态。实际生产环境还可以继续强化把代码写入改为通过 Docker API 的put_archive避免exec_run拼接命令带来的注入风险把网络白名单做成持久化配置而不是每次手动设置。但上面的最小实现已经足够演示沙箱机制的核心思路。4.3 沙箱管理的核心原则默认拒绝 临时授权沙箱配置不是“尽量严一点”而是“默认拒绝一切按需临时授权”。这个原则具体到实现上是默认无网络不对全部端口开放默认无宿主机目录挂载需要读写的数据通过 API 显式传入默认无环境变量需要的凭证单独注入默认无特权模式不用--privileged。任何“为方便调试而放开”的配置都应该在代码评审阶段被打回。Sandbox 是安全边界不是开发环境。5. 自进化 Skill让 Agent 的能力在运行中沉淀如果说沙箱是 Harness Engineering 的“安全带”那自进化 Skill 就是它的“发动机”。自进化 Skill 是指Skill 能够在实际运行后基于效果反馈自动或半自动地更新自身版本。5.1 为什么需要自进化 Skill没有自进化机制时Skill 是一次性开发的软件组件写完后固定不变效果如何完全取决于最初的开发者思考得是否周全。但 Agent 的调用场景是开放的。一个“报告生成 Skill”第一次运行时可能只会输出纯文本报告第二次运行时用户反馈“希望加上数据可视化图表”第三次运行时又发现“PDF 格式更适合汇报”。如果每次需求变化都要开发者手动改代码Skill 就和普通程序没有区别了。自进化 Skill 的流程是运行 → 收集反馈 → 评估 → 更新版本 → 重新注册。5.2 自进化 Skill 的架构与代码一个实用的自进化 Skill 系统包含两个部分Skill 存储和进化管线。下面给出 Skill 定义模型和更新逻辑示例。# 文件路径harness/skill/skill_registry.py from dataclasses import dataclass, field from typing import Dict, Callable, Any import datetime import json dataclass class Skill: Skill 数据模型 name: str # Skill 名称 version: str # 语义化版本 description: str # 自然语言描述用于模型判断何时调用 parameters: Dict[str, Any] # 参数 schema implement: Callable # 实际执行函数 author: str system tags: list field(default_factorylist) create_time: str field(default_factorylambda: datetime.datetime.now().isoformat()) success_count: int 0 # 成功调用次数 fail_count: int 0 # 失败调用次数 feedback: list field(default_factorylist) # 反馈记录 def to_desc(self) - str: 生成给模型的结构化描述 return json.dumps({ name: self.name, description: self.description, parameters: self.parameters, version: self.version, tags: self.tags }, ensure_asciiFalse) class SkillRegistry: Skill 注册中心支持版本管理和自进化 def __init__(self): self._skills: Dict[str, Skill] {} def register(self, skill: Skill): 注册新 Skill 或覆盖旧版本 self._skills[skill.name] skill def get(self, name: str) - Skill: return self._skills[name] def list_all(self) - list: return [s.to_desc() for s in self._skills.values()] def record_result(self, name: str, success: bool, feedback: str ): 记录一次调用结果用于进化评估 skill self._skills.get(name) if not skill: return if success: skill.success_count 1 else: skill.fail_count 1 # 失败日志可以作为 Skill 进化的输入 skill.feedback.append({ time: datetime.datetime.now().isoformat(), success: success, detail: feedback }) def evolve(self, name: str, new_impl: Callable, new_version: str): 发布 Skill 新版本 old self._skills.get(name) if not old: raise ValueError(fSkill {name} not found) # 保留历史统计替换实现和版本 new_skill Skill( nameold.name, versionnew_version, descriptionold.description, parametersold.parameters, implementnew_impl, authorold.author, tagsold.tags, success_countold.success_count, fail_countold.fail_count, feedbackold.feedback ) self._skills[name] new_skill这套模型并不复杂但它定义了一条关键规则Skill 的进化必须是版本化的旧的统计信息不能丢。如果新版本效果反而变差你需要有能力回滚到旧版本并且知道旧版本之前的成功率是多少。5.3 自进化 Skill 的三种进化模式根据人工参与程度自进化 Skill 可以分为三档第一档人工编辑式。Agent 运行后系统把效果不佳的调用日志和用户反馈汇集成建议推送给人。人修改 Skill 代码后以新版本号发布。这是最稳妥的模式适合对准确性要求极高的场景。第二档模型辅助编辑式。系统把失败的调用参数、输出、用户反馈发给大模型让大模型生成新的 Skill 实现代码然后由人确认后发布。这是当前生产环境里最推荐的折中方案既有进化速度又有安全兜底。第三档全自动编辑式。系统根据反馈自动修改 Skill 代码并发布。这种模式适合低风险、高频迭代的 Skill例如日志格式化、数据清洗。全自动模式需要额外的测试套件否则很容易把错误版本发布上线。5.4 Skill 自进化的关键成功因素实际落地时比代码更重要的是一套评估机制。没有评估的进化等于盲目更新。评估机制至少要包含三个维度成功率调用过程中是否报错、结果质量输出和期望的匹配度通常需要人工或另一个模型评估、效率执行时间、Token 消耗。只有当三个维度都满足阈值时新版本才允许上线。比如成功率大于 95%结果质量评分大于 4.0/5.0执行耗时不高于旧版本的 1.2 倍。这里真正容易踩坑的地方是很多团队只看“模型觉得这个 Skill 效果不错”就发布了。但“模型觉得不错”不等于“用户觉得不错”。在生产环境必须采集用户侧的最终反馈而不是只依赖模型自评。6. 人工介入设计人机协同的审批体系人工介入是 Harness Engineering 中最容易被低估的部分。很多人觉得“有人审批”就安全了但实际上如果审批节点设计得不好要么审批流形同虚设要么严重拖慢流程。6.1 三层介入模型一个合理的人工介入体系通常分为三层。第一层任务级介入。发生在 Agent 开始执行任务之前。例如用户在创建任务时明确选择“这个任务需要人工审批每个写操作”。任务级介入适合那些本来就需要严格控制的业务场景。第二层操作级介入。发生在 Agent 执行某个具体操作之前。这是最常见的介入类型。Harness 层会检测即将执行的命令是否命中高风险规则写数据库、发外网请求、删除文件命中则暂停执行生成审批任务。第三层异常级介入。发生在 Agent 执行出现异常或长时间没有进展时。系统根据告警阈值通知人工介入处理。这一层通常容易被忽略它解决的是“Agent 卡住了没人发现”的问题。6.2 人工介入的代码示例下面用消息队列 审批状态机的思路实现一个操作级介入的最小示例。核心思路Agent 在沙箱中执行高风险命令时不直接执行而是发送审批请求等待审批结果。# 文件路径harness/approval/approval_flow.py import enum import time import uuid from dataclasses import dataclass class ApprovalStatus(enum.Enum): PENDING pending APPROVED approved REJECTED rejected TIMEOUT timeout dataclass class ApprovalRequest: request_id: str agent_name: str action_desc: str # 要执行的操作描述 target_resource: str # 目标资源 requested_by: str # 请求人/触发 Agent status: ApprovalStatus ApprovalStatus.PENDING comment: str expire_at: float 0.0 class ApprovalCenter: 审批中心负责创建审批请求、处理审批结果 def __init__(self, timeout_seconds: int 300): self._requests {} self.timeout_seconds timeout_seconds def create_request(self, agent_name: str, action_desc: str, target_resource: str) - ApprovalRequest: req ApprovalRequest( request_iduuid.uuid4().hex[:12], agent_nameagent_name, action_descaction_desc, target_resourcetarget_resource, expire_attime.time() self.timeout_seconds ) self._requests[req.request_id] req # 实际生产环境将审批请求推送到企业微信/钉钉/邮件 # self._push_notification(req) return req def approve(self, request_id: str, comment: str ) - bool: 人工审批通过 req self._requests.get(request_id) if not req or req.status ! ApprovalStatus.PENDING: return False req.status ApprovalStatus.APPROVED req.comment comment return True def reject(self, request_id: str, comment: str ) - bool: 人工审批拒绝 req self._requests.get(request_id) if not req or req.status ! ApprovalStatus.PENDING: return False req.status ApprovalStatus.REJECTED req.comment comment return True def wait_for_result(self, request_id: str, timeout: int 60) - ApprovalStatus: 阻塞等待审批结果。 实际生产环境建议用异步回调或 websocket而不是轮询。 start time.time() while time.time() - start timeout: req self._requests.get(request_id) if req.status ! ApprovalStatus.PENDING: return req.status if time.time() req.expire_at: req.status ApprovalStatus.TIMEOUT return ApprovalStatus.TIMEOUT time.sleep(1) return ApprovalStatus.TIMEOUT这段代码展示了审批状态机的核心流转。集成到 Agent 执行流程时逻辑是这样的# 在高风险操作前调用审批 if is_high_risk_action(action): req approval_center.create_request( agent_namedata_agent, action_desc写数据库更新订单状态, target_resourceorder_db.orders ) status approval_center.wait_for_result(req.request_id) if status ApprovalStatus.APPROVED: execute_action(action) else: abort_action(action)6.3 实现人工介入的三个工程原则原则一审批请求必须携带足够上下文。审批人需要看到的不只是“Agent 请求写数据库”这句话还应该看到完整的 SQL、影响行数预估、涉及哪些表、由哪个任务触发、Agent 的判断依据。否则审批就是走过场。原则二审批必须有超时机制。如果没有超时Agent 会无限期等待任务队列会堆积。超时后的默认策略通常有两种自动拒绝或自动降级。建议默认自动拒绝把超时当作失败处理。原则三审批过程必须全量审计。谁审批的、什么时候审批的、审批时看到了什么内容、最终结果是什么全部留存。这是企业合规的基础要求也是后续优化审批效率的数据来源。7. 完整落地一个多 Agent 协同项目的工程骨架前面已经拆解了沙箱、Skill、审批三个核心模块。这一节把它们组装成一个可运行的最小工程骨架。假设我们要实现一个简单的“数据分析多 Agent 系统”一个 Planner Agent接收任务拆解成子任务两个 Worker Agent一个负责数据查询一个负责报告生成Harness 层统一托管沙箱执行和 Skill 调用。7.1 工程目录结构ai-harness-project/ ├── main.py # 入口启动编排 ├── requirements.txt # 依赖 ├── config/ │ ├── agents.yaml # Agent 配置 │ └── sandbox.yaml # 沙箱配置 ├── harness/ │ ├── sandbox/ │ │ └── docker_sandbox.py # 沙箱执行器 │ ├── skill/ │ │ └── skill_registry.py # Skill 注册中心 │ └── approval/ │ └── approval_flow.py # 审批中心 ├── agents/ │ ├── planner.py # 主控 Agent │ ├── data_worker.py # 数据分析 Worker │ └── report_worker.py # 报告生成 Worker ├── skills/ │ ├── sql_query.py # SQL 查询 Skill │ └── markdown_report.py # 报告生成 Skill └── logs/ └── audit.log # 审计日志7.2 主控 Agent 的编排逻辑主控 Agent 的职责是拆解任务、分发任务、汇总结果。这里演示一个最简单的 Planner它的核心逻辑是根据任务类型选择 Worker然后调用 Worker 的执行接口。实际生产环境这个“选择 Worker”的动作通常由大模型完成但控制流分发、等待、汇总应该是代码逻辑不能完全依赖模型自由发挥。# 文件路径agents/planner.py import json from harness.sandbox.docker_sandbox import DockerSandbox from harness.skill.skill_registry import SkillRegistry class PlannerAgent: 主控 Agent负责任务分发和结果汇总 def __init__(self, sandbox: DockerSandbox, registry: SkillRegistry): self.sandbox sandbox self.registry registry self.workers {} def register_worker(self, name: str, worker): self.workers[name] worker def run_task(self, task_desc: str): 任务入口。 实际生产环境这里会调用 LLM 做任务拆解。 示例中为了演示控制流直接按固定逻辑分发。 print(f[Planner] 接收任务: {task_desc}) # 步骤 1任务规划演示逻辑实际用 LLM 拆解 plan self._plan(task_desc) print(f[Planner] 任务拆解: {json.dumps(plan, ensure_asciiFalse)}) # 步骤 2分发子任务 results {} for step in plan: worker_name step[worker] worker self.workers.get(worker_name) if not worker: print(f[Planner] 未找到 Worker: {worker_name}) continue print(f[Planner] 分发任务到 {worker_name}: {step[action]}) result worker.execute(step[action]) results[worker_name] result # 步骤 3汇总结果 print([Planner] 所有子任务完成开始汇总) return results def _plan(self, task_desc: str): 示意性规划逻辑。生产环境替换为 LLM 规划。 if 订单 in task_desc or 数据 in task_desc: return [ {worker: data_worker, action: query_order_data}, {worker: report_worker, action: generate_report} ] return [{worker: data_worker, action: unknown}]7.3 Worker Agent 与沙箱执行的集成Worker Agent 是实际执行者。它的执行过程要经过 Harness 层先查 Skill 注册中心再通过沙箱执行执行前检查审批条件。# 文件路径agents/data_worker.py from harness.sandbox.docker_sandbox import DockerSandbox from harness.skill.skill_registry import SkillRegistry class DataWorker: 数据分析 Worker负责执行 SQL 查询等数据操作 def __init__(self, sandbox: DockerSandbox, registry: SkillRegistry): self.sandbox sandbox self.registry registry def execute(self, action: str): if action query_order_data: # 从 Skill 注册中心获取查询 Skill skill self.registry.get(sql_query) # 把 Skill 的执行函数需要的信息打包 code f import sqlite3 # 模拟订单数据查询 conn sqlite3.connect(/workspace/order.db) cursor conn.cursor() cursor.execute(SELECT region, COUNT(*) as cnt FROM orders GROUP BY region) rows cursor.fetchall() # 注意实际项目不要把凭据写死在代码里应通过环境变量注入 for row in rows: print(f{{row[0]}},{{row[1]}}) conn.close() # 在沙箱中执行 result self.sandbox.run_code( codecode, timeout15, memory_limit256m ) return { worker: data_worker, action: action, sandbox_result: result } return {worker: data_worker, action: action, error: unknown action}7.4 入口主程序最后把所有模块组装起来。# 文件路径main.py from harness.sandbox.docker_sandbox import DockerSandbox from harness.skill.skill_registry import SkillRegistry, Skill from agents.planner import PlannerAgent from agents.data_worker import DataWorker from agents.report_worker import ReportWorker def demo_sql_query(): SkillSQL 查询用 sqlite 模拟 import sqlite3 conn sqlite3.connect(/workspace/order.db) cursor conn.cursor() cursor.execute(SELECT region, COUNT(*) as cnt FROM orders GROUP BY region) for row in cursor.fetchall(): print(f{row[0]},{row[1]}) conn.close() def demo_report(): Skill生成 Markdown 报告 print(# 订单数据分析报告) print() def main(): # 初始化沙箱 sandbox DockerSandbox( imagepython:3.11-slim, network_disabledTrue ) # 初始化 Skill 注册中心并注册 Skill registry SkillRegistry() registry.register(Skill( namesql_query, version1.0.0, description执行 SQL 查询语句返回结果集, parameters{query: {type: string}}, implementdemo_sql_query, authorteam )) registry.register(Skill( namemarkdown_report, version1.0.0, description生成 Markdown 格式报告, parameters{content: {type: string}}, implementdemo_report, authorteam )) # 创建 Worker data_worker DataWorker(sandbox, registry) report_worker ReportWorker(sandbox, registry) # 创建主控 Agent 并注册 Worker planner PlannerAgent(sandbox, registry) planner.register_worker(data_worker, data_worker) planner.register_worker(report_worker, report_worker) # 执行任务 results planner.run_task(分析最近订单数据并生成报告) # 打印结果 for worker_name, result in results.items(): print(f\n[{worker_name}] 执行结果:) if sandbox_result in result: sandbox_result result[sandbox_result] print(fexit_code: {sandbox_result[exit_code]}) print(fstdout: {sandbox_result[stdout]}) if sandbox_result[stderr]: print(fstderr: {sandbox_result[stderr]}) if __name__ __main__: main()7.5 运行与验证要运行这个骨架首先确保本机 Docker 可用然后安装依赖pip install docker运行主程序python main.py预期输出效果如下关键部分[Planner] 接收任务: 分析最近订单数据并生成报告 [Planner] 任务拆解: [{worker: data_worker, action: query_order_data}, {worker: report_worker, action: generate_report}] [Planner] 分发任务到 data_worker: query_order_data [data_worker] 执行结果: exit_code: 0 stdout: 华东,120 华南,80 华北,65需要注意示例中的demo_sql_query直接封装在 Python 函数里并没有真正放到沙箱容器中执行。实际生产环境Skill 的implement应该是一个存到 Docker 容器内执行的脚本或函数序列而不是宿主机函数的引用。这个骨架演示的是控制流和数据流隔离细节需要结合第 4 节的沙箱代码一起实现。如果运行失败优先检查三点Docker 服务是否启动docker ps能否正常执行沙箱镜像是否存在docker images | grep python查看是否有python:3.11-slim权限是否足够当前用户是否在docker用户组中。8. 生产环境落地常见问题与排查思路从示例骨架到生产环境中间还有很多坑。下面整理几个高频问题按表格形式给出排查思路。问题现象可能原因排查方式解决方案沙箱容器启动失败提示read-only file system代码在初始化时尝试写系统目录查看容器启动日志和代码路径修改代码只允许写入/workspace等 tmpfs 目录Agent 执行 SQL 时提示连接数据库失败沙箱网络被禁用或数据库账号权限不足先确认数据库账号是否仅包含只读权限再检查网络策略将数据库账号权限改为最小只读需要外网时走白名单Skill 调用时模型选错参数Skill 的 parameters schema 描述不清晰查看模型调用 Skill 时实际传入的参数优化参数 schema 的 description增加示例值人工审批超时大量任务被杀掉审批通道通知不及时审批人未看到请求查看审批中心推送日志增加多通道通知企业微信 短信缩短轮询间隔容器执行任务后残留大量容器异常退出时未清理资源执行docker ps -a看残留容器用finally强制清理增加定时清理任务Skill 进化后效果反而变差新版本没有经过充分评估就上线对比新旧版本的 success_count 和反馈增加灰度发布新版本先给 10% 流量多 Agent 并发任务过多Docker 资源耗尽没有并发上限控制查看docker stats和宿主机资源监控增加信号量控制并发数设置全局容器数量上限9. 企业级落地的关键架构决策很多团队看到前面这些模块后会立刻开始写代码。但真正决定系统能否长期稳定运行的往往是几个架构层的决策。这些决策直接决定了你的 Harness Engineering 系统是“能用”还是“好用”。决策一审批中心必须独立于 Agent 运行。不要把审批逻辑写进某个 Worker Agent 的代码里而是做成独立服务。原因是审批是一个贯穿多个 Agent 的横切关注点它需要统一管理而不是被分散到每个 Agent 中各自实现。独立的审批中心还能积累全系统的审批历史便于后续分析和优化。决策二Skill 仓库最好与代码仓库分离。Skill 是运行时产物它包含的不只是代码还有描述、参数 schema、使用统计、反馈记录。把 Skill 存到独立的数据库或 Git 仓库可以独立于主应用做版本管理和发布。尤其是在自进化机制开启后Skill 的更新频率会远高于主应用代码。决策三审计日志不能被 Agent 自己擦除。这是一个容易被忽略但非常致命的问题。如果日志和沙箱在同一个环境Agent 可以通过执行代码来修改日志文件审计就失去意义。正确做法是日志通过只写通道如专用的消息队列发送到独立的日志系统Agent 所在沙箱没有写日志系统的权限。决策四沙箱镜像和 Skill 的执行环境保持一致。很多团队在开发环境用本机 Python 测试 Skill然后部署到沙箱镜像里结果出现“本地能跑容器里报错”的问题。解决方案是要求 Skill 开发者默认在沙箱镜像环境中做开发验证或者至少保证依赖版本一致。决策五为大模型调用设计“安全兜底”。即使有沙箱、审批、Skill 注册中心大模型仍然可能在调用 Skill 时给出不合理的参数。比如 SQL 查询 Skill 接收到DROP TABLE类型的语句。这种风险不能只靠沙箱解决还需要在 SQL 解析层做语义校验只允许 SELECT 开头的语句禁止多语句执行。其他类型的 Skill 同理应该在输入侧做严格校验。10. 从一个 Harness 骨架到真正可运维的系统回到文章开头的判断2026 年再谈 AI Agent真正的工程门槛在于“把能力关进笼子”。Harness Engineering 不是一种新框架而是一套工程纪律。它不是某个可以 npm install 的包而是你设计 Agent 系统时必须遵守的约束原则默认拒绝、最小权限、全量审计、能力版本化、人工兜底。如果你正在从零开始设计企业级 AI Agent建议按照以下路径逐步推进。第一步先完成最小闭环一个 Planner、一个 Worker、一个沙箱、三个 Skill。不要一上来就追求复杂的多 Agent 编排。先验证“模型能不能在沙箱里稳定调用 Skill”。第二步上线审批流。只接入真正的高风险操作不要全链路加审批。观察审批延迟逐步优化通知渠道。第三步采集数据和反馈。为每个 Skill 增加成功率和反馈记录建立评估基线。这一步是开启自进化的前提。第四步用评估数据驱动 Skill 迭代。新 Skill 先灰度发布成功后放大流量失败则回滚。第五步完善可观测性和审计。把所有 Agent 的执行记录、审批记录、Skill 调用记录打通确保系统出了任何问题都能定位到具体环节。最后回到工程实践本身Harness Engineering 的价值不是让 AI Agent 变得更“聪明”而是让 AI Agent 在复杂的生产环境里变得更可靠、更可控、更可追溯。这三件事恰恰是企业级 AI 系统最难做到的。如果你能把沙箱、Skill、审批这三个机制真正落地你的 Agent 系统就已经超过了大多数还停留在“调 Prompt”阶段的团队。