☰
企业级AI Agent工程化:Harness Engineering如何为Agent套上缰绳
2026/10/14 1:21:56 网站建设 项目流程

在2026年这个时间节点上,如果一个团队还在把“AI Agent开发”当成“写一个调用大模型的脚本”,那么他们很快会在企业级场景里撞上一堵墙:单 Agent 在演示环境里跑得飞起,但一旦进入多 Agent 协同、真实业务接入、安全审计、权限管控、异常恢复这些硬约束条件,系统就会像没有缰绳的马一样不可控。

我见过太多项目卡在这里——编排器把任务发给两个 Agent,结果它们同时去改同一个配置文件;Agent 产生了一段看似合理的 Python 代码,但在生产环境上执行时把临时表数据全清了;你说“让 Agent 自己学着变强”,结果它把一次失败经验固化成了一条错误 Skill,从此每次都在同一个坑里摔倒。

这些问题本质上不是模型智商的问题,而是 AI 工程化的问题。企业级 AI Agent 项目真正需要的,是一整套给 Agent“套缰绳”的工程体系。这套体系的业界说法,就是 Harness Engineering。

本文会从工程落地视角拆解这个主题:什么是 Harness Engineering、企业级多 Agent 协同架构怎么搭、沙箱如何让 Agent 安全执行代码、Skill 如何“自进化”但不失控、人工介入机制如何设计。最后,我会给出一个完整的代码示例,让读者能跑通一条“编排 Agent → 沙箱执行 → Skill 自进化 → 人工审批”的最小链路。

读完这篇文章,你能看清企业级 AI Agent 工程化到底在解决什么问题,也能直接拿到一套可落地的设计思路与代码骨架。

1. 企业级 AI Agent 的失控焦虑:单 Agent 跑通不等于工程落地

很多团队接触 Agent 开发的第一站是 LangChain、Dify、Coze 这类框架或平台。在这些环境里搭一个单 Agent 应用,体验往往很顺滑:连上一个模型 API,给 Agent 配置两个工具,它就能根据用户问题自动选择工具、生成回答、甚至调用外部接口。

但把这个 Agent 搬到企业内网,情况就完全变了。

第一个变化是环境约束。企业不会允许 Agent 直接访问所有内部系统,也不会允许它随便在服务器上执行命令。Agent 产生一段脚本之后,系统必须回答三个问题:这段脚本能跑在哪里?它能访问哪些网络和文件?它出错之后会不会污染其他服务?没有沙箱,Agent 的能力越大,破坏半径就越大。

第二个变化是协同复杂度。企业级场景很少只有一个 Agent。通常会有任务规划 Agent、代码生成 Agent、数据分析 Agent、执行操作 Agent、审查 Agent 等等。你会发现,Agent 之间沟通的消息结构、谁负责最终决策、任务失败后谁来兜底,都需要一套非常明确的机制。否则就会出现“两个 Agent 同时改一个文件”“Agent 之间来回踢皮球”“错误信息被多次包装后失真”等混乱局面。

第三个变化是责任与审计。在演示环境里,模型输出错了可以道歉重来。但在企业里,一条自动生成的数据变更命令,一旦被执行,影响的是真实业务。系统必须有完整记录:这条命令是谁生成的、基于什么上下文、经过了谁的审批、在什么沙箱环境里执行、执行结果如何。没有审计链路,AI 项目很难通过安全合规评审。

第四个变化是经验沉淀。个人用 Agent,今天不好用,明天换个提示词就行。但企业级系统需要 Agent 持续积累业务知识:客户工单的处理偏好、内部系统的常见故障规律、历史变更的风险模式。这些知识如果只留在模型上下文里,换一个会话就丢;只有沉淀成结构化的 Skill,Agent 的能力才能“越用越强”。

把这些变化放在一起,可以得出一个明确判断:2026 年的 AI Agent 工程化,核心矛盾已经从“如何让 Agent 更聪明”,转移到了“如何让 Agent 在一个受控、安全、可审计、可迭代的工程框架里稳定工作”。这就是 Harness Engineering 登场的背景。

2. Harness Engineering 到底解决什么问题

2.1 从“马具”到 AI 工程方法论

