☰
智能体可控性设计:HumanLayer人工审批机制与harness engineering实践
2026/10/5 9:18:36 网站建设 项目流程

这次我们来看一个热度很高的智能体构建分享:HumanLayer。一个主题能在三周内拿到 20 万观看,大概率不是因为某个模型又刷新了榜单,而是它戳中了做 Agent 落地的人最头疼的问题——不可控。从标题来看,HumanLayer 想表达的核心很直接:智能体不是越自动越好,而是在关键执行节点加一层“人类批准”,让 AI 干活、让人把关。

这篇文章不打算去复述某一次分享的每一帧画面,而是把 HumanLayer 背后的智能体构建思路拆开:为什么需要人工层,harness engineering 如何把 Agent 装进可控的框架里,长期工作记忆应该怎么做,以及如果自己要搭一套可验证、可批量、可接口化的智能体系统,应该从哪些地方入手。如果你正在做客服机器人、内部流程自动化、内容生产流水线,或者准备把大模型接进真实业务,这篇文章建议先收藏。

先说结论:HumanLayer 不是一个复杂的模型,而是一种工程思路。它的核心价值是让智能体在执行“写入类操作”之前停下来等确认,同时保留完整的审计日志。换句话说,它解决的不是“模型是否能生成内容”,而是“生成之后是否可以直接执行”。这个区别,在 demo 阶段看不出问题,一旦接到支付、邮件、数据库、CRM 里,就变成生死线。

1. 核心能力速览

下面的表格基于公开分享主题、相关热词和常见的智能体工程实践整理。因为目前公开材料里没有一份完整的官方 API 文档,所以具体参数要以你实际拿到的项目版本为准。

能力项说明
项目类型智能体构建中的人工协同层方案,强调人在回路的审批机制
核心功能工具调用前审批、执行状态挂起、审计日志、策略控制、批量任务接入
硬件要求不依赖固定 GPU;接入本地模型时看模型显存,接入云端 API 时主要看网络和 Token 消耗
启动方式作为 SDK 或独立服务集成,需要配合已有的 Agent 框架使用
是否支持 API通常以 HTTP 接口或 SDK 方式暴露,具体以官方文档为准
是否支持批量任务可以接入任务队列,适合批量执行前先审批再放行的场景
依赖环境Python 或 Node.js 项目,需要 LLM 推理服务或云端模型 API
适合场景客服系统、内部流程自动化、内容生成后人工复核、操作类 Agent 的安全控制
不适合场景完全无人值守、低延迟高吞吐的纯自动决策链路

从材料看,HumanLayer 的定位偏“守门员”。它不太关心你用的模型是 GPT 还是本地开源模型,更关心的是工具调用是否被夹在可控的流程里。所以如果你已经有 Agent 在跑,只是担心它乱调工具,这个思路可以直接套用。

2. 适用场景与使用边界

HumanLayer 适合的团队,一般已经跨过了“大模型能不能生成”的阶段,进入“生成之后能不能安全执行”的阶段。典型场景有四个。

第一,客服机器人需要查订单、改地址、发优惠券。这类操作如果全自动,容易出现误操作。人工层介入后,机器人先给出建议动作,客服确认后执行,既保留效率,又降低风险。第二,内部 OA 流程自动化。比如差旅报销、物料申请、合同审批,智能体负责整理材料,最终提交动作由人确认。第三,内容生产流水线。AI 生成文章、图片、短视频脚本后,进入人工复核流程,通过后再发布。第四,数据后台操作。删除记录、批量更新字段、导出敏感数据,这些操作天然需要审批。

使用边界要单独说。HumanLayer 不适合用来做全自动高频交易,也不适合做没有人工复核条件的无人值守任务。审批本身有延迟,如果业务要求毫秒级响应,人工层会成为瓶颈。另外,审批机制不能替代安全策略,敏感操作依然需要更细粒度的权限控制,比如角色权限、双人复核、操作留痕。

