掘金 人工智能 07月11日 09:54
使用GoHumanLoop拓展AI Agent人机协同边界,这次连接到飞书
index_new5.html
../../../zaker_core/zaker_tpl_static/wap/tpl_guoji1.html

 

本文介绍了GoHumanLoop-FeiShu,一个用于拓展AI Agent人机协同边界的工具。通过与飞书的集成,AI Agent能够接入飞书审批流程,实现信息获取和审批操作。文章提供了详细的示例演示,展示了如何使用LangGraph框架结合GoHumanLoop和飞书,以及项目的架构设计、实现细节和部署方法,包括手动部署和Docker部署。最终,GoHumanLoop-FeiShu能够帮助用户更高效地构建AI Agent。

🚀 GoHumanLoop-FeiShu 扩展了AI Agent的人机协同能力,使其能够与飞书集成,实现审批流程。

💡 通过LangGraph框架,GoHumanLoop-FeiShu能够连接飞书,实现信息传递和审批操作的自动化。

🧩 该项目基于Go语言开发,采用Beego框架和飞书SDK,提供了API接口,方便与GoHumanLoop对接。

🛠️ GoHumanLoop-FeiShu支持两种部署方式:手动部署和Docker部署,用户可以根据自己的需求选择。

📝 用户需要配置飞书企业自建应用,包括获取应用ID、应用密钥、创建审批模板等,才能正常使用该服务。

一. 前言

Hi,大家好,继续我们GoHumanLoop系列, 继上一次我开发了GoHumanLoop-Wework,将目前Agent系统的人机协同方式与企业微信连接,使你的AI Agent能接入到企业微信的审批流程中,方便随时控制你的 Agent。这次我又开发了GoHumanLoop-FeiShu,继续拓展AI Agent人机协同边界,这次我们连接到飞书。

这是GoHumanLoop系列的第五篇文章,之前文章如下:

感兴趣的朋友可以阅读往期文章,进一步了解GoHumanLoop

本期我将给大家介绍GoHumanLoop-FeiShuGoHumanLoop如何搭配使用,我们上车吧~ 🚗

二. 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 运行完成~

以上就是使用GoHumanLoopGoHumanLoop-FeiShu拓展AI Agent人机协同到飞书的一个示例,接下来看一下该如何配置实现

四. 项目介绍

架构设计


GoHumanLoop与gohumanloop-feishu架构关系

实现介绍

gohumanloop-feishu采用Beego作为 Web 框架。sqlite作为简单的数据存储。飞书 SDK作为飞书 API 实现。提供一个可拓展的 GoHumanLoop 飞书审批示例服务。

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 中

配置文件

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

飞书审批模板

目前这个版本中,支持审批和信息获取。分别使用两个模板,模板格式固定,需要参考以下配置:


审批模板

以上字段由 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 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

六. 总结

以上就是使用GoHumanLoopGoHumanLoop-FeiShu搭配使用,快速拓展你的Agent智能体人机协同边界,发货更大AI效能

更多信息欢迎你前往查看相关代码仓库

欢迎您的 Star 🌟🌟🌟~

Fish AI Reader

Fish AI Reader

AI辅助创作,多种专业模板,深度分析,高质量内容生成。从观点提取到深度思考,FishAI为您提供全方位的创作支持。新版本引入自定义参数,让您的创作更加个性化和精准。

FishAI

FishAI

鱼阅,AI 时代的下一个智能信息助手,助你摆脱信息焦虑

联系邮箱 441953276@qq.com

相关标签

GoHumanLoop 飞书 AI Agent 人机协同
相关文章