Harness 的英文原意是马具、缰绳、安全带。放在 AI 工程语境里,它的核心思想非常直白:Agent 可以自由地思考、规划、调用工具、生成内容,但它的行为必须被约束在预先设定的边界之内。边界之外的动作,要么被拦截,要么必须经过人工授权。

这个理念和传统软件开发有本质区别。传统软件的行为是确定的:代码经过测试、评审、发布流程,每一步都可以预期。Agent 的行为是不确定的:同一个 Prompt,可能因为模型版本、上下文长度、甚至随机采样参数的变化产生不同结果。因此,传统软件工程那套“写完就锁定”的思路,没法直接套用在 Agent 上。Harness Engineering 的思路不是消灭不确定性,而是为不确定性设置围栏。

从工程实践角度,Harness Engineering 包含四个核心支柱:

支柱解决的问题典型实现手段
可控性Agent 的代码、命令、工具调用会不会越界沙箱执行、工具网关、权限白名单
成长性Agent 的经验能否沉淀、复用、持续迭代Skill 注册中心、版本管理、评估机制
人机协作高风险动作谁说了算,故障时如何兜底分级审批、人工介入接口、超时熔断
可观测性出问题时能否追溯根因和完整链路链路追踪、结构化日志、审计报表

2.2 它不是 Prompt Engineering 的替代品

做 AI 应用的同学对 Prompt Engineering 已经很熟悉。它关注的是“如何让模型更准确地理解用户意图”。RAG 和微调关注的是“如何让模型获得更好的知识或能力”。这些都属于提升模型输出质量的范畴。

Harness Engineering 关注的是另一个层面:模型输出质量提升之后,这个输出如何安全、可靠、合规地接入真实业务系统。换句话说,没有 Prompt Engineering,Agent 可能答非所问;没有 Harness Engineering,Agent 可能直接闯祸。两者不是替代关系,而是上下游关系。

再展开一点:Prompt Engineering 像是在训练一匹更聪明的马,而 Harness Engineering 是在设计马具、缰绳、赛道围栏和骑手指令系统。一匹聪明的马如果没有缰绳,跑得越快,风险越高。

2.3 适合谁阅读和使用

如果你是 AI 研究者,关心模型效果刷新,Harness Engineering 不是你的主战场。但如果你是 AI 应用开发者、平台工程师、后端架构师、或者准备把 Agent 落地到企业业务场景的技术负责人,这套体系就是你绕不开的工作内容。

尤其是在 2026 年的技术背景下,AI 编程助手、企业内部知识助手、自动化运维工单系统、智能客服、数据分析平台,都已经从“单点 Demo”走向“生产系统交付”。生产系统交付意味着要有稳定性承诺、安全边界、审计能力和迭代机制。这些要求,全部落在 Harness Engineering 的范围里。

3. 企业级多 Agent 协同的架构拆解

3.1 为什么需要多 Agent,而不是一个超级 Agent

先回答一个常见疑问:为什么企业级项目一定要做多 Agent 协同?一个 Agent 配上几十个工具不行吗?

从经验来看,单 Agent 塞满工具,会让模型在每次任务规划时面临巨大的选择空间,不仅影响响应速度,还容易在工具调用之间“精神分裂”。更关键的是,企业里不同模块的权限边界不同。数据分析 Agent 可以读数据仓库,但不能执行变更;运维 Agent 可以执行变更,但需要经过审批。把权限边界直接做在 Agent 职责划分上,比在工具层做细粒度控制要清晰得多。

多 Agent 架构还有一个实际收益:模型调用成本可控。每个 Agent 只需要携带与自己职责相关的上下文和工具定义,Prompt 长度更短,Token 消耗更低,响应速度也更快。在 2026 年的模型定价下,这依然是不可忽视的工程优化点。

3.2 分层架构与控制流

一个典型的企业级多 Agent 协同系统,可以分成三个层次:

第一层是控制层,也叫编排层。它接收一个任务,负责拆解子任务、决定调用哪些 Agent、监控每个 Agent 的执行状态、处理失败重试。这一层通常由一个 Orchestrator Agent 配合一个状态机或工作流引擎实现。控制层是多 Agent 系统的“大脑”,它的稳定性直接决定整个系统的可靠性。