合规方面要特别注意。涉及用户隐私数据、人脸、声音、版权素材时,必须提前确认授权范围。智能体读取的每一份数据、调用的每一个外部接口,都应该在项目文档里列清楚。公开分享之所以能火,除了技术新鲜感,还因为它提醒开发者:AI 真正进入业务系统时,约束机制比生成能力更稀缺。

3. 从 HumanLayer 看智能体构建的关键设计

我们先不急着写代码,把 HumanLayer 的设计意图拆开看。智能体系统里,模型只是“大脑”,真正干活的是工具调用链。一个 Agent 收到用户指令后,通常会经历“理解需求 - 规划步骤 - 调用工具 - 返回结果”四个阶段。问题在于,规划阶段和调用阶段之间,缺少一个可编程的关卡。

HumanLayer 提出的思路是在这里加一层人工策略。这个策略要回答三个问题:哪些操作需要人批?审批超时怎么办?审批之后如何恢复执行?围绕这三个问题,一个最小可用的审批网关就成型了。

先看哪些操作需要批。写文件、发邮件、删除数据、创建订单、修改权限,这些属于高风险操作,应该默认需要人工确认。而读取文章、搜索资料、计算数字,这类只读操作可以自动放行。策略不应该写死在业务代码里,而是做成配置,方便随时调整。

再看审批超时。真实场景里,人工不可能 7x24 小时盯审批队列。如果审批人不在,任务应该有一个明确状态:挂起、超时自动拒绝、或者转交给另一个审批人。没有超时机制的审批系统,会在批量任务里积累大量僵尸请求。

最后是恢复执行。审批通过后,系统要能从挂起点继续执行,而不是从头再来。这要求每一步工具调用都有可恢复的状态记录。简单来说,Agent 每一步的状态、输入、输出、审批结果都要落盘,这些数据同时是审计日志,也是失败重试的依据。

下面的代码是一个极简审批策略示例。它只表达设计思路,不是某个项目的真实 SDK。

# 极简审批策略,演示“哪些工具需要人工批准” # 实际项目中应改为配置文件或远程策略服务 class ApprovalPolicy: def __init__(self): self.requires_approval = { "send_email", "delete_record", "create_order", "update_balance", "export_user_data" } self.timeout_seconds = 3600 def check(self, tool_name: str, args: dict): if tool_name in self.requires_approval: return { "status": "pending", "reason": f"tool {tool_name} requires human approval", "suggested_approver": "owner" } return { "status": "auto_allowed", "reason": "read-only operation or safe scope" } policy = ApprovalPolicy() print(policy.check("send_email", {"to": "user@example.com", "body": "hello"})) print(policy.check("search_docs", {"query": "报销流程"}))

这段代码看起来简单,但它是整个可控智能体的基石:把“能不能执行”从模型推理中抽离出来,变成一个独立策略判断。只要这个判断存在,Agent 就不可能绕过人的约束。

4. Harness Engineering:构建可控 AI 智能体的系统工程实践

最近智能体圈子里流行一个词叫 harness engineering。它说的不是“提示词工程”,也不是简单的 function calling,而是把智能体当成一套完整系统来设计。Harness 可以理解成“缰绳”,它包含了提示词、工具注册、执行状态机、权限策略、记忆管理、日志评估这些模块。HumanLayer 只是这整套 harness 里的一个人工审批节点。

一个完整的 harness 至少要包含六个部分。

第一,模型接入层。无论是 OpenAI、Claude,还是本地部署的 Qwen、Llama,都通过统一的接口接入,方便替换和降级。第二,工具注册层。所有可被 Agent 调用的函数都要有名字、参数 schema、权限标签。没有注册的工具不允许调用。第三,执行状态机。每一步任务都是可挂起、可恢复、可失败重试的。第四,策略层。规则引擎判断工具是否被允许执行,是否需要人工审批。第五,记忆层。保存跨会话的关键信息,减少重复提问。第六,观测层。日志、追踪、指标,方便复盘模型决策。

