HumanLayer Slack集成实战：AI代理通过Slack通道获取人类审批的最佳实践
·
HumanLayer Slack集成实战：AI代理通过Slack通道获取人类审批的最佳实践
项目地址: https://gitcode.com/GitHub_Trending/hu/humanlayer 引言：为什么需要人类审批？
在AI代理（AI Agent）日益普及的今天，我们面临着一个关键挑战：如何确保AI在做出高风险决策时能够得到人类的监督和审批？无论是发送重要邮件、执行财务操作还是处理敏感客户数据，纯粹的自动化都可能带来不可预见的风险。
HumanLayer正是为解决这一问题而生——它让AI代理能够通过Slack、Email等渠道与人类进行工具化、异步的工作流交互。本文将深入探讨如何通过Slack集成实现AI代理与人类的高效协作。
Slack集成架构概览
两种Slack集成方案对比
| 方案 | 适用场景 | 配置复杂度 | 自定义程度 | 企业合规性 |
|---|---|---|---|---|
| HumanLayer官方App | 快速启动、个人项目 | ⭐☆☆☆☆ | ⭐☆☆☆☆ | ⭐⭐☆☆☆ |
| 自定义Slack App | 企业级应用、品牌定制 | ⭐⭐⭐⭐☆ | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐⭐ |
实战一：使用自定义Slack App集成
步骤1：创建Slack应用
首先需要在Slack工作区创建新的应用，使用以下manifest配置：
{
"display_information": {
"name": "AI审批助手",
"description": "AI代理人类审批工作流",
"background_color": "#000000"
},
"features": {
"bot_user": {
"display_name": "AI审批助手",
"always_online": false
}
},
"oauth_config": {
"scopes": {
"bot": [
"app_mentions:read",
"users.profile:read",
"users:read",
"commands",
"channels:history",
"channels:read",
"chat:write",
"groups:history",
"groups:write",
"im:history",
"im:read",
"im:write"
]
}
},
"settings": {
"event_subscriptions": {
"request_url": "https://api.humanlayer.dev/humanlayer/v1/slack/events",
"bot_events": [
"message.channels",
"message.groups",
"message.im",
"app_mention"
]
},
"interactivity": {
"is_enabled": true,
"request_url": "https://api.humanlayer.dev/humanlayer/v1/slack/interactions"
}
}
}
步骤2：Python代码集成示例
import os
from dotenv import load_dotenv
from humanlayer import HumanLayer, ContactChannel, SlackContactChannel
# 加载环境变量
load_dotenv()
class SlackApprovalManager:
def __init__(self):
# 获取Slack Bot Token
self.bot_token = os.getenv("SLACK_BOT_TOKEN")
# 初始化HumanLayer客户端
self.hl = HumanLayer(
api_key=os.getenv("HUMANLAYER_API_KEY"),
verbose=True,
contact_channel=ContactChannel(
slack=SlackContactChannel(
bot_token=self.bot_token,
channel_or_user_id="", # 留空表示DM给安装应用的用户
experimental_slack_blocks=True, # 启用富文本消息
)
),
)
@self.hl.require_approval()
def process_sensitive_data(self, data: dict, operation: str) -> dict:
"""
处理敏感数据 - 需要人类审批
Args:
data: 要处理的数据
operation: 操作类型（update/delete/export）
Returns:
处理后的数据或错误信息
"""
# 这里实现具体的数据处理逻辑
if operation == "update":
return {"status": "updated", "data": data}
elif operation == "delete":
return {"status": "deleted"}
else:
return {"status": "exported", "data": data}
# 使用示例
if __name__ == "__main__":
manager = SlackApprovalManager()
# 模拟敏感数据操作
sensitive_data = {"user_id": "123", "email": "user@example.com"}
result = manager.process_sensitive_data(sensitive_data, "update")
print(f"操作结果: {result}")
步骤3：环境变量配置
创建.env文件：
# HumanLayer API配置
HUMANLAYER_API_KEY=your_humanlayer_api_key_here
# Slack配置
SLACK_BOT_TOKEN=xoxb-your-bot-token-here
SLACK_SIGNING_SECRET=your_signing_secret_here
SLACK_APP_ID=your_app_id_here
实战二：精细化权限控制
基于角色的审批控制
from humanlayer import ContactChannel, SlackContactChannel
class RoleBasedApproval:
def __init__(self):
self.hl = HumanLayer(api_key=os.getenv("HUMANLAYER_API_KEY"))
# 定义不同角色的审批通道
self.admin_channel = ContactChannel(
slack=SlackContactChannel(
channel_or_user_id="C123456", # 管理员频道
allowed_responder_ids=["U123456", "U789012"], # 仅管理员可审批
context_about_channel_or_user="管理员审批频道"
)
)
self.finance_channel = ContactChannel(
slack=SlackContactChannel(
channel_or_user_id="C345678", # 财务频道
allowed_responder_ids=["U345678", "U901234"], # 仅财务人员可审批
context_about_channel_or_user="财务审批频道"
)
)
@self.hl.require_approval(contact_channel=self.admin_channel)
def system_config_change(self, config: dict) -> dict:
"""系统配置变更 - 需要管理员审批"""
return {"status": "配置已更新", "config": config}
@self.hl.require_approval(contact_channel=self.finance_channel)
def process_payment(self, amount: float, recipient: str) -> dict:
"""处理付款 - 需要财务审批"""
return {"status": "付款已处理", "amount": amount, "recipient": recipient}
审批流程状态管理
实战三：LangChain集成示例
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
import langchain.tools
from humanlayer import ContactChannel, HumanLayer, SlackContactChannel
load_dotenv()
class LangChainSlackIntegration:
def __init__(self):
self.hl = HumanLayer(
verbose=True,
run_id="langchain-slack-integration",
)
# 配置Slack审批通道
self.slack_channel = ContactChannel(
slack=SlackContactChannel(
channel_or_user_id="C07HR5JL15F",
context_about_channel_or_user="AI代理审批频道",
experimental_slack_blocks=True
)
)
@self.hl.require_approval(contact_channel=self.slack_channel)
def approve_customer_refund(self, customer_id: str, amount: float, reason: str) -> dict:
"""审批客户退款请求"""
return {
"status": "refund_approved",
"customer_id": customer_id,
"amount": amount,
"reason": reason
}
def create_agent(self):
tools = [
langchain.tools.StructuredTool.from_function(self.approve_customer_refund),
]
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", "你是客户服务AI助手，处理退款审批请求。"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
agent = create_tool_calling_agent(llm, tools, prompt)
return AgentExecutor(agent=agent, tools=tools, verbose=True)
# 使用示例
if __name__ == "__main__":
integration = LangChainSlackIntegration()
agent = integration.create_agent()
# 模拟退款审批请求
result = agent.invoke({
"input": "客户ID为12345要求退款100元，原因是商品质量问题，请处理审批"
})
print(f"审批结果: {result}")
最佳实践与故障排除
配置检查清单
-
Slack应用权限验证
- 确保所有必需的OAuth作用域都已正确配置
- 验证事件订阅URL指向HumanLayer API
- 检查签名密钥是否正确配置
-
网络连接验证
# 测试Slack API连接 curl -X POST https://slack.com/api/auth.test \ -H "Authorization: Bearer $SLACK_BOT_TOKEN" # 测试HumanLayer API连接 curl -X GET https://api.humanlayer.dev/health \ -H "Authorization: Bearer $HUMANLAYER_API_KEY" -
调试模式启用
hl = HumanLayer( api_key=os.getenv("HUMANLAYER_API_KEY"), verbose=True, # 启用详细日志 debug=True # 启用调试模式 )
常见问题解决方案
| 问题现象 | 可能原因 | 解决方案 |
|---|---|---|
| 审批消息未发送 | Slack Bot Token无效 | 重新生成Bot Token并更新配置 |
| 人类响应未被处理 | 事件订阅配置错误 | 检查Slack应用的事件订阅URL |
| 权限错误 | OAuth作用域缺失 | 添加缺少的作用域并重新授权 |
| 超时问题 | 网络延迟或审批人未响应 | 调整超时设置或添加提醒机制 |
高级特性：自定义审批界面
def create_custom_approval_ui(function_name: str, kwargs: dict) -> dict:
"""创建自定义Slack审批消息界面"""
# 根据函数类型定制消息格式
if function_name == "process_payment":
return {
"blocks": [
{
"type": "header",
"text": {
"type": "plain_text",
"text": "💰 付款审批请求"
}
},
{
"type": "section",
"fields": [
{
"type": "mrkdwn",
"text": f"*金额:* ${kwargs.get('amount', 0)}"
},
{
"type": "mrkdwn",
"text": f"*收款人:* {kwargs.get('recipient', 'N/A')}"
}
]
},
{
"type": "actions",
"elements": [
{
"type": "button",
"text": {
"type": "plain_text",
"text": "✅ 批准"
},
"value": "approve",
"style": "primary"
},
{
"type": "button",
"text": {
"type": "plain_text",
"text": "❌ 拒绝"
},
"value": "reject",
"style": "danger"
}
]
}
]
}
# 其他函数类型的UI定制...
性能优化建议
-
异步处理
import asyncio async def async_approval_flow(): # 使用异步版本的SDK（如果可用） approval = await hl.async_fetch_approval(spec) # 处理审批结果 -
批量审批请求
- 对多个相关操作使用复合审批通道
- 实现审批请求的队列处理机制
-
缓存策略
- 缓存频繁使用的审批配置
- 实现审批结果的本地缓存
总结
通过HumanLayer的Slack集成，我们能够构建一个既保持AI自动化效率又确保人类监督的安全系统。关键优势包括：
- 无缝集成：与现有Slack工作流完美融合
- 精细控制：基于角色的权限管理和审批流程定制
- 企业级安全：支持自定义Slack应用，满足合规要求
- 开发者友好：简洁的API设计和丰富的文档支持
无论你是构建客户服务机器人、财务审批系统还是内部管理工具，HumanLayer的Slack集成都能为你提供强大而灵活的人类审批解决方案。
下一步行动建议：
- 从简单的数学运算审批示例开始熟悉基本流程
- 逐步引入角色权限控制和自定义UI
- 在生产环境中实施监控和告警机制
- 定期审查和优化审批工作流
记住，最好的AI系统不是完全自动化的系统，而是能够智能地与人类协作的系统。HumanLayer正是实现这一目标的强大工具。
欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！
更多推荐
20
0- 0
已为社区贡献5条内容
所有评论(0)