第二层是执行层。这一层是真正干活的 Agent 集合。每一个执行 Agent 有明确的职责边界,并且持有自己专属的 Skill 库和工具白名单。举例来说,一个日志分析 Agent,它只能读取日志、运行分析脚本、输出结果,不能直接修改系统配置。执行 Agent 与工具之间通常还会插入一个工具网关,统一做参数校验和权限校验。

第三层是基础设施层。包括沙箱运行环境、模型网关(统一管理模型 API Key、路由、限流)、知识库与向量数据库、Skill 注册中心、审批服务、审计日志系统。这一层决定了多 Agent 系统是否具备企业级交付条件。

三层之间的典型控制流是:

任务进入编排层 → 编排层拆分子任务 → 根据子任务调度执行 Agent → 执行 Agent 在沙箱里调用 Skill 和工具 → 工具网关完成权限校验和审计埋点 → 如果是高风险操作则挂起并等待人工审批 → 执行结果回传给编排层 → 编排层汇总结果并判断任务是否完成

从这套流程可以看出,Agent 与 Agent 之间最好不直接通信,而是通过编排层间接协同。直接通信虽然看起来简洁,但会产生蜘蛛网式的调用关系,审计困难,出了问题也难以定位。

4. 沙箱:让 Agent 的代码和命令在安全边界内执行

4.1 为什么 Agent 执行代码必须有沙箱

Agent 在推理过程中可能会生成 Python 脚本、Shell 命令、SQL 查询,或者试图调用内部 API。如果一个 Agent 生成的代码直接在宿主机上执行,后果不堪设想。最典型的风险包括:

  • 代码自带网络请求,试图把内部数据外传。
  • 代码清理临时文件时路径拼接错误,删掉了生产目录。
  • 代码陷入死循环,耗尽服务器 CPU 和内存。
  • 代码试图读取环境变量或密钥文件,造成敏感信息泄露。

沙箱的核心价值,就是在 Agent 与真实系统之间加一层隔离。即使 Agent 生成了恶意或错误的代码,它的影响范围也被限制在沙箱内部,宿主系统不会受到直接破坏。

4.2 沙箱技术选型对比

从工程实现角度,沙箱并不是一个单一技术,而是一组隔离手段的统称。

沙箱方案隔离粒度适合场景典型工具
容器沙箱进程级隔离执行 Agent 生成的脚本、安装依赖、跑批任务Docker、Podman
WebAssembly 沙箱内存级隔离执行不可信的第三方 Skill 插件,限制系统调用Wasmtime、WasmEdge
函数计算沙箱云平台级隔离按次执行、弹性伸缩的短时任务云厂商 FaaS 服务
Linux 命名空间内核级隔离轻量隔离、定制化强nsjail、firejail

选型没有绝对标准,取决于你的安全和成本要求。在企业内网中,比较常见的是容器沙箱为主、WebAssembly 沙箱为辅的组合方式。Agent 生成的脚本如果涉及完整 Python 环境和依赖,放进容器里执行;如果只是执行一段不可信的纯计算逻辑,用 Wasm 沙箱可以做到毫秒级冷启动,成本低,隔离性也好。

4.3 沙箱策略配置示例

无论选择哪种沙箱,都要对 Agent 能做什么、不能做什么做明确限制。下面是一个容器沙箱策略配置的示例:

{ "sandbox": { "runtime": "docker", "image": "sandbox-python:3.11", "network": "none", "memory_limit": "512m", "cpu_limit": "1.0", "pids_limit": 64, "readonly_rootfs": true, "tmpfs": { "/tmp": "size=128m" }, "volumes": [ { "host_path": "/data/agent_workspace", "container_path": "/workspace", "read_only": false } ], "environment": { "APP_ENV": "sandbox", "DATA_API": "http://internal-data-api:8080" }, "allowed_capabilities": [], "blocked_paths": [ "/etc/hosts", "/home/production", "/var/lib/mysql" ] } }

这个配置有几个关键点:

网络设置成none,即沙箱内的代码无法主动访问外部网络。如果需要访问内部数据 API,可以把网络模式改成桥接,并配合防火墙做白名单限制。readonly_rootfs为true,根文件系统只读,防止 Agent 在容器里安装后门或修改系统文件。blocked_paths用来声明宿主目录中绝对不能触碰的敏感路径。

4.4 WebAssembly 沙箱的轻量场景