下面用一个简化状态机来描述 Agent 执行过程。

# 简化版 Agent 执行状态机,演示审批后恢复 # 状态:pending / waiting_approval / running / done / rejected class AgentHarness: def __init__(self, llm, policy, tools, memory): self.llm = llm self.policy = policy self.tools = tools self.memory = memory def run(self, user_request: str, session_id: str): steps = self.llm.plan(user_request) for step in steps: decision = self.policy.check(step.tool, step.args) if decision["status"] == "pending": step.status = "waiting_approval" self._wait_for_approval(step, timeout=60) if not step.approved: return {"status": "rejected", "step": step.tool} if step.status == "approved": result = self.tools.call(step.tool, step.args) self.memory.add(session_id, step.tool, result) step.status = "done" return {"status": "done", "steps": steps}

注意,这里的_wait_for_approval可以是阻塞等待,也可以是把任务写入 Redis 队列后轮询。工程上更推荐后者,因为异步队列不会占用模型实例资源。审批结果可以来自一个简单的 Web 后台,也可以来自企微、钉钉、飞书的审批接口。

Harness engineering 最容易被忽略的部分是“失败重试”。模型调用超时、接口偶尔抖动,并不代表整个任务失败。状态机里一定要给每一步设定最大重试次数,并且允许从失败点恢复。没有重试机制的 Agent,在批量任务里会非常脆弱。

5. 长期工作记忆实现:OpenClaw Active Memory 思路

和 HumanLayer 经常一起被讨论的,还有智能体的长期工作记忆。最近的分享里也提到“OpenClaw Active Memory 高阶指南”,核心观点是:智能体不能只靠上下文窗口活着,它需要主动记忆。

短期记忆就是当前对话上下文,结束后就清空。长期记忆是持久化的知识,比如用户偏好、项目背景、历史操作记录。工作记忆更像是一个动态缓冲:系统从长期记忆里检索出当前任务最相关的片段,放进上下文,任务完成后再把新结论写回长期存储。Active Memory 指的就是这套“主动写入 - 主动检索 - 主动遗忘”的机制。

设计记忆系统时,三个问题最关键。

第一,记什么。不是所有对话都值得记。用户主动强调的偏好、中间产出的关键结论、任务完成后的最终状态,这三类优先级最高。第二,怎么存。轻量方案用 JSON 文件或 SQLite 就够,规模大了再换成向量数据库。第三,怎么取。不能每次把全部记忆塞给模型,要按相关度排序,取 top-k 段作为上下文补充。

下面是一个简化版记忆管理实现,它用关键词重合度代替向量检索,目的是演示流程。

# 简化版 Active Memory,演示写入与检索 import json import hashlib from pathlib import Path class ActiveMemory: def __init__(self, db_path="memory.json"): self.db_path = Path(db_path) if self.db_path.exists(): self.items = json.loads(self.db_path.read_text(encoding="utf-8")) else: self.items = [] def add(self, session_id, content, importance=0.5): item = { "id": hashlib.sha1(content.encode("utf-8")).hexdigest()[:8], "session_id": session_id, "content": content, "importance": importance } self.items.append(item) self._save() def search(self, query, top_k=3): query_words = set(query.split()) scored = [] for item in self.items: words = set(item["content"].split()) overlap = len(query_words & words) score = overlap * item["importance"] scored.append((score, item)) scored.sort(key=lambda x: -x[0]) return [item for score, item in scored[:top_k]] def _save(self): self.db_path.write_text( json.dumps(self.items, ensure_ascii=False, indent=2), encoding="utf-8" ) memory = ActiveMemory() memory.add("session-1", "用户偏好简洁回复,不超过200字", 0.9) memory.add("session-1", "当前项目使用 FastAPI 开发", 0.6) print(memory.search("回复风格", top_k=1))

真实项目里,应该用 embedding 模型生成向量,再通过余弦相似度排序。但核心架构不变:记忆层独立于模型,所有写入和读取都走统一接口。这样即使模型从 GPT 换成 Qwen,记忆也不会丢。

