一. 前言
Hi,大家好,继续我们GoHumanLoop系列, 继上一次我开发了GoHumanLoop-Wework
,将目前Agent系统的人机协同方式与企业微信连接,使你的AI Agent能接入到企业微信的审批流程中,方便随时控制你的 Agent。这次我又开发了GoHumanLoop-FeiShu
,继续拓展AI Agent人机协同边界,这次我们连接到飞书。
这是GoHumanLoop
系列的第五篇文章,之前文章如下:
- 使用GoHumanLoop将你的AI Agent连接到企业微信(二)使用GoHumanLoop将你的AI Agent连接到企业微信(一)使用GoHumanLoop 在 CrewAI 高效构建Human-in-the-Loop流程使用GoHumanLoop增强你的Agent控制 - LangGraph 中的实践
感兴趣的朋友可以阅读往期文章,进一步了解GoHumanLoop
本期我将给大家介绍GoHumanLoop-FeiShu
与GoHumanLoop
如何搭配使用,我们上车吧~ 🚗
二. LangGraph与飞书搭配的简单示例
我们先上来看一个实际演示示例,然后再看具体怎么部署
我编写一个最小化的AI Agent,并且使用GoHumanLoop
库,增强人机协同Human-in-the-Loop
功能。
# /// script# requires-python = ">=3.10"# dependencies = [# "gohumanloop>=0.0.12",# "langgraph>=0.4.7",# "langchain-openai>=0.3.12"]# ///"""LangGraph 与 GoHumanLoop 简单集成示例,并对接飞书应用。这是一个最小化示例,展示如何在 LangGraph 中使用 GoHumanLoopManager 和 LangGraphAdapter进行人机交互。配置:- API 地址: http://localhost:8000/api- API KEY: gohumanloop"""import osimport timefrom typing import TypedDict, List, Dictfrom dotenv import load_dotenv# 导入 LangGraph 相关库from langgraph.graph import StateGraph, END# 导入 GoHumanLoop 相关库from gohumanloop.adapters.langgraph_adapter import HumanloopAdapterfrom gohumanloop.core.interface import HumanLoopStatusfrom gohumanloop import DefaultHumanLoopManager, APIProviderfrom gohumanloop.utils import get_secret_from_envimport logginglogging.basicConfig(level=logging.INFO)# 设置环境变量os.environ["GOHUMANLOOP_API_KEY"] = "46cf87c5-9d08-4027-b72a-f0a91f27298a"# 定义简单状态类型class SimpleState(TypedDict): messages: List[Dict[str, str]]# 创建 GoHumanLoopManager 实例manager = DefaultHumanLoopManager( APIProvider( name="ApiProvider", api_base_url="http://127.0.0.1:9800/api", # 换成自己飞书应用的URL api_key=get_secret_from_env("GOHUMANLOOP_API_KEY"), default_platform="feishu" ))# 创建 LangGraphAdapter 实例adapter = HumanloopAdapter( manager=manager, default_timeout=300, # 默认超时时间为5分钟)# 定义需要人工审批的节点@adapter.require_info( task_id="simple-information-test", additional="这是一个简单的获取信息的示例。",)def get_information_node(state: SimpleState, info_result={}) -> SimpleState: """获取信息的节点""" print("获取人工信息...") print(f"info_result: {info_result}") state["messages"].append({ "role": "system", "content": f"已获取信息: {info_result.get('response')}" }) return state# 定义需要人工审批的节点@adapter.require_approval( task_id="simple-approval-test", additional="这是一个简单的审批示例。", execute_on_reject=True,)def human_approval_node(state: SimpleState, approval_result=None) -> SimpleState: """需要人工审批的节点""" print("人工审批完成中...") print(f"approval_result: {approval_result}") # 处理审批结果 if approval_result: status = approval_result.get("status") response = approval_result.get("response", {}) if status == HumanLoopStatus.APPROVED: state["messages"].append( { "role": "human", "content": f"审批已通过!理由: {response}", } ) elif status == HumanLoopStatus.REJECTED: state["messages"].append( { "role": "human", "content": f"审批被拒绝。理由: {response}", } ) return statedef final_node(state: SimpleState) -> SimpleState: """最终节点""" state["messages"].append({"role": "system", "content": "工作流程已完成!"}) return state# 构建工作流图def build_simple_graph(): """构建简单工作流图""" graph = StateGraph(SimpleState) # 添加节点 graph.add_node("get_info", get_information_node) graph.add_node("human_approval", human_approval_node) graph.add_node("final", final_node) # 设置边 graph.add_edge("get_info", "human_approval") graph.add_edge("human_approval", "final") graph.add_edge("final", END) # 设置入口 graph.set_entry_point("get_info") return graph.compile()# 运行工作流def run_simple_workflow(): """运行简单工作流""" with adapter: # 构建工作流图 workflow = build_simple_graph() # 初始化状态 initial_state = SimpleState( messages=[{"role": "system", "content": "开始简单工作流..."}], ) # 运行工作流 for output in workflow.stream(initial_state, stream_mode="values"): print(f"状态: {output}") # 等待一下,便于观察 time.sleep(1)# 主函数if __name__ == "__main__": # 加载环境变量 load_dotenv() # 运行工作流 run_simple_workflow()
上述代码使用LangGraph框架,整个工作流比较简单,如下图所示:
graph TDStart --> get_info --> human_approval --> END
get_info
节点,模拟通过GoHumanLoop
连接飞书获取到审批人提供的外部输入信息human_approval
节点,模拟Agent在执行中需要审批的执行操作# 创建 GoHumanLoopManager 实例manager = DefaultHumanLoopManager( APIProvider( name="ApiProvider", api_base_url="http://127.0.0.1:9800/api", # 换成自己飞书应用的URL api_key=get_secret_from_env("GOHUMANLOOP_API_KEY"), default_platform="feishu" ))
代码中,定义了APIProvider
构造了默认HumanLoopManager,这个APIProvider
就是对接到企业微信中部署的GoHumanLoop-FeiShu
服务。用来发送请求和接收审批结果
# 创建 LangGraphAdapter 实例adapter = HumanloopAdapter( manager=manager, default_timeout=300, # 默认超时时间为5分钟)# 定义需要获人工信息的节点@adapter.require_info( task_id="simple-information-test", additional="这是一个简单的获取信息的示例。",)# 定义需要人工审批的节点@adapter.require_approval( task_id="simple-approval-test", additional="这是一个简单的审批示例。", execute_on_reject=True,)
LangGraphAdapter
定义了适配LangGraph
框架的适配对象,该对象的两个装饰器方法分别来封装到各自的节点中来说实现控制节点获取信息和审批操作。
- 运行示例代码
uv run langgraph_wework.py
三. 示例展示
连接到飞书,可以通过飞书随时向AI Agent传递信息,帮助Agent顺利执行
在 AI Agent 开始高危操作时,通过向飞书发起审批,获取高危操作授权是非常有必要的。 GoHumanLoop
工具就是拓展AI Agent人机协同边界,帮助人们更好的开发自己的 AI Agent。
Agent 运行完成~
以上就是使用GoHumanLoop
和GoHumanLoop-FeiShu
拓展AI Agent人机协同到飞书的一个示例,接下来看一下该如何配置实现
四. 项目介绍
架构设计
GoHumanLoop与gohumanloop-feishu架构关系
GoHumanLoop
提供了一套统一的 API 接口,通过API Provider
对外提供。gohumanloop-feishu
实现了API Consumer
的功能,通过API Provider
来获取审批相关的信息,并且通过飞书 SDK 实现了与用户的飞书进行交互,发送审批请求和获取审批事件回调等人机交互协同的操作。实现介绍
gohumanloop-feishu
采用Beego作为 Web 框架。sqlite
作为简单的数据存储。飞书 SDK作为飞书 API 实现。提供一个可拓展的 GoHumanLoop 飞书审批示例服务。
- 访问 Swagger 文档:
go run main.go
http://127.0.0.1:9800/docs
gohumanloop-feishu
提供两大类接口,一个对接GoHumanLoop
的 API Provider 的 HumanLoop
接口。另一个是获取 API Key 的接口,方便安全认证。
五. 项目部署
💡需要用户提前准备好飞书企业自建应用详情见:open.feishu.cn/
用户需要获取飞书
企业ID
用户需要在飞书中创建应用GoHumanLoop
,获取应用ID
和应用Secret
用户需要在应用中,开启 API 接收消息事件。(使用使用长连接接收回调)用户需要将应用ID
和应用Secret
配置到 GoHumanLoop 中用户需要在飞书中创建审批模板,获取审批模板ID
和信息模板ID
用户需要将审批模板ID
、信息模板ID
、创建人OpenId
、审批人OpenId
配置到 GoHumanLoop 中
配置文件
- 项目配置样例文件在
conf/app.conf.example
中appname = gohumanloop-feishuhttpport = 9800 # HTTP 端口按需配置# weworkappid = XXX # 企业应用IDappsecret = XXX # 应用Secret# templateapprove_template_id = B102B40F-E288-4A48-8220-FB0E69990222 # 审批模板IDinfo_template_id = 4B131588-A5E2-4CE6-91D1-5A208CA62EDAcreator_userid = ou_ed1d616f3023f7781808ce479b5f0fdd # 创建人ID,用户在应用中的OpenIdapprover_userid = ou_ed1d616f3023f7781808ce479b5f0fdd # 审批人ID,用户在应用中的OpenId# databasedatapath = ./data/gohumanloop.db # 数据库路径
- 修改配置文件
mv conf/app.conf.example conf/app.conf
飞书审批模板
目前这个版本中,支持审批和信息获取。分别使用两个模板,模板格式固定,需要参考以下配置:
审批模板
- 参考图片内的字段,都是文本控件和多行文本控件。包括以下字段
- 任务 ID对话 ID请求 IDHumanLoop 类型申请内容申请问题申请说明
以上字段由 GoHumanLoop 库来传输并自动填充并自动发起审批流程
审批模板
- 审批流程可以参考上图设置,审批人设置为自选
信息获取模板
- 参考图片内的字段,都是文本控件和多行文本控件。详情同审批流程模板和说明
信息获取模板
- 信息获取流程不需要具体审批,只需要获取具体信息,设置必须填写回执信息,否则无法获取信息。
部署方式
GoHumanLoop FeiShu 支持两种部署方式手动部署和 Docker 部署。
[!WARNING]通过采用飞书 SDK 的 Websocket 长连接来接收回调消息,需要在飞书应用中开启长连接接收消息事件,支持本地部署。区别于企业微信的 Webhook 方案需要注册同企微主体下的服务器。服务器和域名需要已备案等这些繁琐的操作
1. 手动部署
Go 版本要求:1.23.0
- 下载代码
git clone https://github.com/ptonlix/gohumanloop-feishu.git
- 编译
make build
- 运行
./gohumanloop-feishu
2. Docker 部署
- 提前安装好 Docker 服务
docker pull ptonlix/gohumanloop-feishu:latest
- 运行容器
docker run -d \ --name gohumanloop-feishu \ -v /path/to/local/conf:/app/conf \ -v /path/to/local/data:/app/data \ -p 9800:9800 \ ptonlix/gohumanloop-feishu:latest
六. 总结
以上就是使用GoHumanLoop
和GoHumanLoop-FeiShu
搭配使用,快速拓展你的Agent智能体人机协同边界,发货更大AI效能
更多信息欢迎你前往查看相关代码仓库
- GoHumanLoop: Perfecting AI workflows with human intelligencegohumanloop-examples : GoHumanLoop使用示例仓库gohumanloop-feishu GoHumanLoop飞书服务示例仓库
欢迎您的 Star 🌟🌟🌟~