如果你的 Skill 代码主要是纯计算任务(数据清洗、文本处理、逻辑校验),用 WebAssembly 沙箱会更轻量。WASM 沙箱的实现原理可以简化为三层:线性内存隔离、无宿主权限、能力导入显式化。代码运行在一个独立线性内存空间中,无法访问宿主进程内存;系统调用被替换成宿主显式导入的函数,你只导入它需要用到的几个函数,其他一律不提供。

下面展示一个用 Wasmtime 运行不可信 Skill 代码的 Rust 示例思路,在实际项目中,同样的模型可以嵌入 Python 服务:

use wasmtime::{Engine, Module, Store, Linker}; use wasmtime::WasmBacktrace; fn main() -> anyhow::Result<()> { let engine = Engine::default(); let module = Module::from_file(&engine, "skill_sample.wasm")?; let mut store = Store::new(&engine, ()); let mut linker = Linker::new(&engine); // 仅显式导入白名单函数,例如一个可用的日志函数 linker.func_wrap("env", "log", |msg_ptr: i32, msg_len: i32| { println!("[skill log] ptr={}, len={}", msg_ptr, msg_len); Ok(()) })?; let instance = linker.instantiate(&mut store, &module)?; let main = instance.get_typed_func::<(), i32>(&mut store, "main")?; let result = main.call(&mut store, ())?; println!("skill result code: {}", result); Ok(()) }

这段代码的关键在于,Skill 模块只有在链接器里显式导入过的能力才可以使用。没有导入文件系统、没有导入网络、没有导入任意命令执行,Skill 再“狡猾”也接触不到外部世界。

5. 自进化 Skill:让 Agent 在工程约束下积累能力

5.1 Skill 到底是什么

在 Agent 工程里,Skill 可以理解为一个可复用的能力单元。它不只是提示词,而是“提示词模板 + 工具调用流程 + 输入输出约束 + 验证用例”的组合。举个例子,一个“读取 MySQL 慢查询日志并自动分类”的 Skill,可能包含:

  • 触发该 Skill 的适用场景描述。
  • 调用数据库查询工具的具体参数模板。
  • 生成分类报告的输出格式。
  • 验证逻辑:查询是否成功、结果是否符合预期。

Skill 的价值在于把模型“临场发挥”变成“标准化动作”。企业业务中 80% 的重复任务,最终都应该沉淀为 Skill,而不是每次都让 Agent 重新思考一遍。

5.2 自进化的工程闭环

所谓自进化 Skill,并不是让 Agent 在运行时偷偷改自己的代码。它的正确工程化路径是:运行采集 → 候选生成 → 沙箱验证 → 人工评审 → 灰度发布 → 效果反馈。

运行采集阶段,系统记录 Agent 每次任务的过程轨迹,包括用户请求、Agent 规划、工具调用、最终结果、用户反馈或业务结果。当某个任务模式反复出现时,触发候选 Skill 生成器。生成器可以是另一个 Agent,它读取多条相似轨迹,提炼出通用步骤和模板,产出一个候选 Skill。

这个候选 Skill 不能直接上线。它先进沙箱,用历史数据集跑一遍验证用例,确认输出质量。通过验证之后,进入人工评审流程。评审通过后,Skill 进入灰度状态,只对少量任务生效。灰度一段时间后,根据效果指标决定是正式发布、继续优化还是下架。

下图用文字描述就是:

收集 → 提炼 → 验证 → 评审 → 灰度 → 发布 → 反馈 → 再收集

5.3 Skill 定义与注册示例

一个 Skill 建议用结构化文件定义,放在 Git 仓库里管理,方便版本回溯和评审。下面是一个 YAML 示例:

apiVersion: agent.skill/v1 kind: Skill metadata: name: mysql-slow-query-classifier version: 1.2.0 owner:># approval_gate.py import time import uuid import requests class ApprovalGate: """审批门:将高风险工具调用挂起,等待人工审批结果""" def __init__(self, approval_service_url: str, timeout_seconds: int = 300): self.approval_service_url = approval_service_url self.timeout_seconds = timeout_seconds def request_approval( self, agent_name: str, tool_name: str, tool_params: dict, risk_reason: str, trace_id: str, ) -> dict: approval_id = str(uuid.uuid4()) payload = { "approval_id": approval_id, "agent_name": agent_name, "tool_name": tool_name, "tool_params": tool_params, "risk_reason": risk_reason, "trace_id": trace_id, } # 请求审批服务创建待办 resp = requests.post( f"{self.approval_service_url}/approvals", json=payload, timeout=10, ) resp.raise_for_status() # 轮询审批结果,可替换为 WebSocket 或消息推送 deadline = time.time() + self.timeout_seconds while time.time() < deadline: result = requests.get( f"{self.approval_service_url}/approvals/{approval_id}", timeout=10, ).json() if result["status"] == "approved": return {"status": "approved", "approval_id": approval_id} if result["status"] == "rejected": return {"status": "rejected", "approval_id": approval_id} time.sleep(3) return {"status": "timeout", "approval_id": approval_id}

这里的核心思路是:Agent 不是自己决定能不能继续,而是必须把操作的“证据链”提交给审批服务,等外部裁决。这样设计之后,即便最终出了问题,审计系统也能说清楚是哪一位审批人在什么时间基于什么信息做的决定。

审批结果返回后,Agent 侧需要做分支处理。审批通过就继续执行;审批拒绝就终止操作并生成说明;审批超时就按超时策略处理,一般是中止任务并通知人工管理员。

7. 完整示例:一个企业级多 Agent 协同工单处理项目

前面讲了很多理论,这一节我们把核心思路拼装成一个最小可运行项目。这里以一个内部 IT 工单系统为例:用户提交“查询订单表数据异常”的工单,系统通过多 Agent 协同完成日志分析、SQL 检查、变更审批、沙箱执行、结果反馈的完整流程。

7.1 项目结构

agent-platform/ ├── main.py # 主入口,模拟一份工单的完整处理流程 ├── orchestrator.py # 编排 Agent:任务拆解与子任务分发 ├── agents/ │ ├── __init__.py │ ├── analysis_agent.py # 分析 Agent:定位问题原因 │ └── execution_agent.py # 执行 Agent:生成变更 SQL 和脚本 ├── sandbox.py # 沙箱执行器:隔离运行 Agent 产生的代码 ├── skill_registry.py # Skill 注册中心:查询与调用 Skill ├── approval_gate.py # 人工审批门:高风险操作挂起等待审批 └── skills/ ├── slow_query_analyzer.yaml # 慢查询分析 Skill └── sql_change_validator.yaml

7.2 第一步:编排 Agent 拆解任务

编排层是整个系统的入口。它接收一份工单,拆解成分析子任务和执行子任务,然后按顺序调度执行 Agent。

# orchestrator.py import json from agents.analysis_agent import AnalysisAgent from agents.execution_agent import ExecutionAgent class Orchestrator: def __init__(self): self.analysis_agent = AnalysisAgent() self.execution_agent = ExecutionAgent() def handle_ticket(self, ticket: dict) -> dict: trace_id = ticket["trace_id"] print(f"[orchestrator] trace_id={trace_id}, 开始处理工单: {ticket['title']}") # 子任务一:交给分析 Agent 定位问题 analysis_result = self.analysis_agent.run( trace_id=trace_id, ticket=ticket, ) if analysis_result["risk_level"] == "high": print("[orchestrator] 检测到高风险问题,需要执行变更") # 子任务二:交给执行 Agent 生成变更方案 execution_plan = self.execution_agent.generate_plan( trace_id=trace_id, analysis=analysis_result, ) # 子任务三:在沙箱中验证变更方案 sandbox_result = self.run_in_sandbox(execution_plan) return { "trace_id": trace_id, "analysis": analysis_result, "execution_plan": execution_plan, "sandbox_verification": sandbox_result, "status": "pending_approval", } return { "trace_id": trace_id, "analysis": analysis_result, "status": "no_action_required", } def run_in_sandbox(self, execution_plan: dict) -> dict: from sandbox import SandboxExecutor sandbox = SandboxExecutor() return sandbox.execute(script=execution_plan["sql_script"], language="sql")

这段代码里,编排层先调分析 Agent,再根据风险等级决定是否进入执行链路。注意,编排层不负责写 SQL,它只负责流程控制。实际的分析和方案生成全在 Agent 层。

7.3 第二步:分析 Agent 与执行 Agent

分析 Agent 模拟了“调用大模型进行推理”的过程。为了能直接运行,我们用一个方法包装模型调用接口,实际项目里替换成你使用的模型 SDK 即可。

# agents/analysis_agent.py from skill_registry import SkillRegistry class AnalysisAgent: def __init__(self): self.registry = SkillRegistry() def run(self, trace_id: str, ticket: dict) -> dict: # 实际项目中,这里会调用大模型并携带相关 Skill 上下文 # 此处用一个示例推理结果代替真实模型调用 problem = ticket["title"] print(f"[analysis] trace_id={trace_id}, 分析工单: {problem}") # 示例:读取慢查询分析 Skill skill = self.registry.get_skill("slow-query-analyzer") print(f"[analysis] 加载 Skill: {skill['version']}") # 模拟模型判断 if "订单表" in problem and "数据异常" in problem: return { "risk_level": "high", "reason": "疑似订单表慢查询导致数据写入阻塞,需要检查 SQL 索引", "suggested_action": "优化慢查询 SQL 并重建索引", } return { "risk_level": "low", "reason": "暂未发现明显风险", "suggested_action": "无需变更", }

执行 Agent 的作用是生成可执行的变更计划。真实系统中,它会把分析结论、相关表结构、历史变更记录一起送给模型,由模型生成 SQL 和回滚脚本。

# agents/execution_agent.py class ExecutionAgent: def generate_plan(self, trace_id: str, analysis: dict) -> dict: print(f"[execution] trace_id={trace_id}, 基于分析结果生成执行计划") # 示例:生成一条 SQL 变更和对应的回滚 SQL sql_script = """ -- 添加索引,优化订单查询性能 CREATE INDEX idx_order_create_time ON orders(create_time); """ rollback_script = """ -- 回滚脚本:删除索引 DROP INDEX idx_order_create_time ON orders; """ return { "sql_script": sql_script, "rollback_script": rollback_script, "estimated_impact": "orders 表增加一个索引,可能短暂持锁", "risk_level": "high", "requires_approval": True, }

7.4 第三步:沙箱执行器

沙箱执行器负责把 Agent 生成的 SQL 或脚本放到隔离环境里,至少完成语法校验和影响行数估算。下面我们用一个简化实现来演示:它先在沙箱容器中执行 Dry Run,只有拿到“语法通过”和“影响行数在预期范围”两个条件后,才允许进入审批环节。

# sandbox.py import subprocess import json class SandboxExecutor: def __init__(self, image: str = "sandbox-mysql-client:latest"): self.image = image def execute(self, script: str, language: str) -> dict: print(f"[sandbox] 在隔离容器中执行 {language} 代码") if language == "sql": # 实际项目中,这里会挂载只读数据库快照或使用临时库 # 并用 EXPLAIN 验证语法和风险 result = self._dry_run_sql(script) return result # 其他语言的执行逻辑 result = self._dry_run_script(script) return result def _dry_run_sql(self, sql: str) -> dict: # 真实场景下,会调用 mysql client 的 EXPLAIN 或 START TRANSACTION; ROLLBACK; # 这里用本地命令模拟校验过程 cmd = [ "docker", "run", "--rm", "--network", "none", self.image, "mysql", "--dry-run", ] try: completed = subprocess.run( cmd, input=sql, text=True, capture_output=True, timeout=30, ) if completed.returncode != 0: return {"status": "failed", "error": completed.stderr} return { "status": "passed", "dry_run_output": completed.stdout, "suggestion": "语法校验通过,影响行数待审批前二次确认", } except subprocess.TimeoutExpired: return {"status": "timeout", "error": "沙箱执行超时"}

这个类的设计核心是:所有不安全操作都在容器内完成,宿主机永远不会直接执行 Agent 生成的 SQL 或脚本。--network none参数保证沙箱容器内没有任何网络访问能力,避免数据外传风险。

7.5 第四步:人工审批流程

主流程里,当沙箱验证通过后,系统会创建审批请求,等待负责人审批。这里我们把审批门的调用串起来:

# main.py import uuid from orchestrator import Orchestrator from approval_gate import ApprovalGate def main(): ticket = { "title": "订单表数据异常,疑似慢查询导致写入阻塞", "description": "业务方反馈每天 14:00 后订单写入延迟明显", "trace_id": str(uuid.uuid4()), "submitter": "ops-user-01", } orchestrator = Orchestrator() result = orchestrator.handle_ticket(ticket) if result["status"] == "pending_approval": gate = ApprovalGate( approval_service_url="http://approval-service.internal:9000" ) approval_result = gate.request_approval( agent_name="execution-agent", tool_name="mysql_executor", tool_params=result["execution_plan"], risk_reason="生产库执行 DDL,需确认影响窗口", trace_id=ticket["trace_id"], ) if approval_result["status"] == "approved": print("[main] 审批通过,通知执行 Agent 继续完成变更。") print("[main] 变更执行后,系统自动将变更记录写入审计日志。") else: print(f"[main] 审批未通过,任务终止。审批结果: {approval_result}") print("[main] 流程结束") if __name__ == "__main__": main()

对比一下,如果没有审批门,执行 Agent 生成 SQL 后就会直接执行。现在加入了人工审批,整个链路变成:Agent 提方案 → 沙箱验证 → 人等审批 → 通过才执行。这多出来的几步,正是 Harness Engineering 最有价值的部分。

8. 运行结果与验证方式

8.1 运行命令

把上面的文件按目录结构放好后,建议先创建一个虚拟环境,再运行:

cd agent-platform python3 -m venv venv source venv/bin/activate pip install requests python main.py

8.2 预期输出

如果一切正常,你会看到类似下面的输出:

[orchestrator] trace_id=xxx, 开始处理工单: 订单表数据异常,疑似慢查询导致写入阻塞 [analysis] trace_id=xxx, 分析工单: 订单表数据异常,疑似慢查询导致写入阻塞 [analysis] 加载 Skill: 1.2.0 [execution] trace_id=xxx, 基于分析结果生成执行计划 [sandbox] 在隔离容器中执行 sql 代码 [sandbox] 语法校验通过 [main] 审批通过,通知执行 Agent 继续完成变更。 [main] 变更执行后,系统自动将变更记录写入审计日志。 [main] 流程结束

8.3 如何判断链路是否健康

这套最小项目可以从三个层面判断是否跑通:

第一,编排正确。分析 Agent 识别出高风险工单后,编排层没有擅自跳过执行环节,而是正确地进入执行 Agent 和沙箱验证。

第二,隔离有效。沙箱执行任何 Agent 生成的脚本时,都没有直接接触宿主机系统和生产网络,容器内部失败也不会传染给主进程。

第三,审批闭环。高风险操作必须经过审批门,审批人看不到、不批准的操作不会被执行。把approval_gate.py中的审批服务地址指向一个模拟服务,就可以验证整个审批状态机。

如果运行失败,第一优先检查的是依赖和网络。这个示例依赖requests库,审批服务地址是内部域名,如果你的环境无法解析,程序会在这里抛异常。第二要检查的是沙箱 Docker 命令是否可用,如果本机没有 Docker,sandbox.py会执行失败。生产环境里,这一步应当使用独立的沙箱集群,而不是依赖开发者本机 Docker。

9. 常见问题与排查思路

在实际搭建类似系统时,团队踩坑最多的往往是下面几个位置:

问题现象可能原因排查方式解决方案
Agent 任务一直卡住不往下走等待人工审批但审批服务没有响应,或超时时间过长检查审批服务日志和回调接口给审批门设置合理超时,超时后熔断并通知管理员
沙箱里无法安装 Python 依赖沙箱网络被限制为 none,且基础镜像里没有预装包查看沙箱构建日志和依赖清单在基础镜像中预装业务依赖,或配置内部 PyPI 镜像白名单
Skill 版本混乱,A 任务和 B 任务用的不是同一个 Skill 版本Skill 没有走版本管理系统,注册中心直接覆盖查看 Skill Registry 的版本记录和调用链 trace_id引入语义化版本号,所有调用按版本锁定
Agent 权限过大,能访问不需要的内部系统工具网关只做了 Agent 级认证,没有做工具级授权审计网关日志,查看工具调用白名单按最小权限原则逐工具配置授权,禁止开放全部 API
审计日志太多,出了问题反而查不到链路日志只有 Agent 级别,没有统一的 trace_id 贯穿按 trace_id 检索日志,观察是否有断裂在编排入口生成 trace_id,所有 Agent、沙箱、审批环节透传
同一类任务反复失败,Skill 没有有效改进自进化流程没有闭环,效果反馈没有回到 Skill 生成器检查反馈数据是否落库,是否触发候选生成器把每次任务的成功/失败、用户评分记录到反馈表,触发再学习流程
多 Agent 同时调用同一工具导致冲突控制层缺少并发控制或任务锁观察工具网关的并发调用日志为关键工具加分布式锁,或按 Agent 职责分片调度

这里特别提醒一个问题:审批服务自身不能是 Agent 系统的简化版。很多团队在演示环境中把审批模拟成“直接 return approved”,结果到了生产环境才发现,审批人收不到通知、审批记录没有落库、审批回调地址是写死的。人工介入机制看起来简单,但要像对待支付系统一样对待它,必须有待办、有通知、有超时、有重试、有审计。

10. 最佳实践与工程建议

10.1 把 Agent 配置当作代码管理

Agent 的 Skill 定义、沙箱策略、工具白名单、审批规则,都应该像代码一样放进 Git 仓库。这样每个改动都有记录,每条规则都能评审,出问题可以快速回滚。最忌讳的是在平台 Web 页面上直接修改线上配置,改完没有版本历史,事后无法追溯。

建议目录结构如下:

config/ ├── agents/ │ ├── analysis-agent.yaml │ └── execution-agent.yaml ├── skills/ │ ├── slow-query-analyzer.yaml │ └── sql-change-validator.yaml ├── sandbox-policies/ │ └── default-policy.json └── approval-rules/ └── risk-matrix.yaml

10.2 从最小权限开始,逐步放权

Agent 的权限设计应该遵守两个原则:默认拒绝,按需授权。初始搭建时,只给 Agent 它完成职责所必需的最小工具集。运行一段时间,根据日志分析哪些工具调用是被合理拒绝的,再逐步扩大白名单。千万不要为了省事,给所有 Agent 一把万能钥匙。

具体到工具网关,建议每次调用都做三层检查:调用方身份是否合法(Agent 名和密钥)、工具参数是否符合 Schema 校验、该操作是否落在当前 Agent 的授权范围内。三层通过才允许执行。

10.3 全链路可观测性是排障的基础

多 Agent 系统的排障难度比单体应用高一个数量级。一个任务可能经过编排层、三个 Agent、两次工具调用、一次人工审批。任何一环出问题,都需要靠链路数据定位。

工程上建议:入口生成一个trace_id,在编排层、Agent 层、沙箱执行器、审批门、审计日志中全量透传。每次工具调用记录入参摘要、出参摘要、耗时、状态。每个 Skill 版本变更记录责任人、时间、验证结果。有了这套基础数据,后续做效果评估、成本分析、风险审计才有据可依。

10.4 成本控制:多 Agent 会放大模型调用费用

多 Agent 协同带来的第一个副作用是模型调用次数成倍增加。一个任务可能触发编排模型一次、分析模型一次、执行模型一次、审批通知模型一次。在大规模业务下,这是一笔不小的成本。

缓解措施主要有三个:第一,为每个 Agent 设置独立的模型路由,低风险任务用便宜的小模型,高风险复杂任务才用强模型。第二,开启语义缓存,相同或近似的问题直接命中缓存,不再重复调用模型。第三,在 Skill 层做模板复用,能用固定逻辑跑完的流程,不要每次都让 Agent 重新推理。

10.5 不要为了多 Agent 而多 Agent

最后一条建议是提醒。单 Agent 能解决的问题,不要硬拆成多 Agent。多 Agent 的收益在于权限隔离、上下文裁剪和并行加速,但代价是系统复杂度上升、延迟增加、故障点变多。如果你的业务只有两三个工具、一个模型就能胜任,直接用单 Agent 加工具网关即可。

在 2026 年的 AI 工程语境下,多 Agent 协同的价值已经被放得很大,但工程能力体现在“什么时候该拆、什么时候不该拆”的判断上。Harness Engineering 的最终目标不是让系统看起来更复杂,而是让复杂系统运转得更可靠。对开发者来说,掌握沙箱隔离、Skill 版本管理、人工审批闭环和全链路观测这四件事,比追逐任何新模型版本都更重要。在实际项目中,可以先从一条最小业务链路开始,把这些机制一项一项加上去,跑稳之后再扩展 Agent 数量,这样踩坑成本最低。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询