有趣的是,HumanLayer 与 Active Memory 可以结合。审批记录本身也是记忆的一部分。例如“上次客户要求删除某个账号,最终被否决”,这类信息写入长期记忆后,下一次 Agent 就不会再用同样的方式提请求。

6. 本地部署与验证:一套通用的智能体环境准备

对于想自己跑起来验证的开发者,环境准备可以先不追求完整平台,而是最小闭环。先准备好 Python 3.10 以上的环境,再选择一个 LLM 推理入口。如果没有云端 API,可以先用 Ollama 或 vLLM 拉一个 7B-14B 模型做本地推理。硬件方面,如果是 NVIDIA 显卡,建议先看一下显卡驱动和 CUDA 是否正常。显存大小会影响你能跑的模型规模,但具体占用要以模型版本和请求并发为准。

以下是通用检查清单,可以直接复制到本地核对。

# 检查 Python 版本 python --version # 检查 CUDA 是否可用(如果走本地 GPU 推理) nvidia-smi # 如果使用 Node.js 生态 node --version npm --version

依赖安装建议使用虚拟环境,避免污染系统 Python。以 Python 项目为例,创建虚拟环境后安装基础库。

# 创建并激活虚拟环境 python -m venv .venv source .venv/bin/activate # 安装基础依赖,示例项目需要 requests、pydantic、fastapi pip install requests pydantic fastapi uvicorn

如果你接入的是云端模型 API,准备好 API Key 后,先写一个最小调用脚本,确认模型推理链路通不通。然后再接入工具调用和审批策略。如果你接入的是本地模型,启动推理服务后,同样先用一个 curl 请求验证响应时间。确认这两条链路都没有问题,再开始搭 Agent。

部署时最好把配置独立出来,不要写死在代码里。下面是一个配置文件模板。

llm: provider: openai_compatible base_url: http://127.0.0.1:8000/v1 api_key: sk-local-test model: qwen2.5-7b-instruct approval: timeout_seconds: 3600 auto_approve_tools: - search_docs - get_weather require_approval_tools: - send_email - delete_record memory: type: json path: ./memory.json

这个模板的好处是,策略调整不需要改代码,只需要改配置。后续接入 Redis、PostgreSQL 或向量库时,替换 memory 类型即可。

7. 功能测试与效果验证

智能体系统上线前,功能测试不能只看“能不能回答问题”,要重点看“能不能被控制”。尤其是加入了 HumanLayer 思路之后,审批拦截、挂起恢复、记忆持久化这三块必须单独测。

建议准备如下测试用例。第一,基础任务完成测试:给 Agent 一个只读任务,比如“搜索文档中关于报销流程的内容”,确认它能够自动完成。第二,审批拦截测试:给 Agent 一个高风险任务,比如“发送邮件给客户”,确认在未审批前它不会执行。第三,审批恢复测试:模拟审批通过后,确认 Agent 能从挂起状态继续执行,而不是重新开始。第四,记忆保持测试:在第一个会话里告诉 Agent“用户偏好简短回复”,在第二个会话里提问“用户偏好什么”,确认它能回忆起。第五,批量任务测试:准备 10 个任务,混入 3 个需要审批的操作,确认它们都被挂起,批准后才执行。第六,异常恢复测试:模拟模型请求超时,确认任务会重试而不是崩溃。

每一步测试都要有明确的判断标准。如果审批拦截失败,说明策略没有生效,第一时间检查工具注册表里的权限标签是否写错。如果记忆没生效,检查写入和检索是否使用同一个存储目录。如果批量任务卡住,检查是否出现死锁,比如审批队列没人处理。

下面是测试记录表模板。

