如果你最近在关注 AI Agent 方向,大概率刷到过那篇关于 HumanLayer 的智能体构建分享。三周 20 万观看,这个数据放在技术圈不算小数目。更值得注意的是,它刷屏的节点刚好卡在“智能体热”从概念走向工程化的阶段——大量开发者已经能跑通 OpenAI Function Calling 或 LangChain 的简单 Agent,但一放到真实业务里就卡住:Agent 乱改数据怎么办?关键操作没人确认怎么办?自动化流程跑偏了谁来兜底?
这篇文章就来拆解一下,HumanLayer 这类工具到底解决了什么问题,以及构建一个“真实可用”的智能体,核心步骤到底是什么。不是泛泛讲概念,而是从工程视角走一遍流程:为什么需要 HumanLayer 这样的设计、环境怎么搭、代码怎么写、上线有哪些坑。
1. 这篇文章真正要解决的问题
先说一个真实痛点:很多人构建智能体,第一反应是“选个大模型、写个 Prompt、调几个工具”,然后用 LangChain 或 Dify 把这些串起来。这个流程跑通一个 Demo 不难,难的是把智能体放到业务流程里。
问题是这样的:
- Agent 不可控:大模型生成的工具调用序列很难保证完全不出错,一旦调用了不该调用的接口,后果不可逆。
- 缺少人工干预时机:传统 Agent 流程是“模型自动决定下一步”,但真实业务里很多操作需要人来确认,比如下单、退款、发邮件、删除数据。
- 权限边界模糊:给 Agent 一个数据库账号,它能查数据,也能改数据;给它一个 API Key,它能读信息,也能写操作。怎么限制?
- 可观测性差:模型为什么调用这个函数?中间经过了哪些推理?出了问题无法回溯。
HumanLayer 为什么值得关注?因为它切入了一个非常具体的点:在不把 Agent 变成纯自动化黑盒的前提下,让人能在关键节点介入审批和确认。
这个设计思路对构建智能体有很强的参考价值:它不是教你“怎么让 Agent 做更多事”,而是教你怎么“让 Agent 做得住手”。
2. HumanLayer 的核心概念与适用场景
2.1 HumanLayer 是什么
从项目定位来看,HumanLayer 是一个让 AI 智能体具备“人类参与节点”的框架。它解决的是 Agent 执行链路上的“人工确认”问题。
通俗解释:普通 Agent 是“接到任务 → 调用模型 → 执行工具 → 返回结果”,HumanLayer 在中间插入了一个环节:“遇到关键操作 → 向人类发送审批请求 → 人工确认后继续执行”。
这个机制拆开看有三个关键组件:
| 组件 | 作用 | 类比 |
|---|---|---|
| Action 注册 | 把 Agent 要执行的操作注册成可审批的动作 | 工单系统里的事前申请 |
| 审批流 | 人类确认或拒绝某个操作 | 双人复核机制 |
| Agent Runtime | 运行时把模型决策映射到已注册的操作上 | 权限控制层 |
2.2 适用场景
从实际需求倒推,HumanLayer 最适合以下场景:
- 业务 Agent:比如销售助手要发报价单、客服机器人要发起退款,这类操作需要人工确认。
- 内部自动化:Agent 操作内部系统,比如 CRM、ERP,关键数据变更不能全自动。
- 代码生成与执行:Agent 生成脚本后自动执行,你可能希望先看 diff 再确认。
- 多智能体协作:多个 Agent 分工协作时,某些跨 Agent 操作需要全局人工确认,避免自动联动出错。
2.3 不适用场景
也要说清楚边界。如果你的场景是“完全不需要人工介入的批处理任务”,HumanLayer 这种设计反而会增加流程成本。比如日志分析、数据清洗、文本分类这类低风险操作,加人工审批反而影响效率。
| 场景类型 | 适合 HumanLayer? | 原因 |
|---|---|---|
| 发邮件给客户 | 适合 | 内容错了影响很大,需要确认 |
| 删除数据库记录 | 适合 | 高危险操作,必须有审批 |
| 爬虫抓取公开数据 | 不一定 | 操作本身可回滚,优先级不高 |
| 批量生成图片描述 | 不适合 | 低风险高并发,人工确认成本太高 |
这个判断很重要:HumanLayer 的价值不是让 Agent 更“聪明”,而是让 Agent 更“听话”。
3. 环境准备与前置条件
在开始搭建带 HumanLayer 流程的智能体之前,先明确环境。不同项目的版本可能有差异,这里讲的是通用思路。
3.1 基础环境
- 操作系统:Windows / Linux / macOS 都可以,推荐 Linux 服务器做生产部署。
- Python 版本:3.9 以上,建议 3.10 或 3.11。
- 包管理工具:pip 或 poetry。
- 模型服务:OpenAI API、Anthropic API,或者通过 Ollama 本地部署的模型都可以,HumanLayer 的接入方式是模型无关的。
- 开发框架:可以直接裸调用 LLM API,也可以基于 LangChain、LlamaIndex 等框架集成。
3.2 项目初始化
mkdir humanlayer-agent-demo cd humanlayer-agent-demo python -m venv venv source venv/bin/activate # Windows 下执行 venv\Scripts\activate pip install humanlayer langchain openai这里需要说明:具体包名和版本以官方最新文档为准。重点是理解整个构建流程,而不是被版本号卡住。
3.3 环境变量配置
创建一个.env文件,存放 API Key 和 HumanLayer 配置:
OPENAI_API_KEY=sk-xxxxxx HUMANLAYER_API_KEY=hl_xxxxxx HUMANLAYER_BASE_URL=https://api.humanlayer.dev注意:HumanLayer 的配置项可能随版本调整,运行时如果报配置缺失,根据报错信息添加对应参数即可。
4. 核心流程拆解:构建智能体需要哪几步
4.1 第一步:定义工具集
Agent 能做什么,取决于你给它注册了哪些工具。这一步非常关键,因为 HumanLayer 的审批流也是基于工具集的。
def send_email(to: str, subject: str, body: str) -> dict: """发送邮件(模拟实现)""" print(f"发送邮件到:{to},主题:{subject}") return {"status": "sent", "to": to, "subject": subject}注册后,模型才知道有send_email这个函数可以调用。
4.2 第二步:定义审批规则
这里就是 HumanLayer 发挥作用的地方。对于send_email这类敏感操作,你应该把它标记为“需要人工审批”。
伪代码逻辑:
# 注册需要审批的动作 humanlayer_action = client.actions.register( action_name="send_email", description="发送邮件给客户", require_approval=True, approvers=["admin@example.com"], )含义是:当 Agent 调用send_email时,系统会先向审批人发送请求,审批人同意后才会真正执行。
4.3 第三步:组装 Agent 执行循环
核心循环是:模型生成动作 → 检查是否需要审批 → 需要则暂停并通知审批人 → 审批通过后执行 → 结果返回给模型。
这一设计有一个很大的好处:模型不需要知道自己“被审批了”。它只负责生成调用意图,执行层负责判断是否拦截。这种解耦让代码逻辑更清晰,也更容易测试。
4.4 第四步:失败回退
审批被拒绝时怎么办?要设计回退逻辑。比如邮件发送被拒绝,Agent 应该能收到“操作被拒绝”的反馈,并尝试其他方案或向用户说明。
这其实是很多智能体工程容易忽略的点:只设计了“成功路径”,没有设计“拒绝路径”和“异常路径”。
5. 完整示例代码实现
下面用一个贴近真实的示例展示完整流程。这里使用 Flask 搭建一个带路由的 Web 服务,演示一个“客户意向确认 → 发送邮件”的智能体流程。
5.1 项目结构
humanlayer-agent-demo/ ├── .env ├── app.py ├── tools.py ├── agent.py ├── requirements.txt5.2 工具定义
# 文件路径:tools.py import smtplib import os from email.mime.text import MIMEText def send_email(to: str, subject: str, body: str) -> dict: """发送邮件,真实项目中替换为邮件服务 SDK""" smtp_server = os.getenv("SMTP_SERVER", "smtp.example.com") smtp_port = int(os.getenv("SMTP_PORT", "465")) sender = os.getenv("SMTP_USER", "sender@example.com") password = os.getenv("SMTP_PASSWORD", "") msg = MIMEText(body, "plain", "utf-8") msg["Subject"] = subject msg["From"] = sender msg["To"] = to try: with smtplib.SMTP_SSL(smtp_server, smtp_port) as server: server.login(sender, password) server.sendmail(sender, [to], msg.as_string()) return {"status": "sent", "to": to, "subject": subject} except Exception as e: return {"status": "error", "message": str(e)}说明:这里的邮件发送逻辑是完整可用的,实际使用要配置正确的 SMTP 参数。如果你的环境不方便测邮件,可以先用print模拟。
5.3 HumanLayer 审批封装
# 文件路径:agent.py import os import json from typing import Callable # 伪代码:HumanLayer 客户端初始化的通用写法 # 不同版本的 SDK 初始化方式可能不同,以官方文档为准 try: from humanlayer import HumanLayer HUMANLAYER_AVAILABLE = True except ImportError: HUMANLAYER_AVAILABLE = False class ApprovalToolWrapper: """包装工具函数,使敏感操作需要审批后才执行""" def __init__(self, func: Callable, require_approval: bool = False): self.func = func self.require_approval = require_approval self.pending_approvals = {} def run(self, *args, **kwargs): if not self.require_approval: return self.func(*args, **kwargs) call_id = self._create_approval_request(args, kwargs) self.pending_approvals[call_id] = { "func": self.func, "args": args, "kwargs": kwargs, } return { "status": "pending_approval", "approval_id": call_id, "message": "该操作需要人工审批,已发送审批请求", } def _create_approval_request(self, args, kwargs) -> str: return f"call_{id(self)}_{len(self.pending_approvals)}" def approve(self, approval_id: str): """审批通过后执行真实调用""" if approval_id not in self.pending_approvals: return {"status": "error", "message": "审批ID不存在"} target = self.pending_approvals.pop(approval_id) result = target["func"](*target["args"], **target["kwargs"]) return {"status": "approved_and_executed", "result": result} def reject(self, approval_id: str): """审批拒绝""" if approval_id not in self.pending_approvals: return {"status": "error", "message": "审批ID不存在"} self.pending_approvals.pop(approval_id) return {"status": "rejected"}这里不要苛求真实 HumanLayer SDK 的准确类名,因为官方 API 可能会变,重点是理解“包装器+审批队列”的实现原理。
5.4 Agent 主服务
# 文件路径:app.py import json from flask import Flask, request, jsonify from tools import send_email from agent import ApprovalToolWrapper app = Flask(__name__) def build_agent_tools(): """组装工具集:普通工具 + 需要审批的工具""" tools = { "send_email": ApprovalToolWrapper( send_email, require_approval=True, ), } return tools @app.route("/api/execute", methods=["POST"]) def execute(): """模拟 Agent 执行入口""" data = request.get_json(force=True) action = data.get("action") params = data.get("params", {}) tools = build_agent_tools() if action not in tools: return jsonify({"status": "error", "message": f"未知操作:{action}"}) wrapper = tools[action] result = wrapper.run(**params) return jsonify(result) @app.route("/api/approve", methods=["POST"]) def approve(): """审批通过接口:生产环境必须加鉴权""" data = request.get_json(force=True) approval_id = data.get("approval_id") tools = build_agent_tools() # 简化处理,实际需要根据 approval_id 找到对应的 wrapper for tool in tools.values(): if hasattr(tool, "approve"): result = tool.approve(approval_id) if result.get("status") != "error": return jsonify(result) return jsonify({"status": "error", "message": "审批失败"}) @app.route("/api/reject", methods=["POST"]) def reject(): """审批拒绝接口""" data = request.get_json(force=True) approval_id = data.get("approval_id") tools = build_agent_tools() for tool in tools.values(): if hasattr(tool, "reject"): result = tool.reject(approval_id) if result.get("status") != "error": return jsonify(result) return jsonify({"status": "error", "message": "拒绝失败"}) if __name__ == "__main__": app.run(host="0.0.0.0", port=8000, debug=True)5.5 依赖文件
# 文件路径:requirements.txt flask==3.0.0 openai==1.30.0 python-dotenv==1.0.1再次提醒:具体版本请以当前环境实际可用版本为准,不要盲目锁定。
5.6 引入大模型决策层
上面的服务只是一个“工具执行框架”,要构成真正的 Agent,还需要模型来决策“该调用哪个 action”。伪代码如下:
# 文件路径:agent_loop.py import os from openai import OpenAI client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) function_schemas = [ { "type": "function", "function": { "name": "send_email", "description": "发送邮件给指定客户", "parameters": { "type": "object", "properties": { "to": {"type": "string", "description": "收件人邮箱"}, "subject": {"type": "string", "description": "邮件主题"}, "body": {"type": "string", "description": "邮件正文"}, }, "required": ["to", "subject", "body"], }, }, } ] def run_agent(user_message: str): response = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": user_message}], tools=function_schemas, ) message = response.choices[0].message if message.tool_calls: tool_call = message.tool_calls[0] function_name = tool_call.function.name arguments = json.loads(tool_call.function.arguments) return {"action": function_name, "params": arguments} return {"message": message.content}这样,整个链路就完整了:用户输入 → 模型决策 → 工具调用(需要审批则挂起) → 人工确认 → 真实执行。
6. 运行结果与效果验证
6.1 启动服务
source venv/bin/activate python app.py启动成功后,控制台会输出 Flask 的监听地址,默认是http://127.0.0.1:8000。
6.2 测试执行接口
curl -X POST http://127.0.0.1:8000/api/execute \ -H "Content-Type: application/json" \ -d '{"action": "send_email", "params": {"to": "test@example.com", "subject": "测试邮件", "body": "Hello"}}'预期输出:
{ "status": "pending_approval", "approval_id": "call_0", "message": "该操作需要人工审批,已发送审批请求" }注意:这里邮件并没有真正发送,而是进入了审批队列。这是 HumanLayer 设计的关键:先挂起,后执行。
6.3 测试审批通过
curl -X POST http://127.0.0.1:8000/api/approve \ -H "Content-Type: application/json" \ -d '{"approval_id": "call_0"}'预期输出(取决于真实邮件发送结果):
{ "status": "approved_and_executed", "result": { "status": "sent", "to": "test@example.com", "subject": "测试邮件" } }如果 SMTP 配置有问题,会返回status: error,此时去检查 SMTP 账号、端口、SSL 设置。
6.4 如何判断构建成功
从三个维度验证:
- 功能维度:Agent 能根据用户意图选择正确的工具。
- 审批维度:敏感操作在未审批前不会真正执行。
- 异常维度:审批拒绝时,流程能正确中断并返回提示。
如果这三个维度都符合预期,说明你的智能体已经具备“可控执行”的基本能力。
7. 常见问题与排查思路
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| Agent 调用了错误的工具 | 模型 Prompt 意图理解不准确 | 打印模型的 tool_calls 输出,检查函数描述是否清晰 | 优化函数描述,增加边界说明和示例 |
| 审批请求没有发送 | HumanLayer 配置错误或网络不通 | 检查 API Key、Base URL、日志中的 HTTP 状态码 | 确认配置项,临时加日志打印请求和响应 |
| 审批通过后执行失败 | 工具函数内部逻辑错误 | 逐个测试工具函数,不经过 Agent 直接调用 | 定位到具体工具函数,修复后重新测试 |
| 邮件发送超时 | SMTP 服务器网络不通或端口被安全组拦截 | telnet 测试 SMTP 端口连通性 | 检查网络安全策略,确认 SMTP 端口已开放 |
| Agent 循环重试导致多次调用审批 | 没有对“挂起”状态做判断 | 查看循环代码是否对 pending 状态做了终止判断 | 在 Agent 循环里加入“挂起则等待”的逻辑,不要盲目重试 |
| 审批 ID 在分布式环境下不唯一 | 多实例部署共用本地内存队列 | 检查 deploy 架构,审批状态如果存内存会导致丢失 | 生产环境建议把审批状态存 Redis 或数据库 |
8. 最佳实践与工程建议
8.1 工具注册要显式声明危险等级
不要把“读操作”和“写操作”混在一起。建议给工具加一个危险等级字段:
{ "name": "delete_user", "danger_level": "HIGH", "require_approval": True }低风险操作直接放行,高风险操作强制审批,中风险操作可以根据上下文决定。
8.2 审批超时处理
真实业务里,审批人不可能一直在线。要给审批加超时机制:
- 审批超时后,默认拒绝还是通过?建议默认拒绝。
- 超时后是否通知申请方?要通知。
- 是否支持审批人委托?复杂场景才需要。
8.3 可观测性建设
记录 Agent 每次调用的完整链路信息:
- 用户原始输入
- 模型选择动作的推理过程(可记录 token 或摘要)
- 工具调用参数
- 审批人是谁、何时审批、结果如何
- 工具返回结果
有了这份日志,才能审计 Agent 的行为。
8.4 生产环境安全边界
生产环境至少要满足这些要求:
- 鉴权:审批接口必须加身份认证,不能让任何人调用。
- 限流:防止恶意刷审批请求。
- 审计:所有审批动作留痕,不可删除。
- 最小权限:Agent 使用的数据库账号、API Key 权限要尽量小,不要给 admin。
8.5 从 Demo 到生产的路径
建议按这个思路演进:
- 第一阶段:本地 CLI + 打印日志,验证 Agent 能正确决策。
- 第二阶段:接入 Web 服务 + 模拟审批,验证审批流程跑通。
- 第三阶段:接入真实工具 + 真实人工审批,小范围联调。
- 第四阶段:完善权限、监控、日志,部署到生产。
不要一步到位,先让流程跑通,再逐步加安全措施。
9. 构建智能体的下一步方向
回到开头那个话题:HumanLayer 的分享为什么能火?因为它踩中了智能体从“能跑”到“可控”之间的关键过渡期。现在不缺能生成文本、调用工具的 Agent Demo,缺的是能放在业务里让人放心使用的 Agent 工程体系。
如果你正在做智能体开发,建议从这篇文章里的最小流程开始:先定义一个工具、加一个审批节点、跑通一次“模型决策 → 人工确认 → 工具执行”的完整链路。这个过程会让你切身体会到,Agent 工程的核心不只是模型能力,更是对执行边界的控制。
接下来可以深入研究的方向:
- HumanLayer 官方文档中更完整的 API 用法和多实例部署方案。
- 审批流与现有企业内部审批系统(如飞书审批、钉钉审批、企业微信)的集成。
- 用 LangGraph 或 Dify 实现更复杂的 Agent 状态机,把审批节点融入图编排。
- 多智能体协作中如何统一审批权限体系,避免出现“责任真空”。
智能体构建这件事,最难的不是让模型“做”,而是让系统“知道什么时候不该做”。希望这篇文章能帮你在构建智能体时少走一段弯路。