用例名称输入预期结果实际结果是否通过
只读任务自动执行查询库存数量返回结果,无需人工返回结果通过
发邮件拦截发送催款邮件给客户状态为 waiting_approval状态为 waiting_approval通过
审批后恢复批准发送邮件邮件发送成功,记录日志发送成功通过
记忆跨会话第二会话问用户偏好召回第一会话偏好召回成功通过
批量任务混合10 个任务含 3 个审批3 个等待审批,7 个完成符合预期通过

测试环境建议先关闭真实外部服务,用 mock 工具代替邮件、支付、数据库写入。否则测试时误发真实邮件或删掉真实数据,后悔都来不及。

8. 接口 API 与批量任务设计

如果要把智能体开放给其他系统使用,最直接的方式是封装一层 HTTP 接口。请求进来后,先创建任务,再根据策略决定是直接执行还是等待审批。响应里需要带一个任务状态字段,便于调用方轮询或者接收回调。

下面是一个 FastAPI 风格的接口示例。注意,这是通用演示,不是某个项目的实际 SDK。

from fastapi import FastAPI, HTTPException from pydantic import BaseModel app = FastAPI() class TaskRequest(BaseModel): request_id: str prompt: str force_approval: bool = False @app.post("/agent/run") async def run_agent(task: TaskRequest): decision = policy.check_from_prompt(task.prompt) if decision["status"] == "pending" or task.force_approval: return { "request_id": task.request_id, "status": "pending", "message": "Task is waiting for human approval", "approval_endpoint": f"/approve/{task.request_id}" } result = harness.run(task.prompt) return { "request_id": task.request_id, "status": "done", "result": result } @app.post("/approve/{request_id}") async def approve_task(request_id: str, approver: str): # 此处应更新 Redis 或数据库中的任务状态 return {"request_id": request_id, "status": "approved", "approver": approver}

接口调用方的用法很简单:先发起任务,如果返回 pending,就展示给人工确认;人工点击通过后,再调用审批接口。这样设计的好处是,智能体本身不需要一直常驻,审批期间释放资源。

批量任务建议用目录加队列的方式管理。输入任务放一个目录,审批通过的任务进入执行目录,失败任务进入重试目录。下面是一个任务输入示例。

{ "tasks": [ { "id": "task-001", "type": "generate_draft", "prompt": "生成本周工作周报", "requires_approval": false }, { "id": "task-002", "type": "send_email", "prompt": "给客户发送合同附件", "requires_approval": true } ] }

批量任务的并发数不宜一开始就拉满。先设置 1-2 个并发,确认模型接口和工具调用都稳定后,再逐步增加。每次批量执行都要保留日志,记录每个子任务的开始时间、结束时间、状态变化,方便失败复盘。

9. 资源占用与性能观察

HumanLayer 这类人工审批层本身占用的资源极低,真正的资源瓶颈在底层模型和工具调用上。如果是云端模型 API,重点观察延迟、Token 消耗和超时次数。如果是本地 GPU 推理,重点观察显存占用、GPU 利用率和推理并发数。

显存观察可以直接用nvidia-smi命令。

# 每隔 2 秒刷新一次 GPU 状态 watch -n 2 nvidia-smi

本地模型推理时,显存占用主要取决于模型参数量和量化方式。7B 模型和 70B 模型之间的显存差异很大,具体占用需要看模型运行时的实测数字。如果显存不够,优先降低并发数,或者换量化版本,再不行就缩小上下文长度。

接口服务的性能观察要关注三个指标:请求排队时间、单任务执行时长、审批挂起比例。请求排队时间过长,说明模型推理速度跟不上,需要扩容或限流。单任务执行时长异常,先看是不是外部工具响应慢。审批挂起比例太高,说明策略配置过严,或者审批人没有及时处理。后者可以用消息提醒缓解。

日志是排查性能问题最重要的依据。建议每个任务都输出一个结构化日志,包含任务 ID、状态、延迟、Token 数、审批人、错误信息。日志格式尽量统一,方便之后接入 ELK 或 Loki。

{ "timestamp": "2025-06-01T10:00:00Z", "request_id": "task-001", "status": "waiting_approval", "tool": "send_email", "delay_ms": 1250, "approver": "" }

无论资源占用多紧张,都不要在正式环境里直接修改策略文件。改策略后要先在测试环境跑一轮审批拦截用例,确认没有绕过风险,再发布到生产。

10. 常见问题与排查方法

智能体系统最容易出的问题不是模型不聪明,而是链路太长,排查看不到全貌。下面整理几个高频问题。

问题现象可能原因排查方式解决方案
Agent 不调用任何工具工具注册表未加载或提示词里没有工具说明检查工具列表是否打印到模型上下文打印注册工具列表,确认 schema 正确
模型开始乱调工具步骤规划和策略校验不在同一层查看日志中每一步的决策在工具调用前强制过审批网关
审批后任务没有继续任务状态没有持久化,重启后丢失检查 Redis/数据库中任务状态审批通过后更新状态,再从挂起点恢复
记忆跨会话失效写入路径与读取路径不一致检查 memory.json 文件位置统一记忆存储路径,避免本地和容器路径不一致
批量任务卡住审批队列无消费端查看队列堆积数量启动审批消费worker,增加超时自动拒绝
API 请求超时模型推理慢或外部工具无响应拆分检查模型耗时与工具耗时给工具调用加超时和重试,必要时降级
显存不足模型过大或并发过高看 nvidia-smi 显存占用降低并发、换量化模型、减小 max_tokens
权限策略被绕过策略校验只在客户端,而非服务端检查接口是否可被直接调用将策略校验放到后端服务统一执行

排查的第一原则:先看日志,不要凭感觉猜。状态机类的 bug,日志里一般都有明确的状态变化记录。第二原则:不要在生产环境直接改代码,先还原最小复现路径。

11. 最佳实践与使用建议

把 HumanLayer 的思路真正落地,下面几条建议值得坚持。

第一,第一次接入先小参数测试。不要一上来就接邮件、支付、数据库写入,先用 mock 工具跑通审批流程,确认挂起、恢复、日志都正常。第二,保留一套最小可运行配置。把模型、策略、记忆、工具四部分的最小配置单独存一份,后续改出问题时可以快速还原。第三,高风险工具默认拒绝,而不是默认通过。人工审批不是摆设,宁可多拦一次,也不要漏放一次。第四,审批记录要完整。谁批的、什么时候批的、依据是什么,都要可追溯。第五,批量任务必须加日志和失败重试。没有重试的批量任务,一次模型抖动就会浪费大量时间。第六,接口服务要限制访问范围。如果 Agent 服务暴露到内网或公网,一定要加鉴权,避免任何人都能触发审批任务。

在数据合规层面,任何人脸、声音、版权素材进入智能体处理链路前,都要确认授权。涉及用户隐私数据时,不要在日志里明文打印敏感字段,该脱敏的脱敏,该加密的加密。智能体可以帮你提高效率,但不能替你背合规责任。

12. 总结与下一步

HumanLayer 这个主题能在三周拿到 20 万观看,说明大部分做 Agent 的人不是缺模型,而是缺“安全感”。从工程角度说,人工审批层、harness engineering、长期工作记忆这三块,足够支撑起一个可控智能体的最小骨架。

如果你想亲手验证,最先应该测的不是花哨的功能,而是“审批拦截是否真正生效”。这个测试几十行代码就能完成,但它决定了后续所有业务接入是否安全。最容易踩的坑有两个:一是审批状态没有持久化,服务一重启就丢任务;二是策略校验只写在客户端,绕过前端直接调接口就能跳过审批。把这两个坑填平,再扩展批量任务和记忆模块,整个系统就会稳很多。

下一步可以继续做三件事:把审批模块接进企业微信或钉钉,让人在手机上审批;把记忆存储从 JSON 换成向量数据库,提升语义检索能力;再加上评估集,每次修改提示词或策略后自动跑回归测试。这样,一个从 demo 到可上线的智能体系统就基本成型了。

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

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

立即咨询